feat(pfs): CY2027 NPRM addenda capture — fetch_nprm_addenda --year (refs #610)

dev/scripts/fetch_nprm_addenda.py

@@ -1,25 +1,16 @@
-"""Fetch + register the CY2026 PFS NPRM (proposed rule) addenda.
+"""Fetch + register PFS NPRM (proposed rule) addenda, by rule year.
 
-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).
+Downloads a CY{year} PFS Proposed Rule "Addenda" ZIP, which contains
+Addendum B (proposed RVUs by HCPCS), and optionally registers Addendum B
+in bib against that year's NPRM rule item, tagged ``sup:{year}_PFS_NPRM``
+so the loader can discover the *proposed* Addendum B without touching
+final-rule files (which carry ``sup:{year}_PFS_FR`` — the two namespaces
+must never collide).
 
 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
+    uv run python dev/scripts/fetch_nprm_addenda.py --year 2027
+    uv run python dev/scripts/fetch_nprm_addenda.py --year 2027 --register
 """
 
 from __future__ import annotations
@@ -27,36 +18,87 @@ from __future__ import annotations
 import argparse
 import re
 import zipfile
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
 
 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:
-#   https://www.cms.gov/medicare/payment/fee-schedules/physician/federal-regulation-notices/cms-1832-p
-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"]
+
+@dataclass(frozen=True)
+class YearConfig:
+    """Per-rule-year config: where to fetch the addenda ZIP and how to
+    register Addendum B against the corresponding bib rule item."""
+
+    year: int
+    detail_page_url: str
+    addenda_url: str
+    dest_dir: Path
+    addenda_zip_name: str
+    rule_item_key: str
+    addendum_b_title: str
+    register_tags: list[str]
 
 
-def download_zip(dest: Path, *, retries: int = 3) -> bool:
+CONFIGS: dict[int, YearConfig] = {
+    2026: YearConfig(
+        year=2026,
+        detail_page_url=(
+            "https://www.cms.gov/medicare/payment/fee-schedules/physician/"
+            "federal-regulation-notices/cms-1832-p"
+        ),
+        # CY2026 PFS Proposed Rule (CMS-1832-P) Addenda ZIP — resolved from
+        # the CMS-1832-P detail page above (the canonical
+        # ".../pfs-federal-regulation-notices/cms-1832-p" URL 404s; this is
+        # the current path found via the federal-regulation-notices
+        # listing). 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).
+        addenda_url=(
+            "https://www.cms.gov/files/zip/"
+            "cy-2026-pfs-proposed-rule-addenda-updated-07/29/2025.zip"
+        ),
+        dest_dir=ROOT / "data" / "cms" / "pfs_nprm" / "2026",
+        addenda_zip_name="cy2026_pfs_nprm_addenda.zip",
+        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"],
+    ),
+    2027: YearConfig(
+        year=2027,
+        detail_page_url=(
+            "https://www.cms.gov/medicare/payment/fee-schedules/physician/"
+            "federal-regulation-notices/cms-1848-p"
+        ),
+        # CY2027 PFS Proposed Rule (CMS-1848-P) Addenda ZIP — resolved from
+        # the CMS-1848-P detail page above (fallback if that 404s: the
+        # federal-regulation-notices index page). Download link text on
+        # that page: "CY 2027 PFS Proposed Rule Addenda (Updated
+        # 07/21/2026)".
+        addenda_url=(
+            "https://www.cms.gov/files/zip/"
+            "cy-2027-pfs-proposed-rule-addenda-updated-07-21-2026.zip"
+        ),
+        dest_dir=ROOT / "data" / "cms" / "pfs_nprm" / "2027",
+        addenda_zip_name="cy2027_pfs_nprm_addenda.zip",
+        rule_item_key="5ITGVDJV",
+        addendum_b_title="CY2027 PFS NPRM Addendum B (proposed RVUs)",
+        register_tags=["sup:2027_PFS_NPRM", "module:pfs", "file:rvu", "year:2027"],
+    ),
+}
+
+
+def download_zip(url: str, dest: Path, *, retries: int = 3) -> bool:
     """Download the addenda ZIP if it doesn't already exist."""
     import time
 
