diff --git a/dev/scripts/import_lead_corpus.py b/dev/scripts/import_lead_corpus.py new file mode 100644 index 0000000..14ce602 --- /dev/null +++ b/dev/scripts/import_lead_corpus.py @@ -0,0 +1,413 @@ +"""Vendor the staged LEAD corpus documents into the bib library. + +``lead_corpus/`` (staged 2026-10-23 for the LEAD repository) holds two +kinds of document: + +- ``lead_research//*`` — methodology papers, CMS model + specifications and research articles, catalogued in ``sources.csv`` + (title, authors, year, URL, sha256, topic). +- ``lead_enrichment//**/{dictionary,reference}__*`` — data + dictionaries and methodology references for the staged datasets, + catalogued in ``_catalogs/*_jobs.json`` (title, publisher, release, + download URL, how LEAD could use it), plus the eCQM HTML specs. + +Each distinct document (by sha256) becomes one bib ``source`` item with +its file attached, tagged ``source:lead-corpus`` and filed under the +"LEAD corpus" collection. The datasets themselves are not documents and +stay out of the library. + +Skipped on purpose: + +- ``dictionary__*`` files that are byte-identical copies of + data.cms.gov's JavaScript app shell ("You need to enable JavaScript + to run this app") — the fetch saved the page skeleton, not the + dictionary, so there is nothing to file. +- text extracts that sit next to a research PDF of the same name — they + are derived from the PDF, which the library indexes itself. + +Re-running is safe: items match on URL (or, without a public URL, on +title within ``source:lead-corpus``) and attachments on file name. + +Usage (from the main checkout, so ``db.bib`` is the real library):: + + uv run python dev/scripts/import_lead_corpus.py [--root lead_corpus] [--dry-run] + +Then push to Zotero:: + + docker exec api uv run --no-sync \\ + stack bib sync-zotero --tag source:lead-corpus --hold-zotero +""" + +from __future__ import annotations + +import argparse +import csv +import hashlib +import json +import re +from collections import Counter +from dataclasses import dataclass, field +from pathlib import Path + +TAG = "source:lead-corpus" +COLLECTION = "LEAD corpus" + +# data.cms.gov's SPA shell, saved 81 times in place of real dictionaries. +SPA_SHELL_MD5 = "71f2ae0beed162e2dbcb8d8d9445c890" + +RESEARCH_TOPICS = { + "benchmark": "Benchmark", + "cms_primary": "CMS primary sources", + "ml_sota": "Machine learning methods", + "quality": "Quality", + "risk_adjustment": "Risk adjustment", + "settlement": "Settlement", +} +DOC_SUFFIXES = {".pdf", ".html", ".htm", ".txt", ".xlsx", ".xls", ".zip", ".docx"} +INSTITUTION = re.compile( + r"\b(CMS|CMMI|MedPAC|NORC|RTI|Yale|YNHHSC|CORE|Board|eCFR|Congress|ASPE|PTAC|" + r"Office|Center|University|International|Commission)\b" +) + + +@dataclass +class Doc: + path: Path + sha256: str + title: str + url: str = "" + institution: str = "" + authors: list[str] = field(default_factory=list) + date: str = "" + doc_type: str = "" + abstract: str = "" + tags: list[str] = field(default_factory=list) + collection: str = "" + + +def _sha256(path: Path) -> str: + h = hashlib.sha256() + with path.open("rb") as f: + for block in iter(lambda: f.read(1 << 20), b""): + h.update(block) + return h.hexdigest() + + +def _md5(path: Path) -> str: + return hashlib.md5(path.read_bytes()).hexdigest() # noqa: S324 — identity, not security + + +def split_authors(raw: str) -> tuple[list[str], str]: + """Split a sources.csv author cell into (people, institution). + + People come back as ``"Last First"`` strings, the shape + ``bib.sync`` parses from an ``Authors:`` line. A cell naming only an + organisation ("CMS Innovation Center") is an institution; a trailing + parenthetical on a list of people ("... (RTI International)") is the + institution behind them. + """ + raw = raw.strip() + inst = "" + m = re.search(r"\(([^)]*)\)\s*$", raw) + if m: + inst, raw = m.group(1).strip(), raw[: m.start()].strip() + if not raw: + return [], inst + if INSTITUTION.search(raw) and not inst: + return [], raw + people = [] + for part in raw.split(","): + name = re.sub(r"\bet al\.?$", "", part).strip() + if name: + people.append(name) + return people, inst + + +def research_docs(root: Path) -> tuple[list[Doc], list[str]]: + base = root / "lead_research" + rows = list(csv.DictReader((base / "sources.csv").open(encoding="utf-8"))) + # sources.csv lists a document once per topic folder it was filed in + # (same bytes, same title), so a file can stand for several rows. + by_sha: dict[str, list[dict]] = {} + by_name: dict[str, list[dict]] = {} + for r in rows: + if r["sha256"].strip(): + by_sha.setdefault(r["sha256"].strip(), []).append(r) + by_name.setdefault(Path(r["local_path"]).name, []).append(r) + docs, notes = [], [] + seen_rows: set[int] = set() + for path in sorted(base.rglob("*")): + if not path.is_file() or path.suffix.lower() not in DOC_SUFFIXES: + continue + if path.suffix.lower() == ".txt" and any( + path.with_suffix(s).exists() for s in (".pdf", ".PDF") + ): + continue # text extract of a PDF we attach anyway + topic_dir = path.relative_to(base).parts[0] + sha = _sha256(path) + matched = by_sha.get(sha) or by_name.get(path.name) or [] + if matched: + row = matched[0] + seen_rows.update(id(r) for r in matched) + people, inst = split_authors(row["authors"]) + topics = list( + dict.fromkeys( + t.strip() + for r in matched + for t in r["topic"].split(";") + if t.strip() + ) + ) + short = [t for t in topics if re.fullmatch(r"[a-z0-9_]+", t)] + detail = [t for t in topics if t not in short] + url = row["url"].strip() + docs.append( + Doc( + path=path, + sha256=sha, + title=row["title"].strip(), + url=url if url.startswith("http") else "", + institution=inst, + authors=people, + date=row["year"].strip(), + doc_type="methodology / research", + abstract=("LEAD use: " + "; ".join(detail)) if detail else "", + tags=[ + f"lead-topic:{t}" for t in dict.fromkeys([topic_dir, *short]) + ], + collection=RESEARCH_TOPICS.get(topic_dir, "Research"), + ) + ) + else: + notes.append(f"research file not in sources.csv: {path.relative_to(root)}") + docs.append( + Doc( + path=path, + sha256=sha, + title=path.stem.replace("_", " "), + doc_type="methodology / research", + tags=[f"lead-topic:{topic_dir}", "lead:uncatalogued"], + collection=RESEARCH_TOPICS.get(topic_dir, "Research"), + ) + ) + for r in rows: + if id(r) not in seen_rows: + notes.append(f"sources.csv row without a file: {r['title'][:80]}") + return docs, notes + + +def _jobs(root: Path) -> list[dict]: + jobs: list[dict] = [] + for name in ("pdc_jobs.json", "cms_datajson_jobs.json", "other_jobs.json"): + p = root / "lead_enrichment" / "_catalogs" / name + if p.exists(): + jobs.extend(json.loads(p.read_text(encoding="utf-8"))) + return jobs + + +def enrichment_docs(root: Path) -> tuple[list[Doc], list[str]]: + base = root / "lead_enrichment" + # A job's relpath is either relative to the dataset folder + # ("2026_10_01/dictionary__X.pdf") or a bare/ID-prefixed file name + # ("dictionary__DOC_Data_Dictionary.pdf"); match on either. + jobs: dict[tuple[str, str], dict] = {} + for j in _jobs(root): + rel = j.get("relpath", "") + jobs.setdefault((j.get("source_id", ""), rel), j) + jobs.setdefault((j.get("source_id", ""), Path(rel).name), j) + docs, notes = [], [] + shells = 0 + candidates = [ + p + for p in sorted(base.rglob("*")) + if p.is_file() + and ( + p.name.startswith(("dictionary__", "reference__")) + or (p.parent.name == "ecqi_ecqm_specs" and p.suffix.lower() == ".html") + ) + ] + for path in candidates: + if path.stat().st_size < 10_000 and _md5(path) == SPA_SHELL_MD5: + shells += 1 + continue + source_id = path.relative_to(base).parts[0] + kind, _, relpath = path.name.partition("__") + if not relpath: # eCQM spec, no kind prefix + kind, relpath = "specification", path.name + rel = path.relative_to(base / source_id).as_posix() + job = jobs.get((source_id, rel)) or jobs.get((source_id, path.name)) + # No row for this file (the Provider Data Catalog dictionaries): + # borrow publisher and release from the dataset's other entries. + sibling = job or next( + (j for (sid, _), j in jobs.items() if sid == source_id), {} + ) + if job: + dataset = job.get("title", source_id).split(" : ")[0].strip() + else: + dataset = ( + source_id.removeprefix("cms_") + .removeprefix("pdc_") + .replace("_", " ") + .strip() + .capitalize() + ) + if source_id.startswith("pdc_"): + dataset = f"Provider Data Catalog: {dataset.lower()}" + label = { + "dictionary": "data dictionary", + "reference": "methodology reference", + "specification": "eCQM specification", + }.get(kind, kind) + if kind == "specification": + m = re.match(r"(CMS\d+)-v([\d.]+?)\.0+\.000-(\w+)", path.stem) + title = ( + f"{m.group(1)}v{m.group(2)} eCQM specification ({m.group(3)})" + if m + else path.stem + ) + else: + title = f"{dataset} — {label}: {Path(relpath).stem.replace('_', ' ')}" + docs.append( + Doc( + path=path, + sha256=_sha256(path), + title=title, + url=(job or {}).get("url", ""), + institution=sibling.get("publisher", ""), + date=sibling.get("release", ""), + doc_type=label, + abstract=("LEAD use: " + job["use"]) if job and job.get("use") else "", + tags=[f"lead-dataset:{source_id}", f"doc:{kind}"], + collection=( + "Measure specifications" + if kind == "specification" + else "Dataset documentation" + ), + ) + ) + if job is None: + notes.append(f"no catalog entry: {path.relative_to(root)}") + notes.append( + f"skipped {shells} data.cms.gov app-shell placeholders (no dictionary content)" + ) + return docs, notes + + +def dedupe(docs: list[Doc]) -> tuple[list[Doc], int]: + """One item per distinct file: the same dictionary often sits in every + release folder of a dataset. Keep the newest release's copy, but carry + every dataset tag.""" + best: dict[str, Doc] = {} + for d in docs: + cur = best.get(d.sha256) + if cur is None: + best[d.sha256] = d + continue + keep, other = (d, cur) if d.date > cur.date else (cur, d) + keep.tags = list(dict.fromkeys([*keep.tags, *other.tags])) + best[d.sha256] = keep + return list(best.values()), len(docs) - len(best) + + +def _existing_by_title(store, title: str) -> str | None: + row = ( + store._con() + .execute( + """SELECT i.key FROM items i + JOIN item_tags it ON it.item_id = i.id + JOIN tags t ON t.id = it.tag_id + WHERE t.name = ? AND i.title = ? LIMIT 1""", + (TAG, title), + ) + .fetchone() + ) + return row["key"] if row else None + + +def import_docs(docs: list[Doc]) -> Counter: + from bib.item import Source + from bib.store import Store + + store = Store() + cols = store.ensure_collections( + { + COLLECTION: { + "Research": {name: {} for name in RESEARCH_TOPICS.values()}, + "Dataset documentation": {}, + "Measure specifications": {}, + } + } + ) + stats: Counter = Counter() + for d in docs: + extra = f"Authors: {'; '.join(d.authors)}" if d.authors else "" + extra = "\n".join(filter(None, [extra, f"sha256: {d.sha256}"])) + item = Source( + title=d.title, + url=d.url, + institution=d.institution, + date_published=d.date, + abstract=d.abstract, + extra=extra, + doc_type=d.doc_type, + ) + if not d.url and (key := _existing_by_title(store, d.title)): + item.key = key + store.add_tags(key, [TAG, *d.tags]) + status = "unchanged" + else: + key, status = store.upsert_status( + item, tags=[TAG, *d.tags], collection=cols[d.collection] + ) + stats[status] += 1 + before = ( + store._con() + .execute( + "SELECT count(*) FROM attachments a JOIN items i ON i.id = a.item_id " + "WHERE i.key = ?", + (key,), + ) + .fetchone()[0] + ) + store.attach_file(key, d.path) + after = ( + store._con() + .execute( + "SELECT count(*) FROM attachments a JOIN items i ON i.id = a.item_id " + "WHERE i.key = ?", + (key,), + ) + .fetchone()[0] + ) + stats["attached" if after > before else "attachment-present"] += 1 + return stats + + +def main() -> int: + ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) + ap.add_argument("--root", type=Path, default=Path("lead_corpus")) + ap.add_argument("--dry-run", action="store_true") + args = ap.parse_args() + + research, rnotes = research_docs(args.root) + enrich, enotes = enrichment_docs(args.root) + docs, dupes = dedupe(research + enrich) + for n in rnotes + enotes: + print("note:", n) + by_col = Counter(d.collection for d in docs) + print( + f"{len(docs)} distinct documents ({len(research)} research files, " + f"{len(enrich)} dataset docs, {dupes} duplicate copies folded)" + ) + for col, n in sorted(by_col.items()): + print(f" {col}: {n}") + print(f" without a public URL: {sum(not d.url for d in docs)}") + if args.dry_run: + return 0 + stats = import_docs(docs) + print("result:", dict(stats)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/bib/sync.py b/src/bib/sync.py index d5e4161..5de9ac7 100644 --- a/src/bib/sync.py +++ b/src/bib/sync.py @@ -287,47 +287,50 @@ def push_to_zotero( collection_id = db.find_collection(collection_key) for item in items: - # Skip (update-in-place) if URL already exists in Zotero - if item.url: - existing_id = db.find_item_by_url(item.url) - if existing_id is not None: - # Refresh fields from scratch — earlier syncs used - # a stale FIELD_MAP that landed values in wrongly- - # named columns (title→sessionTitle, seriesTitle→ - # ISBN, etc.). replace_fields clears the item's - # entire itemData row set before rewriting with the - # corrected map. - db.replace_fields(existing_id, _item_to_zotero_fields(item)) - db.sync_tags(existing_id, item.tags) - if store is not None: - stats["attachments"] += _sync_attachments( - db, - store, - item, - existing_id, - storage_dir, - ) - stats["notes"] += _sync_notes( - db, - store, - item, - existing_id, - ) - stats["fr_links"] += _sync_fr_links( - db, - store, - item, - existing_id, - ) - # Backfill collection membership for items that were - # created before sync learned about the hierarchy. - path = _zotero_collection_path(item) - if path: - key = _resolve_path(tuple(path)) - if db.add_to_collection(existing_id, collection_key=key): - stats["collections"] += 1 - stats["skipped"] += 1 - continue + # Update in place when the item is already in Zotero: by URL, + # else by key (Zotero keys mirror bib keys). Matching on URL + # alone re-created every URL-less item on every sync. + existing_id = db.find_item_by_url(item.url) if item.url else None + if existing_id is None: + existing_id = _same_item_by_key(db, item) + if existing_id is not None: + # Refresh fields from scratch — earlier syncs used + # a stale FIELD_MAP that landed values in wrongly- + # named columns (title→sessionTitle, seriesTitle→ + # ISBN, etc.). replace_fields clears the item's + # entire itemData row set before rewriting with the + # corrected map. + db.replace_fields(existing_id, _item_to_zotero_fields(item)) + db.sync_tags(existing_id, item.tags) + if store is not None: + stats["attachments"] += _sync_attachments( + db, + store, + item, + existing_id, + storage_dir, + ) + stats["notes"] += _sync_notes( + db, + store, + item, + existing_id, + ) + stats["fr_links"] += _sync_fr_links( + db, + store, + item, + existing_id, + ) + # Backfill collection membership for items that were + # created before sync learned about the hierarchy. + path = _zotero_collection_path(item) + if path: + key = _resolve_path(tuple(path)) + if db.add_to_collection(existing_id, collection_key=key): + stats["collections"] += 1 + stats["skipped"] += 1 + continue # Resolve Zotero item type ej = json.loads(item.to_row().get("extra_json", "{}")) @@ -405,6 +408,21 @@ def push_to_zotero( return stats +def _same_item_by_key(db: Db, item: Item) -> int | None: + """The Zotero item carrying *item*'s key, when it is this item. + + Keys mirror bib keys, but a bare key hit can be an unrelated row that + happens to hold the key — the title must match too, otherwise the + caller creates a fresh item under a new key rather than clobber it. + """ + if not is_valid_key(item.key): + return None + item_id = db.find_item_by_key(item.key) + if item_id is None or db.get_field(item_id, "title") != item.title: + return None + return item_id + + def _zotero_collection_path(item: Item) -> list[str]: """Derive the Zotero collection hierarchy an item should live under. @@ -456,9 +474,35 @@ def _zotero_collection_path(item: Item) -> list[str]: if guidance: return [*base, guidance] return base + if "source:lead-corpus" in item.tags: + # dev/scripts/import_lead_corpus.py: research papers by the topic + # folder they were staged in (first lead-topic: tag), dataset + # dictionaries and eCQM specs in their own branches. + base = ["LEAD corpus"] + if "doc:specification" in item.tags: + return [*base, "Measure specifications"] + if any(t.startswith("lead-dataset:") for t in item.tags): + return [*base, "Dataset documentation"] + for t in item.tags: + if t.startswith("lead-topic:"): + topic = t.split(":", 1)[1] + if topic in _LEAD_TOPICS: + return [*base, "Research", _LEAD_TOPICS[topic]] + return [*base, "Research"] return [] +# lead_corpus/lead_research topic folders → Zotero collection names. +_LEAD_TOPICS = { + "benchmark": "Benchmark", + "cms_primary": "CMS primary sources", + "ml_sota": "Machine learning methods", + "quality": "Quality", + "risk_adjustment": "Risk adjustment", + "settlement": "Settlement", +} + + def _sync_attachments( db: Db, store: Store, diff --git a/tests/bib/test_sync_deeper.py b/tests/bib/test_sync_deeper.py index c0fdf56..3298b63 100644 --- a/tests/bib/test_sync_deeper.py +++ b/tests/bib/test_sync_deeper.py @@ -130,6 +130,27 @@ class TestPushToZotero: stats = push_to_zotero([item], zotero_db=path) assert stats["skipped"] >= 1 + def test_url_less_item_matches_on_key(self, tmp_path): + """A staged document with no public URL is found again by its key + (Zotero keys mirror bib keys) instead of being re-created per sync.""" + path = _setup(tmp_path) + item = Source(key="AB2CD3EF", title="Data dictionary", doc_type="dictionary") + + first = push_to_zotero([item], zotero_db=path) + second = push_to_zotero([item], zotero_db=path) + + assert first["created"] == 1 + assert second == {**second, "created": 0, "skipped": 1} + with Db(path) as db: + assert db.find_item_by_key("AB2CD3EF") is not None + assert ( + db.con.execute( + "SELECT count(*) FROM items WHERE itemID NOT IN " + "(SELECT itemID FROM itemAttachments)" + ).fetchone()[0] + == 1 + ) + def test_collection_routing(self, tmp_path): path = _setup(tmp_path) item = Manual(title="IOM Ch1", manual_name="Claims Processing") @@ -195,6 +216,38 @@ class TestCollectionPathEdge: path = _zotero_collection_path(item) assert "Hospitals" in path[-1] + def test_lead_corpus_research_by_first_known_topic(self): + item = Source(title="CMS-HCC calibration") + for t in ("source:lead-corpus", "lead-topic:cms_primary", "lead-topic:ratchet"): + item.add_tag(t) + assert _zotero_collection_path(item) == [ + "LEAD corpus", + "Research", + "CMS primary sources", + ] + + def test_lead_corpus_unknown_topic_lands_in_research(self): + item = Source(title="Paper") + item.add_tag("source:lead-corpus") + item.add_tag("lead-topic:ratchet") + assert _zotero_collection_path(item) == ["LEAD corpus", "Research"] + + def test_lead_corpus_dataset_docs_and_specs(self): + dictionary = Source(title="ACO participants — data dictionary") + for t in ("source:lead-corpus", "lead-dataset:cms_aco", "doc:dictionary"): + dictionary.add_tag(t) + spec = Source(title="CMS122v14 eCQM specification (QDM)") + for t in ("source:lead-corpus", "lead-dataset:ecqi", "doc:specification"): + spec.add_tag(t) + assert _zotero_collection_path(dictionary) == [ + "LEAD corpus", + "Dataset documentation", + ] + assert _zotero_collection_path(spec) == [ + "LEAD corpus", + "Measure specifications", + ] + def test_oig_no_sector_no_guidance(self): """Line 421: OIG with neither sector nor guidance → base only.""" item = Source(title="OIG doc")