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

feat(pfs): CPT EPUB parser — sections, code entries with symbols and elements, parenthetical instructions, CPT Changes references (refs #687)

Browse Source 
Claude-Session: https://claude.ai/code/session_01Aum3pEMAM3yQVdFSdVe6Gc
This commit is contained in:
kert
2026-09-09 17:18:46 -04:00
parent 1a40861d8e
commit dd2c909af6
4 changed files with 1070 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

634
src/pfs/cpt_epub.py Normal file
View File

@@ -0,0 +1,634 @@
"""CPT EPUB parser (pure). Reads the AMA CPT Professional codebook's own
EPUB markup — chapter headings, code-entry tables, parenthetical
instructions, CPT Changes/Assistant citations, and the authoritative
appendix lists — into the dataclasses in ``pfs.cpt_model``. Stdlib only
(``zipfile``, ``html.parser``, ``html``, ``re``); no DuckDB, no bib, no
network. See docs/superpowers/specs/2026-09-09-cpt-canonical-schema-design.md
§1/§3 for what each structure means.
The 2021/2022/2024 Professional editions share one publisher template,
but the exact CSS class names drift release to release (2024 uses
``table-slist``/``table-para2`` for elements/tail; 2022 uses
``table-square`` for elements and reuses ``table-para1`` for both the
tail *and* the parentheticals that follow it; 2019 has no ``id`` on the
code cell at all, keying instead on the ``<b>NNNNN</b>`` inside
``div.table-para``, and its content spans several ``<tr>`` sharing one
``rowspan`` cell rather than one ``<tr>`` per code). Rather than chase
each edition's class names, the grouping pass below is *positional*:
every code entry starts at its ``div.table-para`` glyph cell (the one
class name that stayed stable across all four editions checked) and
runs until the next one, classifying the blocks in between by what kind
of thing they are (an element bullet, a reference line, or prose) and
by their position (first prose block = stem, later ones = tail/
instructions) rather than by a fixed class string. This also makes the
``<tr>``/``rowspan`` structure irrelevant — the flat token stream never
looks at row boundaries at all, only at document order.
"""
from __future__ import annotations
import html
import re
import zipfile
from dataclasses import replace
from html.parser import HTMLParser
from pathlib import Path
from pfs.cpt_model import (
CptCode,
CptCrosswalk,
CptEdition,
CptInstruction,
CptListEntry,
CptReference,
CptSection,
)
# One CPT code: 4 digits + a digit/letter (99490, 0202U, 99213F, 0042T, 0001A).
_CODE = re.compile(r"\b\d{4}[0-9A-Z]\b")
_CODE_TOKEN = re.compile(r"^\d{4}[0-9A-Z]$")
_CODE_OR_RANGE = re.compile(r"\b(\d{4}[0-9A-Z])(?:-(\d{4}[0-9A-Z]))?\b")
_YEAR = re.compile(r"(?:19|20)\d{2}")
# Decorative glyphs stripped from all prose text: ▶/◀ new-or-revised-text
# wrappers, ➲ the citation bullet, ■ the element-bullet marker.
_DECORATIVE = "▶◀➲■"
# Legend glyphs (span.ama-en) kept in text and checked for by character.
_GLYPHS = {
"resequenced": "#",
"addon": "✚", # ✚
"new": "●", # ●
"revised": "▲", # ▲
"telemedicine": "★", # ★
"mod51_exempt": "⦸", # ⦸
"fda_pending": "⚡", # ⚡
}
# The Legend text (Appendix B) names these image files for audio-only /
# duplicate-PLA; the actual body-table glyphs observed on code rows use a
# *different* pair of filenames for the same two symbols (a production
# quirk — see task-2-report.md). Both pairs are accepted.
_AUDIO_ONLY_IMGS = {"enta.png", "entag.png"}
_PLA_IMGS = {"ent.png", "entg.png"}
_HEADING_RE = re.compile(r"^h([1-5])a?$")
_TOC_HEADING_RE = re.compile(r"^h([1-5])a?-toc$")
_TOC_RANGE_RE = re.compile(r"\(([^()]*)\)\s*$")
_APPENDIX_LETTER_RE = re.compile(r"[Aa]ppendix_?([A-Z])(?:[_.]|$)")
_LIST_APPENDICES = {"D", "E", "F", "G", "K", "N", "P", "T"}
def _clean(text: str) -> str:
"""Unescape entities, strip decorative glyphs, collapse whitespace."""
text = html.unescape(text)
for ch in _DECORATIVE:
text = text.replace(ch, "")
return re.sub(r"\s+", " ", text).strip()
def _strip_tags(fragment: str) -> str:
return _clean(re.sub(r"<[^>]+>", " ", fragment))
def _expand_range(lo: str, hi: str, *, cap: int = 20) -> tuple[str, ...]:
"""Expand a ``LO-HI`` code range into individual codes, capped at
``cap`` so a malformed or huge range can't blow up. Falls back to
the two endpoints when the codes aren't same-width all-digit (a
letter-suffixed range like ``0001A-0005A`` isn't expanded)."""
if lo.isdigit() and hi.isdigit() and len(lo) == len(hi):
width = len(lo)
start, end = int(lo), int(hi)
if start <= end:
n = min(end - start + 1, cap)
return tuple(str(start + i).zfill(width) for i in range(n))
return (lo,) if lo == hi else (lo, hi)
def _extract_targets(text: str) -> tuple[str, ...]:
"""Every code named in ``text``, ranges expanded, order preserved,
de-duplicated."""
out: list[str] = []
seen: set[str] = set()
for m in _CODE_OR_RANGE.finditer(text):
lo, hi = m.group(1), m.group(2)
codes = _expand_range(lo, hi) if hi else (lo,)
for c in codes:
if c not in seen:
seen.add(c)
out.append(c)
return tuple(out)
_USE_WITH_RE = re.compile(r"in conjunction with(.*)", re.IGNORECASE | re.DOTALL)
_NOT_WITH_TIME_RE = re.compile(
r"for service time reported with(.*)", re.IGNORECASE | re.DOTALL
)
_WITH_RE = re.compile(r"\bwith\b", re.IGNORECASE)
_SEE_RE = re.compile(r"\bsee\b(.*)", re.IGNORECASE | re.DOTALL)
_USE_ALSO_RE = re.compile(r"\buse\b(.*)", re.IGNORECASE | re.DOTALL)
_USE_CODE_RE = re.compile(r"\buse\s+(?:also\s+)?\d{4}[0-9A-Z]", re.IGNORECASE)
def _classify_instruction(text: str) -> tuple[str, tuple[str, ...]]:
"""Kind + targets for one parenthetical instruction, per the
brief's grammar: ``(Use X in conjunction with Y)`` = use-with;
``(Do not report … for service time reported with …)`` =
not-with-time; ``(Do not report … with …)`` (the CCM example's
"in the same calendar month with" is one phrasing of this, but
"in conjunction with" — by far the more common one across the
book — and a bare "with" are the same relationship) = not-with;
``(For …, see …)`` / ``(… use NNNNN)`` = see; else other."""
body = text.strip()
if body.startswith("(") and body.endswith(")"):
body = body[1:-1]
low = body.lower()
if low.startswith("use") and (m := _USE_WITH_RE.search(body)):
return "use-with", _extract_targets(m.group(1))
if m := _NOT_WITH_TIME_RE.search(body):
return "not-with-time", _extract_targets(m.group(1))
if low.startswith("do not report"):
withs = list(_WITH_RE.finditer(body))
if withs:
return "not-with", _extract_targets(body[withs[-1].end() :])
if m := _SEE_RE.search(body):
return "see", _extract_targets(m.group(1))
if (m := _USE_CODE_RE.search(body)) and (
m2 := _USE_ALSO_RE.search(body[m.start() :])
):
return "see", _extract_targets(m2.group(1))
return "other", _extract_targets(body)
def _parse_toc_range(text: str) -> tuple[str, str]:
"""``"Chronic Care Management Services* (99490-99437)"`` -> ("99490",
"99437"). Handles the 2022 template's comma list plus range form
(``"(99437, 99439, 99490-99491)"``) by taking the first code of the
first item and the last code of the last item. Returns ("", "")
when there's no trailing parenthetical or it isn't a code list."""
m = _TOC_RANGE_RE.search(text.strip())
if not m:
return "", ""
items = [i.strip() for i in m.group(1).split(",") if i.strip()]
if not items:
return "", ""
lo = items[0].split("-")[0].strip()
hi = items[-1].split("-")[-1].strip()
if not _CODE_TOKEN.match(lo) or not _CODE_TOKEN.match(hi):
return "", ""
return lo, hi
class _Frame:
__slots__ = ("classes", "id", "buf", "imgs")
def __init__(self, classes: tuple[str, ...]):
self.classes = classes
self.id: str | None = None
self.buf: list[str] = []
self.imgs: list[str] = []
_DivEvent = tuple # ("div", classes, id, text, imgs) | ("table", "start"|"end")
class _Tokenizer(HTMLParser):
"""Flat event stream: one ``("div", classes, id, text, imgs)`` per
``<div class=...>`` (text = all descendant text, entities already
decoded; imgs = basenames of any ``<img src>`` inside), plus
``("table", "start"/"end")`` at each ``<table>`` boundary. Row
(``<tr>``/``<td>``) structure is deliberately not modelled — the
grouping pass in ``_group`` doesn't need it (see module docstring)."""
def __init__(self) -> None:
super().__init__(convert_charrefs=True)
self.events: list[tuple] = []
self._stack: list[_Frame] = []
def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
amap = {k: v for k, v in attrs if v is not None}
if tag == "table":
self.events.append(("table", "start"))
return
if tag == "div" and "class" in amap:
frame = _Frame(tuple(amap["class"].split()))
frame.id = amap.get("id")
self._stack.append(frame)
return
if tag == "img":
src = amap.get("src", "")
base = src.rsplit("/", 1)[-1]
for f in self._stack:
f.imgs.append(base)
return
frag = amap.get("id")
if frag is None and "href" in amap and "#" in amap["href"]:
frag = amap["href"].rsplit("#", 1)[-1]
if frag:
for f in reversed(self._stack):
if f.id is None:
f.id = frag
break
def handle_endtag(self, tag: str) -> None:
if tag == "table":
self.events.append(("table", "end"))
return
if tag == "div" and self._stack:
frame = self._stack.pop()
text = _clean("".join(frame.buf))
self.events.append(
("div", frame.classes, frame.id, text, tuple(frame.imgs))
)
def handle_data(self, data: str) -> None:
for f in self._stack:
f.buf.append(data)
def _toc_ranges_from_events(events: list[tuple]) -> dict[str, tuple[str, str]]:
out: dict[str, tuple[str, str]] = {}
for ev in events:
if ev[0] != "div":
continue
_, classes, id_, text, _imgs = ev
if id_ and any(_TOC_HEADING_RE.match(c) for c in classes):
lo, hi = _parse_toc_range(text)
if lo or hi:
out[id_] = (lo, hi)
return out
def _category(code: str) -> str:
if re.fullmatch(r"\d{4}F", code):
return "II"
if re.fullmatch(r"\d{4}T", code):
return "III"
return "I"
def _group(
events: list[tuple], *, year: int
) -> tuple[list[CptSection], list[CptCode], list[CptInstruction], list[CptReference]]:
sections: list[CptSection] = []
codes: list[CptCode] = []
instructions: list[CptInstruction] = []
references: list[CptReference] = []
toc_ranges = _toc_ranges_from_events(events)
heading_stack: list[tuple[int, str]] = []
current: dict | None = None
guideline_buf: list[str] = []
in_guideline_zone = False
entry: dict | None = None
last_primary: CptCode | None = None
def finalize_entry() -> None:
nonlocal entry, last_primary
if entry is None:
return
code = entry["code"]
if not code:
entry = None
return
glyph_text = entry["glyph_text"]
flags = {name: (ch in glyph_text) for name, ch in _GLYPHS.items()}
imgs = set(entry["imgs"])
audio_only = bool(imgs & _AUDIO_ONLY_IMGS)
pla = bool(imgs & _PLA_IMGS)
prose: list[str] = entry["prose"]
is_sub = entry["is_sub"]
if is_sub and last_primary is not None:
stem = last_primary.stem
elements = last_primary.elements
parent = last_primary.code
own_prose = prose
else:
stem = prose[0] if prose else ""
elements = tuple(entry["elements"])
parent = ""
own_prose = prose[1:]
tail = ""
for block in own_prose:
if block.startswith("("):
kind, targets = _classify_instruction(block)
instructions.append(
CptInstruction(code=code, kind=kind, text=block, targets=targets)
)
elif not tail:
tail = block
else:
tail = f"{tail} {block}".strip()
parts = [p for p in (stem, "; ".join(elements), tail) if p]
descriptor = " ".join(parts).strip()
for ref_text in entry["refs"]:
low = ref_text.lower()
if "cpt changes" in low:
years = tuple(sorted({int(y) for y in _YEAR.findall(ref_text)}))
references.append(
CptReference(
code=code, kind="cpt-changes", years=years, text=ref_text
)
)
elif "cpt assistant" in low:
references.append(
CptReference(
code=code, kind="cpt-assistant", years=(), text=ref_text
)
)
cpt_code = CptCode(
code=code,
sec_id=current["sec_id"] if current else "",
descriptor=descriptor,
stem=stem,
elements=elements,
tail=tail,
addon=flags["addon"],
resequenced=flags["resequenced"],
new=flags["new"],
revised=flags["revised"],
telemedicine=flags["telemedicine"],
parent=parent,
mod51_exempt=flags["mod51_exempt"],
audio_only=audio_only,
fda_pending=flags["fda_pending"],
pla=pla,
category=_category(code),
)
codes.append(cpt_code)
if not is_sub:
last_primary = cpt_code
entry = None
def finalize_section() -> None:
nonlocal current, guideline_buf
if current is not None:
lo, hi = toc_ranges.get(current["sec_id"], ("", ""))
sections.append(
CptSection(
sec_id=current["sec_id"],
level=current["level"],
title=current["title"],
path=current["path"],
code_lo=lo,
code_hi=hi,
guideline=_clean(" ".join(guideline_buf)),
)
)
current = None
guideline_buf = []
for ev in events:
if ev[0] == "table":
finalize_entry()
if ev[1] == "start":
in_guideline_zone = False
last_primary = None
continue
_, classes, id_, text, imgs = ev
if any(_TOC_HEADING_RE.match(c) for c in classes):
continue # consumed by the toc_ranges pre-pass
if any(_HEADING_RE.match(c) for c in classes):
finalize_entry()
finalize_section()
level = int(
next(m for c in classes if (m := _HEADING_RE.match(c))).group(1)
)
title = text.rstrip("*").strip()
while heading_stack and heading_stack[-1][0] >= level:
heading_stack.pop()
heading_stack.append((level, title))
path = tuple(t for _, t in heading_stack)
current = {
"sec_id": id_ or "",
"level": level,
"title": title,
"path": path,
}
guideline_buf = []
in_guideline_zone = True
last_primary = None
continue
if "noindent" in classes:
if in_guideline_zone and current is not None and text:
guideline_buf.append(text)
continue
if "table-para" in classes:
finalize_entry()
code_num = ""
if id_ and id_.startswith("code_"):
code_num = id_[len("code_") :]
if not code_num:
m = _CODE.search(text)
if m:
code_num = m.group(0)
entry = {
"code": code_num,
"glyph_text": text,
"imgs": set(imgs),
"prose": [],
"elements": [],
"refs": [],
"is_sub": False,
}
in_guideline_zone = False
continue
if entry is None:
continue
if any("sub" in c for c in classes):
entry["is_sub"] = True
if any(
c.startswith("table-slist") or c.startswith("table-square") for c in classes
):
if text:
entry["elements"].append(text)
elif any(c.startswith("table-RT") for c in classes):
if text:
entry["refs"].append(text)
elif any(
c.startswith("table-para1")
or c == "table-paral"
or c.startswith("table-para2")
for c in classes
):
if text:
entry["prose"].append(text)
# else: unrecognised block class (e.g. a special-format table
# cell) — not part of the primary code-entry shape, ignored.
finalize_entry()
finalize_section()
return sections, codes, instructions, references
def parse_xhtml(
text: str, *, year: int
) -> tuple[list[CptSection], list[CptCode], list[CptInstruction], list[CptReference]]:
"""Parse one body-chapter xhtml string (optionally with its chapter
TOC snippet concatenated in, for lo/hi — see module docstring and
the synthetic fixture) into sections/codes/instructions/references."""
tok = _Tokenizer()
tok.feed(text)
tok.close()
return _group(tok.events, year=year)
# --- Appendices -------------------------------------------------------
_APP_PARA_RE = re.compile(r'<div class="app-para(?:-t)?">\s*<b>([^<]*)</b>')
_TR_RE = re.compile(r"<tr>.*?</tr>", re.DOTALL)
_APP_TABLE1_RE = re.compile(r'<div class="app-table-text1">(.*?)</div>', re.DOTALL)
_APP_TABLE2_RE = re.compile(
r'<div class="app-table-text2">.*?<b>([^<]+)</b>', re.DOTALL
)
def _appendix_letter(basename: str) -> str | None:
m = _APPENDIX_LETTER_RE.search(basename)
return m.group(1) if m else None
def _parse_appendix_list(xhtml: str, letter: str) -> list[CptListEntry]:
"""D/E/F/G/K/P/T are a flat ``div.app-para(-t)`` list of codes; N is
a 2-column table (resequenced code -> corresponding locations) — we
take only the first column."""
if letter == "N":
out = []
for row in _TR_RE.findall(xhtml):
if "app-table-th2" in row:
continue
m = _APP_TABLE2_RE.search(row)
if m:
out.append(CptListEntry(appendix=letter, code=_clean(m.group(1))))
return out
return [
CptListEntry(appendix=letter, code=_clean(m.group(1)))
for m in _APP_PARA_RE.finditer(xhtml)
]
def _parse_appendix_crosswalk(xhtml: str) -> list[CptCrosswalk]:
"""Appendix M: current code, former code, year deleted, citations —
a plain 4-column ``table.table8``."""
out = []
for row in _TR_RE.findall(xhtml):
if "app-table-th1" in row:
continue
cells = _APP_TABLE1_RE.findall(row)
if len(cells) >= 4:
current, former, year_deleted, citations = (
_strip_tags(c) for c in cells[:4]
)
out.append(
CptCrosswalk(
current_code=current,
former_code=former,
year_deleted=year_deleted,
citations=citations,
)
)
return out
def _infer_year(path: Path) -> int:
m = re.search(r"(19|20)\d{2}", path.name)
if m:
return int(m.group(0))
raise ValueError(
f"cannot infer CPT edition year from {path.name!r}; pass year= explicitly"
)
def parse_epub(path: Path, *, year: int | None = None) -> CptEdition:
"""Parse one CPT Professional edition EPUB into a ``CptEdition``.
Reads every chapter-TOC file (``*_Toc.xhtml``) first to build a
book-wide ``sec_id -> (lo, hi)`` map — a chapter's TOC can point
into a *different* chapter's body file, so ranges are matched by
id across the whole book, not per source file — then parses every
chapter body file and every appendix file (D/E/F/G/K/M/N/P/T)."""
path = Path(path)
if year is None:
year = _infer_year(path)
with zipfile.ZipFile(path) as zf:
names = zf.namelist()
toc_ranges: dict[str, tuple[str, str]] = {}
for name in names:
base = name.rsplit("/", 1)[-1]
if name.startswith("OPS/") and base.endswith(".xhtml") and "_Toc" in base:
text = zf.read(name).decode("utf-8", errors="replace")
tok = _Tokenizer()
tok.feed(text)
tok.close()
toc_ranges.update(_toc_ranges_from_events(tok.events))
sections: list[CptSection] = []
codes: list[CptCode] = []
instructions: list[CptInstruction] = []
references: list[CptReference] = []
for name in names:
base = name.rsplit("/", 1)[-1]
if not (name.startswith("OPS/") and base.endswith(".xhtml")):
continue
if "Chapter" not in base or "_Toc" in base:
continue
text = zf.read(name).decode("utf-8", errors="replace")
s, c, i, r = parse_xhtml(text, year=year)
sections.extend(s)
codes.extend(c)
instructions.extend(i)
references.extend(r)
fixed_sections = []
for sec in sections:
if sec.code_lo == "" and sec.code_hi == "" and sec.sec_id in toc_ranges:
lo, hi = toc_ranges[sec.sec_id]
sec = replace(sec, code_lo=lo, code_hi=hi)
fixed_sections.append(sec)
crosswalks: list[CptCrosswalk] = []
lists: list[CptListEntry] = []
for name in names:
base = name.rsplit("/", 1)[-1]
if not (
name.startswith("OPS/")
and base.endswith(".xhtml")
and "ppendix" in base
):
continue
letter = _appendix_letter(base)
if letter is None:
continue
text = zf.read(name).decode("utf-8", errors="replace")
if letter == "M":
crosswalks.extend(_parse_appendix_crosswalk(text))
elif letter in _LIST_APPENDICES:
lists.extend(_parse_appendix_list(text, letter))
return CptEdition(
year=year,
sections=tuple(fixed_sections),
codes=tuple(codes),
instructions=tuple(instructions),
references=tuple(references),
crosswalks=tuple(crosswalks),
lists=tuple(lists),
)

119
src/pfs/cpt_model.py Normal file
View File

@@ -0,0 +1,119 @@
"""Dataclasses for the CPT organizing-principles schema (docs/superpowers/
specs/2026-09-09-cpt-canonical-schema-design.md §3). Pure — no DuckDB, no
bib, no I/O. ``cpt_epub.parse_epub``/``parse_xhtml`` build these from the
AMA EPUB markup; nothing here reads a file.
"""
from __future__ import annotations
from dataclasses import dataclass
@dataclass(frozen=True)
class CptSection:
"""One heading (``div.hN``) in the codebook's own hierarchy.
``path`` is the ancestor titles plus this heading's own title, root
first — e.g. ``("Care Management Services", "Chronic Care Management
Services")``. ``code_lo``/``code_hi`` come from the chapter TOC's
print-order code span (``"Title* (LO-HI)"``); both are ``""`` when
the TOC didn't carry a range for this heading (headings without a
trailing ``*`` carry no guideline and often no range either).
"""
sec_id: str
level: int
title: str
path: tuple[str, ...]
code_lo: str
code_hi: str
guideline: str
@dataclass(frozen=True)
class CptCode:
"""One code entry (a table row keyed by its ``table-para`` glyph
cell). ``stem``/``elements`` are inherited from the nearest preceding
non-child code when this entry is an indented child or an add-on
(``parent`` non-empty) — CPT's own semicolon-continuation rule, which
add-on rows use too even though their own sentence reads standalone.
``tail`` is always this entry's own trailing prose (the non-
parenthetical text after the stem/elements, e.g. a time clause), and
``descriptor`` is the assembled reading: stem + elements + tail.
"""
code: str
sec_id: str
descriptor: str
stem: str
elements: tuple[str, ...]
tail: str
addon: bool
resequenced: bool
new: bool
revised: bool
telemedicine: bool
parent: str
mod51_exempt: bool
audio_only: bool
fda_pending: bool
pla: bool
category: str # "I" | "II" | "III" — from the code's own shape
@dataclass(frozen=True)
class CptInstruction:
"""One parenthetical note (``div.table-para2`` — or, in editions that
don't use that class, any trailing prose block — beginning with
``"("``). ``targets`` are the codes it names, ranges expanded (capped
at 20 codes per range so a malformed range can't blow up)."""
code: str
kind: str # "use-with" | "not-with" | "not-with-time" | "see" | "other"
text: str
targets: tuple[str, ...]
@dataclass(frozen=True)
class CptReference:
"""One ``div.table-RT*`` citation line under a code."""
code: str
kind: str # "cpt-changes" | "cpt-assistant"
years: tuple[int, ...]
text: str
@dataclass(frozen=True)
class CptCrosswalk:
"""One row of Appendix M — a deleted/renumbered code's citations
crosswalked to its current code."""
current_code: str
former_code: str
year_deleted: str
citations: str
@dataclass(frozen=True)
class CptListEntry:
"""One code in one authoritative appendix list (D add-on codes, E
modifier-51 exempt, F modifier-63 exempt, G moderate sedation
included, K FDA-pending, N resequenced, P audio-video telemedicine,
T audio-only telemedicine)."""
appendix: str
code: str
@dataclass(frozen=True)
class CptEdition:
"""Everything parsed from one CPT codebook edition."""
year: int
sections: tuple[CptSection, ...]
codes: tuple[CptCode, ...]
instructions: tuple[CptInstruction, ...]
references: tuple[CptReference, ...]
crosswalks: tuple[CptCrosswalk, ...]
lists: tuple[CptListEntry, ...]

43
tests/pfs/fixtures/cpt_sample.xhtml Normal file
View File

@@ -0,0 +1,43 @@
<html xmlns="http://www.w3.org/1999/xhtml">
<head>
<title>Synthetic CPT sample</title>
<link href="stylesheet.css" rel="stylesheet" type="text/css"/>
</head>
<body>
<div class="h1-toc"><a href="body.xhtml#sec_1"><span class="green">Care Coordination Services* (55000-55010)</span></a></div>
<div class="h2-toc"><a href="body.xhtml#sec_2"><span class="red">Sample Chronic Wellness Services* (55000-55003)</span></a></div>
<div class="h1-toc"><a href="body.xhtml#sec_3"><span class="green">Other Sample Services* (55100-55101)</span></a></div>
<div class="h2-toc"><a href="body.xhtml#sec_4"><span class="red">Empty Category Services*</span></a></div>
<div class="h1" id="sec_1">Care Coordination Services</div>
<div class="noindent">Sample guideline text for the invented care-coordination family of services, used only to test parsing &amp; not copied from any real codebook.</div>
<div class="h2" id="sec_2">Sample Chronic Wellness Services</div>
<div class="noindent">Sample per-category guideline text describing invented wellness-coordination activities.</div>
<table class="table1">
<tbody>
<tr>
<td class="td-w1" id="code_55000"><div class="table-para"><span class="ama-en">#</span> <b>55000</b></div></td>
<td class="td"><div class="table-para1">Sample wellness coordination services with the following required elements:</div>
<div class="table-slist">■<span class="ssp"/>first invented element of the service,</div>
<div class="table-slist1">■<span class="ssp"/>second invented element of the service,</div>
<div class="table-slist1">■<span class="ssp"/>third invented element of the service;</div>
<div class="table-para2">first 20 minutes of invented staff time, per calendar month.</div>
<div class="table-RT"><span class="blue"><span class="ama-en">➲</span></span><span class="sp1"/><i>CPT Changes: An Insider's View</i> 2015, 2021, 2022</div>
<div class="table-RT1"><span class="green"><span class="ama-en">➲</span></span><span class="sp1"/><i>CPT Assistant</i> Jan 21:5</div></td>
</tr>
<tr>
<td class="td-w1" id="code_55001"><div class="table-para"><span class="ama-en">#</span><span class="ama-en">✚</span> <b>55001</b></div></td>
<td class="td"><div class="table-para1-sub">each additional 20 minutes of invented staff time, per calendar month (List separately in addition to code for primary procedure)</div>
<div class="table-RT-sub"><span class="blue"><span class="ama-en">➲</span></span><span class="sp1"/><i>CPT Changes: An Insider's View</i> 2022</div>
<div class="table-para2">(Use 55001 in conjunction with 55000)</div>
<div class="table-para2">(Do not report 55001, 55000 in the same calendar month with 90951-90970, 55555)</div>
<div class="table-para2">(Sample wellness services of less than 20 minutes duration are not reported separately)</div></td>
</tr>
</tbody>
</table>
<div class="h1" id="sec_3">Other Sample Services</div>
<div class="h2" id="sec_4">Empty Category Services</div>
<div class="noindent">Guideline text with no code rows, to test that a section with zero codes still gets a path.</div>
</body>
</html>

274
tests/pfs/test_cpt_epub.py Normal file
View File

@@ -0,0 +1,274 @@
"""Tests for pfs.cpt_epub — the pure CPT EPUB parser.
The synthetic fixture (tests/pfs/fixtures/cpt_sample.xhtml) is written
in the 2024 template's markup style with invented text; it is never
copied from the real AMA codebook. The integration tests at the bottom
run the real parser against the actual EPUB files on disk (read-only,
in data/zotero/data/storage/) and are skipped when a file is absent —
they assert structure and counts only, never descriptor text.
"""
from __future__ import annotations
from pathlib import Path
import pytest
from pfs.cpt_epub import parse_epub, parse_xhtml
FIXTURE = Path(__file__).parent / "fixtures" / "cpt_sample.xhtml"
REPO_ROOT = Path(__file__).resolve().parents[2]
EPUB_2024 = (
REPO_ROOT
/ "data/zotero/data/storage/EIGIRKRK/CPT Professional 2024 - American Medical Association.epub"
)
EPUB_2022 = REPO_ROOT / "data/zotero/data/storage/UG4R55FW/CPT Professional 2022.epub"
EPUB_2021 = (
REPO_ROOT / "data/zotero/data/storage/NM3NJZV5/CPT 2021 Professional Edition.epub"
)
EPUB_2019 = REPO_ROOT / "data/zotero/data/storage/VA34EEIM/CPT 2019.epub"
@pytest.fixture(scope="module")
def parsed():
text = FIXTURE.read_text(encoding="utf-8")
sections, codes, instructions, references = parse_xhtml(text, year=2024)
return {
"sections": {s.sec_id: s for s in sections},
"sections_list": sections,
"codes": {c.code: c for c in codes},
"instructions": instructions,
"references": references,
}
class TestSections:
def test_two_top_level_and_two_nested(self, parsed):
levels = sorted((s.sec_id, s.level) for s in parsed["sections_list"])
assert levels == [
("sec_1", 1),
("sec_2", 2),
("sec_3", 1),
("sec_4", 2),
]
def test_paths_include_ancestors(self, parsed):
sec2 = parsed["sections"]["sec_2"]
assert sec2.title == "Sample Chronic Wellness Services"
assert sec2.path == (
"Care Coordination Services",
"Sample Chronic Wellness Services",
)
sec4 = parsed["sections"]["sec_4"]
assert sec4.path == ("Other Sample Services", "Empty Category Services")
def test_lo_hi_from_toc_snippet(self, parsed):
sec1 = parsed["sections"]["sec_1"]
assert (sec1.code_lo, sec1.code_hi) == ("55000", "55010")
sec2 = parsed["sections"]["sec_2"]
assert (sec2.code_lo, sec2.code_hi) == ("55000", "55003")
sec3 = parsed["sections"]["sec_3"]
assert (sec3.code_lo, sec3.code_hi) == ("55100", "55101")
def test_heading_without_range_is_empty(self, parsed):
sec4 = parsed["sections"]["sec_4"]
assert (sec4.code_lo, sec4.code_hi) == ("", "")
def test_section_with_no_codes_still_has_a_path(self, parsed):
sec4 = parsed["sections"]["sec_4"]
assert sec4.path
assert sec4.guideline
def test_entities_and_whitespace_cleaned(self, parsed):
sec1 = parsed["sections"]["sec_1"]
assert "&amp;" not in sec1.guideline
assert "&" in sec1.guideline # entity decoded, not dropped
assert " " not in sec1.guideline
class TestPrimaryCode:
def test_three_elements_and_tail(self, parsed):
code = parsed["codes"]["55000"]
assert code.sec_id == "sec_2"
assert len(code.elements) == 3
assert code.elements[0] == "first invented element of the service,"
assert (
code.tail == "first 20 minutes of invented staff time, per calendar month."
)
assert (
code.stem
== "Sample wellness coordination services with the following required elements:"
)
def test_symbols(self, parsed):
code = parsed["codes"]["55000"]
assert code.resequenced is True
assert code.addon is False
assert code.new is False
assert code.revised is False
assert code.telemedicine is False
assert code.parent == ""
def test_category_i(self, parsed):
assert parsed["codes"]["55000"].category == "I"
def test_descriptor_assembled(self, parsed):
code = parsed["codes"]["55000"]
assert code.stem in code.descriptor
assert code.tail in code.descriptor
assert code.elements[0] in code.descriptor
class TestAddonCode:
def test_addon_flag_and_parent(self, parsed):
code = parsed["codes"]["55001"]
assert code.addon is True
assert code.resequenced is True
assert code.parent == "55000"
def test_inherits_stem_and_elements(self, parsed):
addon = parsed["codes"]["55001"]
primary = parsed["codes"]["55000"]
assert addon.stem == primary.stem
assert addon.elements == primary.elements
def test_own_tail(self, parsed):
addon = parsed["codes"]["55001"]
assert "each additional 20 minutes" in addon.tail
assert "primary procedure" in addon.tail
class TestInstructions:
def test_use_with_targets_primary(self, parsed):
use_with = [
i
for i in parsed["instructions"]
if i.code == "55001" and i.kind == "use-with"
]
assert len(use_with) == 1
assert use_with[0].targets == ("55000",)
def test_not_with_targets_expanded_range_plus_listed(self, parsed):
not_with = [
i
for i in parsed["instructions"]
if i.code == "55001" and i.kind == "not-with"
]
assert len(not_with) == 1
targets = not_with[0].targets
expanded_range = tuple(f"909{n}" for n in range(51, 71))
assert len(expanded_range) == 20
for t in expanded_range:
assert t in targets
assert "55555" in targets
def test_plain_parenthetical_is_other(self, parsed):
others = [
i for i in parsed["instructions"] if i.code == "55001" and i.kind == "other"
]
assert len(others) == 1
assert others[0].text.startswith("(Sample wellness services")
def test_three_instructions_on_addon(self, parsed):
addon_instructions = [i for i in parsed["instructions"] if i.code == "55001"]
assert len(addon_instructions) == 3
class TestReferences:
def test_cpt_changes_years(self, parsed):
refs = [
r
for r in parsed["references"]
if r.code == "55000" and r.kind == "cpt-changes"
]
assert len(refs) == 1
assert refs[0].years == (2015, 2021, 2022)
def test_cpt_assistant_kept_as_text(self, parsed):
refs = [
r
for r in parsed["references"]
if r.code == "55000" and r.kind == "cpt-assistant"
]
assert len(refs) == 1
assert "Jan 21:5" in refs[0].text
def test_addon_reference_year(self, parsed):
refs = [
r
for r in parsed["references"]
if r.code == "55001" and r.kind == "cpt-changes"
]
assert refs[0].years == (2022,)
# --- Real-file integration tests (skipped when the file is absent) ---
@pytest.mark.skipif(not EPUB_2024.exists(), reason="real 2024 CPT EPUB not on disk")
class TestReal2024:
@pytest.fixture(scope="class")
def edition(self):
return parse_epub(EPUB_2024, year=2024)
def test_at_least_8000_codes(self, edition):
assert len(edition.codes) >= 8000
def test_every_section_has_a_path(self, edition):
assert edition.sections
for sec in edition.sections:
assert sec.path
assert sec.path[-1] == sec.title
def test_chronic_care_management_section_has_expected_codes(self, edition):
by_code = {c.code: c for c in edition.codes}
sec = next(
s for s in edition.sections if s.title == "Chronic Care Management Services"
)
codes_in_sec = {c.code for c in edition.codes if c.sec_id == sec.sec_id}
for code in ("99490", "99439", "99491", "99437"):
assert code in codes_in_sec, code
assert code in by_code
def test_99439_is_addon_using_99490(self, edition):
by_code = {c.code: c for c in edition.codes}
code_99439 = by_code["99439"]
assert code_99439.addon is True
use_with = [
i
for i in edition.instructions
if i.code == "99439" and i.kind == "use-with"
]
assert use_with
assert "99490" in use_with[0].targets
def test_appendix_d_addon_codes(self, edition):
d_entries = [e for e in edition.lists if e.appendix == "D"]
assert len(d_entries) >= 500
def test_appendix_m_crosswalk_former_codes_end_in_t(self, edition):
assert edition.crosswalks
assert any(row.former_code.endswith("T") for row in edition.crosswalks)
@pytest.mark.skipif(not EPUB_2022.exists(), reason="real 2022 CPT EPUB not on disk")
def test_real_2022_parses():
edition = parse_epub(EPUB_2022, year=2022)
assert len(edition.codes) >= 5000
assert edition.sections
@pytest.mark.skipif(not EPUB_2021.exists(), reason="real 2021 CPT EPUB not on disk")
def test_real_2021_parses():
edition = parse_epub(EPUB_2021, year=2021)
assert len(edition.codes) >= 5000
assert edition.sections
@pytest.mark.skipif(not EPUB_2019.exists(), reason="real 2019 CPT EPUB not on disk")
def test_real_2019_parses_without_code_ids():
edition = parse_epub(EPUB_2019, year=2019)
assert len(edition.codes) >= 3000
assert edition.sections