feat(pfs): register NPRM Addendum E (proposed GPCIs) in bib (refs #629)

dev/scripts/fetch_nprm_addenda.py

@@ -32,6 +32,7 @@ USER_AGENT = (
 )
 
 ADDENDUM_B_RE = re.compile(r"(?i)addendum[_ ]?b")
+ADDENDUM_E_RE = re.compile(r"(?i)addendum[_ ]?e")
 
 
 @dataclass(frozen=True)
@@ -46,6 +47,7 @@ class YearConfig:
     addenda_zip_name: str
     rule_item_key: str
     addendum_b_title: str
+    addendum_e_title: str
     register_tags: list[str]
 
 
@@ -72,6 +74,7 @@ CONFIGS: dict[int, YearConfig] = {
         addenda_zip_name="cy2026_pfs_nprm_addenda.zip",
         rule_item_key="2KVJ2HKX",
         addendum_b_title="CY2026 PFS NPRM Addendum B (proposed RVUs)",
+        addendum_e_title="CY2026 PFS NPRM Addendum E (proposed GPCIs)",
         register_tags=["sup:2026_PFS_NPRM", "module:pfs", "file:rvu", "year:2026"],
     ),
     2027: YearConfig(
@@ -93,6 +96,7 @@ CONFIGS: dict[int, YearConfig] = {
         addenda_zip_name="cy2027_pfs_nprm_addenda.zip",
         rule_item_key="5ITGVDJV",
         addendum_b_title="CY2027 PFS NPRM Addendum B (proposed RVUs)",
+        addendum_e_title="CY2027 PFS NPRM Addendum E (proposed GPCIs)",
         register_tags=["sup:2027_PFS_NPRM", "module:pfs", "file:rvu", "year:2027"],
     ),
 }
@@ -137,8 +141,8 @@ def extract_zip(zip_path: Path, dest_dir: Path) -> list[Path]:
     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.
+def find_addendum(files: list[Path], pattern: re.Pattern[str]) -> Path | None:
+    """Find the extracted file whose name matches ``pattern``.
 
     Excludes macOS ``__MACOSX/._*`` resource-fork junk that CMS's zips
     sometimes carry alongside the real file — those AppleDouble files
@@ -148,7 +152,7 @@ def find_addendum_b(files: list[Path]) -> Path | None:
     candidates = [
         f
         for f in files
-        if ADDENDUM_B_RE.search(f.name)
+        if pattern.search(f.name)
         and f.suffix.lower() in (".xlsx", ".csv")
         and not f.name.startswith("._")
         and "__MACOSX" not in f.parts
@@ -160,6 +164,11 @@ def find_addendum_b(files: list[Path]) -> Path | None:
     return candidates[0]
 
 
+def find_addendum_b(files: list[Path]) -> Path | None:
+    """Backward-compatible Addendum B finder."""
+    return find_addendum(files, ADDENDUM_B_RE)
+
+
 def preview_addendum_b(path: Path) -> None:
     """Print the header row (and a couple of data rows) for verification."""
     import polars as pl
@@ -185,35 +194,42 @@ def _existing_attachment(store: Any, item_key: str, filename: str) -> str | None
     return row["key"] if row else None
 
 
-def register(cfg: YearConfig, addendum_b_path: Path) -> None:
-    """Attach Addendum B to the NPRM rule item and tag it.
+def _attach_guarded(store: Any, item_key: str, path: Path, title: str) -> None:
+    """Attach ``path`` to ``item_key`` under ``title``, unless an
+    attachment with that title already exists (attach_file has no dedup
+    of its own, so every --register re-run would otherwise duplicate)."""
+    existing_key = _existing_attachment(store, item_key, title)
+    if existing_key is not None:
+        print(
+            f"SKIP attach — {item_key} already has an attachment "
+            f"titled {title!r} (key {existing_key})"
+        )
+        return
+    att_key = store.attach_file(item_key, path, title=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 {path.name} (key {att_key}) -> {storage_path}")
 
-    Idempotent: if an attachment with the Addendum B title already exists
-    on the rule item, skip attach_file (which has no dedup of its own and
-    would otherwise create a duplicate on every --register re-run) and
-    only (re-)apply the tags, which are already idempotent via add_tag's
-    INSERT OR IGNORE.
+
+def register(
+    cfg: YearConfig, addendum_b_path: Path, addendum_e_path: Path | None = None
+) -> None:
+    """Attach Addendum B (and E when present) to the NPRM rule item and
+    tag it.
+
+    Idempotent: attaches are title-guarded (see ``_attach_guarded``) and
+    the tags re-apply harmlessly via add_tag's INSERT OR IGNORE.
     """
     from conf import connect
 
     store = connect.bib()
 
-    existing_key = _existing_attachment(store, cfg.rule_item_key, cfg.addendum_b_title)
-    if existing_key is not None:
-        print(
-            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(
-            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(
-            "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}")
+    _attach_guarded(store, cfg.rule_item_key, addendum_b_path, cfg.addendum_b_title)
+    if addendum_e_path is not None:
+        _attach_guarded(store, cfg.rule_item_key, addendum_e_path, cfg.addendum_e_title)
 
     for tag in cfg.register_tags:
         store.add_tag(cfg.rule_item_key, tag)
@@ -287,9 +303,18 @@ def main() -> None:
     print(f"\nAddendum B: {addendum_b.relative_to(cfg.dest_dir)}")
     preview_addendum_b(addendum_b)
 
+    # Addendum E (proposed GPCIs) is optional-but-expected: every NPRM
+    # addenda ZIP to date has carried one, but its absence shouldn't
+    # block the Addendum B registration this script exists for.
+    addendum_e = find_addendum(files, ADDENDUM_E_RE)
+    if addendum_e is None:
+        print(f"\nNo Addendum E file matched {ADDENDUM_E_RE.pattern!r} — skipping.")
+    else:
+        print(f"\nAddendum E: {addendum_e.relative_to(cfg.dest_dir)}")
+
     if args.register:
         print()
-        register(cfg, addendum_b)
+        register(cfg, addendum_b, addendum_e)
 
 
 if __name__ == "__main__":