diff --git a/bookmark/catalogue.py b/bookmark/catalogue.py new file mode 100644 index 0000000..f21a551 --- /dev/null +++ b/bookmark/catalogue.py @@ -0,0 +1,78 @@ +import csv +import uuid +from dataclasses import asdict, dataclass +from pathlib import Path + +CATALOGUE_COLUMNS = [ + "book_id", + "timestamp", + "size_bucket", + "front_wide_path", + "front_cropped_path", + "back_wide_path", + "back_cropped_path", + "title", + "author", + "isbn", + "match_confidence", + "match_status", + "status", + "checked_out_at", +] + + +@dataclass +class BookRecord: + book_id: str + timestamp: str + size_bucket: str + front_wide_path: str = "" + front_cropped_path: str = "" + back_wide_path: str = "" + back_cropped_path: str = "" + title: str = "" + author: str = "" + isbn: str = "" + match_confidence: str = "" + match_status: str = "" + status: str = "captured" + checked_out_at: str = "" + + def to_row(self) -> dict: + return asdict(self) + + +class Catalogue: + def __init__(self, csv_path: Path) -> None: + self.csv_path = Path(csv_path) + if not self.csv_path.exists(): + self.csv_path.parent.mkdir(parents=True, exist_ok=True) + with open(self.csv_path, "w", newline="") as f: + writer = csv.DictWriter(f, fieldnames=CATALOGUE_COLUMNS) + writer.writeheader() + + def new_book_id(self) -> str: + return uuid.uuid4().hex[:12] + + def append(self, record: BookRecord) -> None: + with open(self.csv_path, "a", newline="") as f: + writer = csv.DictWriter(f, fieldnames=CATALOGUE_COLUMNS) + writer.writerow(record.to_row()) + + def read_all(self) -> list[dict]: + with open(self.csv_path, newline="") as f: + return list(csv.DictReader(f)) + + def update_row(self, book_id: str, **fields_to_update) -> None: + rows = self.read_all() + found = False + for row in rows: + if row["book_id"] == book_id: + row.update({k: str(v) for k, v in fields_to_update.items()}) + found = True + if not found: + raise KeyError(f"book_id {book_id} not found in catalogue") + with open(self.csv_path, "w", newline="") as f: + writer = csv.DictWriter(f, fieldnames=CATALOGUE_COLUMNS) + writer.writeheader() + writer.writerows(rows) diff --git a/tests/test_catalogue.py b/tests/test_catalogue.py new file mode 100644 index 0000000..3be939d --- /dev/null +++ b/tests/test_catalogue.py @@ -0,0 +1,49 @@ +import pytest + +from bookmark.catalogue import Catalogue, BookRecord, CATALOGUE_COLUMNS + + +def test_creating_catalogue_writes_header(tmp_path): + csv_path = tmp_path / "catalogue.csv" + Catalogue(csv_path) + assert csv_path.exists() + header_line = csv_path.read_text().splitlines()[0] + assert header_line.split(",") == CATALOGUE_COLUMNS + + +def test_append_and_read_all_round_trip(tmp_path): + catalogue = Catalogue(tmp_path / "catalogue.csv") + record = BookRecord(book_id="abc123", timestamp="2026-07-13T00:00:00+00:00", size_bucket="medium") + catalogue.append(record) + + rows = catalogue.read_all() + + assert len(rows) == 1 + assert rows[0]["book_id"] == "abc123" + assert rows[0]["size_bucket"] == "medium" + assert rows[0]["status"] == "captured" + + +def test_new_book_id_is_unique(tmp_path): + catalogue = Catalogue(tmp_path / "catalogue.csv") + ids = {catalogue.new_book_id() for _ in range(100)} + assert len(ids) == 100 + + +def test_update_row_modifies_matching_row(tmp_path): + catalogue = Catalogue(tmp_path / "catalogue.csv") + catalogue.append(BookRecord(book_id="abc123", timestamp="t", size_bucket="small")) + catalogue.append(BookRecord(book_id="xyz789", timestamp="t", size_bucket="large")) + + catalogue.update_row("abc123", title="Charlotte's Web", status="ready") + + rows = {row["book_id"]: row for row in catalogue.read_all()} + assert rows["abc123"]["title"] == "Charlotte's Web" + assert rows["abc123"]["status"] == "ready" + assert rows["xyz789"]["title"] == "" + + +def test_update_row_raises_for_unknown_book_id(tmp_path): + catalogue = Catalogue(tmp_path / "catalogue.csv") + with pytest.raises(KeyError): + catalogue.update_row("does-not-exist", title="x")