homelab/stack
1
0
Fork 0
Code Issues 262 Pull Requests 2 Actions Packages Projects 1 Releases Wiki Activity

fix(pfs): empty CPT parse must not wipe an edition (F4)

Browse Source 
write_cpt_edition is delete-then-insert per edition_year; writing an
empty parse would silently erase whatever real data is already on file
for that year. (parse_epub's own guard — raising when _chapter_groups
finds no chapter content at all — landed in the prior commit, since it
shares the same content_root refactor in pfs/cpt_epub.py.)

write_cpt_edition now refuses (raises ValueError) an edition with zero
codes unless force=True. cpt_load.ingest wraps each edition's parse +
write in try/except ValueError, surfaces the failure as
out[year] = {"error": <message>} and continues with the other
requested editions instead of aborting the whole run.
This commit is contained in:
kert
2026-09-09 21:11:39 -04:00
parent 77fd48eb31
commit a44c9982fc
4 changed files with 154 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
src/pfs/codetables.py
View File

@@ -332,7 +332,9 @@ _CPT_TABLES = (
)
def write_cpt_edition(con: Any, edition: "CptEdition", item_key: str) -> dict[str, int]:
def write_cpt_edition(
con: Any, edition: "CptEdition", item_key: str, *, force: bool = False
) -> 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)
@@ -340,7 +342,21 @@ def write_cpt_edition(con: Any, edition: "CptEdition", item_key: str) -> dict[st
slipped past the parser's own filter (``pfs.cpt_epub``). C9's
guideline-reprint/lower-score losers are resolved by the parser
before this ever runs (``edition.alternates``, written to
``pfs.cpt_code_alt``), so that assertion should now always pass."""
``pfs.cpt_code_alt``), so that assertion should now always pass.
F4: refuses (raises ``ValueError``) an edition with zero codes
unless ``force=True`` — this is delete-then-insert, so writing an
empty parse would silently wipe whatever real edition data is
already on file for that year. ``pfs.cpt_epub.parse_epub`` already
raises before returning an edition this empty in the ordinary case
(no chapter content found at all); this is the second guard, for an
edition that parsed some structure but named not one actual code."""
if not edition.codes and not force:
raise ValueError(
f"pfs.cpt_code: edition_year={edition.year} parsed zero codes — "
"refusing to overwrite any existing edition data; pass force=True "
"to write it anyway"
)
ensure_tables(con)
year = edition.year

49
src/pfs/cpt_load.py
View File

@@ -19,6 +19,7 @@ open the DuckDB single-writer lock for a preview (#508-#514).
from __future__ import annotations
import logging
import re
from pathlib import Path
from typing import Any, Sequence
@@ -26,6 +27,8 @@ from typing import Any, Sequence
from pfs.codetables import write_cpt_edition
from pfs.cpt_epub import parse_epub
log = logging.getLogger(__name__)
# "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
@@ -64,6 +67,12 @@ def find_editions(store: Any) -> list[tuple[int, str, Path]]:
title = row["title"] or ""
m = _TITLE_RE.match(title)
if not m:
log.warning(
"source:ama item %s carries an .epub but its title %r "
"doesn't match a CPT Professional edition — skipping",
item_key,
title,
)
continue
year = int(next(g for g in m.groups() if g))
out[item_key] = (year, item_key, Path(row["storage_path"]))
@@ -76,7 +85,7 @@ def ingest(
*,
years: Sequence[int] | None = None,
dry_run: bool = False,
) -> dict[int, dict[str, int]]:
) -> dict[int, dict[str, Any]]:
"""Parse and (unless ``dry_run``) write every matched CPT edition.
Returns ``{year: {table_short_name: row_count}}`` — parsed counts
@@ -84,6 +93,12 @@ def ingest(
``dry_run``, written counts (``write_cpt_edition``'s return value)
otherwise. Raises ``ValueError`` if ``years`` names a year with no
matching bib item/attachment.
F4: an edition that fails to parse (``parse_epub`` found no chapter
content — a bad/empty EPUB) or fails to write (``write_cpt_edition``
refusing a zero-code edition) is a per-edition failure, not a whole-
run one — its entry is ``{"error": <message>}`` and every other
requested edition is still parsed/written.
"""
editions = find_editions(store)
by_year = {year: (item_key, path) for year, item_key, path in editions}
@@ -99,20 +114,24 @@ def ingest(
f"no CPT EPUB found for edition(s) {missing} (known editions: {known})"
)
out: dict[int, dict[str, int]] = {}
out: dict[int, dict[str, Any]] = {}
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),
"alternates": len(edition.alternates),
}
else:
out[year] = write_cpt_edition(con, edition, item_key)
try:
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),
"alternates": len(edition.alternates),
}
else:
out[year] = write_cpt_edition(con, edition, item_key)
except ValueError as exc:
log.error("CPT edition %s (%s) failed: %s", year, item_key, exc)
out[year] = {"error": str(exc)}
return out

