fix(pfs): CPT parser — semicolon-rule descriptors, owner excluded from instruction targets, synthetic sec_id for id-less headings (refs #687)

Claude-Session: https://claude.ai/code/session_01Aum3pEMAM3yQVdFSdVe6Gc

src/pfs/cpt_epub.py

@@ -72,7 +72,11 @@ _GLYPHS = {
 _AUDIO_ONLY_IMGS = {"enta.png", "entag.png"}
 _PLA_IMGS = {"ent.png", "entg.png"}
 
-_HEADING_RE = re.compile(r"^h([1-5])a?$")
+# "-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])(?:[_.]|$)")
@@ -269,7 +273,7 @@ def _category(code: str) -> str:
 
 
 def _group(
-    events: list[tuple], *, year: int
+    events: list[tuple], *, year: int, source: str = ""
 ) -> tuple[list[CptSection], list[CptCode], list[CptInstruction], list[CptReference]]:
     sections: list[CptSection] = []
     codes: list[CptCode] = []
@@ -284,9 +288,11 @@ def _group(
     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
+        nonlocal entry, last_primary, last_primary_semicolon
         if entry is None:
             return
         code = entry["code"]
@@ -302,20 +308,48 @@ def _group(
         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:
-            stem = prose[0] if prose 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 = prose[1:]
+            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)
                 )
@@ -324,8 +358,17 @@ def _group(
             else:
                 tail = f"{tail} {block}".strip()
 
-        parts = [p for p in (stem, "; ".join(elements), tail) if p]
+        # Each element line carries its own trailing list punctuation
-        descriptor = " ".join(parts).strip()
+        # (",", ";") 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()
 
         for ref_text in entry["refs"]:
             low = ref_text.lower()
@@ -365,6 +408,7 @@ def _group(
         codes.append(cpt_code)
         if not is_sub:
             last_primary = cpt_code
+            last_primary_semicolon = semicolon_style
         entry = None
 
     def finalize_section() -> None:
@@ -391,6 +435,7 @@ def _group(
             if ev[1] == "start":
                 in_guideline_zone = False
                 last_primary = None
+                last_primary_semicolon = False
             continue
 
         _, classes, id_, text, imgs = ev
@@ -398,19 +443,35 @@ def _group(
         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):
+        is_heading = any(_HEADING_RE.match(c) for c in classes)
+        # A few chapters (e.g. "Category III Codes") never carry a
+        # div.hN at all — the chapter's own div.ch-title is the only
+        # heading before the code table. Treat it as a level-1 heading
+        # too, so its codes land under a real, titled section instead
+        # of collapsing onto sec_id="" (it's superseded immediately,
+        # same level, the moment a real h1 shows up in chapters that
+        # have both).
+        is_chapter_title = "ch-title" in classes
+        if is_heading or is_chapter_title:
             finalize_entry()
             finalize_section()
+            if is_heading:
                 level = int(
                     next(m for c in classes if (m := _HEADING_RE.match(c))).group(1)
                 )
+            else:
+                level = 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)
+            sec_id = id_
+            if not sec_id:
+                heading_counter += 1
+                sec_id = f"{source}:h{heading_counter}"
             current = {
-                "sec_id": id_ or "",
+                "sec_id": sec_id,
                 "level": level,
                 "title": title,
                 "path": path,
@@ -418,6 +479,7 @@ def _group(
             guideline_buf = []
             in_guideline_zone = True
             last_primary = None
+            last_primary_semicolon = False
             continue
 
         if "noindent" in classes:
@@ -478,15 +540,20 @@ def _group(
 
 
 def parse_xhtml(
-    text: str, *, year: int
+    text: str, *, year: int, source: 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."""
+    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``."""
     tok = _Tokenizer()
     tok.feed(text)
     tok.close()
-    return _group(tok.events, year=year)
+    return _group(tok.events, year=year, source=source)
 
 
 # --- Appendices -------------------------------------------------------
@@ -555,6 +622,44 @@ def _infer_year(path: Path) -> int:
     )
 
 
+_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]) -> 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."""
+    order: list[str] = []
+    grouped: dict[str, list[str]] = {}
+    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
+        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)
+    return [sorted(grouped[key]) for key in order]
+
+
 def parse_epub(path: Path, *, year: int | None = None) -> CptEdition:
     """Parse one CPT Professional edition EPUB into a ``CptEdition``.
 
@@ -584,14 +689,14 @@ def parse_epub(path: Path, *, year: int | None = None) -> CptEdition:
         codes: list[CptCode] = []
         instructions: list[CptInstruction] = []
         references: list[CptReference] = []
-        for name in names:
+        for group in _chapter_groups(names):
-            base = name.rsplit("/", 1)[-1]
+            bodies = [
-            if not (name.startswith("OPS/") and base.endswith(".xhtml")):
+                _body_inner(zf.read(name).decode("utf-8", errors="replace"))
-                continue
+                for name in group
-            if "Chapter" not in base or "_Toc" in base:
+            ]
-                continue
+            combined = "<html><body>" + "\n".join(bodies) + "</body></html>"
-            text = zf.read(name).decode("utf-8", errors="replace")
+            source = group[0].rsplit("/", 1)[-1]
-            s, c, i, r = parse_xhtml(text, year=year)
+            s, c, i, r = parse_xhtml(combined, year=year, source=source)
             sections.extend(s)
             codes.extend(c)
             instructions.extend(i)

tests/pfs/fixtures/cpt_sample.xhtml

@@ -31,7 +31,17 @@
 <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>
+<div class="table-para2">(Sample wellness services of less than 20 minutes duration are not reported separately)</div>
+<div class="table-para2">(Do not report 55001 more than twice per calendar month)</div>
+<div class="table-para2">(Do not report 55001 in addition to 55002 more than once per calendar month)</div></td>
+</tr>
+<tr>
+<td class="td-w1" id="code_56000"><div class="table-para"><b>56000</b></div></td>
+<td class="td"><div class="table-para1">Sample incision, deep fascial plane; with exploratory biopsy</div></td>
+</tr>
+<tr>
+<td class="td-w1" id="code_56005"><div class="table-para"><b>56005</b></div></td>
+<td class="td"><div class="table-para1-sub">with synovial debridement</div></td>
 </tr>
 </tbody>
 </table>
@@ -39,5 +49,16 @@
 <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>
+
+<div class="h1">No Id Category Services</div>
+<div class="noindent">Sample guideline text for a section whose heading carries no id attribute at all, used to test the synthetic sec_id fallback.</div>
+<table class="table1">
+<tbody>
+<tr>
+<td class="td-w1" id="code_57000"><div class="table-para"><b>57000</b></div></td>
+<td class="td"><div class="table-para1">Sample invented service with no indentation family.</div></td>
+</tr>
+</tbody>
+</table>
 </body>
 </html>

tests/pfs/test_cpt_epub.py

@@ -33,7 +33,9 @@ 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)
+    sections, codes, instructions, references = parse_xhtml(
+        text, year=2024, source="cpt_sample.xhtml"
+    )
     return {
         "sections": {s.sec_id: s for s in sections},
         "sections_list": sections,
@@ -45,13 +47,19 @@ def parsed():
 
 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"])
+        levels = sorted(
+            (s.sec_id, s.level)
+            for s in parsed["sections_list"]
+            if s.sec_id in {"sec_1", "sec_2", "sec_3", "sec_4"}
+        )
         assert levels == [
             ("sec_1", 1),
             ("sec_2", 2),
             ("sec_3", 1),
             ("sec_4", 2),
         ]
+        # plus the id-less "No Id Category Services" h1 (TestSecIdFallback)
+        assert len(parsed["sections_list"]) == 5
 
     def test_paths_include_ancestors(self, parsed):
         sec2 = parsed["sections"]["sec_2"]
@@ -88,6 +96,38 @@ class TestSections:
         assert "  " not in sec1.guideline
 
 
+class TestSecIdFallback:
+    """ "No Id Category Services" (the fixture's last h1) has no ``id``
+    attribute and no nested anchor carrying one — same as most 2019
+    headings. It must still get a unique, non-empty ``sec_id`` (never
+    collapse onto ``""`` alongside every other id-less heading), and
+    the code under it must carry that same id."""
+
+    def test_no_id_heading_gets_a_synthetic_sec_id(self, parsed):
+        no_id_section = next(
+            s for s in parsed["sections_list"] if s.title == "No Id Category Services"
+        )
+        assert no_id_section.sec_id != ""
+        assert no_id_section.sec_id.startswith("cpt_sample.xhtml:h")
+
+    def test_synthetic_sec_ids_are_unique(self, parsed):
+        synthetic = [
+            s.sec_id
+            for s in parsed["sections_list"]
+            if s.sec_id.startswith("cpt_sample.xhtml:h")
+        ]
+        assert len(synthetic) == len(set(synthetic))
+        assert synthetic  # at least the one id-less heading produced one
+
+    def test_code_under_id_less_heading_gets_the_synthetic_sec_id(self, parsed):
+        no_id_section = next(
+            s for s in parsed["sections_list"] if s.title == "No Id Category Services"
+        )
+        code = parsed["codes"]["57000"]
+        assert code.sec_id == no_id_section.sec_id
+        assert code.sec_id != ""
+
+
 class TestPrimaryCode:
     def test_three_elements_and_tail(self, parsed):
         code = parsed["codes"]["55000"]
@@ -118,7 +158,47 @@ class TestPrimaryCode:
         code = parsed["codes"]["55000"]
         assert code.stem in code.descriptor
         assert code.tail in code.descriptor
-        assert code.elements[0] in code.descriptor
+        assert code.elements[0].rstrip(",;") in code.descriptor
+
+    def test_descriptor_element_join_has_no_double_punctuation(self, parsed):
+        code = parsed["codes"]["55000"]
+        assert ",;" not in code.descriptor
+        assert ";;" not in code.descriptor
+
+
+class TestSemicolonRule:
+    """25100 "Arthrotomy, wrist joint; with biopsy" / 25105 "with
+    synovectomy" is the book's own worked example of the indentation
+    convention (Introduction, "Format of the Terminology"): an
+    indented child's descriptor is the parent's pre-semicolon stem
+    plus the child's own fragment — never the parent's *whole* stem
+    (which would duplicate the parent's own post-semicolon text)."""
+
+    def test_parent_stem_excludes_own_continuation(self, parsed):
+        parent = parsed["codes"]["56000"]
+        assert parent.stem == "Sample incision, deep fascial plane"
+        assert "with exploratory biopsy" not in parent.stem
+
+    def test_parent_descriptor_has_the_continuation(self, parsed):
+        parent = parsed["codes"]["56000"]
+        assert (
+            parent.descriptor
+            == "Sample incision, deep fascial plane; with exploratory biopsy"
+        )
+
+    def test_child_inherits_only_the_shared_stem(self, parsed):
+        child = parsed["codes"]["56005"]
+        assert child.stem == "Sample incision, deep fascial plane"
+        assert child.parent == "56000"
+
+    def test_child_descriptor_substitutes_its_own_fragment(self, parsed):
+        child = parsed["codes"]["56005"]
+        assert (
+            child.descriptor
+            == "Sample incision, deep fascial plane; with synovial debridement"
+        )
+        # the parent's own fragment must not leak into the child's descriptor
+        assert "exploratory biopsy" not in child.descriptor
 
 
 class TestAddonCode:
@@ -168,12 +248,46 @@ class TestInstructions:
         others = [
             i for i in parsed["instructions"] if i.code == "55001" and i.kind == "other"
         ]
-        assert len(others) == 1
+        assert len(others) == 3
         assert others[0].text.startswith("(Sample wellness services")
 
-    def test_three_instructions_on_addon(self, parsed):
+    def test_five_instructions_on_addon(self, parsed):
         addon_instructions = [i for i in parsed["instructions"] if i.code == "55001"]
-        assert len(addon_instructions) == 3
+        assert len(addon_instructions) == 5
+
+    def test_owner_code_excluded_from_targets(self, parsed):
+        # "(Do not report 55001 more than twice per calendar month)" —
+        # 55001 is the instruction's own owning code (a subject), not
+        # a target of itself.
+        matches = [
+            i
+            for i in parsed["instructions"]
+            if i.code == "55001" and "more than twice" in i.text
+        ]
+        assert len(matches) == 1
+        assert matches[0].kind == "other"
+        assert matches[0].targets == ()
+
+    def test_sibling_code_kept_as_target(self, parsed):
+        # "(Do not report 55001 in addition to 55002 more than once
+        # per calendar month)" — 55001 is the owner (excluded), 55002
+        # is a genuine sibling code and must survive the owner filter.
+        matches = [
+            i
+            for i in parsed["instructions"]
+            if i.code == "55001" and "in addition to 55002" in i.text
+        ]
+        assert len(matches) == 1
+        assert matches[0].targets == ("55002",)
+
+    def test_not_with_never_includes_the_owner(self, parsed):
+        not_with = [
+            i
+            for i in parsed["instructions"]
+            if i.code == "55001" and i.kind == "not-with"
+        ]
+        assert "55001" not in not_with[0].targets
+        assert "55555" in not_with[0].targets
 
 
 class TestReferences:
@@ -252,6 +366,24 @@ class TestReal2024:
         assert edition.crosswalks
         assert any(row.former_code.endswith("T") for row in edition.crosswalks)
 
+    def test_25100_25105_semicolon_rule(self, edition):
+        # The book's own worked example (Introduction, "Format of the
+        # Terminology"): 25100 "Arthrotomy, wrist joint; with biopsy"
+        # / 25105 "with synovectomy". Structure only — never asserts
+        # or prints the actual descriptor text.
+        by_code = {c.code: c for c in edition.codes}
+        parent = by_code["25100"]
+        child = by_code["25105"]
+        assert child.parent == "25100"
+        assert child.stem == parent.stem
+        assert child.descriptor.startswith(parent.stem)
+        parent_own_fragment = parent.descriptor[len(parent.stem) :].lstrip("; ").strip()
+        assert parent_own_fragment  # the parent really did have a post-semicolon part
+        assert parent_own_fragment not in child.descriptor
+
+    def test_no_code_has_an_empty_sec_id(self, edition):
+        assert all(c.sec_id != "" for c in edition.codes)
+
 
 @pytest.mark.skipif(not EPUB_2022.exists(), reason="real 2022 CPT EPUB not on disk")
 def test_real_2022_parses():
@@ -272,3 +404,6 @@ def test_real_2019_parses_without_code_ids():
     edition = parse_epub(EPUB_2019, year=2019)
     assert len(edition.codes) >= 3000
     assert edition.sections
+    # 2019 headings mostly lack an id in the source markup — every
+    # code must still land under a non-empty (real or synthetic) sec_id.
+    assert all(c.sec_id != "" for c in edition.codes)