feat(pfs): pfs.cpt_* tables + stack pfs cpt-ingest (refs #687)
Six delete-then-insert-per-edition_year tables (schema spec §3): cpt_section/cpt_code/cpt_instruction/cpt_reference/cpt_crosswalk/ cpt_list, keyed by (edition_year, item_key). cpt_section carries a new path_key = " > ".join(path) column (C7) so consumers can group the same heading title's several sec_ids (2019's running-header repeats) together. codetables.write_cpt_edition(con, edition, item_key) asserts one pfs.cpt_code row per (edition_year, code) before inserting and raises with the offending codes otherwise (C8's loader-side backstop, paired with the parser fix). read_cpt_sections/read_cpt_codes/ read_cpt_instructions/cpt_years round out the readers. pfs.cpt_load.find_editions(store) locates source:ama-tagged bib items whose title names a CPT Professional edition year and carry an .epub attachment (2019/2021/2022/2024 on disk today); the 2018 PDF-only edition, the 2023 "CPT Changes" book and Netter's Atlas are excluded by the title/attachment match itself, no special-casing needed. ingest(store, con, years=None, dry_run=False) parses each matched edition and writes it, or — dry_run — only counts, never touching con so a preview can't contend for the DuckDB single-writer lock (#508-#514). `stack pfs cpt-ingest [--edition YEAR]... [--all] [--dry-run]` follows the `elements` command's shape: dry-run opens no connection at all, the write path goes through duckdb_batch + ensure_tables + publish_replica.
This commit is contained in:
kert
@@ -6,6 +6,8 @@
|
||||
uv run stack pfs lineage --code G2058 [--write]
|
||||
uv run stack pfs families [--write]
|
||||
uv run stack pfs review [--code G0556]
|
||||
uv run stack pfs cpt-ingest --edition 2024 [--edition 2022 …] [--dry-run]
|
||||
uv run stack pfs cpt-ingest --all
|
||||
|
||||
Writes go through ``duckdb_batch`` (single-writer rule) and republish
|
||||
the read-only replica so the chat and notebooks see them.
|
||||
@@ -26,6 +28,7 @@ from pfs.codetables import (
|
||||
write_events,
|
||||
write_families,
|
||||
)
|
||||
from pfs.cpt_load import ingest as cpt_ingest
|
||||
from pfs.extract import extract_code
|
||||
from pfs.families import FAMILIES, HAND_FAMILIES, derive_families, refresh_from
|
||||
from pfs.lineage import lineage
|
||||
@@ -278,3 +281,43 @@ def review(code: str = typer.Option("", "--code")) -> None:
|
||||
sql + " ORDER BY code, p_id", params
|
||||
).fetchall():
|
||||
typer.echo(f"{c} {key} ¶{p_id} [{proposed or '?'}] {text[:120]}")
|
||||
|
||||
|
||||
def _print_cpt_counts(counts: dict[int, dict[str, int]]) -> None:
|
||||
for year in sorted(counts):
|
||||
c = counts[year]
|
||||
detail = ", ".join(f"{k}={v}" for k, v in c.items())
|
||||
typer.echo(f"{year}: {detail}")
|
||||
|
||||
|
||||
@app.command("cpt-ingest")
|
||||
def cpt_ingest_cmd(
|
||||
edition: list[int] = typer.Option(
|
||||
[], "--edition", help="CPT edition year to ingest (repeatable)."
|
||||
),
|
||||
all_editions: bool = typer.Option(
|
||||
False, "--all", help="Every CPT EPUB edition found in bib (source:ama)."
|
||||
),
|
||||
dry_run: bool = typer.Option(
|
||||
False, "--dry-run", help="Parse and report counts; write nothing."
|
||||
),
|
||||
) -> None:
|
||||
"""Parse CPT EPUB editions into pfs.cpt_section/cpt_code/cpt_instruction/
|
||||
cpt_reference/cpt_crosswalk/cpt_list (docs/superpowers/specs/
|
||||
2026-09-09-cpt-canonical-schema-design.md §3)."""
|
||||
if not edition and not all_editions:
|
||||
raise typer.BadParameter("pass --edition YEAR (repeatable) or --all")
|
||||
years = None if all_editions else sorted(set(edition))
|
||||
store = _store()
|
||||
if dry_run:
|
||||
# A preview never needs the DuckDB single-writer lock a notebook
|
||||
# may be holding (#508-#514) — parsing alone answers a dry run,
|
||||
# so no connection is opened at all (I4).
|
||||
counts = cpt_ingest(store, None, years=years, dry_run=True)
|
||||
_print_cpt_counts(counts)
|
||||
return
|
||||
with _batch() as con:
|
||||
ensure_tables(con)
|
||||
counts = cpt_ingest(store, con, years=years, dry_run=False)
|
||||
_print_cpt_counts(counts)
|
||||
_publish()
|
||||
|
||||
@@ -5,16 +5,30 @@
|
||||
``pfs.code_event`` dated lineage events (created, replaced, …)
|
||||
``pfs.code_family`` derived family membership with roles
|
||||
|
||||
Writers are delete-then-insert per code (events, elements) or full
|
||||
replace (families) so a re-run is idempotent. Callers own the
|
||||
connection (``conf.connect.duckdb_batch`` for writes) and republish the
|
||||
replica afterwards; nothing here opens a database.
|
||||
``pfs.cpt_*`` (docs/superpowers/specs/2026-09-09-cpt-canonical-schema-design.md
|
||||
§3) hold one CPT codebook edition's own organizing structure —
|
||||
``cpt_section``/``cpt_code``/``cpt_instruction``/``cpt_reference``/
|
||||
``cpt_crosswalk``/``cpt_list`` — written by ``pfs.cpt_load.ingest`` from
|
||||
``pfs.cpt_epub.parse_epub``. ``cpt_section.path_key`` is
|
||||
``" > ".join(path)`` (C7): the same heading title can occur as several
|
||||
``sec_id``s (2019's running-header repeats), so ``cpt_section`` keeps
|
||||
every row and consumers group by ``path_key`` to treat those repeats as
|
||||
one heading.
|
||||
|
||||
Writers are delete-then-insert per code (events, elements) or per
|
||||
``edition_year`` (cpt_*), or full replace (families), so a re-run is
|
||||
idempotent. Callers own the connection (``conf.connect.duckdb_batch``
|
||||
for writes) and republish the replica afterwards; nothing here opens a
|
||||
database.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import astuple, dataclass
|
||||
from typing import Any, Sequence
|
||||
from typing import TYPE_CHECKING, Any, Sequence
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pfs.cpt_model import CptEdition
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
@@ -68,6 +82,81 @@ class FamilyRow:
|
||||
p_id: int
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CptSectionRow:
|
||||
edition_year: int
|
||||
item_key: str
|
||||
sec_id: str
|
||||
level: int
|
||||
title: str
|
||||
path: list[str]
|
||||
path_key: str # " > ".join(path) — see module docstring (C7)
|
||||
code_lo: str
|
||||
code_hi: str
|
||||
guideline: str
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CptCodeRow:
|
||||
edition_year: int
|
||||
item_key: str
|
||||
code: str
|
||||
sec_id: str
|
||||
category: str # "I" | "II" | "III"
|
||||
descriptor: str
|
||||
stem: str
|
||||
elements: list[str]
|
||||
tail: str
|
||||
parent: str
|
||||
addon: bool
|
||||
resequenced: bool
|
||||
new: bool
|
||||
revised: bool
|
||||
telemedicine: bool
|
||||
mod51_exempt: bool
|
||||
audio_only: bool
|
||||
fda_pending: bool
|
||||
pla: bool
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CptInstructionRow:
|
||||
edition_year: int
|
||||
item_key: str
|
||||
code: str
|
||||
kind: str
|
||||
text: str
|
||||
targets: list[str]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CptReferenceRow:
|
||||
edition_year: int
|
||||
item_key: str
|
||||
code: str
|
||||
kind: str # "cpt-changes" | "cpt-assistant"
|
||||
years: list[int]
|
||||
text: str
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CptCrosswalkRow:
|
||||
edition_year: int
|
||||
item_key: str
|
||||
current_code: str
|
||||
former_code: str
|
||||
year_deleted: str
|
||||
citations: str
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CptListRow:
|
||||
edition_year: int
|
||||
item_key: str
|
||||
appendix: str
|
||||
code: str
|
||||
|
||||
|
||||
_DDL = """
|
||||
CREATE SCHEMA IF NOT EXISTS pfs;
|
||||
CREATE TABLE IF NOT EXISTS pfs.code_element (
|
||||
@@ -82,6 +171,25 @@ CREATE TABLE IF NOT EXISTS pfs.code_event (
|
||||
CREATE TABLE IF NOT EXISTS pfs.code_family (
|
||||
key VARCHAR, name VARCHAR, code VARCHAR, role VARCHAR, since INTEGER, until INTEGER,
|
||||
item_key VARCHAR, p_id INTEGER);
|
||||
CREATE TABLE IF NOT EXISTS pfs.cpt_section (
|
||||
edition_year INTEGER, item_key VARCHAR, sec_id VARCHAR, level INTEGER, title VARCHAR,
|
||||
path VARCHAR[], path_key VARCHAR, code_lo VARCHAR, code_hi VARCHAR, guideline VARCHAR);
|
||||
CREATE TABLE IF NOT EXISTS pfs.cpt_code (
|
||||
edition_year INTEGER, item_key VARCHAR, code VARCHAR, sec_id VARCHAR, category VARCHAR,
|
||||
descriptor VARCHAR, stem VARCHAR, elements VARCHAR[], tail VARCHAR, parent VARCHAR,
|
||||
addon BOOLEAN, resequenced BOOLEAN, new BOOLEAN, revised BOOLEAN, telemedicine BOOLEAN,
|
||||
mod51_exempt BOOLEAN, audio_only BOOLEAN, fda_pending BOOLEAN, pla BOOLEAN);
|
||||
CREATE TABLE IF NOT EXISTS pfs.cpt_instruction (
|
||||
edition_year INTEGER, item_key VARCHAR, code VARCHAR, kind VARCHAR, text VARCHAR,
|
||||
targets VARCHAR[]);
|
||||
CREATE TABLE IF NOT EXISTS pfs.cpt_reference (
|
||||
edition_year INTEGER, item_key VARCHAR, code VARCHAR, kind VARCHAR, years INTEGER[],
|
||||
text VARCHAR);
|
||||
CREATE TABLE IF NOT EXISTS pfs.cpt_crosswalk (
|
||||
edition_year INTEGER, item_key VARCHAR, current_code VARCHAR, former_code VARCHAR,
|
||||
year_deleted VARCHAR, citations VARCHAR);
|
||||
CREATE TABLE IF NOT EXISTS pfs.cpt_list (
|
||||
edition_year INTEGER, item_key VARCHAR, appendix VARCHAR, code VARCHAR);
|
||||
"""
|
||||
|
||||
|
||||
@@ -171,3 +279,152 @@ def read_families(con: Any) -> list[FamilyRow]:
|
||||
"SELECT * FROM pfs.code_family ORDER BY key, role, code"
|
||||
).fetchall()
|
||||
return [FamilyRow(*r) for r in rows]
|
||||
|
||||
|
||||
_CPT_TABLES = (
|
||||
"pfs.cpt_section",
|
||||
"pfs.cpt_code",
|
||||
"pfs.cpt_instruction",
|
||||
"pfs.cpt_reference",
|
||||
"pfs.cpt_crosswalk",
|
||||
"pfs.cpt_list",
|
||||
)
|
||||
|
||||
|
||||
def write_cpt_edition(con: Any, edition: "CptEdition", item_key: str) -> dict[str, int]:
|
||||
"""Delete-then-insert one CPT codebook edition's parsed rows into
|
||||
``pfs.cpt_*``, keyed by ``edition.year``. Returns rows written per
|
||||
table. C8: asserts one ``pfs.cpt_code`` row per (edition_year, code)
|
||||
before inserting — a duplicate means a resequenced placeholder row
|
||||
slipped past the parser's own filter (``pfs.cpt_epub``)."""
|
||||
ensure_tables(con)
|
||||
year = edition.year
|
||||
|
||||
seen: dict[str, int] = {}
|
||||
for c in edition.codes:
|
||||
seen[c.code] = seen.get(c.code, 0) + 1
|
||||
dupes = sorted(code for code, n in seen.items() if n > 1)
|
||||
if dupes:
|
||||
raise ValueError(
|
||||
f"pfs.cpt_code: duplicate rows for edition_year={year}, "
|
||||
f"codes: {', '.join(dupes)}"
|
||||
)
|
||||
|
||||
section_rows = [
|
||||
CptSectionRow(
|
||||
edition_year=year,
|
||||
item_key=item_key,
|
||||
sec_id=s.sec_id,
|
||||
level=s.level,
|
||||
title=s.title,
|
||||
path=list(s.path),
|
||||
path_key=" > ".join(s.path),
|
||||
code_lo=s.code_lo,
|
||||
code_hi=s.code_hi,
|
||||
guideline=s.guideline,
|
||||
)
|
||||
for s in edition.sections
|
||||
]
|
||||
code_rows = [
|
||||
CptCodeRow(
|
||||
edition_year=year,
|
||||
item_key=item_key,
|
||||
code=c.code,
|
||||
sec_id=c.sec_id,
|
||||
category=c.category,
|
||||
descriptor=c.descriptor,
|
||||
stem=c.stem,
|
||||
elements=list(c.elements),
|
||||
tail=c.tail,
|
||||
parent=c.parent,
|
||||
addon=c.addon,
|
||||
resequenced=c.resequenced,
|
||||
new=c.new,
|
||||
revised=c.revised,
|
||||
telemedicine=c.telemedicine,
|
||||
mod51_exempt=c.mod51_exempt,
|
||||
audio_only=c.audio_only,
|
||||
fda_pending=c.fda_pending,
|
||||
pla=c.pla,
|
||||
)
|
||||
for c in edition.codes
|
||||
]
|
||||
instruction_rows = [
|
||||
CptInstructionRow(
|
||||
edition_year=year,
|
||||
item_key=item_key,
|
||||
code=i.code,
|
||||
kind=i.kind,
|
||||
text=i.text,
|
||||
targets=list(i.targets),
|
||||
)
|
||||
for i in edition.instructions
|
||||
]
|
||||
reference_rows = [
|
||||
CptReferenceRow(
|
||||
edition_year=year,
|
||||
item_key=item_key,
|
||||
code=r.code,
|
||||
kind=r.kind,
|
||||
years=list(r.years),
|
||||
text=r.text,
|
||||
)
|
||||
for r in edition.references
|
||||
]
|
||||
crosswalk_rows = [
|
||||
CptCrosswalkRow(
|
||||
edition_year=year,
|
||||
item_key=item_key,
|
||||
current_code=x.current_code,
|
||||
former_code=x.former_code,
|
||||
year_deleted=x.year_deleted,
|
||||
citations=x.citations,
|
||||
)
|
||||
for x in edition.crosswalks
|
||||
]
|
||||
list_rows = [
|
||||
CptListRow(
|
||||
edition_year=year, item_key=item_key, appendix=e.appendix, code=e.code
|
||||
)
|
||||
for e in edition.lists
|
||||
]
|
||||
|
||||
for table in _CPT_TABLES:
|
||||
con.execute(f"DELETE FROM {table} WHERE edition_year = ?", [year])
|
||||
|
||||
return {
|
||||
"sections": _insert(con, "pfs.cpt_section", section_rows),
|
||||
"codes": _insert(con, "pfs.cpt_code", code_rows),
|
||||
"instructions": _insert(con, "pfs.cpt_instruction", instruction_rows),
|
||||
"references": _insert(con, "pfs.cpt_reference", reference_rows),
|
||||
"crosswalks": _insert(con, "pfs.cpt_crosswalk", crosswalk_rows),
|
||||
"lists": _insert(con, "pfs.cpt_list", list_rows),
|
||||
}
|
||||
|
||||
|
||||
def read_cpt_sections(con: Any, year: int) -> list[CptSectionRow]:
|
||||
rows = con.execute(
|
||||
"SELECT * FROM pfs.cpt_section WHERE edition_year = ? ORDER BY sec_id", [year]
|
||||
).fetchall()
|
||||
return [CptSectionRow(*r) for r in rows]
|
||||
|
||||
|
||||
def read_cpt_codes(con: Any, year: int) -> list[CptCodeRow]:
|
||||
rows = con.execute(
|
||||
"SELECT * FROM pfs.cpt_code WHERE edition_year = ? ORDER BY code", [year]
|
||||
).fetchall()
|
||||
return [CptCodeRow(*r) for r in rows]
|
||||
|
||||
|
||||
def read_cpt_instructions(con: Any, year: int) -> list[CptInstructionRow]:
|
||||
rows = con.execute(
|
||||
"SELECT * FROM pfs.cpt_instruction WHERE edition_year = ? ORDER BY code", [year]
|
||||
).fetchall()
|
||||
return [CptInstructionRow(*r) for r in rows]
|
||||
|
||||
|
||||
def cpt_years(con: Any) -> list[int]:
|
||||
rows = con.execute(
|
||||
"SELECT DISTINCT edition_year FROM pfs.cpt_section ORDER BY edition_year"
|
||||
).fetchall()
|
||||
return [r[0] for r in rows]
|
||||
|
||||
117
src/pfs/cpt_load.py
Normal file
117
src/pfs/cpt_load.py
Normal file
@@ -0,0 +1,117 @@
|
||||
"""CPT EPUB editions -> ``pfs.cpt_*`` (task 3, refs #687).
|
||||
|
||||
``find_editions`` locates the bib items tagged ``source:ama`` whose
|
||||
title names a CPT Professional edition year and that carry an EPUB
|
||||
attachment (docs/superpowers/specs/2026-09-09-cpt-canonical-schema-design.md
|
||||
§5, C2): 2019, 2021, 2022 and 2024 are on disk today; 2018 is PDF-only,
|
||||
the "CPT Changes" book and Netter's Atlas are different books entirely
|
||||
— none of the three match the title pattern or (for 2018) carry an
|
||||
EPUB, so they're skipped without a special case. The edition year comes
|
||||
from the item's own title, never the bib ``year:`` tag (that's Zotero's
|
||||
publication date, which runs a year behind the edition).
|
||||
|
||||
``ingest`` parses each matched edition (``pfs.cpt_epub.parse_epub``) and
|
||||
writes it (``pfs.codetables.write_cpt_edition``), delete-then-insert per
|
||||
``edition_year`` so a re-run is idempotent. ``dry_run=True`` only
|
||||
parses and counts — ``con`` is never touched, so the CLI never has to
|
||||
open the DuckDB single-writer lock for a preview (#508-#514).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import Any, Sequence
|
||||
|
||||
from pfs.codetables import write_cpt_edition
|
||||
from pfs.cpt_epub import parse_epub
|
||||
|
||||
# "CPT Professional 2024", "CPT 2021 Professional Edition", "CPT 2019".
|
||||
# Deliberately does *not* match "CPT Changes 2023" (no "Professional"/
|
||||
# bare-year token follows "CPT " there) or "Netter's Atlas for CPT
|
||||
# Coding 2015" (doesn't start with "CPT").
|
||||
_TITLE_RE = re.compile(
|
||||
r"^CPT\s+(?:Professional\s+(\d{4})|(\d{4})\s+Professional(?:\s+Edition)?\b|(\d{4})\b)"
|
||||
)
|
||||
|
||||
|
||||
def find_editions(store: Any) -> list[tuple[int, str, Path]]:
|
||||
"""(year, item_key, epub path) for every ``source:ama``-tagged bib
|
||||
item whose title names a CPT Professional edition and carries an
|
||||
``.epub`` attachment, sorted by year. One entry per item (the first
|
||||
matching ``.epub`` attachment wins if an item somehow has more than
|
||||
one)."""
|
||||
con = store._con()
|
||||
rows = con.execute(
|
||||
"SELECT i.key AS item_key, i.title AS title, a.filename AS filename, "
|
||||
" a.storage_path AS storage_path "
|
||||
"FROM items i "
|
||||
"JOIN item_tags it ON it.item_id = i.id "
|
||||
"JOIN tags t ON t.id = it.tag_id "
|
||||
"JOIN attachments a ON a.item_id = i.id "
|
||||
"WHERE t.name = 'source:ama' "
|
||||
"ORDER BY i.key, a.id"
|
||||
).fetchall()
|
||||
|
||||
out: dict[str, tuple[int, str, Path]] = {}
|
||||
for row in rows:
|
||||
item_key = row["item_key"]
|
||||
if item_key in out:
|
||||
continue
|
||||
filename = row["filename"] or ""
|
||||
if not filename.lower().endswith(".epub"):
|
||||
continue
|
||||
title = row["title"] or ""
|
||||
m = _TITLE_RE.match(title)
|
||||
if not m:
|
||||
continue
|
||||
year = int(next(g for g in m.groups() if g))
|
||||
out[item_key] = (year, item_key, Path(row["storage_path"]))
|
||||
return sorted(out.values())
|
||||
|
||||
|
||||
def ingest(
|
||||
store: Any,
|
||||
con: Any,
|
||||
*,
|
||||
years: Sequence[int] | None = None,
|
||||
dry_run: bool = False,
|
||||
) -> dict[int, dict[str, int]]:
|
||||
"""Parse and (unless ``dry_run``) write every matched CPT edition.
|
||||
|
||||
Returns ``{year: {table_short_name: row_count}}`` — parsed counts
|
||||
(sections/codes/instructions/references/crosswalks/lists) when
|
||||
``dry_run``, written counts (``write_cpt_edition``'s return value)
|
||||
otherwise. Raises ``ValueError`` if ``years`` names a year with no
|
||||
matching bib item/attachment.
|
||||
"""
|
||||
editions = find_editions(store)
|
||||
by_year = {year: (item_key, path) for year, item_key, path in editions}
|
||||
|
||||
if years is None:
|
||||
wanted = sorted(by_year)
|
||||
else:
|
||||
wanted = sorted(set(years))
|
||||
missing = [y for y in wanted if y not in by_year]
|
||||
if missing:
|
||||
known = ", ".join(str(y) for y in sorted(by_year)) or "none"
|
||||
raise ValueError(
|
||||
f"no CPT EPUB found for edition(s) {missing} (known editions: {known})"
|
||||
)
|
||||
|
||||
out: dict[int, dict[str, int]] = {}
|
||||
for year in wanted:
|
||||
item_key, path = by_year[year]
|
||||
edition = parse_epub(path, year=year)
|
||||
if dry_run:
|
||||
out[year] = {
|
||||
"sections": len(edition.sections),
|
||||
"codes": len(edition.codes),
|
||||
"instructions": len(edition.instructions),
|
||||
"references": len(edition.references),
|
||||
"crosswalks": len(edition.crosswalks),
|
||||
"lists": len(edition.lists),
|
||||
}
|
||||
else:
|
||||
out[year] = write_cpt_edition(con, edition, item_key)
|
||||
return out
|
||||
@@ -332,6 +332,70 @@ class TestFamilies:
|
||||
assert key in res.output
|
||||
|
||||
|
||||
class TestCptIngest:
|
||||
def test_dry_run_never_opens_batch(self, con, monkeypatch):
|
||||
seen = {}
|
||||
|
||||
def fake_ingest(store, con_, *, years=None, dry_run=False):
|
||||
seen["store"] = store
|
||||
seen["con"] = con_
|
||||
seen["years"] = years
|
||||
seen["dry_run"] = dry_run
|
||||
return {2024: {"sections": 3, "codes": 5}}
|
||||
|
||||
monkeypatch.setattr(pfs_cli, "cpt_ingest", fake_ingest)
|
||||
|
||||
def fail_batch():
|
||||
raise AssertionError("--dry-run must not open a RW duckdb_batch connection")
|
||||
|
||||
monkeypatch.setattr(pfs_cli, "_batch", fail_batch)
|
||||
|
||||
res = runner.invoke(
|
||||
app, ["pfs", "cpt-ingest", "--edition", "2024", "--dry-run"]
|
||||
)
|
||||
assert res.exit_code == 0, res.output
|
||||
assert seen["dry_run"] is True
|
||||
assert seen["con"] is None
|
||||
assert seen["years"] == [2024]
|
||||
assert "2024: sections=3, codes=5" in res.output
|
||||
assert con.published == []
|
||||
|
||||
def test_all_flag_passes_years_none(self, con, monkeypatch):
|
||||
seen = {}
|
||||
monkeypatch.setattr(
|
||||
pfs_cli,
|
||||
"cpt_ingest",
|
||||
lambda store, con_, *, years=None, dry_run=False: (
|
||||
seen.update(years=years) or {}
|
||||
),
|
||||
)
|
||||
res = runner.invoke(app, ["pfs", "cpt-ingest", "--all", "--dry-run"])
|
||||
assert res.exit_code == 0, res.output
|
||||
assert seen["years"] is None
|
||||
|
||||
def test_requires_edition_or_all(self, con):
|
||||
res = runner.invoke(app, ["pfs", "cpt-ingest"])
|
||||
assert res.exit_code != 0
|
||||
|
||||
def test_write_path_opens_batch_ensures_tables_and_publishes(
|
||||
self, con, monkeypatch
|
||||
):
|
||||
seen = {}
|
||||
|
||||
def fake_ingest(store, con_, *, years=None, dry_run=False):
|
||||
seen["con"] = con_
|
||||
seen["dry_run"] = dry_run
|
||||
return {2024: {"sections": 1, "codes": 2}}
|
||||
|
||||
monkeypatch.setattr(pfs_cli, "cpt_ingest", fake_ingest)
|
||||
res = runner.invoke(app, ["pfs", "cpt-ingest", "--edition", "2024"])
|
||||
assert res.exit_code == 0, res.output
|
||||
assert seen["dry_run"] is False
|
||||
assert seen["con"] is not None
|
||||
assert "2024: sections=1, codes=2" in res.output
|
||||
assert con.published == [True]
|
||||
|
||||
|
||||
class TestReview:
|
||||
def test_lists_queue(self, con):
|
||||
from pfs.codetables import ReviewRow
|
||||
|
||||
@@ -10,16 +10,30 @@ from pfs.codetables import (
|
||||
EventRow,
|
||||
FamilyRow,
|
||||
ReviewRow,
|
||||
cpt_years,
|
||||
ensure_tables,
|
||||
read_all_elements,
|
||||
read_all_events,
|
||||
read_cpt_codes,
|
||||
read_cpt_instructions,
|
||||
read_cpt_sections,
|
||||
read_elements,
|
||||
read_events,
|
||||
read_families,
|
||||
write_cpt_edition,
|
||||
write_elements,
|
||||
write_events,
|
||||
write_families,
|
||||
)
|
||||
from pfs.cpt_model import (
|
||||
CptCode,
|
||||
CptCrosswalk,
|
||||
CptEdition,
|
||||
CptInstruction,
|
||||
CptListEntry,
|
||||
CptReference,
|
||||
CptSection,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
@@ -50,6 +64,12 @@ class TestDDL:
|
||||
"code_element_review",
|
||||
"code_event",
|
||||
"code_family",
|
||||
"cpt_section",
|
||||
"cpt_code",
|
||||
"cpt_instruction",
|
||||
"cpt_reference",
|
||||
"cpt_crosswalk",
|
||||
"cpt_list",
|
||||
} <= names
|
||||
|
||||
|
||||
@@ -179,3 +199,197 @@ class TestFamilies:
|
||||
)
|
||||
fams = read_families(con)
|
||||
assert len(fams) == 1 and fams[0].code == "99439" and fams[0].role == "add-on"
|
||||
|
||||
|
||||
def _tiny_edition(year=2024, codes=None):
|
||||
sections = (
|
||||
CptSection(
|
||||
sec_id="sec_1",
|
||||
level=2,
|
||||
title="Chronic Care Management Services",
|
||||
path=(
|
||||
"Evaluation and Management",
|
||||
"Care Management Services",
|
||||
"Chronic Care Management Services",
|
||||
),
|
||||
code_lo="99490",
|
||||
code_hi="99491",
|
||||
guideline="Sample guideline text.",
|
||||
),
|
||||
)
|
||||
if codes is None:
|
||||
codes = (
|
||||
CptCode(
|
||||
code="99490",
|
||||
sec_id="sec_1",
|
||||
descriptor="Chronic care management services first 20 minutes.",
|
||||
stem="Chronic care management services",
|
||||
elements=("first element;",),
|
||||
tail="first 20 minutes.",
|
||||
addon=False,
|
||||
resequenced=False,
|
||||
new=False,
|
||||
revised=False,
|
||||
telemedicine=False,
|
||||
parent="",
|
||||
mod51_exempt=False,
|
||||
audio_only=False,
|
||||
fda_pending=False,
|
||||
pla=False,
|
||||
category="I",
|
||||
),
|
||||
CptCode(
|
||||
code="99439",
|
||||
sec_id="sec_1",
|
||||
descriptor="each additional 20 minutes.",
|
||||
stem="Chronic care management services",
|
||||
elements=("first element;",),
|
||||
tail="each additional 20 minutes.",
|
||||
addon=True,
|
||||
resequenced=False,
|
||||
new=False,
|
||||
revised=False,
|
||||
telemedicine=False,
|
||||
parent="99490",
|
||||
mod51_exempt=False,
|
||||
audio_only=False,
|
||||
fda_pending=False,
|
||||
pla=False,
|
||||
category="I",
|
||||
),
|
||||
)
|
||||
instructions = (
|
||||
CptInstruction(
|
||||
code="99439",
|
||||
kind="use-with",
|
||||
text="(Use 99439 in conjunction with 99490)",
|
||||
targets=("99490",),
|
||||
),
|
||||
)
|
||||
references = (
|
||||
CptReference(
|
||||
code="99490",
|
||||
kind="cpt-changes",
|
||||
years=(2015, 2022),
|
||||
text="CPT Changes: An Insider's View 2015, 2022",
|
||||
),
|
||||
)
|
||||
crosswalks = (
|
||||
CptCrosswalk(
|
||||
current_code="99490",
|
||||
former_code="99091T",
|
||||
year_deleted="2022",
|
||||
citations="CPT Changes 2022",
|
||||
),
|
||||
)
|
||||
lists = (CptListEntry(appendix="D", code="99439"),)
|
||||
return CptEdition(
|
||||
year=year,
|
||||
sections=sections,
|
||||
codes=codes,
|
||||
instructions=instructions,
|
||||
references=references,
|
||||
crosswalks=crosswalks,
|
||||
lists=lists,
|
||||
)
|
||||
|
||||
|
||||
class TestCptEdition:
|
||||
def test_write_returns_counts_per_table(self, con):
|
||||
counts = write_cpt_edition(con, _tiny_edition(), "GQGTPGYV")
|
||||
assert counts == {
|
||||
"sections": 1,
|
||||
"codes": 2,
|
||||
"instructions": 1,
|
||||
"references": 1,
|
||||
"crosswalks": 1,
|
||||
"lists": 1,
|
||||
}
|
||||
|
||||
def test_round_trip_reads(self, con):
|
||||
write_cpt_edition(con, _tiny_edition(), "GQGTPGYV")
|
||||
|
||||
secs = read_cpt_sections(con, 2024)
|
||||
assert len(secs) == 1
|
||||
assert secs[0].item_key == "GQGTPGYV"
|
||||
assert secs[0].path == [
|
||||
"Evaluation and Management",
|
||||
"Care Management Services",
|
||||
"Chronic Care Management Services",
|
||||
]
|
||||
assert (
|
||||
secs[0].path_key
|
||||
== "Evaluation and Management > Care Management Services > Chronic Care Management Services"
|
||||
)
|
||||
|
||||
codes = read_cpt_codes(con, 2024)
|
||||
assert {c.code for c in codes} == {"99490", "99439"}
|
||||
addon = next(c for c in codes if c.code == "99439")
|
||||
assert addon.addon is True
|
||||
assert addon.parent == "99490"
|
||||
assert addon.elements == ["first element;"]
|
||||
|
||||
instr = read_cpt_instructions(con, 2024)
|
||||
assert len(instr) == 1
|
||||
assert instr[0].targets == ["99490"]
|
||||
|
||||
assert cpt_years(con) == [2024]
|
||||
|
||||
def test_delete_then_insert_per_edition_year(self, con):
|
||||
write_cpt_edition(con, _tiny_edition(year=2021), "AJM4KF4G")
|
||||
write_cpt_edition(con, _tiny_edition(year=2024), "GQGTPGYV")
|
||||
assert cpt_years(con) == [2021, 2024]
|
||||
|
||||
write_cpt_edition(
|
||||
con, _tiny_edition(year=2024), "GQGTPGYV"
|
||||
) # idempotent re-run
|
||||
assert cpt_years(con) == [2021, 2024]
|
||||
assert len(read_cpt_codes(con, 2021)) == 2 # untouched by the 2024 rewrite
|
||||
assert len(read_cpt_codes(con, 2024)) == 2
|
||||
|
||||
def test_duplicate_code_raises_and_writes_nothing(self, con):
|
||||
# C8: the loader's own last-ditch guard against a resequenced
|
||||
# placeholder row slipping past the parser's filter.
|
||||
dup_codes = (
|
||||
CptCode(
|
||||
code="99490",
|
||||
sec_id="sec_1",
|
||||
descriptor="a",
|
||||
stem="a",
|
||||
elements=(),
|
||||
tail="",
|
||||
addon=False,
|
||||
resequenced=False,
|
||||
new=False,
|
||||
revised=False,
|
||||
telemedicine=False,
|
||||
parent="",
|
||||
mod51_exempt=False,
|
||||
audio_only=False,
|
||||
fda_pending=False,
|
||||
pla=False,
|
||||
category="I",
|
||||
),
|
||||
CptCode(
|
||||
code="99490",
|
||||
sec_id="sec_1",
|
||||
descriptor="b",
|
||||
stem="b",
|
||||
elements=(),
|
||||
tail="",
|
||||
addon=False,
|
||||
resequenced=False,
|
||||
new=False,
|
||||
revised=False,
|
||||
telemedicine=False,
|
||||
parent="",
|
||||
mod51_exempt=False,
|
||||
audio_only=False,
|
||||
fda_pending=False,
|
||||
pla=False,
|
||||
category="I",
|
||||
),
|
||||
)
|
||||
with pytest.raises(ValueError, match="99490"):
|
||||
write_cpt_edition(con, _tiny_edition(codes=dup_codes), "GQGTPGYV")
|
||||
assert read_cpt_codes(con, 2024) == []
|
||||
|
||||
199
tests/pfs/test_cpt_load.py
Normal file
199
tests/pfs/test_cpt_load.py
Normal file
@@ -0,0 +1,199 @@
|
||||
"""Tests for pfs.cpt_load — CPT EPUB editions -> pfs.cpt_* (task 3, #687)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
|
||||
import duckdb
|
||||
import pytest
|
||||
|
||||
from bib.item import Item
|
||||
from bib.store import Store
|
||||
from pfs.cpt_load import find_editions, ingest
|
||||
|
||||
_CHAPTER_2024 = """
|
||||
<div class="h1" id="sec_1">Care Management Services</div>
|
||||
<div class="h2" id="sec_2">Chronic Care Management Services</div>
|
||||
<div class="noindent">Sample guideline text.</div>
|
||||
<table class="table1"><tbody>
|
||||
<tr>
|
||||
<td class="td-w1" id="code_99490"><div class="table-para"><b>99490</b></div></td>
|
||||
<td class="td"><div class="table-para1">Chronic care management services with the following required elements:</div>
|
||||
<div class="table-slist">first element;</div>
|
||||
<div class="table-para2">first 20 minutes of clinical staff time.</div></td>
|
||||
</tr>
|
||||
<tr>
|
||||
<td class="td-w1" id="code_99439"><div class="table-para"><span class="ama-en">✚</span> <b>99439</b></div></td>
|
||||
<td class="td"><div class="table-para1-sub">each additional 20 minutes of clinical staff time (List separately in addition to code for primary procedure)</div>
|
||||
<div class="table-para2">(Use 99439 in conjunction with 99490)</div></td>
|
||||
</tr>
|
||||
</tbody></table>
|
||||
"""
|
||||
|
||||
_CHAPTER_2021 = """
|
||||
<div class="h1" id="sec_1">Care Management Services</div>
|
||||
<div class="h2" id="sec_2">Chronic Care Management Services</div>
|
||||
<div class="noindent">Sample guideline text.</div>
|
||||
<table class="table1"><tbody>
|
||||
<tr>
|
||||
<td class="td-w1" id="code_99490"><div class="table-para"><b>99490</b></div></td>
|
||||
<td class="td"><div class="table-para1">Chronic care management services with the following required elements:</div>
|
||||
<div class="table-slist">first element;</div>
|
||||
<div class="table-para2">first 20 minutes of clinical staff time.</div></td>
|
||||
</tr>
|
||||
</tbody></table>
|
||||
"""
|
||||
|
||||
|
||||
def _build_epub(path: Path, chapter_body: str) -> None:
|
||||
with zipfile.ZipFile(path, "w") as zf:
|
||||
zf.writestr("mimetype", "application/epub+zip")
|
||||
zf.writestr("OPS/Chapter01.xhtml", f"<html><body>{chapter_body}</body></html>")
|
||||
|
||||
|
||||
def _store(tmp_path) -> Store:
|
||||
return Store(tmp_path / "bib.sqlite3", storage_dir=tmp_path / "storage")
|
||||
|
||||
|
||||
def _add_edition(
|
||||
store: Store, tmp_path: Path, *, title: str, filename: str, chapter_body: str
|
||||
) -> str:
|
||||
item_key = store.create(Item(title=title, item_type="book"), tags=["source:ama"])
|
||||
src = tmp_path / filename
|
||||
_build_epub(src, chapter_body)
|
||||
store.attach_file(item_key, src, title=filename)
|
||||
return item_key
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def store_with_editions(tmp_path):
|
||||
s = _store(tmp_path)
|
||||
key_2024 = _add_edition(
|
||||
s,
|
||||
tmp_path,
|
||||
title="CPT Professional 2024 - American Medical Association",
|
||||
filename="CPT Professional 2024 - American Medical Association.epub",
|
||||
chapter_body=_CHAPTER_2024,
|
||||
)
|
||||
key_2021 = _add_edition(
|
||||
s,
|
||||
tmp_path,
|
||||
title="CPT 2021 Professional Edition",
|
||||
filename="CPT 2021 Professional Edition.epub",
|
||||
chapter_body=_CHAPTER_2021,
|
||||
)
|
||||
# A 2024 PDF sibling attachment — never matched (no .epub suffix).
|
||||
pdf = tmp_path / "CPT Professional 2024.pdf"
|
||||
pdf.write_bytes(b"%PDF-1.4 not a real pdf")
|
||||
s.attach_file(key_2024, pdf, title="CPT Professional 2024.pdf")
|
||||
# The 2023 "CPT Changes" book — tagged source:ama, has an .epub, but
|
||||
# its title never matches the edition pattern (C2: skip).
|
||||
key_changes = s.create(
|
||||
Item(title="CPT Changes 2023", item_type="book"), tags=["source:ama"]
|
||||
)
|
||||
changes_epub = tmp_path / "CPT Changes 2023.epub"
|
||||
_build_epub(changes_epub, _CHAPTER_2021)
|
||||
s.attach_file(key_changes, changes_epub, title="CPT Changes 2023.epub")
|
||||
# Netter's Atlas — tagged source:ama, has an .epub, title doesn't
|
||||
# start with "CPT" at all.
|
||||
key_netter = s.create(
|
||||
Item(title="Netter's Atlas for CPT Coding 2015", item_type="book"),
|
||||
tags=["source:ama"],
|
||||
)
|
||||
netter_epub = tmp_path / "Netter.epub"
|
||||
_build_epub(netter_epub, _CHAPTER_2021)
|
||||
s.attach_file(key_netter, netter_epub, title="Netter.epub")
|
||||
# 2018 PDF-only edition — tagged source:ama, title matches, but no
|
||||
# .epub attachment at all (C2: skip).
|
||||
key_2018 = s.create(Item(title="CPT 2018", item_type="book"), tags=["source:ama"])
|
||||
pdf_2018 = tmp_path / "CPT 2018.pdf"
|
||||
pdf_2018.write_bytes(b"%PDF-1.4 not a real pdf")
|
||||
s.attach_file(key_2018, pdf_2018, title="CPT 2018.pdf")
|
||||
# An item tagged something else entirely, with an .epub and a
|
||||
# matching-looking title — must never surface (wrong tag).
|
||||
key_other = s.create(
|
||||
Item(title="CPT Professional 2099", item_type="book"), tags=["module:coding"]
|
||||
)
|
||||
other_epub = tmp_path / "CPT Professional 2099.epub"
|
||||
_build_epub(other_epub, _CHAPTER_2021)
|
||||
s.attach_file(key_other, other_epub, title="CPT Professional 2099.epub")
|
||||
|
||||
yield s, key_2024, key_2021
|
||||
s.close()
|
||||
|
||||
|
||||
class TestFindEditions:
|
||||
def test_finds_only_the_matching_epub_editions(self, store_with_editions):
|
||||
store, key_2024, key_2021 = store_with_editions
|
||||
editions = find_editions(store)
|
||||
assert [(year, key) for year, key, _ in editions] == [
|
||||
(2021, key_2021),
|
||||
(2024, key_2024),
|
||||
]
|
||||
|
||||
def test_paths_point_at_the_epub_not_the_pdf(self, store_with_editions):
|
||||
store, key_2024, _ = store_with_editions
|
||||
editions = find_editions(store)
|
||||
path = next(p for y, k, p in editions if k == key_2024)
|
||||
assert path.suffix == ".epub"
|
||||
|
||||
|
||||
class TestIngestDryRun:
|
||||
def test_dry_run_never_touches_con(self, store_with_editions):
|
||||
store, key_2024, key_2021 = store_with_editions
|
||||
out = ingest(store, None, dry_run=True)
|
||||
assert set(out) == {2021, 2024}
|
||||
assert out[2024] == {
|
||||
"sections": 2,
|
||||
"codes": 2,
|
||||
"instructions": 1,
|
||||
"references": 0,
|
||||
"crosswalks": 0,
|
||||
"lists": 0,
|
||||
}
|
||||
assert out[2021]["codes"] == 1
|
||||
|
||||
def test_dry_run_can_be_scoped_to_one_year(self, store_with_editions):
|
||||
store, _, _ = store_with_editions
|
||||
out = ingest(store, None, years=[2024], dry_run=True)
|
||||
assert set(out) == {2024}
|
||||
|
||||
def test_unknown_year_raises(self, store_with_editions):
|
||||
store, _, _ = store_with_editions
|
||||
with pytest.raises(ValueError, match="2099"):
|
||||
ingest(store, None, years=[2099], dry_run=True)
|
||||
|
||||
|
||||
class TestIngestWrite:
|
||||
def test_writes_rows_per_year(self, store_with_editions):
|
||||
store, key_2024, key_2021 = store_with_editions
|
||||
con = duckdb.connect(":memory:")
|
||||
try:
|
||||
out = ingest(store, con, dry_run=False)
|
||||
assert out[2024]["codes"] == 2
|
||||
assert out[2021]["codes"] == 1
|
||||
|
||||
rows = con.execute(
|
||||
"SELECT code, item_key FROM pfs.cpt_code WHERE edition_year = 2024 ORDER BY code"
|
||||
).fetchall()
|
||||
assert rows == [("99439", key_2024), ("99490", key_2024)]
|
||||
|
||||
rows_2021 = con.execute(
|
||||
"SELECT code, item_key FROM pfs.cpt_code WHERE edition_year = 2021"
|
||||
).fetchall()
|
||||
assert rows_2021 == [("99490", key_2021)]
|
||||
finally:
|
||||
con.close()
|
||||
|
||||
def test_scoped_to_one_year_leaves_others_absent(self, store_with_editions):
|
||||
store, key_2024, _ = store_with_editions
|
||||
con = duckdb.connect(":memory:")
|
||||
try:
|
||||
ingest(store, con, years=[2024], dry_run=False)
|
||||
years = con.execute(
|
||||
"SELECT DISTINCT edition_year FROM pfs.cpt_code"
|
||||
).fetchall()
|
||||
assert years == [(2024,)]
|
||||
finally:
|
||||
con.close()
|
||||
Reference in New Issue
Block a user