19
tests/pfs/test_codetables.py
View File

@@ -407,6 +407,25 @@ def _tiny_edition(year=2024, codes=None, alternates=()):
)
class TestCptEditionEmptyGuardF4:
def test_zero_codes_refused_without_force(self, con):
edition = _tiny_edition(codes=())
with pytest.raises(ValueError, match="zero codes"):
write_cpt_edition(con, edition, "GQGTPGYV")
# And it must never have wiped whatever was already on file —
# write a real edition first, then the refused zero-code write
# for the same year must leave it untouched.
write_cpt_edition(con, _tiny_edition(), "GQGTPGYV")
with pytest.raises(ValueError, match="zero codes"):
write_cpt_edition(con, edition, "GQGTPGYV")
assert len(read_cpt_codes(con, 2024)) == 2
def test_zero_codes_written_with_force(self, con):
edition = _tiny_edition(codes=())
counts = write_cpt_edition(con, edition, "GQGTPGYV", force=True)
assert counts["codes"] == 0
class TestCptEdition:
def test_write_returns_counts_per_table(self, con):
counts = write_cpt_edition(con, _tiny_edition(), "GQGTPGYV")

83
tests/pfs/test_cpt_load.py
View File

@@ -123,6 +123,27 @@ def store_with_editions(tmp_path):
s.close()
class TestTitleMismatchWarnsF4:
def test_unrecognized_title_logs_a_warning_naming_the_item(self, tmp_path, caplog):
s = _store(tmp_path)
key = s.create(
Item(title="Netter's Atlas for CPT Coding 2015", item_type="book"),
tags=["source:ama"],
)
epub = tmp_path / "Netter.epub"
_build_epub(epub, _CHAPTER_2021)
s.attach_file(key, epub, title="Netter.epub")
try:
with caplog.at_level("WARNING", logger="pfs.cpt_load"):
editions = find_editions(s)
assert editions == []
messages = [r.getMessage() for r in caplog.records]
assert any(key in m for m in messages)
assert any("Netter's Atlas" in m for m in messages)
finally:
s.close()
class TestFindEditions:
def test_finds_only_the_matching_epub_editions(self, store_with_editions):
store, key_2024, key_2021 = store_with_editions
@@ -198,3 +219,65 @@ class TestIngestWrite:
assert years == [(2024,)]
finally:
con.close()
def _build_epub_with_no_chapters(path: Path) -> None:
"""An EPUB with content files but none named ``ChapterNN`` — parses
to zero chapter groups, so ``parse_epub`` raises (F4)."""
with zipfile.ZipFile(path, "w") as zf:
zf.writestr("mimetype", "application/epub+zip")
zf.writestr(
"OPS/Frontmatter.xhtml", "<html><body><p>Not a chapter.</p></body></html>"
)
@pytest.fixture
def store_with_one_good_one_bad_edition(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 = s.create(
Item(title="CPT 2021 Professional Edition", item_type="book"),
tags=["source:ama"],
)
bad_epub = tmp_path / "CPT 2021 Professional Edition.epub"
_build_epub_with_no_chapters(bad_epub)
s.attach_file(key_2021, bad_epub, title="CPT 2021 Professional Edition.epub")
yield s, key_2024, key_2021
s.close()
class TestIngestPerEditionFailureF4:
"""A CPT edition that fails to parse is a per-edition failure, not a
whole-run one — ingest surfaces it and continues with the others."""
def test_write_surfaces_the_bad_edition_and_writes_the_good_one(
self, store_with_one_good_one_bad_edition
):
store, key_2024, key_2021 = store_with_one_good_one_bad_edition
con = duckdb.connect(":memory:")
try:
out = ingest(store, con, dry_run=False)
assert out[2024]["codes"] == 2
assert "error" in out[2021]
assert "no chapter content found" in out[2021]["error"]
years = con.execute(
"SELECT DISTINCT edition_year FROM pfs.cpt_code"
).fetchall()
assert years == [(2024,)]
finally:
con.close()
def test_dry_run_surfaces_the_bad_edition_too(
self, store_with_one_good_one_bad_edition
):
store, key_2024, key_2021 = store_with_one_good_one_bad_edition
out = ingest(store, None, dry_run=True)
assert out[2024]["codes"] == 2
assert "error" in out[2021]