@@ -67,7 +109,7 @@ def download_zip(dest: Path, *, retries: int = 3) -> bool:
     for attempt in range(1, retries + 1):
         try:
             resp = httpx.get(
-                ADDENDA_URL,
+                url,
                 headers={"User-Agent": USER_AGENT},
                 timeout=120,
                 follow_redirects=True,
@@ -96,11 +138,20 @@ def extract_zip(zip_path: Path, dest_dir: Path) -> list[Path]:
 
 
 def find_addendum_b(files: list[Path]) -> Path | None:
-    """Find the extracted file whose name matches Addendum B."""
+    """Find the extracted file whose name matches Addendum B.
+
+    Excludes macOS ``__MACOSX/._*`` resource-fork junk that CMS's zips
+    sometimes carry alongside the real file — those AppleDouble files
+    match the same name pattern and suffix, so they'd otherwise be
+    eligible candidates too.
+    """
     candidates = [
         f
         for f in files
-        if ADDENDUM_B_RE.search(f.name) and f.suffix.lower() in (".xlsx", ".csv")
+        if ADDENDUM_B_RE.search(f.name)
+        and f.suffix.lower() in (".xlsx", ".csv")
+        and not f.name.startswith("._")
+        and "__MACOSX" not in f.parts
     ]
     if not candidates:
         return None
@@ -134,7 +185,7 @@ def _existing_attachment(store: Any, item_key: str, filename: str) -> str | None
     return row["key"] if row else None
 
 
-def register(addendum_b_path: Path) -> None:
+def register(cfg: YearConfig, addendum_b_path: Path) -> None:
     """Attach Addendum B to the NPRM rule item and tag it.
 
     Idempotent: if an attachment with the Addendum B title already exists
@@ -147,15 +198,15 @@ def register(addendum_b_path: Path) -> None:
 
     store = connect.bib()
 
-    existing_key = _existing_attachment(store, RULE_ITEM_KEY, ADDENDUM_B_TITLE)
+    existing_key = _existing_attachment(store, cfg.rule_item_key, cfg.addendum_b_title)
     if existing_key is not None:
         print(
-            f"SKIP attach — {RULE_ITEM_KEY} already has an attachment titled "
-            f"{ADDENDUM_B_TITLE!r} (key {existing_key})"
+            f"SKIP attach — {cfg.rule_item_key} already has an attachment "
+            f"titled {cfg.addendum_b_title!r} (key {existing_key})"
         )
     else:
         att_key = store.attach_file(
-            RULE_ITEM_KEY, addendum_b_path, title=ADDENDUM_B_TITLE
+            cfg.rule_item_key, addendum_b_path, title=cfg.addendum_b_title
         )
         con = store._con()  # noqa: SLF001 — same pattern as bib/oig.py, bib/iom.py
         row = con.execute(
@@ -164,16 +215,30 @@ def register(addendum_b_path: Path) -> None:
         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}")
+    for tag in cfg.register_tags:
+        store.add_tag(cfg.rule_item_key, tag)
+    print(f"Tagged {cfg.rule_item_key} with: {', '.join(cfg.register_tags)}")
+    item = store.get(cfg.rule_item_key)
+    print(f"Current tags on {cfg.rule_item_key}: {item.tags}")
 
 
 def main() -> None:
+    years = sorted(CONFIGS)
     parser = argparse.ArgumentParser(
-        description="Fetch + register the CY2026 PFS NPRM addenda"
+        description=(
+            "Fetch + register PFS NPRM addenda for a given rule year "
+            f"(available: {', '.join(str(y) for y in years)})."
+        )
+    )
+    parser.add_argument(
+        "--year",
+        type=int,
+        choices=years,
+        default=2026,
+        help=(
+            "Rule year to fetch — one of "
+            f"{', '.join(str(y) for y in years)} (default: 2026)."
+        ),
     )
     parser.add_argument(
         "--dry-run",
@@ -183,34 +248,35 @@ def main() -> None:
     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.",
+        help="After fetching, attach Addendum B to the rule item and apply "
+        "its tags for the selected --year.",
     )
     args = parser.parse_args()
+    cfg = CONFIGS[args.year]
 
-    zip_dest = DEST_DIR / ADDENDA_ZIP_NAME
+    zip_dest = cfg.dest_dir / cfg.addenda_zip_name
 
     if args.dry_run:
-        print(f"Would GET  {ADDENDA_URL}")
+        print(f"Would GET  {cfg.addenda_url}")
         print(f"Would save {zip_dest}")
-        print(f"Would extract into {DEST_DIR}")
+        print(f"Would extract into {cfg.dest_dir}")
         if args.register:
             print(
-                f"Would attach Addendum B to {RULE_ITEM_KEY} "
-                f"and tag with: {', '.join(REGISTER_TAGS)}"
+                f"Would attach Addendum B to {cfg.rule_item_key} "
+                f"and tag with: {', '.join(cfg.register_tags)}"
             )
         return
 
-    DEST_DIR.mkdir(parents=True, exist_ok=True)
+    cfg.dest_dir.mkdir(parents=True, exist_ok=True)
 
-    if not download_zip(zip_dest):
-        raise SystemExit(f"Failed to download {ADDENDA_URL}")
+    if not download_zip(cfg.addenda_url, zip_dest):
+        raise SystemExit(f"Failed to download {cfg.addenda_url}")
 
-    files = extract_zip(zip_dest, DEST_DIR)
-    print(f"\nExtracted {len(files)} files to {DEST_DIR}:")
+    files = extract_zip(zip_dest, cfg.dest_dir)
+    print(f"\nExtracted {len(files)} files to {cfg.dest_dir}:")
     for f in files:
         size_kb = f.stat().st_size / 1024
-        print(f"  {f.relative_to(DEST_DIR)} ({size_kb:.0f} KB)")
+        print(f"  {f.relative_to(cfg.dest_dir)} ({size_kb:.0f} KB)")
 
     addendum_b = find_addendum_b(files)
     if addendum_b is None:
@@ -218,12 +284,12 @@ def main() -> None:
             "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)}")
+    print(f"\nAddendum B: {addendum_b.relative_to(cfg.dest_dir)}")
     preview_addendum_b(addendum_b)
 
     if args.register:
         print()
-        register(addendum_b)
+        register(cfg, addendum_b)
 
 
 if __name__ == "__main__":