diff --git a/src/cli/pfs.py b/src/cli/pfs.py index 11dedd7..4d0ae7e 100644 --- a/src/cli/pfs.py +++ b/src/cli/pfs.py @@ -18,6 +18,7 @@ the read-only replica so the chat and notebooks see them. from __future__ import annotations import logging +from pathlib import Path from typing import Any import typer @@ -40,6 +41,7 @@ from pfs.codetables import ( write_exposures, write_families, write_guidance, + write_origin, write_reaction, write_rounds, ) @@ -59,6 +61,7 @@ from pfs.families import ( ) from pfs.guidance import build as build_guidance from pfs.lineage import lineage, lineage_all, seen_codes +from pfs.origin import DEFAULT_BOOK, family_origin, origin_inventory, resolve_book from pfs.reaction import series as reaction_series from pfs.rounds import family_rounds, rounds_inventory from pfs.runlog import RunLog, ledger_path @@ -1064,3 +1067,91 @@ def rounds( n = write_rounds(con, fam.key, rows) typer.echo(f"wrote {n} rounds") _publish() + + +def _origin_books(store: Any, keys: list[str]) -> list[tuple[str, str, Path]]: + books = [] + for key in keys or [DEFAULT_BOOK]: + found = resolve_book(store, key) + if found is None: + raise typer.BadParameter(f"no EPUB attachment for bib item {key}") + books.append((key, found[0], found[1])) + return books + + +@app.command() +def origin( + family: str = typer.Option(..., "--family", help="Curated family key."), + book: list[str] = typer.Option( + [], + "--book", + help=f"Bib item of a licensed EPUB to read (default {DEFAULT_BOOK}, " + "CPT Changes 2023); repeatable.", + ), + limit: int = typer.Option(0, "--limit", help="Cap new model calls (0 = all)."), + dry_run: bool = typer.Option( + False, "--dry-run", help="Count candidates; no model calls, no writes." + ), + no_llm: bool = typer.Option( + False, "--no-llm", help="Make no model calls; the finished ledger is replayed." + ), + retry_failed: bool = typer.Option( + False, "--retry-failed", help="Re-run ledger entries that failed." + ), + write: bool = typer.Option(False, "--write"), +) -> None: + """Origin and formation rows per code (pfs.code_origin): the statute, + regulation, program, demonstration, CPT Editorial Panel action, G-code + creation or coverage rule behind them, from Federal Register + paragraphs, the licensed CPT Changes book (paraphrase + page only) and + the guidance rows, through a resumable local-model ledger.""" + fam = curated_family(normalize_key(family)) + if fam is None: + raise typer.BadParameter(f"unknown curated family {family!r}") + store = _store() + books = _origin_books(store, book) + if dry_run: + paras, pages = origin_inventory(store, fam, books) + typer.echo(f"{fam.key}: {len(paras)} candidate Federal Register paragraphs") + for key, labels in pages.items(): + typer.echo(f"{key}: {len(labels)} pages naming a code") + typer.echo("no model calls made") + return + ask = None + if not no_llm: + try: + ask = _asker() + except RuntimeError as exc: + typer.echo(f"no model host available ({exc}); replaying the ledger") + log = RunLog(ledger_path("origin", fam.key)) + con = _read() + try: + rows, counts = family_origin( + con, + store, + fam, + ask=ask, + log=log, + books=books, + limit=limit, + retry_failed=retry_failed, + replay_model=_model_name() if ask is None else None, + ) + finally: + con.close() + for field in ("kind", "cite_kind"): + tally: dict[str, int] = {} + for r in rows: + tally[getattr(r, field)] = tally.get(getattr(r, field), 0) + 1 + for name, n in sorted(tally.items()): + typer.echo(f"{field} {name}: {n}") + typer.echo( + f"origins: {len(rows)}; done {counts['done']}, skipped {counts['skipped']}, " + f"failed {counts['failed']}, failed_pending {counts['failed_pending']}" + ) + if write: + with _batch() as con: + ensure_tables(con) + n = write_origin(con, fam.key, rows) + typer.echo(f"wrote {n} origins") + _publish() diff --git a/src/pfs/codetables.py b/src/pfs/codetables.py index d91b565..047fc61 100644 --- a/src/pfs/codetables.py +++ b/src/pfs/codetables.py @@ -328,6 +328,44 @@ class CptCodeAltRow: reason: str # "guidelines-reprint" | "lower-score" (C9) +ORIGIN_KINDS = ( + "statute", + "regulation", + "program", + "demonstration", + "cpt_panel", + "cms_gcode", + "coverage", +) +ORIGIN_CITE_KINDS = ("fr_para", "cpt_book", "iom", "cfr", "mln") + + +@dataclass(frozen=True) +class OriginRow: + """One origin of a code (``pfs.code_origin``): the statute, regulation, + program, demonstration, CPT Editorial Panel action, CMS G-code creation + or coverage rule behind it. ``summary`` is a paraphrase (at most 240 + characters; never licensed book text) and ``cite_kind`` says what the + citation is: ``fr_para`` = paragraph ``p_id`` of bib item ``item_key`` + (``page_label`` = its FR page), ``cpt_book`` = printed page + ``page_label`` of the licensed book ``item_key`` (``p_id`` 0), + ``iom``/``cfr``/``mln`` = a guidance reference found in the paragraph + ``item_key``/``p_id``. ``locator`` is the human-readable cite.""" + + family: str + code: str + kind: str + reference: str + year: int | None + summary: str + cite_kind: str + item_key: str + p_id: int + page_label: str + locator: str + extractor: str + + _DDL = """ CREATE SCHEMA IF NOT EXISTS pfs; CREATE TABLE IF NOT EXISTS pfs.code_element ( @@ -385,6 +423,10 @@ CREATE TABLE IF NOT EXISTS pfs.valuation_round ( decision VARCHAR, rationale_p_id INTEGER, rationale_quote VARCHAR, pe_note VARCHAR, extractor VARCHAR, final_p_id INTEGER, ruc_p_id INTEGER, field_cites VARCHAR); +CREATE TABLE IF NOT EXISTS pfs.code_origin ( + family VARCHAR, code VARCHAR, kind VARCHAR, reference VARCHAR, year INTEGER, + summary VARCHAR, cite_kind VARCHAR, item_key VARCHAR, p_id INTEGER, + page_label VARCHAR, locator VARCHAR, extractor VARCHAR); """ @@ -782,3 +824,18 @@ def read_rounds(con: Any, family: str) -> list[RoundRow]: [family], ).fetchall() return [RoundRow(*r) for r in rows] + + +def write_origin(con: Any, family: str, rows: Sequence[OriginRow]) -> int: + ensure_tables(con) + con.execute("DELETE FROM pfs.code_origin WHERE family = ?", [family]) + return _insert(con, "pfs.code_origin", rows) + + +def read_origin(con: Any, family: str) -> list[OriginRow]: + rows = con.execute( + "SELECT * FROM pfs.code_origin WHERE family = ? " + "ORDER BY code, kind, year NULLS FIRST, reference, item_key, p_id", + [family], + ).fetchall() + return [OriginRow(*r) for r in rows] diff --git a/src/pfs/origin.py b/src/pfs/origin.py new file mode 100644 index 0000000..3db5ec2 --- /dev/null +++ b/src/pfs/origin.py @@ -0,0 +1,547 @@ +"""Origin and formation of a code family (``pfs.code_origin``): the statute, +regulation, program, demonstration, CPT Editorial Panel action, CMS G-code +creation or coverage rule behind each code. + +Three sources, every row citing the exact place that supports it: + +* Federal Register paragraphs (``fr_para``: item + paragraph), read by the + local model through a resumable ledger. A model item survives only when + its ``reference`` is written in a plain sentence of the paragraph (no + recap of another year's rule, hypothetical or commenter's request), the + codes it names are family codes the paragraph names, and no number in + its summary is absent from the paragraph. +* The licensed AMA book *CPT Changes* (``cpt_book``: item + printed page). + Book text goes only to the local model; a summary is kept only as a + paraphrase (no 8-word run shared with the page), is at most 240 + characters, and the ledger stores only the guarded paraphrases. +* ``pfs.code_guidance`` rows (deterministic): CFR cites become + ``regulation`` rows, IOM and MLN cites ``coverage`` rows. +""" + +from __future__ import annotations + +import re +from collections.abc import Collection, Sequence +from pathlib import Path +from typing import Any + +from pfs.codetables import ORIGIN_KINDS, OriginRow, read_guidance +from pfs.descriptors import rule_year_of +from pfs.families import code_pattern +from pfs.lineage import _expand_ranges +from pfs.rounds import ( + _HYPOTHETICAL, + _PRINTED_PAGE, + _anchor_rows, + _named, + _recap, + _sentences, + rule_items, +) +from pfs.runlog import RunLog, input_hash, ledger_path, run_units + +ORIGIN_PROMPT_VERSION = "origin-v1" +DEFAULT_BOOK = "3ANRT854" # "CPT Changes 2023" + +ORIGIN_SCHEMA: dict = { + "type": "object", + "required": ["origins"], + "properties": { + "origins": { + "type": "array", + "items": { + "type": "object", + "required": ["kind", "reference", "year", "codes", "summary"], + "properties": { + "kind": {"type": "string", "enum": list(ORIGIN_KINDS)}, + "reference": {"type": "string", "maxLength": 120}, + "year": {"type": ["integer", "null"]}, + "codes": {"type": "array", "items": {"type": "string"}}, + "summary": {"type": "string", "maxLength": 240}, + }, + }, + } + }, +} + +_COMMON = ( + "You read one passage about CPT/HCPCS codes and list the ORIGINS the " + "passage states for the listed codes: the statute, regulation, program, " + "demonstration, CPT Editorial Panel action, CMS G-code creation or " + "coverage rule that created, defined or governs them. kind is one of " + "statute, regulation, program, demonstration, cpt_panel, cms_gcode, " + "coverage. List one origin per distinct statute, program, demonstration " + "or panel action; leave out anything the passage only proposes, " + "imagines, recaps from another year, or attributes to a commenter. " + "codes = the listed codes that origin applies to, only ones the passage " + "names; year = the year the passage gives for it, else null; summary = " + "one plain sentence (under 240 characters) in YOUR OWN WORDS of what " + "that origin did for the codes; never add a fact, name or number the " + "passage does not state. These are NOT origins: quality-measure or " + "MIPS denominator lists, additions to the Medicare telehealth list, " + "payment rates, valuations, utilization, a pointer to another section " + "of the same rule, and anything CMS only proposes. If the passage " + "states no origin, return an empty list. " +) +ORIGIN_SYSTEM = { + "fr": _COMMON + + ( + "reference = the statute, rule, program or action written EXACTLY as " + 'the passage writes it (for example "section 1861(m) of the Act"), ' + "so it can be found in the passage word for word. Paraphrase the " + "summary; never quote." + ), + "book": _COMMON + + ( + "The passage is a page of licensed text. A code listed with its " + "descriptor and symbols is NOT an origin; use only prose that says " + "a code was created, revised, deleted, merged or moved (usually in " + "a Rationale paragraph), and say which of those it was. reference = " + 'a short label you write (for example "CPT Editorial Panel ' + 'action"), not a quotation. Paraphrase: never copy eight or more ' + "consecutive words of the passage into the reference or the summary." + ), +} + +#: a commenter's request, not CMS stating something +_REQUEST = re.compile( + r"\b(?:commenters?|requesters?)\b[^.;]{0,60}\b" + r"(?:recommend|request|suggest|urg|ask|support|oppos)\w*", + re.I, +) +#: the FR paragraphs that can state an origin +_ORIGIN_WORDS = re.compile( + r"section 1\d{3}|of the Act\b|statut|demonstration|Independence at Home|" + r"CPT Editorial Panel|establish|creat|benefit|homebound|coverage", + re.I, +) +_PROPOSAL = re.compile(r"\bpropos\w*", re.I) +#: what a reference must look like for its kind: a statute is a section of +#: the Act (or the U.S. Code), a regulation a CFR/section-sign cite, a G-code +#: creation names a HCPCS code; a pointer into the same rule is no origin +_ACT = re.compile( + r"\bof the (?:Social Security )?Act\b|\bsection 1\d{3}\b|U\.S\.C", re.I +) +_CFR = re.compile(r"\bCFR\b|§") +_GCODE = re.compile(r"\b[A-Z]\d{4}\b|\bG[A-Z]{3,4}\d\b|\bHCPCS\b") +_POINTER = re.compile( + r"\bof this (?:proposed |final |interim final )?rule\b|^section [IVX]+\.", re.I +) +#: a book page states an origin only in prose that says what happened to a +#: code; a page that merely lists codes with their descriptors and symbols +#: ("▲" etc.) states none, whatever the model reads into it +_BOOK_PROSE = re.compile( + r"Rationale|\bdeleted\b|\bnew codes?\b|" + r"\b(?:has|have) been (?:revised|created|added|merged|moved|replaced)\b", + re.I, +) +_ANY_CODE = re.compile(r"\b(?:\d{5}|[A-Z]\d{4})\b") +_NUM = re.compile(r"\d+") +_WORD = re.compile(r"[a-z0-9]+") + + +def _words(text: str) -> list[str]: + return _WORD.findall(text.lower()) + + +def shares_run(summary: str, text: str, n: int = 8) -> bool: + """True when *summary* and *text* share a run of *n* consecutive words + (case- and punctuation-insensitive): the copy guard for licensed text.""" + a = _words(summary) + b = _words(text) + grams = {tuple(b[i : i + n]) for i in range(len(b) - n + 1)} + return any(tuple(a[i : i + n]) in grams for i in range(len(a) - n + 1)) + + +def _collapse(text: str) -> str: + return " ".join(_PRINTED_PAGE.sub(" ", text).split()).casefold() + + +def _names(text: str, code: str) -> bool: + return bool(re.search(rf"\b{re.escape(code)}\b", _expand_ranges(text))) + + +def _named_codes(text: str, codes: Sequence[str]) -> list[str]: + return [c for c in codes if _names(text, c)] + + +def origin_user(text: str, codes: Sequence[str], *, label: str) -> str: + return f"CODES: {', '.join(codes)}\nSOURCE: {label}\n\nPASSAGE:\n{text}" + + +def _copy_free(it: dict, text: str) -> bool: + return not (shares_run(it["summary"], text) or shares_run(it["reference"], text)) + + +def _plain(sentence: str, rule_year: int | None) -> bool: + """A sentence that states something: no recap of another year's rule, + hypothetical or commenter's request.""" + return not ( + _recap(sentence, rule_year) + or _HYPOTHETICAL.search(sentence) + or _REQUEST.search(sentence) + ) + + +def _window(ref: str, text: str, rule_year: int | None) -> str | None: + """The sentences of *text* that write *ref* in a plain sentence, each + with its neighbours (where the codes it applies to are named); ``None`` + when no plain sentence writes it.""" + want = _collapse(ref) + sents = _sentences(text) + hit = [ + i for i, s in enumerate(sents) if want in _collapse(s) and _plain(s, rule_year) + ] + if not hit: + return None + near = sorted({j for i in hit for j in (i - 1, i, i + 1) if 0 <= j < len(sents)}) + return " ".join(sents[j] for j in near) + + +def _kind_of(it: dict) -> str | None: + """The item's kind, settled by what its reference is: a CFR cite is a + regulation, a section of the Act a statute; a statute/regulation/G-code + claim the reference does not support is ``None`` (dropped).""" + ref, kind = it["reference"], it["kind"] + if _CFR.search(ref): + return "regulation" + if _ACT.search(ref): + return "statute" + if kind in ("statute", "regulation") or ( + kind == "cms_gcode" and not _GCODE.search(ref) + ): + return None + return kind + + +def origins_from_output( + out: dict, + text: str, + *, + source: str, + codes: Collection[str] | None = None, + rule_year: int | None = None, +) -> list[dict]: + """Guarded origin items from one model reply. Codes outside *codes* (or + the family) and codes the passage does not name are dropped, an item + left with none is dropped; a summary that names codes applies only to + those (a code it does not name is dropped); a summary number the passage lacks drops the + item; a year the passage lacks becomes ``None``. FR source: the + ``reference`` must be written in a plain sentence of *text*, and the + codes near it. Book source: a page with no prose about what happened + to a code yields nothing; an item whose summary or reference shares an 8-word run with + *text* is dropped.""" + if source == "book" and not _BOOK_PROSE.search(text): + return [] + have = set(_NUM.findall(text)) + kept: list[dict] = [] + for it in out.get("origins") or []: + kind = _kind_of(it) + if kind is None or _POINTER.search(it["reference"]): + continue + if source == "fr": + near = _window(it["reference"], text, rule_year) + if near is None or _PROPOSAL.search(it["summary"]): + continue + elif not _copy_free(it, text): + continue + else: + near = text + if set(_NUM.findall(it["summary"])) - have: + continue + said = set(_ANY_CODE.findall(_expand_ranges(it["summary"]))) + mine = [ + c + for c in dict.fromkeys(str(c).strip().upper() for c in it["codes"]) + if (codes is None or c in codes) + and _names(near, c) + and (not said or c in said) + ] + if not mine: + continue + year = ( + it["year"] if it["year"] is not None and str(it["year"]) in have else None + ) + kept.append({**it, "kind": kind, "codes": mine, "year": year}) + return kept + + +def extract_origin( + text: str, + ask: Any, + *, + source: str, + codes: Collection[str] | None = None, + rule_year: int | None = None, +) -> list[dict]: + """Ask the local model for the origins *text* states and apply the + guards of :func:`origins_from_output`. A failed reply gives ``[]``.""" + if source not in ORIGIN_SYSTEM: + raise ValueError(f"unknown origin source {source!r}") + named = sorted(codes) if codes else [] + user = origin_user(text, named, label=source) + out = ask(ORIGIN_SYSTEM[source], user, ORIGIN_SCHEMA) + if out is None: + return [] + return origins_from_output( + out, text, source=source, codes=codes, rule_year=rule_year + ) + + +# ── candidates ─────────────────────────────────────────────────────── + + +def fr_origin_candidates( + store: Any, codes: Sequence[str], phrases: Sequence[str] +) -> list[dict]: + """``fr_anchors`` paragraphs of the (de-duplicated, non-correction) + rules that name a family code or phrase and talk about a statute, + program, creation or coverage; oldest rule first. Each carries + ``rule_kind`` and ``rule_year``.""" + if not codes and not phrases: + return [] + pattern = code_pattern(codes) if codes else None + low = [x.lower() for x in phrases] + anchors = _anchor_rows(store, codes, phrases) + kinds = {it["item_key"]: it["kind"] for it in rule_items(store, anchors)} + out = [] + for r in anchors: + text = r["text"] or "" + if r["item_key"] not in kinds: + continue + named = bool(pattern and _named(text, pattern)) or any( + x in text.lower() for x in low + ) + if not named or not _ORIGIN_WORDS.search(text): + continue + out.append( + { + **r, + "rule_kind": kinds[r["item_key"]], + "rule_year": rule_year_of(r["title"], r["date_published"]), + } + ) + return out + + +def book_candidates(path: Path, codes: Sequence[str]) -> list[tuple[str, str]]: + """Printed pages of the EPUB naming a family code. Text before the + first page marker has no page to cite and is left out.""" + if not codes: + return [] + # lazy: importing ``rex`` pulls the lake extras, which ``stack`` must + # start without + from rex.comments.epub_text import epub_pages # noqa: PLC0415 + + return [ + (label, text) + for label, text in epub_pages(path) + if label and _named_codes(text, codes) + ] + + +def resolve_book(store: Any, item_key: str) -> tuple[str, Path] | None: + """``(title, EPUB path)`` of a bib item with an ``.epub`` attachment.""" + con = store._con() # noqa: SLF001 + for title, fn, sp in con.execute( + "SELECT i.title, a.filename, 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(): + if str(fn).lower().endswith(".epub") and sp and Path(sp).is_file(): + return str(title), Path(sp) + return None + + +# ── guidance (deterministic) ───────────────────────────────────────── + +_GUIDANCE_KIND = {"cfr": "regulation", "iom": "coverage", "mln": "coverage"} + + +def guidance_origins(con: Any, family: str) -> list[OriginRow]: + """One ``regulation`` (CFR) or ``coverage`` (IOM, MLN) row per distinct + ``(code, cite)`` in ``pfs.code_guidance``, citing the first paragraph + that carries it.""" + rows: dict[tuple[str, str, str], OriginRow] = {} + for g in read_guidance(con, family): + key = (g.code, g.kind, g.locator) + if key in rows: + continue + rows[key] = OriginRow( + family=family, + code=g.code, + kind=_GUIDANCE_KIND[g.kind], + reference=g.locator, + year=None, + summary=f"CMS guidance for {g.code} points to {g.locator}.", + cite_kind=g.kind, + item_key=g.item_key_src, + p_id=g.p_id_src, + page_label=str(g.page_src) if g.page_src else "", + locator=g.locator, + extractor="guidance", + ) + return list(rows.values()) + + +# ── family run ─────────────────────────────────────────────────────── + + +def _fr_job(p: dict, family: Any) -> dict: + codes = _named_codes(_PRINTED_PAGE.sub(" ", p["text"]), family.codes) + label = f"Federal Register, CY{p['rule_year']} {p['rule_kind']} rule" + return { + "unit": f"fr:{p['item_key']}:{p['p_id']}", + "source": "fr", + "text": p["text"], + "codes": codes, + "user": origin_user(p["text"], codes, label=label), + "year": p["rule_year"], + "cite_kind": "fr_para", + "item_key": p["item_key"], + "p_id": p["p_id"], + "page_label": str(p["page"]), + "locator": f"FR p. {p['page']}, ¶{p['p_id']}", + } + + +def _book_job(item_key: str, edition: str, label: str, text: str, family: Any) -> dict: + codes = _named_codes(text, family.codes) + return { + "unit": f"book:{item_key}:{label}", + "source": "book", + "text": text, + "codes": codes, + "user": origin_user(text, codes, label=f"{edition}, page {label}"), + "year": None, + "cite_kind": "cpt_book", + "item_key": item_key, + "p_id": 0, + "page_label": label, + "locator": f"{edition}, p. {label}", + } + + +def origin_inventory( + store: Any, family: Any, books: Sequence[tuple[str, str, Path]] +) -> tuple[list[dict], dict[str, list[str]]]: + """What a run would work on without a model call: the candidate FR + paragraphs and, per book item, the labels of the pages naming a code.""" + paras = [j for j in _jobs(store, family, ()) if j["source"] == "fr"] + pages = { + key: [label for label, _ in book_candidates(path, family.codes)] + for key, _edition, path in books + } + return paras, pages + + +def _jobs( + store: Any, family: Any, books: Sequence[tuple[str, str, Path]] +) -> list[dict]: + jobs = [ + _fr_job(p, family) + for p in fr_origin_candidates(store, family.codes, family.synonyms) + ] + for key, edition, path in books: + jobs += [ + _book_job(key, edition, label, text, family) + for label, text in book_candidates(path, family.codes) + ] + return [j for j in jobs if j["codes"]] + + +def _ask_job(ask: Any, job: dict) -> dict | None: + """One model call. A book reply is reduced to its paraphrases before + it reaches the ledger, so the ledger never holds copied book text.""" + out = ask(ORIGIN_SYSTEM[job["source"]], job["user"], ORIGIN_SCHEMA) + if out is None or job["source"] != "book": + return out + return { + "origins": [ + it for it in out.get("origins") or [] if _copy_free(it, job["text"]) + ] + } + + +def _rows(job: dict, out: dict, family: Any, model: str) -> list[OriginRow]: + return [ + OriginRow( + family=family.key, + code=code, + kind=it["kind"], + reference=it["reference"], + year=it["year"], + summary=it["summary"], + cite_kind=job["cite_kind"], + item_key=job["item_key"], + p_id=job["p_id"], + page_label=job["page_label"], + locator=job["locator"], + extractor=f"model:{model}@{ORIGIN_PROMPT_VERSION}", + ) + for it in origins_from_output( + out, + job["text"], + source=job["source"], + codes=set(family.codes), + rule_year=job["year"], + ) + for code in it["codes"] + ] + + +def family_origin( + con: Any, + store: Any, + family: Any, + *, + ask: Any | None, + log: RunLog | None, + books: Sequence[tuple[str, str, Path]], + limit: int = 0, + retry_failed: bool = False, + replay_model: str | None = None, +) -> tuple[list[OriginRow], dict[str, int]]: + """Guidance rows plus model rows from FR paragraphs and book pages. + ``books`` is ``(item_key, edition_label, epub path)``. The model runs + through the ledger *log* (resumable, ``limit`` caps new calls); without + *ask* nothing is asked but the finished ledger for *replay_model* is + replayed, so ``--write`` never drops model rows. A model row repeating + an earlier FR ``(code, kind, reference)`` is dropped: the oldest + paragraph is the one cited (book rows are kept per page). Returns ``(rows, run counts)``.""" + counts = {"done": 0, "skipped": 0, "failed": 0, "failed_pending": 0} + model_rows: list[OriginRow] = [] + model = ask.model if ask is not None else replay_model + if model is not None: + jobs = _jobs(store, family, books) + for j in jobs: + j["hash"] = input_hash(j["user"], ORIGIN_SYSTEM[j["source"]]) + log = log or RunLog(ledger_path("origin", family.key)) + if ask is not None: + units = [(j["unit"], j["hash"], lambda j=j: _ask_job(ask, j)) for j in jobs] + counts = run_units( + units, + log, + model=model, + version=ORIGIN_PROMPT_VERSION, + limit=limit, + retry_failed=retry_failed, + ) + for j in jobs: + entry = log.get(j["unit"], j["hash"], model, ORIGIN_PROMPT_VERSION) + if entry is not None and entry.output is not None: + model_rows += _rows(j, entry.output, family, model) + seen: set[tuple[str, str, str]] = set() + rows: list[OriginRow] = [] + for r in model_rows: + key = (r.code, r.kind, r.reference.casefold()) + if r.cite_kind == "cpt_book": + key += (r.item_key, r.page_label) + if key not in seen: + seen.add(key) + rows.append(r) + rows += guidance_origins(con, family.key) + rows.sort( + key=lambda r: (r.code, r.kind, r.year or 0, r.reference, r.item_key, r.p_id) + ) + return rows, counts diff --git a/src/rex/comments/epub_text.py b/src/rex/comments/epub_text.py index d4d453d..19808bd 100644 --- a/src/rex/comments/epub_text.py +++ b/src/rex/comments/epub_text.py @@ -106,25 +106,58 @@ def _to_text(raw: str) -> str: return text.strip() +def _member_names(zf: zipfile.ZipFile) -> list[str]: + """Content members in reading order: the spine when the OPF resolves, + else every ``.xhtml``/``.html`` member sorted by name.""" + names = spine_paths(zf) + if not names: + names = sorted( + n + for n in zf.namelist() + if n.lower().endswith((".xhtml", ".html", ".htm")) + and not n.startswith("META-INF/") + ) + return names + + def extract_text(path: str | Path) -> str: """Whole-book plain text, in reading order: spine order when the OPF resolves, else every ``.xhtml``/``.html`` member sorted by name. Sections are joined with a blank line; empty sections (nav/cover files with no text) are dropped.""" with zipfile.ZipFile(path) as zf: - names = spine_paths(zf) - if not names: - names = sorted( - n - for n in zf.namelist() - if n.lower().endswith((".xhtml", ".html", ".htm")) - and not n.startswith("META-INF/") - ) parts = [] - for name in names: + for name in _member_names(zf): try: raw = zf.read(name).decode("utf-8", errors="replace") except KeyError: continue parts.append(_to_text(raw)) return "\n\n".join(p for p in parts if p.strip()) + + +#: the AMA books mark each printed page with an empty ```` +#: (labels are roman numerals for front matter, arabic after) +_PAGE_SPAN = re.compile(r'|>)') +_PAGE_MARK = re.compile(r"\x00([^\x00]*)\x00") + + +def epub_pages(path: str | Path) -> list[tuple[str, str]]: + """``(page_label, plain text)`` in reading order, split at the + ```` markers. The whole book is one stream, + so a page that runs on into the next content file keeps its text; + text before the first marker is labelled ``""``. Pages with no text + are dropped.""" + with zipfile.ZipFile(path) as zf: + parts = [] + for name in _member_names(zf): + try: + raw = zf.read(name).decode("utf-8", errors="replace") + except KeyError: + continue + parts.append( + _to_text(_PAGE_SPAN.sub(lambda m: f"\x00{m.group(1)}\x00", raw)) + ) + pieces = _PAGE_MARK.split("\n\n".join(p for p in parts if p.strip())) + pages = [("", pieces[0])] + list(zip(pieces[1::2], pieces[2::2], strict=True)) + return [(label, text.strip()) for label, text in pages if text.strip()] diff --git a/tests/cli/test_pfs_cli.py b/tests/cli/test_pfs_cli.py index 029fe34..fa682a4 100644 --- a/tests/cli/test_pfs_cli.py +++ b/tests/cli/test_pfs_cli.py @@ -1461,3 +1461,153 @@ class TestRounds: ) assert res.exit_code == 0, res.output assert "3 candidate paragraphs, 2 table rows; no model calls made" in res.output + + +class TestOrigin: + def _row(self, **kw): + from pfs.codetables import OriginRow + + base = dict( + family="HOME-VISITS", + code="99341", + kind="statute", + reference="Section 1861(m) of the Act", + year=None, + summary="A paraphrase.", + cite_kind="fr_para", + item_key="K", + p_id=2, + page_label="100", + locator="FR p. 100, ¶2", + extractor="m", + ) + return OriginRow(**{**base, **kw}) + + def _patch(self, monkeypatch, order): + rows = [ + self._row(), + self._row(code="99350", kind="regulation", cite_kind="cfr"), + self._row(code="99342", cite_kind="cpt_book", item_key="B", p_id=0), + ] + seen = {} + + def fake(con, store, fam, **kw): + order.append("family_origin") + seen.update(kw, key=fam.key) + return rows, {"done": 2, "skipped": 1, "failed": 0, "failed_pending": 3} + + monkeypatch.setattr(pfs_cli, "family_origin", fake) + monkeypatch.setattr(pfs_cli, "_asker", lambda: "ASK") + monkeypatch.setattr( + pfs_cli, + "resolve_book", + lambda store, key: ("Title " + key, Path("/x") / f"{key}.epub"), + ) + monkeypatch.setattr( + pfs_cli, "ledger_path", lambda t, f: Path("/nonexistent") / f"{t}-{f}" + ) + return seen + + def test_origin_cli_write_after_extraction(self, con, monkeypatch): + from pfs.codetables import read_origin, write_origin + + order = [] + seen = self._patch(monkeypatch, order) + orig_write = write_origin + + def write(c, fam, rows): + order.append("write_origin") + return orig_write(c, fam, rows) + + class _B: + def __enter__(self): + order.append("_batch") + return con._real + + def __exit__(self, *a): + return False + + monkeypatch.setattr(pfs_cli, "write_origin", write) + monkeypatch.setattr(pfs_cli, "_batch", lambda: _B()) + monkeypatch.setattr(pfs_cli, "_publish", lambda: order.append("_publish")) + res = runner.invoke( + app, + ["pfs", "origin", "--family", "home-visits", "--limit", "4", "--write"], + ) + assert res.exit_code == 0, res.output + assert order == ["family_origin", "_batch", "write_origin", "_publish"] + assert ( + seen["ask"] == "ASK" and seen["limit"] == 4 and seen["key"] == "HOME-VISITS" + ) + assert seen["books"] == [ + ("3ANRT854", "Title 3ANRT854", Path("/x/3ANRT854.epub")) + ] + assert seen["replay_model"] is None + assert "statute: 2" in res.output and "regulation: 1" in res.output + assert "cpt_book: 1" in res.output and "fr_para: 1" in res.output + assert "done 2, skipped 1, failed 0, failed_pending 3" in res.output + assert "wrote 3 origins" in res.output + assert len(read_origin(con._real, "HOME-VISITS")) == 3 + + def test_origin_without_write_writes_nothing(self, con, monkeypatch): + order = [] + seen = self._patch(monkeypatch, order) + monkeypatch.setattr(pfs_cli, "_batch", lambda: 1 / 0) + res = runner.invoke( + app, + ["pfs", "origin", "--family", "HOME-VISITS", "--book", "A", "--book", "B"], + ) + assert res.exit_code == 0, res.output + assert order == ["family_origin"] and "wrote" not in res.output + assert [b[0] for b in seen["books"]] == ["A", "B"] + + def test_no_llm_replays_the_ledger(self, con, monkeypatch): + seen = self._patch(monkeypatch, []) + monkeypatch.setattr(pfs_cli, "_asker", lambda: 1 / 0) + monkeypatch.setattr(pfs_cli, "_model_name", lambda: "qwen-x") + res = runner.invoke( + app, ["pfs", "origin", "--family", "HOME-VISITS", "--no-llm"] + ) + assert res.exit_code == 0, res.output + assert seen["ask"] is None and seen["replay_model"] == "qwen-x" + + def test_no_model_host_falls_back_to_replay(self, con, monkeypatch): + seen = self._patch(monkeypatch, []) + + def down(): + raise RuntimeError("no Ollama host serves qwen") + + monkeypatch.setattr(pfs_cli, "_asker", down) + monkeypatch.setattr(pfs_cli, "_model_name", lambda: "qwen-x") + res = runner.invoke(app, ["pfs", "origin", "--family", "HOME-VISITS"]) + assert res.exit_code == 0, res.output + assert seen["ask"] is None and seen["replay_model"] == "qwen-x" + assert "replaying the ledger" in res.output + + def test_origin_unknown_family_and_missing_book(self, con, monkeypatch): + res = runner.invoke(app, ["pfs", "origin", "--family", "NOPE"]) + assert res.exit_code != 0 and "unknown curated family" in res.output + self._patch(monkeypatch, []) + monkeypatch.setattr(pfs_cli, "resolve_book", lambda store, key: None) + res = runner.invoke(app, ["pfs", "origin", "--family", "HOME-VISITS"]) + assert res.exit_code != 0 and "no EPUB attachment" in res.output + + def test_origin_dry_run_makes_no_model_calls(self, con, monkeypatch): + self._patch(monkeypatch, []) + monkeypatch.setattr( + pfs_cli, + "origin_inventory", + lambda store, fam, books: ( + [{"unit": "a"}, {"unit": "b"}], + {"3ANRT854": ["12", "13", "20"]}, + ), + ) + monkeypatch.setattr(pfs_cli, "_asker", lambda: 1 / 0) + monkeypatch.setattr(pfs_cli, "family_origin", lambda *a, **k: 1 / 0) + monkeypatch.setattr(pfs_cli, "_batch", lambda: 1 / 0) + res = runner.invoke( + app, ["pfs", "origin", "--family", "HOME-VISITS", "--dry-run"] + ) + assert res.exit_code == 0, res.output + assert "2 candidate Federal Register paragraphs" in res.output + assert "3ANRT854: 3 pages" in res.output and "no model calls made" in res.output diff --git a/tests/pfs/test_origin.py b/tests/pfs/test_origin.py new file mode 100644 index 0000000..6f61b13 --- /dev/null +++ b/tests/pfs/test_origin.py @@ -0,0 +1,747 @@ +"""pfs.origin — origin and formation rows (dossier task 6). + +The EPUBs below are built in ``tmp_path`` from synthetic strings: the repo +holds no AMA text. +""" + +from __future__ import annotations + +import zipfile +from pathlib import Path +from types import SimpleNamespace + +import duckdb +import pytest + +from bib.item import Rule +from bib.store import Store +from pfs.codetables import ( + ORIGIN_KINDS, + GuidanceRow, + OriginRow, + ensure_tables, + read_origin, + write_guidance, + write_origin, +) +from pfs.origin import ( + ORIGIN_PROMPT_VERSION, + ORIGIN_SCHEMA, + book_candidates, + extract_origin, + family_origin, + fr_origin_candidates, + guidance_origins, + origin_inventory, + origins_from_output, + resolve_book, + shares_run, +) +from pfs.runlog import RunLog + +FAM = SimpleNamespace( + key="HOME-VISITS", + codes=("99341", "99342", "99350"), + synonyms=("home or residence services",), +) +CODES = FAM.codes + +_CONTAINER = ( + '' + '' + "" +) +_OPF = ( + '' + '' + '' +) + + +def make_epub(path: Path, pages: dict[str, str]) -> Path: + body = "".join( + f'

{text}

' for label, text in pages.items() + ) + with zipfile.ZipFile(path, "w") as zf: + zf.writestr("mimetype", "application/epub+zip") + zf.writestr("META-INF/container.xml", _CONTAINER) + zf.writestr("OPS/book.opf", _OPF) + zf.writestr("OPS/c1.xhtml", f"{body}") + return path + + +# synthetic stand-ins; the sentences are invented for the tests +BOOK_PAGES = { + "ix": "Preface with no code at all.", + "12": ( + "The panel adopted new home visit family entries beginning with 99341 " + "and 99350 for 2023, and the former domiciliary codes were deleted." + ), + "13": "Another page about 99342 only in an unrelated table.", +} + +FR_STATUTE = ( + "Section 1861(m) of the Act establishes the home health benefit, and CPT " + "code 99341 describes a home visit for a new patient." +) +FR_RECAP = ( + "In the CY 2019 PFS final rule, we established the homebound exception " + "for CPT code 99342." +) +FR_REQUEST = ( + "Commenters requested that section 1834(m) of the Act cover CPT code 99341." +) +FR_HYPO = ( + "If we were to repeal section 1862(a)(1) of the Act, CPT code 99350 " + "would need to be revised." +) +FR_PLAIN = "CPT code 99350 is not subject to the demonstration." + + +def ask_with(items, calls=None): + def ask(system, user, schema): + if calls is not None: + calls.append((system, user, schema)) + return {"origins": items} + + ask.model = "qwen-test" + return ask + + +def item(**kw): + base = dict( + kind="statute", + reference="Section 1861(m) of the Act", + year=None, + codes=["99341"], + summary="The Act sets up the home health benefit.", + ) + return {**base, **kw} + + +# ── pure guards ────────────────────────────────────────────────────── + + +def test_shares_run_is_case_and_punctuation_insensitive(): + text = "The Panel, in 2022, adopted these Home-Visit entries for everyone." + assert shares_run( + "it said: the panel in 2022 adopted these home visit entries", text + ) + assert not shares_run("the panel adopted entries about home visits", text) + assert not shares_run("too short", "too short") # fewer words than n + + +def test_book_summary_copy_guard(): + page = BOOK_PAGES["12"] + copied = item( + kind="cpt_panel", + reference="CPT Editorial Panel", + summary="The panel adopted new home visit family entries beginning with 99341.", + ) + fresh = item( + kind="cpt_panel", + reference="CPT Editorial Panel", + codes=["99350"], + summary="Editors created a fresh family of home visit entries.", + ) + out = origins_from_output( + {"origins": [copied, fresh]}, page, source="book", codes=CODES + ) + assert [o["summary"] for o in out] == [fresh["summary"]] + # the reference is guarded as well + leaky = item( + kind="cpt_panel", + reference="adopted new home visit family entries beginning with 99341", + ) + assert origins_from_output({"origins": [leaky]}, page, source="book") == [] + + +def test_fr_reference_must_appear(): + ok = item() + missing = item(reference="section 1861(s)(1) of the Act") + out = origins_from_output( + {"origins": [missing, ok]}, FR_STATUTE, source="fr", codes=CODES + ) + assert [o["reference"] for o in out] == ["Section 1861(m) of the Act"] + + +@pytest.mark.parametrize( + "text", + [FR_RECAP, FR_REQUEST, FR_HYPO], + ids=["recap", "commenter-request", "hypothetical"], +) +def test_fr_reference_must_sit_in_a_plain_sentence(text): + ref = { + FR_RECAP: "the homebound exception", + FR_REQUEST: "section 1834(m) of the Act", + FR_HYPO: "section 1862(a)(1) of the Act", + }[text] + code = {FR_RECAP: "99342", FR_REQUEST: "99341", FR_HYPO: "99350"}[text] + out = origins_from_output( + {"origins": [item(kind="program", reference=ref, codes=[code])]}, + text, + source="fr", + codes=CODES, + rule_year=2023, + ) + assert out == [] + + +def test_fr_recap_of_the_same_year_is_a_plain_statement(): + text = "In the CY 2023 PFS final rule, we established the homebound exception for CPT code 99342." + out = origins_from_output( + { + "origins": [ + item( + kind="program", reference="the homebound exception", codes=["99342"] + ) + ] + }, + text, + source="fr", + codes=CODES, + rule_year=2023, + ) + assert len(out) == 1 + + +def test_codes_outside_the_family_or_not_named_are_dropped(): + out = origins_from_output( + {"origins": [item(codes=["99341", "99999", "99342", "99341"])]}, + FR_STATUTE, + source="fr", + codes=CODES, + ) + assert out[0]["codes"] == ["99341"] # 99999 foreign, 99342 not in the text + gone = origins_from_output( + {"origins": [item(codes=["99999"])]}, FR_STATUTE, source="fr", codes=CODES + ) + assert gone == [] + + +def test_code_ranges_in_the_text_name_every_code(): + text = "Section 1861(m) of the Act covers CPT codes 99341-99342." + out = origins_from_output( + {"origins": [item(codes=["99342"])]}, text, source="fr", codes=CODES + ) + assert len(out) == 1 + + +def test_year_and_numbers_must_be_in_the_text(): + ok = item(year=2023, summary="Section 1861 and 2023 set it up.") + text = "Section 1861(m) of the Act, effective 2023, sets it up for CPT code 99341." + [row] = origins_from_output({"origins": [ok]}, text, source="fr", codes=CODES) + assert row["year"] == 2023 + # a year the text never states is dropped to null, not guessed + [row] = origins_from_output( + {"origins": [item(year=1965)]}, text, source="fr", codes=CODES + ) + assert row["year"] is None + # a summary number the text lacks drops the item + assert ( + origins_from_output( + {"origins": [item(summary="It began in 1988 under the Act.")]}, + text, + source="fr", + codes=CODES, + ) + == [] + ) + + +def test_origins_from_output_tolerates_missing_codes_filter_and_empty_output(): + out = origins_from_output({"origins": [item()]}, FR_STATUTE, source="fr") + assert out[0]["codes"] == ["99341"] + assert origins_from_output({}, FR_STATUTE, source="fr") == [] + assert ( + origins_from_output({"origins": [item(codes=[])]}, FR_STATUTE, source="fr") + == [] + ) + + +def test_schema_and_version(): + assert ORIGIN_PROMPT_VERSION == "origin-v1" + kinds = ORIGIN_SCHEMA["properties"]["origins"]["items"]["properties"]["kind"] + assert tuple(kinds["enum"]) == ORIGIN_KINDS + assert ORIGIN_KINDS == ( + "statute", + "regulation", + "program", + "demonstration", + "cpt_panel", + "cms_gcode", + "coverage", + ) + + +def test_extract_origin_asks_with_the_schema_and_filters(): + calls: list = [] + out = extract_origin( + FR_STATUTE, + ask_with([item(), item(reference="not in the text")], calls), + source="fr", + codes=CODES, + ) + assert [o["reference"] for o in out] == ["Section 1861(m) of the Act"] + system, user, schema = calls[0] + assert schema is ORIGIN_SCHEMA and "99341" in user and FR_STATUTE in user + assert "never quote" in system.lower() or "paraphras" in system.lower() + + +def test_extract_origin_failed_reply_is_empty(): + ask = lambda s, u, sc: None # noqa: E731 + assert extract_origin(FR_STATUTE, ask, source="fr") == [] + + +def test_extract_origin_book_system_differs_from_fr(): + calls: list = [] + extract_origin(BOOK_PAGES["12"], ask_with([], calls), source="book", codes=CODES) + extract_origin(FR_STATUTE, ask_with([], calls), source="fr", codes=CODES) + assert calls[0][0] != calls[1][0] + assert "8" in calls[0][0] or "eight" in calls[0][0].lower() + + +def test_extract_origin_rejects_an_unknown_source(): + with pytest.raises(ValueError): + extract_origin("x", ask_with([]), source="web") + + +# ── candidates ─────────────────────────────────────────────────────── + + +def test_book_candidates(tmp_path): + path = make_epub(tmp_path / "b.epub", {"": "Lead 99341 text.", **BOOK_PAGES}) + got = book_candidates(path, CODES) + assert [label for label, _ in got] == ["12", "13"] # unlabeled text can't be cited + assert book_candidates(path, ()) == [] + + +def test_book_candidates_see_code_ranges(tmp_path): + path = make_epub(tmp_path / "b.epub", {"4": "Codes 99341-99342 were revised."}) + assert [label for label, _ in book_candidates(path, ("99342",))] == ["4"] + + +TITLE = "Medicare Program; CY 2023 PFS Final Rule" + + +def _store(tmp_path): + s = Store(":memory:", storage_dir=tmp_path / "st") + key = s.create( + Rule(title=TITLE, document_number="22-1", date_published="2022-11-18") + ) + other = s.create( + Rule( + title="Home Health Rule", + document_number="22-2", + date_published="2022-06-01", + ) + ) + corr = s.create( + Rule( + title="CY 2023 PFS Final Rule; Correction", + document_number="22-3", + date_published="2023-01-01", + ) + ) + anchors = [ + (key, 1, 100, 1, "An unrelated opening paragraph."), + (key, 2, 100, 2, FR_STATUTE), + (key, 3, 101, 3, "CPT code 99341 appears here with nothing about origins."), + ( + key, + 4, + 101, + 4, + "The home or residence services family includes a new benefit.", + ), + ( + other, + 7, + 55, + 1, + "Section 1895 of the Act created the demonstration for 99350.", + ), + (corr, 9, 1, 1, FR_STATUTE), + ] + s._con().executemany( + "INSERT INTO fr_anchors (item_key, p_id, page, ordinal, text) VALUES (?,?,?,?,?)", + anchors, + ) + for k, doc in ((key, "22-1"), (other, "22-2"), (corr, "22-3")): + s._con().execute( + "INSERT INTO fr_anchor_docs (item_key, document_number, html_url, start_page," + " end_page, fr_volume) VALUES (?,?,'u',1,2,87)", + (k, doc), + ) + s._con().commit() + return s, key, other + + +def test_fr_origin_candidates_filter(tmp_path): + s, key, other = _store(tmp_path) + got = fr_origin_candidates(s, CODES, FAM.synonyms) + assert [(g["item_key"], g["p_id"]) for g in got] == [(other, 7), (key, 2), (key, 4)] + first = got[1] + assert first["page"] == 100 and first["rule_year"] == 2023 + assert first["rule_kind"] == "final" and first["text"] == FR_STATUTE + assert fr_origin_candidates(s, (), ()) == [] + + +def test_resolve_book(tmp_path): + s = Store(":memory:", storage_dir=tmp_path / "st") + from bib.item import Source # noqa: PLC0415 + + key = s.create(Source(title="Synthetic Changes 2023")) + assert resolve_book(s, key) is None # no attachment + epub = make_epub(tmp_path / "book.epub", BOOK_PAGES) + s.attach_file(key, epub) + label, path = resolve_book(s, key) + assert label == "Synthetic Changes 2023" and path.name.endswith(".epub") + txt = tmp_path / "n.txt" + txt.write_text("x") + k2 = s.create(Source(title="No epub")) + s.attach_file(k2, txt) + assert resolve_book(s, k2) is None + + +# ── guidance ───────────────────────────────────────────────────────── + + +def _con(): + c = duckdb.connect(":memory:") + ensure_tables(c) + return c + + +def test_guidance_origins_map_kinds(): + c = _con() + g = [ + GuidanceRow( + "HOME-VISITS", "99341", "cfr", "42 CFR 410.78(a)(3)", "", "K1", 5, 100 + ), + GuidanceRow( + "HOME-VISITS", "99341", "cfr", "42 CFR 410.78(a)(3)", "", "K2", 9, 200 + ), + GuidanceRow( + "HOME-VISITS", + "99341", + "iom", + "100-04 ch.12 §30.6.4", + "IOM1", + "K1", + 6, + 100, + 3, + ), + GuidanceRow("HOME-VISITS", "99350", "mln", "MLN 907166", "", "K3", 0, 0), + ] + write_guidance(c, "HOME-VISITS", g) + rows = guidance_origins(c, "HOME-VISITS") + by = {(r.code, r.cite_kind): r for r in rows} + assert len(rows) == 3 # the CFR cite repeated by two paragraphs collapses + cfr = by[("99341", "cfr")] + assert (cfr.kind, cfr.reference, cfr.extractor) == ( + "regulation", + "42 CFR 410.78(a)(3)", + "guidance", + ) + assert (cfr.item_key, cfr.p_id, cfr.page_label, cfr.year) == ("K1", 5, "100", None) + assert cfr.locator == "42 CFR 410.78(a)(3)" and len(cfr.summary) <= 240 + assert by[("99341", "iom")].kind == "coverage" + mln = by[("99350", "mln")] + assert mln.kind == "coverage" and mln.page_label == "" and mln.p_id == 0 + + +# ── family run ─────────────────────────────────────────────────────── + + +def test_origin_rows_carry_page_label_for_book(tmp_path): + s, key, other = _store(tmp_path) + epub = make_epub(tmp_path / "b.epub", BOOK_PAGES) + book_item = ("BOOK1", "Synthetic Changes 2023", epub) + paraphrase = item( + kind="cpt_panel", + reference="CPT Editorial Panel action", + codes=["99341", "99350"], + summary="Editors introduced a new group of home visit entries.", + year=2023, + ) + ask = ask_with([paraphrase]) + c = _con() + rows, counts = family_origin( + c, s, FAM, ask=ask, log=RunLog(tmp_path / "l.jsonl"), books=[book_item] + ) + book = [r for r in rows if r.cite_kind == "cpt_book"] + assert {(r.code, r.page_label, r.p_id, r.item_key) for r in book} == { + ("99341", "12", 0, "BOOK1"), + ("99350", "12", 0, "BOOK1"), + } + r = book[0] + assert r.locator == "Synthetic Changes 2023, p. 12" and r.year == 2023 + assert r.extractor == f"model:qwen-test@{ORIGIN_PROMPT_VERSION}" + assert all(len(r.summary) <= 240 for r in rows) + assert counts["failed"] == 0 + + +def test_family_origin_fr_rows_cite_the_paragraph_and_dedupe(tmp_path): + s, key, other = _store(tmp_path) + ask = ask_with( + [ + item(codes=["99341"]), + item(reference="not in the paragraph", codes=["99341"]), + ] + ) + c = _con() + write_guidance( + c, + "HOME-VISITS", + [GuidanceRow("HOME-VISITS", "99341", "cfr", "42 CFR 410.78", "", key, 2, 100)], + ) + rows, counts = family_origin( + c, s, FAM, ask=ask, log=RunLog(tmp_path / "l.jsonl"), books=[] + ) + fr = [r for r in rows if r.cite_kind == "fr_para"] + assert [(r.code, r.item_key, r.p_id, r.page_label) for r in fr] == [ + ("99341", key, 2, "100") + ] + assert fr[0].locator == "FR p. 100, ¶2" and fr[0].kind == "statute" + assert [r.cite_kind for r in rows if r.extractor == "guidance"] == ["cfr"] + assert counts == {"done": 2, "skipped": 0, "failed": 0, "failed_pending": 0} + + +def test_family_origin_resumes_and_replays_without_a_model(tmp_path): + s, key, other = _store(tmp_path) + epub = make_epub(tmp_path / "b.epub", BOOK_PAGES) + books = [("BOOK1", "Synthetic Changes 2023", epub)] + log = RunLog(tmp_path / "l.jsonl") + calls: list = [] + ask = ask_with([item()], calls) + first, c1 = family_origin(_con(), s, FAM, ask=ask, log=log, books=books) + n = len(calls) + assert c1["done"] == n and n > 0 + # a second run with the same ledger asks nothing + again, c2 = family_origin(_con(), s, FAM, ask=ask, log=log, books=books) + assert len(calls) == n and c2["skipped"] == n and again == first + # no model: the finished ledger is replayed, model rows are kept + replay, c3 = family_origin( + _con(), + s, + FAM, + ask=None, + log=RunLog(tmp_path / "l.jsonl"), + books=books, + replay_model="qwen-test", + ) + assert replay == first and c3 == { + "done": 0, + "skipped": 0, + "failed": 0, + "failed_pending": 0, + } + # no model and nothing to replay: only the deterministic guidance rows + none, _ = family_origin(_con(), s, FAM, ask=None, log=None, books=books) + assert none == [] + + +def test_family_origin_ledger_holds_no_copied_book_text(tmp_path): + s, key, other = _store(tmp_path) + epub = make_epub(tmp_path / "b.epub", BOOK_PAGES) + copied = item( + kind="cpt_panel", + reference="CPT Editorial Panel", + codes=["99341"], + summary="adopted new home visit family entries beginning with 99341 and 99350", + ) + log = RunLog(tmp_path / "l.jsonl") + rows, _ = family_origin( + _con(), + s, + FAM, + ask=ask_with([copied]), + log=log, + books=[("BOOK1", "Synthetic Changes 2023", epub)], + ) + assert not [r for r in rows if r.cite_kind == "cpt_book"] + lines = (tmp_path / "l.jsonl").read_text().splitlines() + book = [ln for ln in lines if '"unit": "book:BOOK1:12' in ln] + assert book and all("adopted new home visit" not in ln for ln in book) + + +def test_family_origin_limit_failed_and_retry(tmp_path): + s, key, other = _store(tmp_path) + log = RunLog(tmp_path / "l.jsonl") + ask = ask_with([item()]) + _, c = family_origin(_con(), s, FAM, ask=ask, log=log, books=[], limit=1) + assert c["done"] == 1 + failing = lambda sy, u, sc: None # noqa: E731 + failing.model = "qwen-test" + _, c = family_origin(_con(), s, FAM, ask=failing, log=log, books=[]) + assert c["failed"] == 1 and c["skipped"] == 1 + _, c = family_origin(_con(), s, FAM, ask=failing, log=log, books=[]) + assert c["failed_pending"] == 1 + _, c = family_origin(_con(), s, FAM, ask=ask, log=log, books=[], retry_failed=True) + assert c["done"] == 1 + + +def test_origin_inventory(tmp_path): + s, key, other = _store(tmp_path) + epub = make_epub(tmp_path / "b.epub", BOOK_PAGES) + paras, pages = origin_inventory(s, FAM, [("BOOK1", "Synthetic Changes 2023", epub)]) + assert len(paras) == 2 and pages == {"BOOK1": ["12", "13"]} + + +# ── writers ────────────────────────────────────────────────────────── + + +def _row(**kw): + base = dict( + family="HOME-VISITS", + code="99341", + kind="statute", + reference="Section 1861(m) of the Act", + year=None, + summary="A paraphrase.", + cite_kind="fr_para", + item_key="K", + p_id=2, + page_label="100", + locator="FR p. 100, ¶2", + extractor="m", + ) + return OriginRow(**{**base, **kw}) + + +def test_write_read_origin_round_trip_and_replace(): + c = duckdb.connect(":memory:") + rows = [_row(), _row(code="99350", year=2023), _row(year=1965, reference="A")] + assert write_origin(c, "HOME-VISITS", rows) == 3 + write_origin(c, "OTHER", [_row(family="OTHER")]) + got = read_origin(c, "HOME-VISITS") + assert len(got) == 3 and got[0].year is None and got[1].year == 1965 + assert write_origin(c, "HOME-VISITS", [_row()]) == 1 + assert len(read_origin(c, "HOME-VISITS")) == 1 and len(read_origin(c, "OTHER")) == 1 + assert write_origin(c, "HOME-VISITS", []) == 0 + + +def _fr(it, text, **kw): + return origins_from_output({"origins": [it]}, text, source="fr", codes=CODES, **kw) + + +def test_kind_follows_what_the_reference_is(): + # a section of the Act is a statute whatever the model called it + [row] = _fr(item(kind="regulation"), FR_STATUTE) + assert row["kind"] == "statute" + text = "Section 425.20 of 42 CFR defines primary care services for CPT code 99341." + [row] = _fr(item(kind="statute", reference="of 42 CFR"), text) + assert row["kind"] == "regulation" + # a statute/regulation claim no reference supports is dropped + assert ( + _fr( + item(kind="regulation", reference="the CY 2026 MPFS"), + "The CY 2026 MPFS covers CPT code 99341.", + ) + == [] + ) + # a G-code creation must name a HCPCS code + g = "CMS created HCPCS code G2211 to pay for visits with CPT code 99341." + assert _fr(item(kind="cms_gcode", reference="HCPCS code G2211"), g) + assert ( + _fr( + item(kind="cms_gcode", reference="a new add-on"), + "A new add-on exists for CPT code 99341.", + ) + == [] + ) + + +def test_pointers_into_the_same_rule_and_proposals_are_no_origin(): + ptr = "Section II.F. of this proposed rule covers the program for CPT code 99341." + assert ( + _fr(item(kind="program", reference="section II.F. of this proposed rule"), ptr) + == [] + ) + text = "The program for CPT code 99341 is described in the telehealth section." + assert ( + _fr( + item( + kind="program", + reference="The program", + summary="Proposes to add the codes.", + ), + text, + ) + == [] + ) + assert _fr( + item(kind="program", reference="The program", summary="It adds the codes."), + text, + ) + + +def test_codes_must_sit_near_the_sentence_that_writes_the_reference(): + text = ( + "Section 1861(m) of the Act establishes the home health benefit. " + "That is all it does. Unrelated sentence here. " + "Later we discuss CPT code 99341 in a different context." + ) + # 99341 is three sentences from the reference sentence: no origin for it + assert _fr(item(codes=["99341"]), text) == [] + near = ( + "Section 1861(m) of the Act establishes the home health benefit. " + "It covers CPT code 99341." + ) + assert _fr(item(codes=["99341"]), near) + + +def test_book_rows_are_kept_per_page_and_fr_rows_deduped(tmp_path): + s, key, other = _store(tmp_path) + pages = { + "5": "Rationale: 99341 was restructured last year.", + "6": "Rationale: 99341 and 99350 were revised for 2023.", + } + epub = make_epub(tmp_path / "b.epub", pages) + it = item( + kind="cpt_panel", + reference="CPT Editorial Panel action", + codes=["99341"], + summary="Editors revised the visit codes.", + ) + rows, _ = family_origin( + _con(), + s, + FAM, + ask=ask_with([it]), + log=RunLog(tmp_path / "l.jsonl"), + books=[("B", "Synthetic Changes 2023", epub)], + ) + assert [(r.page_label) for r in rows if r.cite_kind == "cpt_book"] == ["5", "6"] + + +def test_a_summary_naming_codes_applies_only_to_those_codes(): + text = "Section 1861(m) of the Act covers CPT codes 99341, 99342 and 99350." + out = _fr( + item( + codes=["99341", "99342", "99350"], + summary="The Act covers codes 99341-99342.", + ), + text, + ) + assert out[0]["codes"] == ["99341", "99342"] + # a summary naming no code keeps every named code + assert _fr(item(codes=["99341", "99350"]), text)[0]["codes"] == ["99341", "99350"] + + +def test_a_book_page_without_prose_states_no_origin(): + table = "99341 Home visit for a new patient, 20 minutes. 99350 Home visit." + it = item( + kind="cpt_panel", + reference="CPT Editorial Panel action", + codes=["99341"], + summary="Editors created the code.", + ) + assert ( + origins_from_output({"origins": [it]}, table, source="book", codes=CODES) == [] + ) + prose = table + " Rationale: 99341 was restructured." + assert origins_from_output({"origins": [it]}, prose, source="book", codes=CODES) diff --git a/tests/rex/comments/test_epub_text.py b/tests/rex/comments/test_epub_text.py index c8ba2a5..48310c3 100644 --- a/tests/rex/comments/test_epub_text.py +++ b/tests/rex/comments/test_epub_text.py @@ -5,7 +5,7 @@ from __future__ import annotations import zipfile from pathlib import Path -from rex.comments.epub_text import content_root, extract_text, spine_paths +from rex.comments.epub_text import content_root, epub_pages, extract_text, spine_paths _CONTAINER_XML = ( '\n' @@ -104,3 +104,62 @@ class TestExtractText: text = extract_text(path) assert "

" not in text assert "CPT® codes & values" in text + + +def _paged_epub(path: Path, a: str, b: str, *, container: bool = True) -> None: + with zipfile.ZipFile(path, "w") as zf: + zf.writestr("mimetype", "application/epub+zip") + if container: + zf.writestr("META-INF/container.xml", _CONTAINER_XML) + zf.writestr("OEBPS/content.opf", _opf(["a", "b"])) + zf.writestr("OEBPS/a.xhtml", a) + zf.writestr("OEBPS/b.xhtml", b) + else: + zf.writestr("OPS/a.xhtml", a) + zf.writestr("OPS/b.xhtml", b) + + +def test_epub_pages_split_on_page_spans(tmp_path): + path = tmp_path / "book.epub" + _paged_epub( + path, + '

Front matter.

Nine text.

' + '

Twelve one.

Twelve two.

', + '

Twelve continues.

' + '

Fourteen.

', + ) + assert epub_pages(path) == [ + ("", "Front matter."), + ("ix", "Nine text."), + ("12", "Twelve one.\nTwelve two.\n\nTwelve continues."), + ("14", "Fourteen."), + ] + + +def test_epub_pages_without_container_sorts_members(tmp_path): + path = tmp_path / "book.epub" + _paged_epub( + path, + '

One.

', + '

Two.

', + container=False, + ) + assert epub_pages(path) == [("1", "One."), ("2", "Two.")] + + +def test_epub_pages_skips_unreadable_spine_member(tmp_path): + path = tmp_path / "book.epub" + with zipfile.ZipFile(path, "w") as zf: + zf.writestr("META-INF/container.xml", _CONTAINER_XML) + zf.writestr("OEBPS/content.opf", _opf(["a", "b"])) # b.xhtml never written + zf.writestr("OEBPS/a.xhtml", "

Only A.

") + assert epub_pages(path) == [("", "Only A.")] + + +def test_extract_text_skips_unreadable_spine_member(tmp_path): + path = tmp_path / "book.epub" + with zipfile.ZipFile(path, "w") as zf: + zf.writestr("META-INF/container.xml", _CONTAINER_XML) + zf.writestr("OEBPS/content.opf", _opf(["a", "b"])) + zf.writestr("OEBPS/a.xhtml", "

Only A.

") + assert extract_text(path) == "Only A."