fix(pfs): FR lineage — same-block range prefilter + letter ranges, directional replaces/replaced_by, no created-on-established (refs #686)

src/pfs/lineage.py

@@ -96,7 +96,7 @@ def _patterns(code: str) -> list[tuple[str, re.Pattern[str], str]]:
         (
             "created",
             re.compile(
-                rf"(?:\b(?:creat|establish)\w*\b.{{0,80}}\b(?:HCPCS|CPT|G-?)\s*codes?\b[^.]{{0,80}}{c})"
+                rf"(?:\bcreat\w*\b.{{0,80}}\b(?:HCPCS|CPT|G-?)\s*codes?\b[^.]{{0,80}}{c})"
                 rf"|(?:new (?:HCPCS|CPT) code\s+{c})",
                 re.I,
             ),
@@ -110,7 +110,8 @@ def _patterns(code: str) -> list[tuple[str, re.Pattern[str], str]]:
         (
             "replaces",
             re.compile(
-                rf"{c}[^.]{{0,120}}\b(?:replace|replacing|replacement for|instead of|with a descriptor identical to)\b",
+                rf"{c}(?:(?!{_CODE})[^.]){{0,120}}"
+                rf"\b(?:replace|replacing|replacement for|instead of|with a descriptor identical to)\b",
                 re.I,
             ),
             "from",
@@ -118,7 +119,7 @@ def _patterns(code: str) -> list[tuple[str, re.Pattern[str], str]]:
         (
             "replaced_by",
             re.compile(
-                rf"(?:\b(?:replace\w*|replacement)\b[^.]{{0,120}}{c})"
+                rf"(?:\b(?:replace\w*|replacement)\b(?:(?!{_CODE})[^.]){{0,120}}{c})"
                 rf"|(?:{c}[^.]{{0,80}}\b(?:as|is being|will be|which we are) (?:finaliz|replac)\w+ (?:in this rule )?as new (?:CPT|HCPCS) code\s+{_CODE})",
                 re.I,
             ),
@@ -158,10 +159,12 @@ def _patterns(code: str) -> list[tuple[str, re.Pattern[str], str]]:
 
 
 _RE_RANGE = re.compile(r"\b(\d{5})-(\d{5})\b")
+_RE_LETTER_RANGE = re.compile(r"\b([A-Z])(\d{4})-\1(\d{4})\b")
 
 
 def _expand_ranges(text: str) -> str:
-    """ "99441-99443" -> "99441 99442 99443" so a range mentions each code."""
+    """ "99441-99443" -> "99441 99442 99443" and "G0008-G0010" -> "G0008
+    G0009 G0010" so a range mentions each code."""
 
     def rep(m: re.Match[str]) -> str:
         a, b = int(m.group(1)), int(m.group(2))
@@ -169,7 +172,18 @@ def _expand_ranges(text: str) -> str:
             " ".join(str(n) for n in range(a, b + 1)) if 0 < b - a <= 20 else m.group(0)
         )
 
-    return _RE_RANGE.sub(rep, text)
+    def rep_letter(m: re.Match[str]) -> str:
+        letter = m.group(1)
+        a, b = int(m.group(2)), int(m.group(3))
+        return (
+            " ".join(f"{letter}{n:04d}" for n in range(a, b + 1))
+            if 0 < b - a <= 20
+            else m.group(0)
+        )
+
+    text = _RE_RANGE.sub(rep, text)
+    text = _RE_LETTER_RANGE.sub(rep_letter, text)
+    return text
 
 
 def fr_events(store: Any, code: str) -> list[EventRow]:
@@ -178,8 +192,9 @@ def fr_events(store: Any, code: str) -> list[EventRow]:
     rows = con.execute(
         "SELECT a.item_key, a.p_id, a.page, a.text, i.title, i.date_published "
         "FROM fr_anchors a JOIN items i ON i.key = a.item_key WHERE a.text LIKE ? OR a.text LIKE ?",
-        (f"%{code}%", f"%{code[:5]}-%"),
+        (f"%{code}%", f"%{code[:2]}___-{code[:2]}___%"),
     ).fetchall()
+    patterns = _patterns(code)
     seen: set[tuple[int, str, str, int]] = set()
     out: list[EventRow] = []
     for r in rows:
@@ -188,15 +203,25 @@ def fr_events(store: Any, code: str) -> list[EventRow]:
             continue
         year = rule_year_of(r["title"], r["date_published"])
         others = tuple(c for c in find_codes(text) if c != code)
-        for kind, pat, direction in _patterns(code):
+        for kind, pat, direction in patterns:
             m = pat.search(text)
             if not m:
                 continue
-            frm = " ".join(others) if direction == "from" else ""
+            frm = ""
             to = ""
-            if direction == "to":
+            if direction == "from":
+                # "replaces": the replaced code(s) are named after the
+                # trigger phrase, up to the end of the sentence — e.g.
+                # "...99439 (with a descriptor identical to G2058)."
+                tail_text = text[m.end() :]
+                period = tail_text.find(".")
+                if period != -1:
+                    tail_text = tail_text[:period]
+                tail_codes = tuple(tc for tc in find_codes(tail_text) if tc != code)
+                frm = " ".join(tail_codes) or " ".join(others)
+            elif direction == "to":
                 tail = find_codes(text[m.start() : m.end()])
-                to = " ".join(c for c in tail if c != code) or " ".join(others)
+                to = " ".join(tc for tc in tail if tc != code) or " ".join(others)
             # A directional kind ("replaces"/"replaced_by") with no other
             # code found nearby is not a real relationship — e.g. "...
             # instead of the G code" names no actual code, so the sentence

tests/pfs/test_lineage.py

@@ -7,7 +7,7 @@ import sqlite3
 import duckdb
 import pytest
 
-from pfs.lineage import fr_events, lineage, rvu_events, rvu_year_span
+from pfs.lineage import _expand_ranges, fr_events, lineage, rvu_events, rvu_year_span
 
 RVU_COLS = "hcpcs VARCHAR, mod VARCHAR, description VARCHAR, status_code VARCHAR, non_fac_total DOUBLE, year INTEGER"
 
@@ -145,6 +145,12 @@ def store():
             61652,
             "The CPT Editorial Panel also deleted three codes (99441-99443) for reporting telephone E/M services. We note that CPT codes 99441, 99442, and 99443, each are assigned provisional status on the Medicare telehealth services list, and would return to bundled status when the telehealth flexibilities expire.",
         ),
+        (
+            "XFGGRBDH",
+            512,
+            61700,
+            "The CPT Editorial Panel also deleted three codes (90867-90869) for reporting transcranial magnetic stimulation treatment.",
+        ),
     ]
     s.con.executemany(
         "INSERT INTO fr_anchors VALUES (?,?,?,?,?)",
@@ -161,17 +167,21 @@ class TestFrEvents:
             (2015, "adopted_cpt", "DE2VH9PD", 1251)
         ]
         assert ev[0].anchored is True and ev[0].source == "fr" and ev[0].page == 67716
+        assert not any(e.kind == "replaces" for e in ev)
 
     def test_replaced_by_and_crosswalk(self, store):
         ev = fr_events(store, "G2058")
-        kinds = {(e.year, e.kind, e.to_codes) for e in ev}
+        kinds = {e.kind for e in ev}
-        assert (2021, "replaced_by", "99439") in kinds
+        assert "created" not in kinds
-        assert (2021, "crosswalk", "") in kinds or any(
+        assert "replaces" not in kinds
-            e.kind == "crosswalk" for e in ev
+        assert "replaced_by" in kinds and "crosswalk" in kinds
-        )
+        to_kinds = {(e.year, e.kind, e.to_codes) for e in ev}
+        assert (2021, "replaced_by", "99439") in to_kinds
 
     def test_replaces_from_the_successor_side(self, store):
         ev = fr_events(store, "99439")
+        kinds = {(e.year, e.kind) for e in ev}
+        assert (2021, "created") in kinds  # ¶686 "new CPT code 99439"
         rep = [e for e in ev if e.kind == "replaces"]
         assert rep and rep[0].from_codes == "G2058"
 
@@ -179,6 +189,18 @@ class TestFrEvents:
         kinds = {e.kind for e in fr_events(store, "99441")}
         assert {"deleted", "bundled", "telehealth_list"} <= kinds
 
+    def test_deleted_via_range_prefilter_middle_member(self, store):
+        # 90868 is a middle member of a "90867-90869" range and appears
+        # nowhere else in the paragraph literally — the SQL prefilter must
+        # still surface the row for a range-block LIKE match.
+        ev = fr_events(store, "90868")
+        assert any(e.kind == "deleted" for e in ev)
+
+
+class TestExpandRanges:
+    def test_letter_range(self):
+        assert _expand_ranges("G0008-G0010") == "G0008 G0009 G0010"
+
 
 class TestCrossCheck:
     def test_rvu_events_anchored_by_nearby_fr_event(self, con, store):