f7c31346bb
Some checks failed
CI / lint (push) Successful in 33s
CI / notebooks-smoke (push) Successful in 1m26s
Deploy / notebooks (push) Has been skipped
Deploy / zotero (push) Has been skipped
Deploy / docs (push) Has been skipped
Deploy / api (push) Has been skipped
Deploy / llm (push) Has been skipped
Deploy / mc (push) Has been skipped
Infra CI / notebooks (push) Successful in 51s
Infra CI / zotero (push) Successful in 13s
Infra CI / docs (push) Successful in 20s
Infra CI / api (push) Successful in 1m13s
Infra CI / llm (push) Successful in 45s
Infra CI / mc (push) Failing after 13s
Deploy / report (push) Successful in 14s
CI / test (push) Successful in 14m20s
1006 lines
40 KiB
Python
1006 lines
40 KiB
Python
"""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``) plus
|
|
``rex.comments.epub_text`` (itself stdlib-only — container/spine
|
|
resolution shared with the corpus indexer's own EPUB text extraction,
|
|
F2); 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 (
|
|
CptAlternate,
|
|
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"}
|
|
|
|
# "-ru" = a running-header repeat of the previous split file's last
|
|
# heading at the top of a continuation file (574 of these in 2019
|
|
# alone, vs. 4 in 2021/2022/2024) — a genuine heading occurrence, not
|
|
# a TOC entry, so it's folded into the same pattern as a plain "hN".
|
|
_HEADING_RE = re.compile(r"^h([1-5])a?(?:-ru)?$")
|
|
_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])(?:[_.]|$)")
|
|
|
|
# C8 (task 3): a resequenced code's placeholder row at its old numeric
|
|
# position ("Code is out of numerical sequence. See 99490-99491") —
|
|
# never a real code entry, see finalize_entry.
|
|
_RESEQUENCED_PLACEHOLDER_RE = re.compile(r"out of numerical sequence", re.IGNORECASE)
|
|
|
|
_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, source: str = "", chapter_title: str = ""
|
|
) -> tuple[
|
|
list[CptSection], list[CptCode], list[CptInstruction], list[CptReference], str
|
|
]:
|
|
"""Group one chapter-group's flat event stream (see module docstring)
|
|
into sections/codes/instructions/references, plus the *active*
|
|
chapter title on exit (5th return value).
|
|
|
|
``chapter_title`` (C10) seeds ``heading_stack`` as a level-0 parent:
|
|
``div.ch-title`` is the book's own chapter banner ("Surgery
|
|
Guidelines", "Surgery", "Anesthesia Guidelines", …) and a real
|
|
``hN`` heading nests *under* it — it is popped only by the next
|
|
``ch-title``, never by an ``hN`` (level ≥ 1 never pops level 0). Most
|
|
chapter files carry no ``ch-title`` of their own at all (a chapter
|
|
split across many numbered files for size, e.g. 2019's Surgery body
|
|
spans Chapter07 through Chapter16 with the banner only on Chapter06)
|
|
— ``parse_epub`` threads the returned title from one chapter-group's
|
|
call into the next one's ``chapter_title`` argument so it keeps
|
|
parenting headings until a new banner replaces it."""
|
|
sections: list[CptSection] = []
|
|
codes: list[CptCode] = []
|
|
instructions: list[CptInstruction] = []
|
|
references: list[CptReference] = []
|
|
|
|
toc_ranges = _toc_ranges_from_events(events)
|
|
|
|
# C10: the (possibly inherited) chapter title is the level-0 base of
|
|
# the stack from the very start of this chapter-group, so the first
|
|
# real heading already nests under it.
|
|
heading_stack: list[tuple[int, str]] = [(0, chapter_title)] if chapter_title else []
|
|
current: dict | None = None
|
|
guideline_buf: list[str] = []
|
|
in_guideline_zone = False
|
|
entry: dict | None = None
|
|
last_primary: CptCode | None = None
|
|
last_primary_semicolon = False
|
|
heading_counter = 0
|
|
|
|
def finalize_entry() -> None:
|
|
nonlocal entry, last_primary, last_primary_semicolon
|
|
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:
|
|
# An indented child (add-on or the classic semicolon
|
|
# continuation, e.g. 25100 "Arthrotomy, wrist joint; with
|
|
# biopsy" / 25105 "with synovectomy") inherits the parent's
|
|
# *shared* stem and elements — never the parent's own
|
|
# post-semicolon completion, which belongs only to the
|
|
# parent (see below).
|
|
stem = last_primary.stem
|
|
elements = last_primary.elements
|
|
parent = last_primary.code
|
|
own_prose = prose
|
|
semicolon_style = last_primary_semicolon
|
|
else:
|
|
first = prose[0] if prose else ""
|
|
if ";" in first:
|
|
# This row's own text is itself a semicolon-family
|
|
# entry: the shared stem is inheritable (by later
|
|
# sub-rows), and this row's own post-semicolon
|
|
# completion is *not* part of that shared stem — it
|
|
# goes through the same tail-building loop below as
|
|
# any other row's own trailing text.
|
|
shared, _, own_fragment = first.partition(";")
|
|
stem = shared.strip()
|
|
own_fragment = own_fragment.strip()
|
|
semicolon_style = True
|
|
else:
|
|
stem = first
|
|
own_fragment = ""
|
|
semicolon_style = False
|
|
elements = tuple(entry["elements"])
|
|
parent = ""
|
|
own_prose = ([own_fragment] if own_fragment else []) + prose[1:]
|
|
|
|
tail = ""
|
|
for block in own_prose:
|
|
if block.startswith("("):
|
|
kind, targets = _classify_instruction(block)
|
|
# The instruction's own owning code is a subject, not a
|
|
# target, even when it's named alongside a sibling
|
|
# code in the same clause (e.g. "Do not report 99439,
|
|
# 99490 ... with ..." — 99490 stays a target, 99439
|
|
# the owner doesn't).
|
|
targets = tuple(t for t in targets if t != code)
|
|
instructions.append(
|
|
CptInstruction(code=code, kind=kind, text=block, targets=targets)
|
|
)
|
|
elif not tail:
|
|
tail = block
|
|
else:
|
|
tail = f"{tail} {block}".strip()
|
|
|
|
# Each element line carries its own trailing list punctuation
|
|
# (",", ";") from the printed page; strip it before joining so
|
|
# the descriptor doesn't read "...service,; ...".
|
|
cleaned_elements = (re.sub(r"[,;]+\s*$", "", el).strip() for el in elements)
|
|
elements_joined = "; ".join(e for e in cleaned_elements if e)
|
|
# A semicolon-family row's descriptor reads "shared stem; own
|
|
# fragment" (elements/tail folded in the same way); every
|
|
# other row keeps the plain space join it always had.
|
|
joiner = "; " if semicolon_style else " "
|
|
parts = [p for p in (stem, elements_joined, tail) if p]
|
|
descriptor = joiner.join(parts).strip()
|
|
|
|
if _RESEQUENCED_PLACEHOLDER_RE.search(descriptor):
|
|
# A resequenced code prints twice: once as a bare pointer at
|
|
# its old numeric position ("Code is out of numerical
|
|
# sequence. See NNNNN-NNNNN") and once with its real entry
|
|
# at its new print-order position. The pointer row carries
|
|
# no descriptor/section of its own worth keeping — C8 (task
|
|
# 3) — so it must never become a pfs.cpt_code row; only the
|
|
# real entry (elsewhere, under its real section) survives.
|
|
entry = None
|
|
return
|
|
|
|
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
|
|
last_primary_semicolon = semicolon_style
|
|
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
|
|
last_primary_semicolon = False
|
|
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
|
|
|
|
is_heading = any(_HEADING_RE.match(c) for c in classes)
|
|
# C10: div.ch-title is the book's own chapter banner ("Surgery
|
|
# Guidelines", "Surgery", …) — the level-0 parent of every
|
|
# heading in this chapter (and every later chapter-group that
|
|
# doesn't carry its own banner, via the seeded heading_stack
|
|
# above), replaced only by the next ch-title, never popped by a
|
|
# real hN (level ≥ 1 never pops level 0). A chapter with no hN
|
|
# of its own at all (e.g. "Category III Codes") still gets a
|
|
# real, titled section this way instead of collapsing onto
|
|
# sec_id="" — it's just level 0 doing that job now, not level 1.
|
|
is_chapter_title = "ch-title" in classes
|
|
if is_heading or is_chapter_title:
|
|
finalize_entry()
|
|
finalize_section()
|
|
title = text.rstrip("*").strip()
|
|
if is_chapter_title:
|
|
level = 0
|
|
chapter_title = title
|
|
heading_stack = [(0, title)]
|
|
else:
|
|
level = int(
|
|
next(m for c in classes if (m := _HEADING_RE.match(c))).group(1)
|
|
)
|
|
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)
|
|
sec_id = id_
|
|
if not sec_id:
|
|
heading_counter += 1
|
|
sec_id = f"{source}:h{heading_counter}"
|
|
current = {
|
|
"sec_id": sec_id,
|
|
"level": level,
|
|
"title": title,
|
|
"path": path,
|
|
}
|
|
guideline_buf = []
|
|
in_guideline_zone = True
|
|
last_primary = None
|
|
last_primary_semicolon = False
|
|
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, chapter_title
|
|
|
|
|
|
def parse_xhtml(
|
|
text: str, *, year: int, source: str = "", chapter_title: str = ""
|
|
) -> 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.
|
|
|
|
``source`` (the chapter file's basename, when called from
|
|
``parse_epub``) seeds the synthetic id given to a heading that has
|
|
none of its own (2019's headings mostly lack one) — see
|
|
``_group``'s ``heading_counter``. ``chapter_title`` (C10) seeds the
|
|
level-0 chapter-banner parent of every heading in ``text`` — direct
|
|
callers (tests) default to none, matching a chapter's own first
|
|
``div.ch-title``; ``parse_epub`` threads it itself via ``_group``
|
|
directly (this wrapper's return stays a 4-tuple, dropping ``_group``'s
|
|
trailing updated-title element, since ``parse_epub`` is the only
|
|
caller that needs to carry it forward)."""
|
|
tok = _Tokenizer()
|
|
tok.feed(text)
|
|
tok.close()
|
|
sections, codes, instructions, references, _chapter_title = _group(
|
|
tok.events, year=year, source=source, chapter_title=chapter_title
|
|
)
|
|
return sections, codes, instructions, references
|
|
|
|
|
|
# --- 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"
|
|
)
|
|
|
|
|
|
_CHAPTER_NUM_RE = re.compile(r"Chapter(\d+)")
|
|
_BODY_RE = re.compile(r"<body[^>]*>(.*)</body>", re.IGNORECASE | re.DOTALL)
|
|
|
|
|
|
def _body_inner(text: str) -> str:
|
|
m = _BODY_RE.search(text)
|
|
return m.group(1) if m else text
|
|
|
|
|
|
def _chapter_groups(names: list[str], root: str = "OPS/") -> list[list[str]]:
|
|
"""Group a book's chapter body files by chapter number, in book
|
|
order. A chapter that's been split across several xhtml files
|
|
(``Chapter10.xhtml`` + ``Chapter10a.xhtml`` in 2019,
|
|
``Chapter02.xhtml`` + ``Chapter02_01.xhtml`` in 2021/2022/2024) is
|
|
one group — some continuation files (2019's, mostly) start with no
|
|
heading of their own at all, not even a running-header repeat, so
|
|
without grouping their leading codes would fall through to
|
|
sec_id="". Concatenating a chapter's files into one document before
|
|
calling ``parse_xhtml`` lets the same heading_stack/`current`
|
|
machinery carry the last open heading across the file boundary,
|
|
with no change to ``_group`` itself.
|
|
|
|
Groups are returned sorted by chapter *number*, not by
|
|
``names``/``zf.namelist()`` order — the zip's own member order is
|
|
not book order in every edition (verified: 2019's happens to match,
|
|
2021/2022/2024's don't, entries interleaved arbitrarily). This
|
|
didn't matter before C10 (each group was parsed independently), but
|
|
C10 threads the active chapter title from one group's call into the
|
|
next's, so processing them out of book order would parent headings
|
|
under the wrong chapter entirely."""
|
|
order: list[str] = []
|
|
grouped: dict[str, list[str]] = {}
|
|
for name in names:
|
|
base = name.rsplit("/", 1)[-1]
|
|
if not (name.startswith(root) and base.endswith(".xhtml")):
|
|
continue
|
|
if "Chapter" not in base or "_Toc" in base:
|
|
continue
|
|
m = _CHAPTER_NUM_RE.search(base)
|
|
key = m.group(1) if m else base
|
|
if key not in grouped:
|
|
grouped[key] = []
|
|
order.append(key)
|
|
grouped[key].append(name)
|
|
|
|
def _chapter_sort_key(k: str) -> tuple[int, str]:
|
|
try:
|
|
return (int(k), "")
|
|
except ValueError:
|
|
return (10**9, k) # no Chapter<N> match at all: keep, sort last
|
|
|
|
return [sorted(grouped[key]) for key in sorted(order, key=_chapter_sort_key)]
|
|
|
|
|
|
def _has_range(sec: CptSection) -> bool:
|
|
return bool(sec.code_lo and sec.code_hi)
|
|
|
|
|
|
def _range_key(v: str) -> tuple[int, str]:
|
|
"""Numeric sort key for a 5-char CPT code: the 4-digit numeric
|
|
prefix as an int, then the whole string as a tiebreaker. The prefix
|
|
alone already orders Category II/III (``NNNNF``/``NNNNT``) codes
|
|
correctly — the trailing letter is constant across one range, so it
|
|
never needs to enter the comparison."""
|
|
digits = v[:4]
|
|
if digits.isdigit():
|
|
return (int(digits), v)
|
|
return (-1, v) # malformed guard: sorts first, never matches a real range
|
|
|
|
|
|
def _in_range(code: str, lo: str, hi: str) -> bool:
|
|
"""Numeric containment check (C10). A resequenced heading's own TOC
|
|
span can print in the book's *display* order rather than numeric
|
|
order — e.g. "Qualifying Circumstances* (99490-99427)" — so ``lo``
|
|
is not guaranteed to be the smaller endpoint; this always resolves
|
|
the true numeric ``min``/``max`` of the pair before comparing,
|
|
rather than assuming ``lo <= hi``."""
|
|
if not lo or not hi:
|
|
return False
|
|
lo_key, hi_key = sorted((_range_key(lo), _range_key(hi)))
|
|
return lo_key <= _range_key(code) <= hi_key
|
|
|
|
|
|
def _is_guideline_path(path: tuple[str, ...]) -> bool:
|
|
return any(p.strip().lower().endswith("guidelines") for p in path)
|
|
|
|
|
|
def _resolve_alternates(
|
|
sections: list[CptSection],
|
|
codes: list[CptCode],
|
|
references: list[CptReference],
|
|
) -> tuple[list[CptCode], list[CptAlternate]]:
|
|
"""C9: some codes print twice with a full, real-looking descriptor
|
|
both times — a guideline "Unlisted Service or Procedure" summary
|
|
table, a "Qualifying Circumstances for Anesthesia" cross-reference
|
|
reprint, etc. — structurally distinct from C8's placeholder rows
|
|
(already dropped upstream, never reaching here: both entries seen
|
|
here carry real text). For each code with more than one ``CptCode``
|
|
entry, score every entry and keep the highest (ties: first in
|
|
document order) as canonical; the rest become ``CptAlternate`` rows
|
|
instead of silently vanishing or duplicating. A code with only one
|
|
entry — even one that lives entirely inside a guideline section —
|
|
is untouched and always canonical."""
|
|
sections_by_id = {s.sec_id: s for s in sections}
|
|
sections_by_path: dict[tuple[str, ...], list[CptSection]] = {}
|
|
for s in sections:
|
|
sections_by_path.setdefault(s.path, []).append(s)
|
|
codes_with_references = {r.code for r in references}
|
|
|
|
def resolved_range_section(sec: CptSection) -> CptSection | None:
|
|
if _has_range(sec):
|
|
return sec
|
|
for k in range(len(sec.path) - 1, 0, -1):
|
|
for cand in sections_by_path.get(sec.path[:k], ()):
|
|
if _has_range(cand):
|
|
return cand
|
|
return None
|
|
|
|
def score(entry: CptCode) -> int:
|
|
s = 0
|
|
sec = sections_by_id.get(entry.sec_id)
|
|
if sec is not None:
|
|
resolved = resolved_range_section(sec)
|
|
if resolved is not None:
|
|
s += 1
|
|
if _in_range(entry.code, resolved.code_lo, resolved.code_hi):
|
|
s += 2
|
|
if _is_guideline_path(sec.path):
|
|
s -= 3
|
|
if entry.elements or entry.code in codes_with_references:
|
|
s += 1
|
|
return s
|
|
|
|
by_code_idx: dict[str, list[int]] = {}
|
|
for i, c in enumerate(codes):
|
|
by_code_idx.setdefault(c.code, []).append(i)
|
|
|
|
drop: set[int] = set()
|
|
alternates: list[CptAlternate] = []
|
|
for code, idxs in by_code_idx.items():
|
|
if len(idxs) < 2:
|
|
continue
|
|
scored = [(score(codes[i]), i) for i in idxs]
|
|
best_score = max(s for s, _ in scored)
|
|
best_i = next(i for s, i in scored if s == best_score)
|
|
for _s, i in scored:
|
|
if i == best_i:
|
|
continue
|
|
drop.add(i)
|
|
entry = codes[i]
|
|
sec = sections_by_id.get(entry.sec_id)
|
|
reason = (
|
|
"guidelines-reprint"
|
|
if sec is not None and _is_guideline_path(sec.path)
|
|
else "lower-score"
|
|
)
|
|
alternates.append(
|
|
CptAlternate(code=code, sec_id=entry.sec_id, reason=reason)
|
|
)
|
|
|
|
canonical = [c for i, c in enumerate(codes) if i not in drop]
|
|
return canonical, alternates
|
|
|
|
|
|
def _dedupe_instructions(instructions: list[CptInstruction]) -> list[CptInstruction]:
|
|
"""Collapse exact-duplicate ``(code, kind, text)`` rows. A losing
|
|
guideline-reprint entry (C9/C10) and its real counterpart often
|
|
print the identical parenthetical note verbatim; there's no
|
|
per-entry ``sec_id`` on ``CptInstruction`` to attribute one row to
|
|
one specific ``CptCode`` occurrence, so this is the loser's
|
|
instructions being dropped in the one case that's unambiguous — an
|
|
exact repeat — leaving any of its instructions with genuinely
|
|
distinct text untouched."""
|
|
seen: set[tuple[str, str, str]] = set()
|
|
out = []
|
|
for i in instructions:
|
|
key = (i.code, i.kind, i.text)
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
out.append(i)
|
|
return out
|
|
|
|
|
|
def _dedupe_references(references: list[CptReference]) -> list[CptReference]:
|
|
"""Collapse exact-duplicate ``(code, kind, text)`` rows — see
|
|
``_dedupe_instructions``."""
|
|
seen: set[tuple[str, str, str]] = set()
|
|
out = []
|
|
for r in references:
|
|
key = (r.code, r.kind, r.text)
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
out.append(r)
|
|
return out
|
|
|
|
|
|
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 (in true chapter-number order, per
|
|
``_chapter_groups`` — the epub's own zip member order is *not* book
|
|
order in every edition) and every appendix file (D/E/F/G/K/M/N/P/T).
|
|
|
|
Accepted C9/C10 outcome, documented rather than "fixed" further: a
|
|
code that's genuinely cross-referenced at two different numeric
|
|
homes in the book resolves to whichever entry carries the TOC's own
|
|
declared numeric range, even when that isn't the chapter a reader
|
|
would expect. Example: 99100-99140 ("Qualifying Circumstances")
|
|
print both under Anesthesia's own guideline text *and* under
|
|
Medicine's "Qualifying Circumstances for Anesthesia" heading — only
|
|
the Medicine heading carries a TOC-declared range (99100-99140), so
|
|
it wins as canonical and the Anesthesia Guidelines copy becomes the
|
|
``CptAlternate``. This is the book's own numbering, not a parser
|
|
quirk — see ``TestReal2024.test_c10_99100_resolves_to_the_ranged_medicine_listing``
|
|
in tests/pfs/test_cpt_epub.py."""
|
|
path = Path(path)
|
|
if year is None:
|
|
year = _infer_year(path)
|
|
|
|
with zipfile.ZipFile(path) as zf:
|
|
names = zf.namelist()
|
|
# Resolved once via container.xml -> the OPF's own path, not
|
|
# hard-coded — falls back to "OPS/" (every edition checked so
|
|
# far) when container.xml is missing/malformed (F2).
|
|
# Lazy: importing rex.comments pulls in the rex package (fsspec),
|
|
# which the api image's `cli` extra does not carry — the CLI must
|
|
# import without it (tests/cli/test_cli_import_without_optional_deps.py).
|
|
from rex.comments.epub_text import content_root
|
|
|
|
root = content_root(zf)
|
|
|
|
toc_ranges: dict[str, tuple[str, str]] = {}
|
|
for name in names:
|
|
base = name.rsplit("/", 1)[-1]
|
|
if name.startswith(root) 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] = []
|
|
chapter_groups = _chapter_groups(names, root)
|
|
if not chapter_groups:
|
|
# F4: an empty parse must never reach write_cpt_edition's
|
|
# delete-then-insert — that would wipe an edition already on
|
|
# file for nothing. A book with genuinely no chapter body
|
|
# files (wrong root resolved, or a malformed/non-CPT EPUB)
|
|
# is a real failure, not a silent zero-row edition.
|
|
raise ValueError(f"no chapter content found in {path}")
|
|
# C10: most chapter-groups carry no div.ch-title of their own (a
|
|
# chapter split across many numbered files for size) — thread
|
|
# the active chapter title from one group's call into the
|
|
# next's so it keeps parenting headings until a real ch-title
|
|
# replaces it. Calls _group directly (not parse_xhtml, whose
|
|
# public 4-tuple return drops this) to carry it.
|
|
chapter_title = ""
|
|
for group in chapter_groups:
|
|
bodies = [
|
|
_body_inner(zf.read(name).decode("utf-8", errors="replace"))
|
|
for name in group
|
|
]
|
|
combined = "<html><body>" + "\n".join(bodies) + "</body></html>"
|
|
source = group[0].rsplit("/", 1)[-1]
|
|
tok = _Tokenizer()
|
|
tok.feed(combined)
|
|
tok.close()
|
|
s, c, i, r, chapter_title = _group(
|
|
tok.events, year=year, source=source, chapter_title=chapter_title
|
|
)
|
|
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(root) 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))
|
|
|
|
# C9/C10: resolve any code that printed more than once (real
|
|
# descriptor both times — never the C8 placeholder, already dropped)
|
|
# down to one canonical CptCode, keeping the rest as CptAlternate
|
|
# rows. Dedupe (code, kind, text) first (minor #4) so an exact-repeat
|
|
# instruction/reference from a losing entry doesn't survive as a
|
|
# spurious duplicate row alongside the real entry's own copy.
|
|
instructions = _dedupe_instructions(instructions)
|
|
references = _dedupe_references(references)
|
|
canonical_codes, alternates = _resolve_alternates(fixed_sections, codes, references)
|
|
|
|
return CptEdition(
|
|
year=year,
|
|
sections=tuple(fixed_sections),
|
|
codes=tuple(canonical_codes),
|
|
instructions=tuple(instructions),
|
|
references=tuple(references),
|
|
crosswalks=tuple(crosswalks),
|
|
lists=tuple(lists),
|
|
alternates=tuple(alternates),
|
|
)
|