stack/dev/scripts/add_cf_history_to_zotero.py

"""Add the AMA Medicare Physician CF history PDF to Zotero.

The document is the authoritative public source for the historical
record of CMS PFS conversion factors from CY1992 through present,
including the four-CF split (QP APM / non-APM / Anesthesia × 2)
that took effect CY2026.  It explains that the Final Rule's
conversion factor for a given year is derived from the prior year's
CF via update factor × budget-neutrality adjustor × performance
adjustment, and then **baked into** the published value — which is
why ``pfs.rules.RULES[year].conversion_factor`` is already BN-
adjusted and should never be multiplied by a separate BN term at
payment-calculation time.

Source:
    https://www.ama-assn.org/system/files/cf-history.pdf

Tags:
    - module:pfs
    - source:ama
    - file:cf-history
    - year:1992 … year:2026  (range covered)

Usage:
    uv run python dev/scripts/add_cf_history_to_zotero.py [--pdf PATH]

The PDF is downloaded from AMA if no local path is supplied.
"""

from __future__ import annotations

import argparse
import subprocess
import urllib.request
from pathlib import Path

from conf import path as _conf_path
from zot.db import TYPE_MAP, Db, generate_key, now_iso

ZOTERO_DB = str(_conf_path("db.zotero"))
ZOTERO_STORAGE = str(_conf_path("storage.zotero"))

CF_HISTORY_URL = "https://www.ama-assn.org/system/files/cf-history.pdf"
CF_HISTORY_TITLE = "History of Medicare Physician Payment Schedule Conversion Factors"
CF_HISTORY_YEARS = range(1992, 2027)  # 1992 → 2026


def _download(dest: Path) -> None:
    """Fetch the PDF from AMA if missing."""
    if dest.exists() and dest.stat().st_size > 0:
        print(f"Using existing {dest}")
        return
    dest.parent.mkdir(parents=True, exist_ok=True)
    print(f"Downloading {CF_HISTORY_URL} → {dest}")
    urllib.request.urlretrieve(CF_HISTORY_URL, dest)  # noqa: S310
    print(f"  {dest.stat().st_size:,} bytes")


def _add_parent_item(db: Db) -> int:
    """Create the parent report item + tags.  Returns item id."""
    now = now_iso()
    item_id = db.create_item(TYPE_MAP["report"], now=now)
    db.set_fields(
        item_id,
        {
            "title": CF_HISTORY_TITLE,
            "date": "2026-01-01",
            "url": CF_HISTORY_URL,
            "accessDate": now,
            "institution": "American Medical Association",
            "reportType": "Data Provenance",
            "abstractNote": (
                "Authoritative historical record of the Medicare Physician "
                "Payment Schedule conversion factor from CY1992 through the "
                "present.  For CY2026, CMS finalised FOUR conversion factors "
                "(APM / non-APM × standard / anesthesia) under MACRA 2015. "
                "The published CF for each year already reflects the "
                "statutory update factor, budget-neutrality adjustor, and "
                "any performance adjustment — callers should NOT apply BN "
                "as a separate multiplier at payment-calculation time."
            ),
        },
    )
    tags = ["module:pfs", "source:ama", "file:cf-history"] + [
        f"year:{y}" for y in CF_HISTORY_YEARS
    ]
    db.sync_tags(item_id, tags)
    print(f"Created parent item {item_id} with {len(tags)} tags")
    return item_id


def _attach_pdf(db: Db, parent_id: int, pdf_path: Path) -> None:
    """Copy the PDF into Zotero storage and create the attachment."""
    att_key = generate_key()
    storage_dir = Path(ZOTERO_STORAGE) / att_key
    subprocess.run(["sudo", "mkdir", "-p", str(storage_dir)], check=True)
    dest = storage_dir / pdf_path.name
    subprocess.run(["sudo", "cp", str(pdf_path), str(dest)], check=True)
    subprocess.run(
        ["sudo", "chown", "-R", "100999:100999", str(storage_dir)],
        check=True,
    )

    att_id = db.add_attachment(
        parent_id,
        key=att_key,
        content_type="application/pdf",
        path=f"storage:{pdf_path.name}",
    )
    db.set_field(att_id, "title", pdf_path.name)
    print(f"Attached {pdf_path.name} (attachment id {att_id}, key {att_key})")


def main() -> None:
    ap = argparse.ArgumentParser(description=__doc__ or "")
    ap.add_argument(
        "--pdf",
        type=Path,
        default=Path("/tmp/cf_history/cf-history.pdf"),
        help="Local path to cf-history.pdf (downloads from AMA if missing).",
    )
    args = ap.parse_args()

    _download(args.pdf)

    with Db(ZOTERO_DB) as db:
        parent_id = _add_parent_item(db)
        _attach_pdf(db, parent_id, args.pdf)
        parent_key = db.con.execute(
            "SELECT key FROM items WHERE itemID = ?", (parent_id,)
        ).fetchone()[0]
        db.commit()

    print()
    print(f"Done.  Parent item key: {parent_key}")
    print("Cite in docstrings with:")
    print(f"    :pincite:`{parent_key}`")


if __name__ == "__main__":
    main()