dev/scripts/add_ama_coding_to_zotero.py

@@ -0,0 +1,233 @@
+"""Mirror the AMA coding/CPT publications into Zotero.
+
+Reads the OpenLibrary-derived title list (``data/ama/ama_coding_titles.json``,
+one entry per OpenLibrary work: title, first publish year, ISBNs, OL key)
+and creates one ``book`` item per work under a top-level collection, then
+attaches whatever ebook LazyLibrarian has already landed under the books
+library for that title. Re-running is safe: items are matched on their
+OpenLibrary URL, attachments on file name.
+
+Tags:
+    - source:ama
+    - module:coding
+    - year:YYYY
+
+Usage:
+    uv run python dev/scripts/add_ama_coding_to_zotero.py [--titles PATH]
+        [--books DIR] [--collection NAME] [--dry-run]
+
+Restart Zotero afterwards so the desktop picks up the new rows.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import re
+import subprocess
+from pathlib import Path
+
+from conf import path as _conf_path
+from zot.db import TYPE_MAP, Db, generate_key, now_iso
+
+ZOTERO_DB = str(_conf_path("db.zotero"))
+ZOTERO_STORAGE = Path(str(_conf_path("storage.zotero")))
+
+DEFAULT_TITLES = Path("data/ama/ama_coding_titles.json")
+DEFAULT_BOOKS = Path("/srv/pool/media/books/American Medical Association")
+DEFAULT_COLLECTION = "AMA Coding Publications"
+PUBLISHER = "American Medical Association"
+CONTENT_TYPES = {
+    ".epub": "application/epub+zip",
+    ".pdf": "application/pdf",
+    ".mobi": "application/x-mobipocket-ebook",
+    ".azw3": "application/vnd.amazon.mobi8-ebook",
+}
+
+
+def _norm(s: str) -> str:
+    return re.sub(r"[^a-z0-9]+", " ", s.lower()).strip()
+
+
+def _ol_url(key: str) -> str:
+    olid = key.rsplit("/", 1)[-1]
+    if key.startswith("/local/"):
+        # Landed file with no OpenLibrary work: a stable local identity.
+        return f"local://ama-coding/{olid}"
+    kind = "works" if olid.endswith("W") else "books"
+    return f"https://openlibrary.org/{kind}/{olid}"
+
+
+_YEAR_RE = re.compile(r"\b(19|20)\d\d\b")
+
+
+def _year_of(s: str) -> str | None:
+    m = _YEAR_RE.search(s)
+    return m.group(0) if m else None
+
+
+def _folder_matches(folder: str, entries: list[dict]) -> dict | None:
+    """Best title entry for a landed folder: exact normalised title, else a
+    title that contains it (or is contained by it) once spaces are dropped,
+    with agreeing years. Containment keeps ICD-10-CM apart from ICD-10-PCS
+    and a 4th edition apart from a 6th; a similarity ratio does not."""
+    target = _norm(folder)
+    by_norm = {_norm(e["title"]): e for e in entries}
+    if target in by_norm:
+        return by_norm[target]
+    year = _year_of(folder)
+    tight = target.replace(" ", "")
+    best, best_len = None, 0
+    for e in entries:
+        ey = _year_of(e["title"]) or (str(e["year"]) if e.get("year") else None)
+        if year and ey and year != ey:
+            continue
+        cand = _norm(e["title"]).replace(" ", "")
+        if cand in tight or tight in cand:
+            if len(cand) > best_len:
+                best, best_len = e, len(cand)
+    return best
+
+
+def _folder_entry(folder: str) -> dict:
+    """A synthetic entry for a landed book no OpenLibrary title covers."""
+    year = _year_of(folder)
+    return {"key": f"/local/{_norm(folder).replace(' ', '-')}", "title": folder,
+            "year": int(year) if year else None, "isbns": []}
+
+
+def _index_books(books_dir: Path) -> dict[str, list[Path]]:
+    """Map normalised folder title -> ebook files LazyLibrarian has landed."""
+    index: dict[str, list[Path]] = {}
+    if not books_dir.is_dir():
+        return index
+    for folder in books_dir.iterdir():
+        if not folder.is_dir():
+            continue
+        files = [p for p in folder.iterdir() if p.suffix.lower() in CONTENT_TYPES]
+        if files:
+            index[_norm(folder.name)] = sorted(files)
+    return index
+
+
+def _existing_attachment_names(db: Db, parent_id: int) -> set[str]:
+    rows = db.con.execute(
+        "SELECT path FROM itemAttachments WHERE parentItemID = ?", (parent_id,)
+    ).fetchall()
+    return {str(r[0]).removeprefix("storage:") for r in rows if r[0]}
+
+
+def _attach(db: Db, parent_id: int, src: Path, *, dry_run: bool) -> None:
+    att_key = generate_key()
+    storage_dir = ZOTERO_STORAGE / att_key
+    if dry_run:
+        print(f"    would attach {src.name}")
+        return
+    subprocess.run(["sudo", "mkdir", "-p", str(storage_dir)], check=True)
+    subprocess.run(["sudo", "cp", str(src), str(storage_dir / src.name)], check=True)
+    subprocess.run(["sudo", "chown", "-R", "100999:100999", str(storage_dir)], check=True)
+    att_id = db.add_attachment(
+        parent_id,
+        key=att_key,
+        content_type=CONTENT_TYPES[src.suffix.lower()],
+        path=f"storage:{src.name}",
+    )
+    db.set_field(att_id, "title", src.name)
+    print(f"    attached {src.name} ({att_key})")
+
+
+def _ensure_item(db: Db, entry: dict, collection_key: str, *, dry_run: bool) -> int | None:
+    url = _ol_url(entry["key"])
+    item_id = db.find_item_by_url(url)
+    if item_id is not None:
+        db.add_to_collection(item_id, collection_key=collection_key)
+        return item_id
+    if dry_run:
+        print(f"  would create: {entry['title']}")
+        return None
+    now = now_iso()
+    item_id = db.create_item(TYPE_MAP["book"], now=now)
+    fields = {
+        "title": entry["title"],
+        "publisher": PUBLISHER,
+        "url": url,
+        "accessDate": now,
+        "language": "en",
+        "libraryCatalog": "Open Library",
+    }
+    if not entry["key"].startswith("/local/"):
+        fields["extra"] = f"OpenLibrary: {entry['key'].rsplit('/', 1)[-1]}"
+    if entry.get("year"):
+        fields["date"] = str(entry["year"])
+    if entry.get("isbns"):
+        fields["ISBN"] = " ".join(entry["isbns"])
+    db.set_fields(item_id, fields)
+    db.add_creators(item_id, [("", PUBLISHER)], creator_type="author")
+    tags = ["source:ama", "module:coding"]
+    if entry.get("year"):
+        tags.append(f"year:{entry['year']}")
+    db.sync_tags(item_id, tags)
+    db.add_to_collection(item_id, collection_key=collection_key)
+    print(f"  created: {entry['title']}")
+    return item_id
+
+
+def main() -> None:
+    ap = argparse.ArgumentParser(description=__doc__ or "")
+    ap.add_argument("--titles", type=Path, default=DEFAULT_TITLES)
+    ap.add_argument("--books", type=Path, default=DEFAULT_BOOKS)
+    ap.add_argument("--collection", default=DEFAULT_COLLECTION)
+    ap.add_argument("--dry-run", action="store_true")
+    args = ap.parse_args()
+
+    entries = json.loads(args.titles.read_text())
+    books = _index_books(args.books)
+    print(f"{len(entries)} titles, {len(books)} ebook folders on disk")
+
+    created = attached = existing = 0
+    with Db(ZOTERO_DB) as db:
+        collection_key = db.ensure_collection(args.collection)
+        for entry in entries:
+            before = db.find_item_by_url(_ol_url(entry["key"]))
+            item_id = _ensure_item(db, entry, collection_key, dry_run=args.dry_run)
+            if item_id is None:
+                continue
+            if before is None:
+                created += 1
+            else:
+                existing += 1
+            have = _existing_attachment_names(db, item_id)
+            for f in books.get(_norm(entry["title"]), []):
+                if f.name in have:
+                    continue
+                _attach(db, item_id, f, dry_run=args.dry_run)
+                attached += 1
+        # Folders the exact-title pass did not consume: fuzzy-match a title
+        # (same year), else mint a local item so every landed file is filed.
+        exact = {_norm(e["title"]) for e in entries}
+        for folder_norm, files in books.items():
+            if folder_norm in exact:
+                continue
+            folder = files[0].parent.name
+            entry = _folder_matches(folder, entries) or _folder_entry(folder)
+            item_id = _ensure_item(db, entry, collection_key, dry_run=args.dry_run)
+            if item_id is None:
+                continue
+            have = _existing_attachment_names(db, item_id)
+            for f in files:
+                if f.name in have:
+                    continue
+                _attach(db, item_id, f, dry_run=args.dry_run)
+                attached += 1
+        if not args.dry_run:
+            db.commit()
+
+    print()
+    print(f"collection '{args.collection}' ({collection_key}): "
+          f"{created} created, {existing} already present, {attached} files attached")
+    if not args.dry_run:
+        print("Restart Zotero so the desktop reloads the database.")
+
+
+if __name__ == "__main__":
+    main()