homelab/stack
1
0
Fork 0
Code Issues 114 Pull Requests 1 Actions Packages Projects 1 Releases Wiki Activity

feat(bib): vendor the staged LEAD corpus documents into the library

Browse Source 
dev/scripts/import_lead_corpus.py files every distinct document in
lead_corpus/ as a bib `source` item with its file attached, tagged
source:lead-corpus, under a "LEAD corpus" collection tree:

- lead_research: 73 methodology papers, CMS model specifications and
  research articles, with title / authors / year / URL from
  sources.csv (people become an `Authors:` line, organisations the
  institution) and the per-topic LEAD notes as the abstract.
- lead_enrichment: 68 data dictionaries, methodology references and
  eCQM specifications, with title / publisher / release / download URL
  and "how LEAD could use it" from the _catalogs job lists.

Identical files (one PDF filed under two topic folders, the same
dictionary in every release folder) fold to one item by sha256: 123
items. Skipped: 81 `dictionary__*` files that are byte-identical
copies of data.cms.gov's JavaScript app shell (no dictionary content),
and text extracts sitting next to their PDF. Re-running is a no-op
(items match on URL, or on title within the tag; attachments on name).

bib.sync, so the import lands properly in Zotero:
- `source:lead-corpus` items route to LEAD corpus / Research /
  <topic>, / Dataset documentation, or / Measure specifications.
- An item already in Zotero is matched by URL, else by key when the
  Zotero row with that key has the same title. Matching on URL alone
  re-created every URL-less item on every sync — the second corpus
  sync duplicated the 11 documents without a public URL. The title
  guard keeps an unrelated row that merely holds the key from being
  overwritten (test_key_collision).
This commit is contained in:
kert
2026-10-27 12:38:39 -04:00
parent a4b4110dc1
commit 2dc0e9c586
3 changed files with 551 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

413
dev/scripts/import_lead_corpus.py Normal file
View File

@@ -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/<topic>/*`` — methodology papers, CMS model
specifications and research articles, catalogued in ``sources.csv``
(title, authors, year, URL, sha256, topic).
- ``lead_enrichment/<dataset>/**/{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())

126
src/bib/sync.py
View File

@@ -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,

53
tests/bib/test_sync_deeper.py
View File

@@ -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")