feat(pfs): fetch + register CY2026 NPRM addenda (refs #590)
This commit is contained in:
kert
198
dev/scripts/fetch_nprm_addenda.py
Normal file
198
dev/scripts/fetch_nprm_addenda.py
Normal file
@@ -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()
|
||||
Reference in New Issue
Block a user