feat(pfs): fetch + register CY2026 NPRM addenda (refs #590)

dev/scripts/fetch_nprm_addenda.py

@@ -0,0 +1,198 @@
+"""Fetch + register the CY2026 PFS NPRM (proposed rule) addenda.
+
+Downloads the CMS-1832-P "CY 2026 PFS Proposed Rule Addenda" ZIP, which
+contains Addendum B (proposed RVUs by HCPCS), and optionally registers
+Addendum B in bib against the CY2026 PFS NPRM rule item (2KVJ2HKX),
+tagged ``sup:2026_PFS_NPRM`` so Task 5's loader can discover the
+*proposed* Addendum B without touching final-rule files (which carry
+``sup:2026_PFS_FR`` — the two namespaces must never collide).
+
+Source page (CMS-1832-P detail page — the canonical
+``.../pfs-federal-regulation-notices/cms-1832-p`` URL 404s; this is the
+current path found via the federal-regulation-notices listing):
+    https://www.cms.gov/medicare/payment/fee-schedules/physician/federal-regulation-notices/cms-1832-p
+Download link text on that page: "CY 2026 PFS Proposed Rule Addenda -
+Updated 07/29/2025" (the URL path segment literally contains
+"07/29/2025" for the update-date suffix — that's CMS's own href, not a
+typo here).
+
+Usage:
+    uv run python dev/scripts/fetch_nprm_addenda.py --dry-run
+    uv run python dev/scripts/fetch_nprm_addenda.py
+    uv run python dev/scripts/fetch_nprm_addenda.py --register
+"""
+
+from __future__ import annotations
+
+import argparse
+import re
+import zipfile
+from pathlib import Path
+
+import httpx
+
+ROOT = Path(__file__).resolve().parents[2]
+DEST_DIR = ROOT / "data" / "cms" / "pfs_nprm" / "2026"
+
+USER_AGENT = (
+    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
+    "(KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36"
+)
+
+# CY2026 PFS Proposed Rule (CMS-1832-P) Addenda ZIP — resolved from the
+# CMS-1832-P detail page (see module docstring for the page URL).
+ADDENDA_URL = (
+    "https://www.cms.gov/files/zip/"
+    "cy-2026-pfs-proposed-rule-addenda-updated-07/29/2025.zip"
+)
+ADDENDA_ZIP_NAME = "cy2026_pfs_nprm_addenda.zip"
+
+ADDENDUM_B_RE = re.compile(r"(?i)addendum[_ ]?b")
+
+RULE_ITEM_KEY = "2KVJ2HKX"
+ADDENDUM_B_TITLE = "CY2026 PFS NPRM Addendum B (proposed RVUs)"
+REGISTER_TAGS = ["sup:2026_PFS_NPRM", "module:pfs", "file:rvu", "year:2026"]
+
+
+def download_zip(dest: Path, *, retries: int = 3) -> bool:
+    """Download the addenda ZIP if it doesn't already exist."""
+    import time
+
+    if dest.exists():
+        print(f"  EXISTS {dest.name}")
+        return True
+    print(f"  GET    {dest.name} ...", end="", flush=True)
+    for attempt in range(1, retries + 1):
+        try:
+            resp = httpx.get(
+                ADDENDA_URL,
+                headers={"User-Agent": USER_AGENT},
+                timeout=120,
+                follow_redirects=True,
+            )
+            if resp.status_code == 200:
+                dest.write_bytes(resp.content)
+                print(f" {len(resp.content) / 1024 / 1024:.1f} MB")
+                return True
+            if resp.status_code < 500 or attempt == retries:
+                print(f" HTTP {resp.status_code}")
+                return False
+        except Exception as e:
+            if attempt == retries:
+                print(f" ERROR: {e}")
+                return False
+        time.sleep(2 * attempt)
+    return False
+
+
+def extract_zip(zip_path: Path, dest_dir: Path) -> list[Path]:
+    """Unzip into dest_dir. Idempotent — extraction overwrites in place."""
+    with zipfile.ZipFile(zip_path) as zf:
+        names = [n for n in zf.namelist() if not n.endswith("/")]
+        zf.extractall(dest_dir)
+    return sorted(dest_dir / n for n in names)
+
+
+def find_addendum_b(files: list[Path]) -> Path | None:
+    """Find the extracted file whose name matches Addendum B."""
+    candidates = [
+        f
+        for f in files
+        if ADDENDUM_B_RE.search(f.name) and f.suffix.lower() in (".xlsx", ".csv")
+    ]
+    if not candidates:
+        return None
+    # Prefer .xlsx over .csv if both are present.
+    candidates.sort(key=lambda f: f.suffix.lower() != ".xlsx")
+    return candidates[0]
+
+
+def preview_addendum_b(path: Path) -> None:
+    """Print the header row (and a couple of data rows) for verification."""
+    import polars as pl
+
+    if path.suffix.lower() == ".xlsx":
+        df = pl.read_excel(path, read_options={"header_row": None})
+    else:
+        df = pl.read_csv(path, n_rows=5, has_header=False)
+    print(f"  columns/first rows of {path.name}:")
+    with pl.Config(tbl_cols=-1, tbl_width_chars=200):
+        print(df.head(5))
+
+
+def register(addendum_b_path: Path) -> None:
+    """Attach Addendum B to the NPRM rule item and tag it."""
+    from conf import connect
+
+    store = connect.bib()
+    att_key = store.attach_file(RULE_ITEM_KEY, addendum_b_path, title=ADDENDUM_B_TITLE)
+    con = store._con()  # noqa: SLF001 — same pattern as bib/oig.py, bib/iom.py
+    row = con.execute(
+        "SELECT storage_path FROM attachments WHERE key = ?", (att_key,)
+    ).fetchone()
+    storage_path = row["storage_path"] if row else "<unknown>"
+    print(f"Attached {addendum_b_path.name} (key {att_key}) -> {storage_path}")
+    for tag in REGISTER_TAGS:
+        store.add_tag(RULE_ITEM_KEY, tag)
+    print(f"Tagged {RULE_ITEM_KEY} with: {', '.join(REGISTER_TAGS)}")
+    item = store.get(RULE_ITEM_KEY)
+    print(f"Current tags on {RULE_ITEM_KEY}: {item.tags}")
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="Fetch + register the CY2026 PFS NPRM addenda"
+    )
+    parser.add_argument(
+        "--dry-run",
+        action="store_true",
+        help="Print what would be downloaded/extracted without doing it.",
+    )
+    parser.add_argument(
+        "--register",
+        action="store_true",
+        help="After fetching, attach Addendum B to bib item "
+        f"{RULE_ITEM_KEY} and apply the {', '.join(REGISTER_TAGS)} tags.",
+    )
+    args = parser.parse_args()
+
+    zip_dest = DEST_DIR / ADDENDA_ZIP_NAME
+
+    if args.dry_run:
+        print(f"Would GET  {ADDENDA_URL}")
+        print(f"Would save {zip_dest}")
+        print(f"Would extract into {DEST_DIR}")
+        if args.register:
+            print(
+                f"Would attach Addendum B to {RULE_ITEM_KEY} "
+                f"and tag with: {', '.join(REGISTER_TAGS)}"
+            )
+        return
+
+    DEST_DIR.mkdir(parents=True, exist_ok=True)
+
+    if not download_zip(zip_dest):
+        raise SystemExit(f"Failed to download {ADDENDA_URL}")
+
+    files = extract_zip(zip_dest, DEST_DIR)
+    print(f"\nExtracted {len(files)} files to {DEST_DIR}:")
+    for f in files:
+        size_kb = f.stat().st_size / 1024
+        print(f"  {f.relative_to(DEST_DIR)} ({size_kb:.0f} KB)")
+
+    addendum_b = find_addendum_b(files)
+    if addendum_b is None:
+        raise SystemExit(
+            "Could not find an Addendum B file (pattern "
+            f"{ADDENDUM_B_RE.pattern!r}) among extracted files."
+        )
+    print(f"\nAddendum B: {addendum_b.relative_to(DEST_DIR)}")
+    preview_addendum_b(addendum_b)
+
+    if args.register:
+        print()
+        register(addendum_b)
+
+
+if __name__ == "__main__":
+    main()