697 lines
24 KiB
Python
697 lines
24 KiB
Python
"""stack pfs — code elements, lineage events and derived families.
|
|
|
|
uv run stack pfs elements --code G0556 --code 99490
|
|
uv run stack pfs elements --family CCM
|
|
uv run stack pfs elements --all-payable --no-llm --dry-run
|
|
uv run stack pfs lineage --code G2058 [--write]
|
|
uv run stack pfs families [--write]
|
|
uv run stack pfs guidance --family CCM [--family …] [--write]
|
|
uv run stack pfs reaction --family CCM [--code G2211] [--stance-sample 30] [--write]
|
|
uv run stack pfs review [--code G0556]
|
|
uv run stack pfs cpt-ingest --edition 2024 [--edition 2022 …] [--dry-run]
|
|
uv run stack pfs cpt-ingest --all
|
|
|
|
Writes go through ``duckdb_batch`` (single-writer rule) and republish
|
|
the read-only replica so the chat and notebooks see them.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any
|
|
|
|
import typer
|
|
|
|
from pfs.codetables import (
|
|
cpt_years,
|
|
ensure_tables,
|
|
is_missing_table_error,
|
|
read_all_elements,
|
|
read_all_events,
|
|
read_all_reviews,
|
|
read_cpt_codes,
|
|
read_cpt_instructions,
|
|
read_cpt_sections,
|
|
read_guidance,
|
|
read_reviews,
|
|
write_elements,
|
|
write_events,
|
|
write_families,
|
|
write_guidance,
|
|
write_reaction,
|
|
)
|
|
from pfs.cpt_load import ingest as cpt_ingest
|
|
from pfs.extract import extract_code
|
|
from pfs.families import FAMILIES, HAND_FAMILIES, derive_families, refresh_from
|
|
from pfs.guidance import build as build_guidance
|
|
from pfs.lineage import lineage, lineage_all
|
|
from pfs.reaction import series as reaction_series
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
app = typer.Typer(no_args_is_help=True)
|
|
|
|
|
|
#: A replica that hasn't had `elements`/`lineage --write` (or an initial
|
|
#: `families --write`) run yet is missing `pfs.code_element`/
|
|
#: `pfs.code_event`/`pfs.code_family` — not a bug, just nothing derived
|
|
#: yet (I4). DuckDB's "table does not exist" errors are Catalog errors —
|
|
#: the predicate itself lives in `pfs.codetables` so `pfs.lineage`/
|
|
#: `pfs.extract` can reuse it for the same replica shape (a CPT-table-
|
|
#: free replica is not a bug either).
|
|
def _missing_table(exc: Exception) -> bool:
|
|
return is_missing_table_error(exc)
|
|
|
|
|
|
# ── indirections the tests monkeypatch ───────────────────────────────
|
|
def _batch() -> Any:
|
|
from conf.connect import duckdb_batch
|
|
|
|
return duckdb_batch("aco")
|
|
|
|
|
|
def _read() -> Any:
|
|
from conf.connect import duckdb
|
|
|
|
return duckdb("aco", read_only=True)
|
|
|
|
|
|
def _store() -> Any:
|
|
from conf.connect import bib
|
|
|
|
return bib()
|
|
|
|
|
|
def _publish() -> None:
|
|
from conf.connect import publish_replica
|
|
|
|
typer.echo(f"replica → {publish_replica('aco')}")
|
|
|
|
|
|
def _classifier() -> Any:
|
|
from llm import config as llm_config
|
|
from llm.classify import closed_vocab_classifier
|
|
from llm.pool import HostPool
|
|
|
|
cfg = llm_config.load()
|
|
return closed_vocab_classifier(cfg, HostPool.from_config(cfg))
|
|
|
|
|
|
def _engine() -> Any:
|
|
# llm.index/llm.evidence expose no public "give me a bare SQLAlchemy
|
|
# engine over the pgvector DB" helper — cli/bib.py and cli/llm.py
|
|
# both already reach into llm.index._engine for the same reason, so
|
|
# this is the established reuse, not a new private import.
|
|
from llm import config as llm_config
|
|
from llm.index import _engine as pgvector_engine
|
|
|
|
return pgvector_engine(llm_config.load())
|
|
|
|
|
|
def _run_elements(
|
|
con: Any, store: Any, targets: list[str], classify: Any, *, write: bool
|
|
) -> None:
|
|
for c in targets:
|
|
x = extract_code(store, con, c, classify=classify)
|
|
if write:
|
|
write_elements(con, c, x.rows, x.reviews)
|
|
confirmed = sum(1 for r in x.rows if r.confirmed_by)
|
|
note = f", {confirmed} confirmed by a second source" if confirmed else ""
|
|
typer.echo(f"{c}: {len(x.rows)} elements, {len(x.reviews)} for review{note}")
|
|
|
|
|
|
def _codes_for(
|
|
con: Any,
|
|
codes: list[str],
|
|
families: list[str],
|
|
all_payable: bool,
|
|
*,
|
|
warn_slow: bool = True,
|
|
) -> list[str]:
|
|
"""*warn_slow*: the "~20 min of SQL before any model call" warning
|
|
describes ``elements``' per-code LLM classification pass — it does
|
|
NOT apply to ``lineage --all-payable`` (one inverted pass over
|
|
``fr_anchors``, seconds not minutes), which passes ``warn_slow=False``."""
|
|
out: list[str] = [c.upper() for c in codes]
|
|
if families:
|
|
refresh_from(con)
|
|
for family in families:
|
|
fam = FAMILIES.get(family.upper())
|
|
if fam is None:
|
|
# List only the hand families here (M3) — FAMILIES can hold
|
|
# ~15k derived keys once `refresh_from` has run, and that
|
|
# dump is useless in an error message.
|
|
raise typer.BadParameter(
|
|
f"unknown family {family!r}; known: {', '.join(HAND_FAMILIES)}"
|
|
)
|
|
out.extend(fam.codes)
|
|
if all_payable:
|
|
rows = con.execute(
|
|
"SELECT DISTINCT hcpcs FROM pfs.rvu WHERE status_code IN ('A','R','T') "
|
|
"AND year = (SELECT max(year) FROM pfs.rvu) ORDER BY hcpcs"
|
|
).fetchall()
|
|
if warn_slow:
|
|
log.warning(
|
|
"--all-payable is experimental: targeting %d codes (~20 min of SQL "
|
|
"before any model call)",
|
|
len(rows),
|
|
)
|
|
out.extend(r[0] for r in rows)
|
|
if not out:
|
|
raise typer.BadParameter("pass --code, --family or --all-payable")
|
|
return sorted(set(out))
|
|
|
|
|
|
@app.command()
|
|
def elements(
|
|
code: list[str] = typer.Option([], "--code", help="HCPCS/CPT code (repeatable)."),
|
|
family: list[str] = typer.Option(
|
|
[], "--family", help="Expand a registered family (CCM, APCM, …); repeatable."
|
|
),
|
|
all_payable: bool = typer.Option(
|
|
False,
|
|
"--all-payable",
|
|
help=(
|
|
"Every A/R/T code in the newest RVU year (experimental: ~20 min "
|
|
"of SQL before any model call)."
|
|
),
|
|
),
|
|
no_llm: bool = typer.Option(
|
|
False,
|
|
"--no-llm",
|
|
help="Skip the local-model classifier (unknown lines go to review).",
|
|
),
|
|
dry_run: bool = typer.Option(
|
|
False, "--dry-run", help="Extract and report; write nothing."
|
|
),
|
|
) -> None:
|
|
"""Extract typed elements for codes into pfs.code_element (+ review queue)."""
|
|
store = _store()
|
|
classify = None if no_llm else _classifier()
|
|
if dry_run:
|
|
# A preview never contends for the DuckDB single-writer lock a
|
|
# notebook may be holding (#508-#514) — read the replica, don't
|
|
# open a batch writer.
|
|
con = _read()
|
|
try:
|
|
targets = _codes_for(con, code, family, all_payable)
|
|
_run_elements(con, store, targets, classify, write=False)
|
|
finally:
|
|
con.close()
|
|
else:
|
|
with _batch() as con:
|
|
ensure_tables(con)
|
|
targets = _codes_for(con, code, family, all_payable)
|
|
_run_elements(con, store, targets, classify, write=True)
|
|
_publish()
|
|
|
|
|
|
def _print_events(events: list[Any]) -> None:
|
|
for e in events:
|
|
arrow = f"{e.from_codes or '·'}→{e.to_codes or '·'}"
|
|
if e.source == "cpt":
|
|
anchor = f"cpt {e.year} {e.item_key}"
|
|
elif e.item_key:
|
|
anchor = f"{e.item_key} ¶{e.p_id}"
|
|
else:
|
|
anchor = f"rvu {e.note}".strip()
|
|
flag = "" if e.anchored else " UNANCHORED"
|
|
typer.echo(f"{e.year} {e.kind:<18}{arrow:<16}{anchor}{flag}")
|
|
|
|
|
|
def _print_lineage_all(by_code: dict[str, list[Any]]) -> None:
|
|
per_source: dict[str, int] = {}
|
|
with_events = 0
|
|
unanchored = 0
|
|
for events in by_code.values():
|
|
if events:
|
|
with_events += 1
|
|
for e in events:
|
|
per_source[e.source] = per_source.get(e.source, 0) + 1
|
|
if e.source == "rvu" and not e.anchored:
|
|
unanchored += 1
|
|
detail = ", ".join(f"{k}={v}" for k, v in sorted(per_source.items())) or "none"
|
|
typer.echo(
|
|
f"lineage: {len(by_code)} codes targeted, {with_events} with >=1 event, "
|
|
f"events per source: {detail}, unanchored rvu events: {unanchored}"
|
|
)
|
|
|
|
|
|
@app.command("lineage")
|
|
def lineage_cmd(
|
|
code: str = typer.Option("", "--code"),
|
|
all_payable: bool = typer.Option(
|
|
False,
|
|
"--all-payable",
|
|
help="Every A/R/T code in the newest RVU year, one inverted pass over fr_anchors.",
|
|
),
|
|
limit: int = typer.Option(
|
|
0,
|
|
"--limit",
|
|
help="Only the first N target codes (sorted) — for smoke runs with --all-payable.",
|
|
),
|
|
write: bool = typer.Option(
|
|
False, "--write", help="Persist to pfs.code_event and republish."
|
|
),
|
|
) -> None:
|
|
"""Timeline of a code: RVU-file diffs and FR paragraphs, cross-checked.
|
|
|
|
--all-payable computes lineage for every payable code in ONE pass over
|
|
fr_anchors instead of one query per code.
|
|
"""
|
|
if not code and not all_payable:
|
|
raise typer.BadParameter("pass --code or --all-payable")
|
|
store = _store()
|
|
|
|
if all_payable:
|
|
# Ruling A13: resolve targets and build every code's lineage
|
|
# against a plain DuckDB read (I4) — never the write lock a
|
|
# notebook may be holding (#508-#514). The batch below only ever
|
|
# calls write_events and publishes.
|
|
con = _read()
|
|
try:
|
|
targets = _codes_for(con, [], [], True, warn_slow=False)
|
|
if limit:
|
|
targets = targets[:limit]
|
|
by_code = lineage_all(con, store, targets)
|
|
finally:
|
|
con.close()
|
|
if write:
|
|
with _batch() as con:
|
|
ensure_tables(con)
|
|
for c in targets:
|
|
write_events(con, c, by_code.get(c, []))
|
|
_publish()
|
|
_print_lineage_all(by_code)
|
|
return
|
|
|
|
if write:
|
|
# Same A13 discipline for the single-code path: build the events
|
|
# against a plain read, then open the batch only to write them.
|
|
con = _read()
|
|
try:
|
|
events = lineage(con, store, code.upper())
|
|
finally:
|
|
con.close()
|
|
_print_events(events)
|
|
with _batch() as con:
|
|
ensure_tables(con)
|
|
write_events(con, code.upper(), events)
|
|
_publish()
|
|
return
|
|
# No --write: a plain lookup never needs the DuckDB single-writer lock
|
|
# a notebook may be holding (#508-#514) — read the replica (I4).
|
|
con = _read()
|
|
try:
|
|
try:
|
|
events = lineage(con, store, code.upper())
|
|
except Exception as exc:
|
|
if _missing_table(exc):
|
|
typer.echo("no derived tables yet — run with --write")
|
|
return
|
|
raise
|
|
_print_events(events)
|
|
finally:
|
|
con.close()
|
|
|
|
|
|
def _cpt_inputs(
|
|
con: Any,
|
|
) -> tuple[list[Any], list[Any], list[Any], dict[str, tuple[int, ...]]]:
|
|
"""The newest CPT edition's rows for ``derive_families`` (grouping/
|
|
naming stays newest-edition-scoped, Ruling C12), plus ``cpt_presence``
|
|
— every ingested edition's ``code -> years present`` map, one extra
|
|
full-table pass, used only for ``since``/``until``: a code an earlier
|
|
edition ingested but the newest one dropped still gets its true
|
|
``since`` (not clamped to the newest edition) and a real ``until``.
|
|
All empty on any replica shape that doesn't have them yet (I4: a
|
|
plain read of an old replica, pre pfs.cpt_*, is not a bug)."""
|
|
try:
|
|
years = cpt_years(con)
|
|
except Exception as exc:
|
|
if _missing_table(exc):
|
|
return [], [], [], {}
|
|
raise
|
|
if not years:
|
|
return [], [], [], {}
|
|
newest = max(years)
|
|
presence: dict[str, list[int]] = {}
|
|
for code, year in con.execute(
|
|
"SELECT code, edition_year FROM pfs.cpt_code"
|
|
).fetchall():
|
|
presence.setdefault(code, []).append(year)
|
|
cpt_presence = {code: tuple(sorted(set(yrs))) for code, yrs in presence.items()}
|
|
return (
|
|
read_cpt_codes(con, newest),
|
|
read_cpt_sections(con, newest),
|
|
read_cpt_instructions(con, newest),
|
|
cpt_presence,
|
|
)
|
|
|
|
|
|
def _derive_rows(con: Any) -> list[Any]:
|
|
codes = [
|
|
r[0]
|
|
for r in con.execute(
|
|
"SELECT DISTINCT code FROM pfs.code_element UNION SELECT DISTINCT code FROM pfs.code_event "
|
|
"UNION SELECT DISTINCT hcpcs FROM pfs.rvu WHERE year = (SELECT max(year) FROM pfs.rvu) AND status_code IN ('A','R','T')"
|
|
).fetchall()
|
|
]
|
|
# Two full-table reads grouped in Python instead of ~2xN per-code
|
|
# queries run inside the write lock (I9).
|
|
all_elements = read_all_elements(con)
|
|
all_events = read_all_events(con)
|
|
elements_by_code = {c: all_elements.get(c, []) for c in codes}
|
|
events_by_code = {c: all_events.get(c, []) for c in codes}
|
|
descriptions = {
|
|
r[0]: r[1] or ""
|
|
for r in con.execute(
|
|
"SELECT hcpcs, arg_max(description, year) FROM pfs.rvu WHERE mod IS NULL OR mod = '' GROUP BY hcpcs"
|
|
).fetchall()
|
|
}
|
|
cpt_codes, cpt_sections, cpt_instructions, cpt_presence = _cpt_inputs(con)
|
|
return derive_families(
|
|
elements_by_code,
|
|
events_by_code,
|
|
descriptions,
|
|
cpt_codes=cpt_codes,
|
|
cpt_sections=cpt_sections,
|
|
cpt_instructions=cpt_instructions,
|
|
cpt_presence=cpt_presence,
|
|
)
|
|
|
|
|
|
def _print_families(rows: list[Any], *, verbose: bool = False) -> None:
|
|
by_key: dict[str, list[Any]] = {}
|
|
for r in rows:
|
|
by_key.setdefault(r.key, []).append(r)
|
|
if verbose:
|
|
for key, members in sorted(by_key.items()):
|
|
typer.echo(f"{key}: {' '.join(f'{r.code}({r.role})' for r in members)}")
|
|
total = len(by_key)
|
|
multi = sum(1 for members in by_key.values() if len(members) >= 2)
|
|
hand = sum(1 for key in by_key if key in HAND_FAMILIES)
|
|
cpt_named = sum(
|
|
1
|
|
for key, members in by_key.items()
|
|
if key not in HAND_FAMILIES and any(r.note for r in members)
|
|
)
|
|
other = total - hand - cpt_named
|
|
typer.echo(
|
|
f"families: total {total}, multi-code {multi}, cpt-named {cpt_named}, "
|
|
f"hand {hand}, other {other}"
|
|
)
|
|
|
|
|
|
@app.command()
|
|
def families(
|
|
write: bool = typer.Option(False, "--write"),
|
|
verbose: bool = typer.Option(
|
|
False, "--verbose", help="Print every family's members, not just the summary."
|
|
),
|
|
) -> None:
|
|
"""Derive families from pfs.code_element / pfs.code_event / pfs.rvu."""
|
|
if write:
|
|
with _batch() as con:
|
|
ensure_tables(con)
|
|
rows = _derive_rows(con)
|
|
write_families(con, rows)
|
|
_publish()
|
|
# refresh_from and printing don't need the write lock open — do
|
|
# them after it's released, from the freshly published replica
|
|
# (the same pattern `stack llm restamp` uses).
|
|
refresh_from(_read())
|
|
_print_families(rows, verbose=verbose)
|
|
return
|
|
# No --write: derive-and-print never needs the write lock (I4).
|
|
con = _read()
|
|
try:
|
|
try:
|
|
rows = _derive_rows(con)
|
|
except Exception as exc:
|
|
if _missing_table(exc):
|
|
typer.echo("no derived tables yet — run with --write")
|
|
return
|
|
raise
|
|
_print_families(rows, verbose=verbose)
|
|
finally:
|
|
con.close()
|
|
|
|
|
|
def _print_guidance(rows: list[Any]) -> None:
|
|
for r in sorted(rows, key=lambda r: (r.family, r.code, r.kind, r.locator)):
|
|
item = r.item_key or "—"
|
|
typer.echo(
|
|
f"{r.family} {r.kind} {r.locator} {item} ← {r.item_key_src} ¶{r.p_id_src}"
|
|
)
|
|
|
|
|
|
def _resolve_guidance_families(con: Any, keys: list[str]) -> list[tuple[str, Any]]:
|
|
"""(key, Family) per requested key, refreshed from *con* — raises
|
|
``typer.BadParameter`` on the first unknown key, before any family's
|
|
rows are built or any write happens (mirrors ``_reaction_targets``'s
|
|
resolve-everything-up-front discipline)."""
|
|
refresh_from(con)
|
|
out: list[tuple[str, Any]] = []
|
|
for key in keys:
|
|
fam = FAMILIES.get(key)
|
|
if fam is None:
|
|
raise typer.BadParameter(
|
|
f"unknown family {key!r}; known: {', '.join(HAND_FAMILIES)}"
|
|
)
|
|
out.append((key, fam))
|
|
return out
|
|
|
|
|
|
@app.command()
|
|
def guidance(
|
|
family: list[str] = typer.Option(
|
|
[], "--family", help="Family key (CCM, APCM, …); repeatable."
|
|
),
|
|
write: bool = typer.Option(False, "--write"),
|
|
) -> None:
|
|
"""CFR/IOM/MLN sub-regulatory crosswalk per code family
|
|
(pfs.code_guidance): which CFR sections and IOM manual chapters the
|
|
family's FR paragraphs and the CPT manual cite, with FR provenance."""
|
|
if not family:
|
|
raise typer.BadParameter("pass --family (repeatable)")
|
|
keys = [f.upper() for f in family]
|
|
store = _store()
|
|
if write:
|
|
# Ruling A13 (mirrors `stack pfs reaction`): resolve every family
|
|
# key up front and build every family's rows against a plain
|
|
# DuckDB read (I4) — build_guidance only reads pfs.cpt_* — never
|
|
# the write lock a notebook may be holding (#508-#514). The batch
|
|
# below only ever calls write_guidance and publishes.
|
|
con = _read()
|
|
try:
|
|
targets = _resolve_guidance_families(con, keys)
|
|
by_family: dict[str, list[Any]] = {}
|
|
rows: list[Any] = []
|
|
for key, fam in targets:
|
|
fam_rows = build_guidance(con, store, fam.codes, family=key)
|
|
by_family[key] = fam_rows
|
|
rows.extend(fam_rows)
|
|
finally:
|
|
con.close()
|
|
|
|
with _batch() as con:
|
|
ensure_tables(con)
|
|
for key, fam_rows in by_family.items():
|
|
write_guidance(con, key, fam_rows)
|
|
_publish()
|
|
_print_guidance(rows)
|
|
return
|
|
# No --write: read-only never needs the DuckDB single-writer lock a
|
|
# notebook may be holding (#508-#514) — read the replica (I4).
|
|
con = _read()
|
|
try:
|
|
refresh_from(con)
|
|
rows = []
|
|
for key in keys:
|
|
if key not in FAMILIES:
|
|
raise typer.BadParameter(
|
|
f"unknown family {key!r}; known: {', '.join(HAND_FAMILIES)}"
|
|
)
|
|
try:
|
|
rows.extend(read_guidance(con, key))
|
|
except Exception as exc:
|
|
if _missing_table(exc):
|
|
typer.echo(f"{key}: no derived guidance yet — run with --write")
|
|
continue
|
|
raise
|
|
_print_guidance(rows)
|
|
finally:
|
|
con.close()
|
|
|
|
|
|
def _print_reaction(rows: list[Any]) -> None:
|
|
for r in sorted(rows, key=lambda r: (r.family, r.period_kind, r.year, r.period)):
|
|
if r.period_kind == "docket":
|
|
stance = ""
|
|
if (
|
|
r.stance_support
|
|
or r.stance_oppose
|
|
or r.stance_modify
|
|
or r.stance_unclear
|
|
):
|
|
stance = (
|
|
f" stance support={r.stance_support} oppose={r.stance_oppose} "
|
|
f"modify={r.stance_modify} unclear={r.stance_unclear}"
|
|
)
|
|
typer.echo(
|
|
f"{r.family} {r.period} ({r.year}) {r.n_items}/{r.n_total} commenters{stance}"
|
|
)
|
|
else:
|
|
typer.echo(
|
|
f"{r.family} {r.period} CY{r.year} "
|
|
f"{r.n_items} comment/response pairs"
|
|
)
|
|
|
|
|
|
def _reaction_targets(
|
|
con: Any, family: list[str], code: list[str]
|
|
) -> list[tuple[str, tuple[str, ...], str | None]]:
|
|
"""(key, codes, display name) per target — *name* is the registered
|
|
family's own name for a ``--family`` target, ``None`` for a
|
|
``--code`` pseudo-family (``pfs.reaction.series`` falls back to the
|
|
key itself, which is already the right label for a bare code)."""
|
|
refresh_from(con)
|
|
targets: list[tuple[str, tuple[str, ...], str | None]] = []
|
|
for f in family:
|
|
key = f.upper()
|
|
fam = FAMILIES.get(key)
|
|
if fam is None:
|
|
raise typer.BadParameter(
|
|
f"unknown family {key!r}; known: {', '.join(HAND_FAMILIES)}"
|
|
)
|
|
targets.append((key, fam.codes, fam.name))
|
|
for c in code:
|
|
key = c.upper()
|
|
targets.append((key, (key,), None))
|
|
return targets
|
|
|
|
|
|
@app.command()
|
|
def reaction(
|
|
family: list[str] = typer.Option(
|
|
[], "--family", help="Family key (CCM, APCM, …); repeatable."
|
|
),
|
|
code: list[str] = typer.Option(
|
|
[],
|
|
"--code",
|
|
help="A single code as its own pseudo-family, keyed by the code; repeatable.",
|
|
),
|
|
stance_sample: int = typer.Option(
|
|
0,
|
|
"--stance-sample",
|
|
help="Classify the newest N commenters per docket into support/oppose/"
|
|
"modify/unclear via the self-hosted classifier (0 = skip; no stance "
|
|
"sample means no LLM call at all).",
|
|
),
|
|
write: bool = typer.Option(False, "--write"),
|
|
) -> None:
|
|
"""Public-reaction series per code family (pfs.code_reaction): comment
|
|
counts per docket (share of the docket's commenters mentioning the
|
|
family), an optional stance sample, and the FR Comment:/Response:
|
|
pair counts per rule as the pre-2017 proxy. Always recomputed live
|
|
from pgvector/bib — there is no persisted-table fallback to read
|
|
without --write, so --write only adds persisting the result."""
|
|
if not family and not code:
|
|
raise typer.BadParameter("pass --family or --code (repeatable)")
|
|
store = _store()
|
|
|
|
# Resolving family keys is a plain DuckDB read (I4) — never the
|
|
# write lock a notebook may be holding (#508-#514) — and it's the
|
|
# only DuckDB touch this command needs before writing.
|
|
con = _read()
|
|
try:
|
|
targets = _reaction_targets(con, family, code)
|
|
finally:
|
|
con.close()
|
|
|
|
# Ruling A13: the classifier and every family's series (docket
|
|
# counts, the stance sample — ~1-2s/sample — and FR pairs) all run
|
|
# against pgvector/bib, not DuckDB, so they run BEFORE any DuckDB
|
|
# batch is opened. The batch below only ever calls write_reaction
|
|
# and publishes — it never holds the single-writer lock (#508-#514)
|
|
# while the slow work happens.
|
|
classify = _classifier() if stance_sample > 0 else None
|
|
engine = _engine()
|
|
by_family: dict[str, list[Any]] = {}
|
|
rows: list[Any] = []
|
|
for key, codes, name in targets:
|
|
fam_rows = reaction_series(
|
|
engine,
|
|
store,
|
|
key,
|
|
codes,
|
|
classify=classify,
|
|
stance_sample_n=stance_sample,
|
|
name=name,
|
|
)
|
|
by_family[key] = fam_rows
|
|
rows.extend(fam_rows)
|
|
|
|
if write:
|
|
with _batch() as con:
|
|
ensure_tables(con)
|
|
for key, fam_rows in by_family.items():
|
|
write_reaction(con, key, fam_rows)
|
|
_publish()
|
|
_print_reaction(rows)
|
|
|
|
|
|
@app.command()
|
|
def review(code: str = typer.Option("", "--code")) -> None:
|
|
"""Element lines the classifier could not place."""
|
|
con = _read()
|
|
rows = read_reviews(con, code.upper()) if code else read_all_reviews(con)
|
|
for r in rows:
|
|
src = f" ({r.source})" if r.source else ""
|
|
typer.echo(
|
|
f"{r.code} {r.item_key} ¶{r.p_id}{src} [{r.proposed_value or '?'}] "
|
|
f"{r.text[:120]}"
|
|
)
|
|
|
|
|
|
def _print_cpt_counts(counts: dict[int, dict[str, int]]) -> None:
|
|
for year in sorted(counts):
|
|
c = counts[year]
|
|
detail = ", ".join(f"{k}={v}" for k, v in c.items())
|
|
typer.echo(f"{year}: {detail}")
|
|
|
|
|
|
@app.command("cpt-ingest")
|
|
def cpt_ingest_cmd(
|
|
edition: list[int] = typer.Option(
|
|
[], "--edition", help="CPT edition year to ingest (repeatable)."
|
|
),
|
|
all_editions: bool = typer.Option(
|
|
False, "--all", help="Every CPT EPUB edition found in bib (source:ama)."
|
|
),
|
|
dry_run: bool = typer.Option(
|
|
False, "--dry-run", help="Parse and report counts; write nothing."
|
|
),
|
|
) -> None:
|
|
"""Parse CPT EPUB editions into pfs.cpt_section/cpt_code/cpt_instruction/
|
|
cpt_reference/cpt_crosswalk/cpt_list (docs/superpowers/specs/
|
|
2026-09-09-cpt-canonical-schema-design.md §3)."""
|
|
if not edition and not all_editions:
|
|
raise typer.BadParameter("pass --edition YEAR (repeatable) or --all")
|
|
years = None if all_editions else sorted(set(edition))
|
|
store = _store()
|
|
if dry_run:
|
|
# A preview never needs the DuckDB single-writer lock a notebook
|
|
# may be holding (#508-#514) — parsing alone answers a dry run,
|
|
# so no connection is opened at all (I4).
|
|
counts = cpt_ingest(store, None, years=years, dry_run=True)
|
|
_print_cpt_counts(counts)
|
|
return
|
|
with _batch() as con:
|
|
# write_cpt_edition (via cpt_ingest) already calls ensure_tables
|
|
# itself — no need to duplicate that here.
|
|
counts = cpt_ingest(store, con, years=years, dry_run=False)
|
|
_print_cpt_counts(counts)
|
|
_publish()
|