fix(pfs): guard fetch_nprm_addenda --register against duplicate attach (refs #590)

dev/scripts/fetch_nprm_addenda.py

@@ -28,6 +28,7 @@ import argparse
 import re
 import zipfile
 from pathlib import Path
+from typing import Any
 
 import httpx
 
@@ -40,7 +41,8 @@ USER_AGENT = (
 )
 
 # CY2026 PFS Proposed Rule (CMS-1832-P) Addenda ZIP — resolved from the
-# CMS-1832-P detail page (see module docstring for the page URL).
+# 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"
@@ -120,18 +122,48 @@ def preview_addendum_b(path: Path) -> None:
         print(df.head(5))
 
 
+def _existing_attachment(store: Any, item_key: str, filename: str) -> str | None:
+    """Return the attachment key on item_key whose filename matches, if any."""
+    con = store._con()  # noqa: SLF001 — same pattern as bib/oig.py, bib/iom.py
+    row = con.execute(
+        """SELECT a.key FROM attachments a
+           JOIN items i ON a.item_id = i.id
+           WHERE i.key = ? AND a.filename = ?""",
+        (item_key, filename),
+    ).fetchone()
+    return row["key"] if row else None
+
+
 def register(addendum_b_path: Path) -> None:
-    """Attach Addendum B to the NPRM rule item and tag it."""
+    """Attach Addendum B to the NPRM rule item and tag it.
+
+    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.
+    """
     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}")
+
+    existing_key = _existing_attachment(store, RULE_ITEM_KEY, 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})"
+        )
+    else:
+        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)}")