From 6bff63b49981a9c5c7ff3821849af7fdbc530d68 Mon Sep 17 00:00:00 2001 From: kert Date: Fri, 11 Sep 2026 17:51:20 -0400 Subject: [PATCH] feat(pfs,llm): resolve MLN products/articles to bib items; locate IOM section pages in chapter PDFs (refs #705) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - pfs.guidance: MLN_RE accepts ICN/MLN product numbers ("ICN MLN909188", "ICN 909289"), MLN_MATTERS_RE captures MM/SE article numbers; mln_refs normalises both to "MLN " / "MLN Matters MM"; resolve_mln finds the bib item whose URL or title carries the identifier (word-bounded). - pfs.guidance.iom_section_page: locates an IOM section heading inside the chapter PDF attached to the resolved item (llm.pages.locate_section — the body heading is the one followed by its "(Rev." line, which the table of contents lacks); cached per item/section and per PDF. - pfs.codetables.GuidanceRow.page (+ column, added in place by ensure_tables; legacy NULL reads 0). - llm.lineage: resolved IOM/MLN guidance rows link to the bib item's URL, with #page=N when the section was located. --- src/llm/lineage.py | 8 ++ src/llm/pages.py | 26 +++++ src/pfs/codetables.py | 14 ++- src/pfs/guidance.py | 163 +++++++++++++++++++++++++++--- tests/llm/test_lineage.py | 54 +++++++++- tests/llm/test_pages.py | 32 ++++++ tests/pfs/test_guidance.py | 197 +++++++++++++++++++++++++++++++++++++ 7 files changed, 474 insertions(+), 20 deletions(-) diff --git a/src/llm/lineage.py b/src/llm/lineage.py index 99978fa..7f53fbe 100644 --- a/src/llm/lineage.py +++ b/src/llm/lineage.py @@ -829,6 +829,14 @@ def _collect_guidance( url = cfrlink.url(cfrlink.parse_cite(r.locator)) except Exception as e: # noqa: BLE001 log.warning("lineage CFR url unresolved (%s): %s", r.locator, e) + elif r.item_key: + # #705: a resolved IOM chapter / MLN product links to the bib + # item's own URL (the CMS PDF), plus ``#page=N`` when the + # guidance build located the section heading in the chapter. + item = _cached_item(store, r.item_key, mtime) + url = (getattr(item, "url", "") or "") if item is not None else "" + if url and r.page: + url = f"{url}#page={r.page}" # IOM section heading, #705 item 2 out.append( GuidanceRef( kind=r.kind, diff --git a/src/llm/pages.py b/src/llm/pages.py index 9b1bba0..860c269 100644 --- a/src/llm/pages.py +++ b/src/llm/pages.py @@ -58,6 +58,32 @@ def locate(pages: list[str], probe: str) -> int: return 0 +#: An IOM chapter heading in the body — "30.6.4 - Evaluation and +#: Management ... (Rev. 12345; Issued: ...)" — is always followed by its +#: revision line; the table of contents on the first pages lists the +#: same "30.6.4 - ..." entry without one, which is how the two are told +#: apart (#705 item 2). +_REV_WINDOW = 240 + + +def locate_section(pages: list[str], section: str) -> int: + """1-based page whose text carries IOM ``section`` ("30.6.4") as a + body heading — the number, a dash, the title, then "(Rev." within + ``_REV_WINDOW`` characters; 0 when no page does (a TOC-only hit is + not a location).""" + section = section.strip().rstrip(".") + if not section: + return 0 + rx = re.compile( + rf"(? list[Chunk]: """Stamp ``attachment`` (+ ``page`` for PDFs) on chunks whose ``section`` names one of ``doc.files``. Identity when ``doc.files`` diff --git a/src/pfs/codetables.py b/src/pfs/codetables.py index b075d30..cb04a76 100644 --- a/src/pfs/codetables.py +++ b/src/pfs/codetables.py @@ -119,7 +119,11 @@ class GuidanceRow: most CFR/IOM citations have no matching library item on hand); ``item_key_src``/``p_id_src``/``page_src`` are the citing paragraph's own provenance (an FR paragraph, or the CPT edition - item with ``p_id_src=0``, ``page_src=0``).""" + item with ``p_id_src=0``, ``page_src=0``); ``page`` is the 1-based + page inside the resolved item's PDF where an IOM section heading + sits (#705 item 2 — ``pfs.guidance.iom_section_page``), ``0`` when + unlocated or not applicable. Trailing default so positional + constructors written before it existed keep working.""" family: str code: str @@ -129,6 +133,7 @@ class GuidanceRow: item_key_src: str p_id_src: int page_src: int + page: int = 0 @dataclass(frozen=True) @@ -261,7 +266,7 @@ CREATE TABLE IF NOT EXISTS pfs.code_family ( item_key VARCHAR, p_id INTEGER, note VARCHAR); CREATE TABLE IF NOT EXISTS pfs.code_guidance ( family VARCHAR, code VARCHAR, kind VARCHAR, locator VARCHAR, item_key VARCHAR, - item_key_src VARCHAR, p_id_src INTEGER, page_src INTEGER); + item_key_src VARCHAR, p_id_src INTEGER, page_src INTEGER, page INTEGER); CREATE TABLE IF NOT EXISTS pfs.code_reaction ( family VARCHAR, period VARCHAR, period_kind VARCHAR, year INTEGER, n_items INTEGER, n_total INTEGER, stance_support INTEGER, stance_oppose INTEGER, @@ -320,6 +325,8 @@ def ensure_tables(con: Any) -> None: con.execute( "ALTER TABLE pfs.code_element_review ADD COLUMN IF NOT EXISTS source VARCHAR" ) + # #705 item 2: the located IOM section page. + con.execute("ALTER TABLE pfs.code_guidance ADD COLUMN IF NOT EXISTS page INTEGER") def _insert(con: Any, table: str, rows: Sequence[Any]) -> int: @@ -429,7 +436,8 @@ def read_guidance(con: Any, family: str) -> list[GuidanceRow]: "SELECT * FROM pfs.code_guidance WHERE family = ? ORDER BY code, kind, locator", [family], ).fetchall() - return [GuidanceRow(*r) for r in rows] + # Rows written before ``page`` existed read back NULL for it. + return [GuidanceRow(*r[:-1], r[-1] or 0) for r in rows] def write_reaction(con: Any, family: str, rows: Sequence[ReactionRow]) -> int: diff --git a/src/pfs/guidance.py b/src/pfs/guidance.py index 6ff3c5d..a9ac53c 100644 --- a/src/pfs/guidance.py +++ b/src/pfs/guidance.py @@ -16,7 +16,7 @@ Two sources, one extraction pass over each hit's text: Both funnel through ``_extract``, which runs three regex families (CFR, IOM, MLN) over one paragraph/guideline's text and resolves each -hit against the bib (``resolve_cfr``/``resolve_iom``) — unresolved +hit against the bib (``resolve_cfr``/``resolve_iom``/``resolve_mln``) — unresolved references keep an empty ``item_key`` (the canonical locator is kept either way, Resolutions §fr_anchors). Rows are written by ``pfs.codetables.write_guidance``/``read_guidance`` into @@ -31,6 +31,7 @@ bare CFR forms firing on the same span) collapses to one. from __future__ import annotations import re +from pathlib import Path from typing import Any, Sequence from bib.cfrlink import canonical, item_for, parse_cite @@ -109,10 +110,38 @@ IOM_RE = re.compile( r"(?:[^.]{0,120}?[Ss]ection\s*([\d.]+))?" ) -#: "MLN907166" / "MLN 907166" booklet numbers (#689: optional/best-effort -#: — no title match against bib items, just the number surfaced as a -#: locator; always unresolved). -MLN_RE = re.compile(r"\bMLN\s*(\d{6,7})\b", re.IGNORECASE) +#: "MLN907166" / "MLN 907166" / "ICN 909289" / "ICN MLN909188" — an MLN +#: product number (booklet, fact sheet, web-based training). CMS renamed +#: the "ICN" (Internet Content Number) prefix to "MLN" in 2020 without +#: renumbering, so both prefixes name the same product series and the +#: stored locator is always the "MLN " form (#705 item 3). +MLN_RE = re.compile(r"\b(?:ICN\s*)?(?:MLN|ICN)\s*(\d{6,7})\b", re.IGNORECASE) +#: "MLN Matters® Number MM9603" / "MLN Matters article SE1316" / "MLN +#: Matters SE 1316" / "MLN Matters article 11268" (a bare number is a +#: change-request article, prefix "MM"). Product numbers (``MLN_RE``) +#: never follow the "Matters" keyword, so the two patterns are disjoint. +MLN_MATTERS_RE = re.compile( + r"\bMLN\s+Matters\W{0,3}(?:(?:article|number|no\.?|#)\s*)*" + r"(MM|SE)?\s?(\d{4,5})\b", + re.IGNORECASE, +) + + +def mln_refs(text: str) -> list[str]: + """Every MLN locator in *text* — ``"MLN 909188"`` for product numbers, + ``"MLN Matters MM9603"`` / ``"MLN Matters SE1316"`` for articles — + in order of appearance, first occurrence wins.""" + out: list[str] = [] + for m in MLN_RE.finditer(text): + loc = f"MLN {m.group(1)}" + if loc not in out: + out.append(loc) + for m in MLN_MATTERS_RE.finditer(text): + prefix = (m.group(1) or "MM").upper() + loc = f"MLN Matters {prefix}{m.group(2)}" + if loc not in out: + out.append(loc) + return out def resolve_cfr(store: Any, title: str, section: str) -> str: @@ -134,6 +163,89 @@ def resolve_iom(store: Any, pub: str, chapter: str) -> str: return "" +def resolve_mln(store: Any, locator: str) -> str: + """The bib item whose URL or title carries the MLN identifier in + *locator* (``"MLN 909188"`` -> ``MLN909188`` or the pre-2020 + ``ICN909188`` spelling; ``"MLN Matters MM9603"`` -> ``MM9603``), or + ``""`` when unresolved. CMS files its products under the number + (``.../MLNProducts/Downloads/eval-mgmt-serv-guide-ICN006764.pdf``, + ``.../MLNMattersArticles/downloads/MM9603.pdf``) and ``bib`` titles + curated booklets with it ("... (MLN909188, June 2025)"), so a + word-bounded match on either column is the whole resolver.""" + tail = locator.split()[-1] if locator else "" + if not tail: + return "" + if tail.isdigit(): + idents = (f"MLN{tail}", f"ICN{tail}") + else: + idents = (tail,) + con = store._con() # noqa: SLF001 — same pattern as ``harvest`` + like = " OR ".join("url LIKE ? OR title LIKE ?" for _ in idents) + params = [p for i in idents for p in (f"%{i}%", f"%{i}%")] + rows = con.execute( + f"SELECT key, title, url FROM items WHERE {like} ORDER BY id", params + ).fetchall() + bounded = [ + re.compile(rf"(? located page; storage path -> normalized page +#: texts. Both live for the process — one ``stack pfs guidance --write`` +#: run locates the same chapter's sections for several families. +_PAGE_CACHE: dict[tuple[str, str], int] = {} +_PDF_CACHE: dict[str, list[str]] = {} + + +def iom_section_page(store: Any, item_key: str, section: str) -> int: + """1-based page of IOM *section* ("30.6.4") inside the chapter PDF + attached to bib item *item_key* (``bib.iom.download_attachments`` + stores one PDF per chapter), or 0 when the item has no PDF on hand, + the section has no body heading, or the PDF toolkit is missing + (#705 item 2). Located at build time on the host so the chat, whose + image has neither the storage tree nor pymupdf, only reads the + stored number.""" + if not item_key or not section: + return 0 + key = (item_key, section) + if key in _PAGE_CACHE: + return _PAGE_CACHE[key] + try: + from llm.pages import locate_section, pdf_pages + except ImportError: # pragma: no cover — llm package always ships with pfs + return 0 + con = store._con() # noqa: SLF001 — same pattern as ``harvest`` + rows = con.execute( + "SELECT a.storage_path FROM attachments a JOIN items i ON i.id = a.item_id " + "WHERE i.key = ? ORDER BY a.id", + (item_key,), + ).fetchall() + page = 0 + for (path,) in rows: + if not path or not path.lower().endswith(".pdf"): + continue + if path not in _PDF_CACHE: + _PDF_CACHE[path] = pdf_pages(Path(path)) + page = locate_section(_PDF_CACHE[path], section) + if page: + break + _PAGE_CACHE[key] = page + return page + + +def _guidance_page(store: Any, kind: str, locator: str, item_key: str) -> int: + """``iom_section_page`` for a resolved IOM locator that names a + section ("100-04 ch.12 §30.6.4"); 0 for everything else.""" + if kind != "iom" or not item_key or "§" not in locator: + return 0 + return iom_section_page(store, item_key, locator.rsplit("§", 1)[1]) + + def _cfr_continuations(text: str, pos: int, pattern: re.Pattern[str]) -> list[str]: """Every list-continuation token starting at *pos* (Ruling A10) — ``pattern`` is ``_CFR_CONT_SECTION_RE`` or ``_CFR_CONT_PART_RE``. @@ -217,8 +329,8 @@ def _extract(text: str, store: Any) -> list[tuple[str, str, str]]: for pub, chapter, section in _iom_refs(text): locator = f"{pub} ch.{chapter}" + (f" §{section}" if section else "") out.append(("iom", locator, resolve_iom(store, pub, chapter))) - for m in MLN_RE.finditer(text): - out.append(("mln", f"MLN {m.group(1)}", "")) + for locator in mln_refs(text): + out.append(("mln", locator, resolve_mln(store, locator))) return out @@ -256,10 +368,19 @@ def harvest(store: Any, codes: Sequence[str], *, family: str) -> list[GuidanceRo if not present: continue for kind, locator, resolved in _extract(text, store): + located = _guidance_page(store, kind, locator, resolved) for code in present: out.append( GuidanceRow( - family, code, kind, locator, resolved, item_key, p_id, page + family, + code, + kind, + locator, + resolved, + item_key, + p_id, + page, + located, ) ) return dedupe(out) @@ -271,8 +392,8 @@ def harvest_cpt( """The newest ingested CPT edition's guideline text and ``see``/ ``other`` instructions for *codes* -> the CFR/IOM/MLN references they cite, anchored to the CPT edition item (``p_id_src=0``, - ``page_src=0`` — no FR paragraph; locating a page inside the PDF - chapter is out of scope, #689). ``[]`` on a replica with no + ``page_src=0`` — no FR paragraph). A resolved IOM section carries + the chapter-PDF page it was located on (``page``, #705 item 2). ``[]`` on a replica with no ``pfs.cpt_*`` tables yet (I4: not a bug, just nothing ingested).""" from pfs.codetables import ( cpt_years, @@ -306,10 +427,19 @@ def harvest_cpt( if section is None or not section.guideline: continue for kind, locator, resolved in _extract(section.guideline, store): + located = _guidance_page(store, kind, locator, resolved) for code in sorted(fam_codes): out.append( GuidanceRow( - family, code, kind, locator, resolved, section.item_key, 0, 0 + family, + code, + kind, + locator, + resolved, + section.item_key, + 0, + 0, + located, ) ) @@ -319,9 +449,18 @@ def harvest_cpt( if not any(k in instr.text for k in ("CFR", "Medicare", "Chapter", "chapter")): continue for kind, locator, resolved in _extract(instr.text, store): + located = _guidance_page(store, kind, locator, resolved) out.append( GuidanceRow( - family, instr.code, kind, locator, resolved, instr.item_key, 0, 0 + family, + instr.code, + kind, + locator, + resolved, + instr.item_key, + 0, + 0, + located, ) ) return dedupe(out) diff --git a/tests/llm/test_lineage.py b/tests/llm/test_lineage.py index 551df3b..4f1fcb3 100644 --- a/tests/llm/test_lineage.py +++ b/tests/llm/test_lineage.py @@ -113,6 +113,12 @@ class _FakeStore: "2026-07-16", "https://www.federalregister.gov/d/2026-2027doc", ), + ( + "PV8APQ4A", + "Chronic Care Management Services (MLN909188, June 2025)", + "2025-06-01", + "https://www.cms.gov/files/document/chroniccaremanagement.pdf", + ), ], ) self.con.executemany( @@ -167,11 +173,13 @@ class _FakeStore: def get(self, key: str) -> SimpleNamespace: row = self.con.execute( - "SELECT title, date_published FROM items WHERE key = ?", (key,) + "SELECT title, date_published, url FROM items WHERE key = ?", (key,) ).fetchone() if row is None: raise KeyError(key) - return SimpleNamespace(title=row["title"], date_published=row["date_published"]) + return SimpleNamespace( + title=row["title"], date_published=row["date_published"], url=row["url"] + ) def close(self): self.con.close() @@ -219,9 +227,11 @@ def _el(code, year, type_, value, item_key="", p_id=0, source="fr") -> ElementRo def _gd( - family, code, kind, locator, item_key_src, p_id_src, item_key="" + family, code, kind, locator, item_key_src, p_id_src, item_key="", page=0 ) -> GuidanceRow: - return GuidanceRow(family, code, kind, locator, item_key, item_key_src, p_id_src, 0) + return GuidanceRow( + family, code, kind, locator, item_key, item_key_src, p_id_src, 0, page + ) class TestKindRank: @@ -636,7 +646,7 @@ class TestGuidance: out = lineage._collect_guidance(c, store, ["APCM"], ["APCM"], 1) assert out == () - def test_iom_mln_have_no_url(self, con, store): + def test_unresolved_iom_mln_have_no_url(self, con, store): c, _ = con write_guidance( c, @@ -647,6 +657,40 @@ class TestGuidance: assert out[0].url == "" assert out[0].label == "CY2021 PFS final 85 FR 84547 ¶686" + def test_resolved_mln_links_to_the_bib_item_url(self, con, store): + # #705 item 3: pfs.guidance.resolve_mln fills item_key; the chat + # link is the booklet PDF itself. #705 item 2: a located IOM + # section adds ``#page=N`` to the chapter PDF's URL. + c, _ = con + write_guidance( + c, + "CCM", + [ + _gd("CCM", "99490", "mln", "MLN 909188", "YBM4IZUS", 686, "PV8APQ4A"), + _gd("CCM", "99490", "iom", "100-04 ch.12", "YBM4IZUS", 686, "GONE1234"), + _gd( + "CCM", + "99490", + "iom", + "100-04 ch.12 §30.6.4", + "YBM4IZUS", + 686, + "PV8APQ4A", + 39, + ), + ], + ) + out = lineage._collect_guidance(c, store, ["CCM"], [], 1) + by = {g.locator: g for g in out} + assert ( + by["MLN 909188"].url + == "https://www.cms.gov/files/document/chroniccaremanagement.pdf" + ) + assert by["100-04 ch.12"].url == "" # item_key set but not in the bib + assert by["100-04 ch.12 §30.6.4"].url == ( + "https://www.cms.gov/files/document/chroniccaremanagement.pdf#page=39" + ) + def test_missing_table_is_swallowed_as_no_guidance(self): bare = duckdb.connect(":memory:") # pfs.code_guidance doesn't exist assert lineage._collect_guidance(bare, None, ["CCM"], [], 1) == () diff --git a/tests/llm/test_pages.py b/tests/llm/test_pages.py index 453b9eb..d9ee772 100644 --- a/tests/llm/test_pages.py +++ b/tests/llm/test_pages.py @@ -3,6 +3,7 @@ import fitz import pytest +from llm import pages as pages_mod from llm.chunk import Chunk, Doc from llm.pages import enrich_pdf_pages, locate, pdf_pages @@ -113,3 +114,34 @@ class TestEnrich: ) (out,) = enrich_pdf_pages(doc, [c]) assert out.metadata["page"] == "9" + + +# ── locate_section (#705 item 2: IOM section headings) ────────────── + + +class TestLocateSection: + TOC = ( + "Table of Contents (Rev. 12780) 30.6.3 - Payment for Immunosuppressive " + "Therapy Management 30.6.4 - Evaluation and Management (E/M) Services " + "Furnished Incident to Physician's Service 30.6.5 - Physicians in Group" + ) + BODY = ( + "visit is for immunosuppressive therapy. 30.6.4 - Evaluation and " + "Management (E/M) Services Furnished Incident to Physician's Service by " + "Nonphysician Practitioners (Rev. 11288; Issued: 03-31-22) A. General" + ) + + def test_body_heading_beats_the_table_of_contents(self): + assert pages_mod.locate_section([self.TOC, "filler", self.BODY], "30.6.4") == 3 + + def test_toc_only_is_not_a_location(self): + assert pages_mod.locate_section([self.TOC], "30.6.4") == 0 + + def test_prefix_numbers_do_not_match(self): + # "30.6.4" must not fire on "130.6.4" or "30.6.4.1" + body = "130.6.4 - Other (Rev. 1) 30.6.4.1 - Sub (Rev. 2)" + assert pages_mod.locate_section([body], "30.6.4") == 0 + + def test_trailing_period_and_blank_section(self): + assert pages_mod.locate_section([self.BODY], "30.6.4.") == 1 + assert pages_mod.locate_section([self.BODY], "") == 0 diff --git a/tests/pfs/test_guidance.py b/tests/pfs/test_guidance.py index 9597d51..5f19baf 100644 --- a/tests/pfs/test_guidance.py +++ b/tests/pfs/test_guidance.py @@ -25,13 +25,17 @@ from pfs.guidance import ( MLN_RE, _cfr_refs, _extract, + _guidance_page, _iom_refs, build, dedupe, harvest, harvest_cpt, + iom_section_page, + mln_refs, resolve_cfr, resolve_iom, + resolve_mln, ) # ── real fr_anchors snippets ────────────────────────────────────────── @@ -601,3 +605,196 @@ class TestBuild: assert harvest_cpt(bare, s, ("99490",), family="CCM") == [] finally: bare.close() + + +# ── MLN (#705 item 3) ─────────────────────────────────────────────── + + +class TestMlnRefs: + def test_icn_prefix_and_icn_mln_combo_normalise_to_mln(self): + text = ( + "MLN Booklet “Chronic Care Management Services” (ICN MLN909188, July " + "2019); see also “Advance Care Planning” (ICN 909289, August 2016)." + ) + assert mln_refs(text) == ["MLN 909188", "MLN 909289"] + + def test_matters_article_forms(self): + assert mln_refs("MLN Matters® Number MM9603: https://x") == [ + "MLN Matters MM9603" + ] + assert mln_refs("as stated in MLN Matters article SE1316, issued") == [ + "MLN Matters SE1316" + ] + assert mln_refs("documentation is available in MLN Matters SE 1316.") == [ + "MLN Matters SE1316" + ] + # a bare number after "Matters" is a change-request article + assert mln_refs("consistent with MLN Matters article 11268.") == [ + "MLN Matters MM11268" + ] + + def test_prose_mentions_without_a_number_yield_nothing(self): + assert ( + mln_refs("We will issue an MLN Matters article once the CR is out.") == [] + ) + assert mln_refs("Medicare Learning Network (MLN) Matters® article,") == [] + + def test_first_occurrence_wins(self): + assert mln_refs("MLN006764 ... again MLN 006764") == ["MLN 006764"] + + +@pytest.fixture +def mln_store(): + s = Store(":memory:") + booklet = s.create( + Manual( + title="Chronic Care Management Services (MLN909188, June 2025) — CCM", + url="https://www.cms.gov/files/document/chroniccaremanagement.pdf", + ) + ) + em_guide = s.create( + Manual( + title="Evaluation and Management Services Guide", + url=( + "https://www.cms.gov/Outreach-and-Education/Medicare-Learning-" + "Network-MLN/MLNProducts/Downloads/eval-mgmt-serv-guide-ICN006764.pdf" + ), + ) + ) + article = s.create( + Manual( + title="MLN Matters MM9603", + url=( + "https://www.cms.gov/Outreach-and-Education/Medicare-Learning-" + "Network-MLN/MLNMattersArticles/Downloads/MM9603.pdf" + ), + ) + ) + decoy = s.create( + Manual(title="Unrelated", url="https://www.cms.gov/files/MM96030.pdf") + ) + yield s, booklet, em_guide, article, decoy + s.close() + + +class TestResolveMln: + def test_title_carries_the_number(self, mln_store): + s, booklet, *_ = mln_store + assert resolve_mln(s, "MLN 909188") == booklet + + def test_pre_2020_icn_spelling_in_url(self, mln_store): + s, _b, em_guide, *_ = mln_store + assert resolve_mln(s, "MLN 006764") == em_guide + + def test_matters_article_by_url_word_bounded(self, mln_store): + s, _b, _e, article, _decoy = mln_store + # MM9603 must not match the decoy's MM96030 + assert resolve_mln(s, "MLN Matters MM9603") == article + assert resolve_mln(s, "MLN Matters MM9604") == "" + + def test_unresolved_and_empty(self, mln_store): + s, *_ = mln_store + assert resolve_mln(s, "MLN 907166") == "" + assert resolve_mln(s, "") == "" + + +class TestExtractMlnResolved: + def test_extract_resolves_against_store(self, mln_store): + s, booklet, *_ = mln_store + rows = _extract("Refer to the CCM booklet (ICN MLN909188, July 2019).", s) + assert rows == [("mln", "MLN 909188", booklet)] + + +# ── IOM section page locating (#705 item 2) ───────────────────────── + + +class TestIomSectionPage: + TOC = "Table of Contents 30.6.4 - E/M Services Incident to 30.6.5 - Groups" + BODY = "30.6.4 - E/M Services Incident to (Rev. 11288; Issued: 03-31-22) A." + + @pytest.fixture(autouse=True) + def _fresh_caches(self, monkeypatch): + monkeypatch.setattr(guidance_mod, "_PAGE_CACHE", {}) + monkeypatch.setattr(guidance_mod, "_PDF_CACHE", {}) + + @pytest.fixture + def chapter(self, tmp_path, monkeypatch): + # Own store: an in-memory Store's default storage dir is ./storage + # relative to the cwd, which attach_file would create in the repo. + s = Store(":memory:", storage_dir=tmp_path / "storage") + manual_key = s.create( + Manual(title="Medicare Claims Processing Manual — Chapter 12: X") + ) + pdf = tmp_path / "clm104c12.pdf" + pdf.write_bytes(b"%PDF-1.4 stub") + s.attach_file(manual_key, pdf) + calls: list[str] = [] + + def fake_pages(path): + calls.append(str(path)) + return [self.TOC, "filler", self.BODY] + + monkeypatch.setattr("llm.pages.pdf_pages", fake_pages) + yield s, manual_key, calls + s.close() + + def test_locates_body_heading_past_the_toc(self, chapter): + s, key, _ = chapter + assert iom_section_page(s, key, "30.6.4") == 3 + + def test_pdf_read_once_and_page_cached(self, chapter): + s, key, calls = chapter + assert iom_section_page(s, key, "30.6.4") == 3 + assert iom_section_page(s, key, "30.6.4") == 3 + assert iom_section_page(s, key, "30.6.5") == 0 # TOC-only + assert len(calls) == 1 + + def test_no_pdf_attachment_or_blank_args(self, store): + s, _sec, manual_key = store + assert iom_section_page(s, manual_key, "30.6.4") == 0 + assert iom_section_page(s, "", "30.6.4") == 0 + assert iom_section_page(s, manual_key, "") == 0 + + def test_guidance_page_only_for_resolved_iom_sections(self, chapter): + s, key, _ = chapter + assert _guidance_page(s, "iom", "100-04 ch.12 §30.6.4", key) == 3 + assert _guidance_page(s, "iom", "100-04 ch.12", key) == 0 + assert _guidance_page(s, "iom", "100-04 ch.12 §30.6.4", "") == 0 + assert _guidance_page(s, "cfr", "42 CFR 410.78", key) == 0 + + def test_harvest_carries_the_located_page(self, store_with_anchors, monkeypatch): + s, _item, _sec, manual_key = store_with_anchors + monkeypatch.setattr( + guidance_mod, + "iom_section_page", + lambda st, k, sec: 39 if k == manual_key else 0, + ) + rows = harvest(s, ("99490", "99439"), family="CCM") + iom = [r for r in rows if r.kind == "iom"] + assert iom and all(r.page == 39 for r in iom) + assert all(r.page == 0 for r in rows if r.kind != "iom") + + +class TestGuidanceTablePage: + def test_page_round_trips(self, con): + row = GuidanceRow( + "CCM", "99490", "iom", "100-04 ch.12 §30.6.4", "K", "S", 1, 2, 39 + ) + write_guidance(con, "CCM", [row]) + assert read_guidance(con, "CCM") == [row] + + def test_legacy_table_without_page_column_reads_zero(self): + c = duckdb.connect(":memory:") + c.execute("CREATE SCHEMA pfs") + c.execute( + "CREATE TABLE pfs.code_guidance (family VARCHAR, code VARCHAR, kind VARCHAR, " + "locator VARCHAR, item_key VARCHAR, item_key_src VARCHAR, p_id_src INTEGER, " + "page_src INTEGER)" + ) + c.execute( + "INSERT INTO pfs.code_guidance VALUES ('CCM','99490','iom','x','','S',1,2)" + ) + ensure_tables(c) # adds the column in place; the old row reads NULL + got = read_guidance(c, "CCM") + assert got == [GuidanceRow("CCM", "99490", "iom", "x", "", "S", 1, 2, 0)] + c.close()