homelab/stack
1
0
Fork 0
Code Issues 210 Pull Requests 2 Actions Packages Projects 1 Releases 1 Wiki Activity

fix(pfs): units without money say why; seat notes stored; seat ingest_log rows idempotent (R89)

Browse Source 
A unit with no billing specialty reads "no billing specialty identified" and a
known specialty with no PSPS row "no family billing on record" (stance stays
n/a) instead of claiming it bills under 1%. The seat page's self-disagreement
notes are stored as ingest_log rows (rule_id note) for the dossier, and a rerun
replaces a year's gap and note rows instead of duplicating them.
This commit is contained in:
kert
2026-11-12 21:40:53 -05:00
parent 6226341ff9
commit 0f331176f2
4 changed files with 141 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
src/cli/pfs.py
View File

@@ -1288,6 +1288,7 @@ def seats(
refusal = ""
staged: list[tuple[int, str, str, Path, list[Any]]] = []
texts: dict[int, str] = {}
year_notes: dict[int, list[str]] = {}
client = _http_dossier()
try:
snaps = seat_lib.cdx_snapshots(client, url, years, gaps=gaps)
@@ -1341,6 +1342,7 @@ def seats(
continue
for note in notes:
typer.echo(f"note: {body} {y}: {note}")
year_notes[y] = notes
texts[y] = text
staged.append(
(
@@ -1374,28 +1376,50 @@ def seats(
if cross_check and staged:
_seats_cross_check(body, staged, texts)
if write and (rows or gaps):
from cms.ingest_log import log_ingest
from cms.ingest_log import ensure_table, log_ingest
def _log(bcon: Any, *, source: str, rule: str = "", **kw: Any) -> None:
"""One ingest_log row, replacing any earlier row for the same
source and kind so a rerun does not duplicate it."""
bcon.execute(
"DELETE FROM cms.ingest_log WHERE module = 'pfs.seats' "
"AND source_file = ? AND rule_id = ?",
[source, rule],
)
log_ingest(
bcon,
module="pfs.seats",
table_name=seat_lib.TABLE,
source_file=source,
rule_id=rule,
**kw,
)
with _batch() as bcon:
ensure_table(bcon)
for y in sorted({*gaps, *(st[0] for st in staged)}):
# a rerun replaces this year's gap and note rows (also the
# earlier unkinded gap rows) instead of adding to them
bcon.execute(
"DELETE FROM cms.ingest_log WHERE module = 'pfs.seats' "
"AND starts_with(source_file, ?)",
[f"{url}#{y} "],
)
if rows:
seat_lib.write_seats(bcon, body, rows)
for y, _ts, ikey, src, got in staged:
log_ingest(
bcon,
module="pfs.seats",
table_name=seat_lib.TABLE,
rows=len(got),
source_file=str(src),
pincite_key=ikey,
)
_log(bcon, source=str(src), rows=len(got), pincite_key=ikey)
for note in year_notes.get(y, []):
# the page disagrees with itself: kept for the dossier
_log(
bcon,
source=f"{url}#{y} note: {note}",
rule="note",
rows=len(got),
pincite_key=ikey,
)
for y in sorted(gaps):
log_ingest(
bcon,
module="pfs.seats",
table_name=seat_lib.TABLE,
rows=0,
source_file=f"{url}#{y} ({gaps[y]})",
)
_log(bcon, source=f"{url}#{y} ({gaps[y]})", rule="gap", rows=0)
_publish()
typer.echo(f"wrote {len(rows)} seats")

16
src/pfs/stakes.py
View File

@@ -481,19 +481,26 @@ _CHANGE = 0.005
def expected_stance(
*,
org_type: str,
share: float,
share: float | None,
delta_pct: float | None,
new_payment: bool,
new_codes: Sequence[str] = (),
specialty: str | None = None,
) -> Expectation:
"""Phase-1 explicit rule. A payer, patient group, "other" or unknown
organization is not a billing stakeholder, nor is a specialty billing
under 1% of the family (both ``n/a``); new payment expects support
organization is not a billing stakeholder; nor is a unit with no billing
specialty (*specialty* ``NONE`` or empty; ``None`` skips the check), a
specialty with no family billing on record (*share* ``None``) or one
billing under 1% of the family (all ``n/a``, each with its own reason); new payment expects support
(naming *new_codes*); otherwise the per-service payment change decides:
at most -0.5% oppose, at least +0.5% support, in between (or unknown)
unclear."""
if org_type in _NON_BILLING:
return Expectation("n/a", "not a billing stakeholder")
if specialty is not None and specialty in ("", "NONE"):
return Expectation("n/a", "no billing specialty identified")
if share is None:
return Expectation("n/a", "no family billing on record")
if share < _MIN_SHARE:
return Expectation("n/a", f"bills under 1% of the family ({share:.1%})")
if new_payment:
@@ -594,7 +601,8 @@ def stakeholder_units(
stance=dict(sorted(counts.items())),
expected=expected_stance(
org_type=org_type,
share=share,
share=m.share if m else None,
specialty=spec,
delta_pct=m.delta_pct if m else None,
new_payment=m.new_payment if m else False,
new_codes=m.new_codes if m else (),

23
tests/cli/test_pfs_cli.py
View File

@@ -3431,6 +3431,29 @@ class TestSeats:
"SELECT count(*) FROM pfs.governance_seat WHERE year = 2020"
).fetchone() == (len(rows),)
def test_rerun_replaces_gap_and_note_rows_and_stores_notes(self, con, env):
store, state, tmp_path = env
state["page"] = self._RUC.replace(b"3 of its 9 members", b"3 of its 12 members")
for _ in range(2):
assert self._run("--write").exit_code == 0
gaps = con.execute(
"SELECT count(*) FROM cms.ingest_log WHERE module = 'pfs.seats' "
"AND rule_id = 'gap' AND source_file LIKE '%#2021 %'"
).fetchone()
assert gaps == (1,)
notes = con.execute(
"SELECT source_file, rows FROM cms.ingest_log WHERE module = 'pfs.seats' "
"AND rule_id = 'note' ORDER BY source_file"
).fetchall()
assert (
len(notes) == 2
and "#2020 note: parsed 9 seats, the page says 12" in notes[0][0]
)
assert con.execute(
"SELECT count(*) FROM cms.ingest_log WHERE module = 'pfs.seats' "
"AND rows > 0 AND rule_id = ''"
).fetchone() == (2,)
def test_limit_caps_snapshots(self, con, env):
store, state, tmp_path = env
res = self._run("--limit", "1")

66
tests/pfs/test_stakes.py
View File

@@ -628,6 +628,50 @@ _NOT_BILLING = "not a billing stakeholder"
"n/a",
"bills under 1% of the family (0.0%)",
),
(
dict(
org_type="health_system",
share=None,
delta_pct=None,
new_payment=False,
specialty="NONE",
),
"n/a",
"no billing specialty identified",
),
(
dict(
org_type="individual_clinician",
share=None,
delta_pct=None,
new_payment=True,
specialty="",
),
"n/a",
"no billing specialty identified",
),
(
dict(
org_type="specialty_society",
share=None,
delta_pct=None,
new_payment=False,
specialty="PODIATRY",
),
"n/a",
"no family billing on record",
),
(
dict(
org_type="payer",
share=None,
delta_pct=None,
new_payment=False,
specialty="NONE",
),
"n/a",
"not a billing stakeholder",
),
(
dict(
org_type="specialty_society",
@@ -853,3 +897,25 @@ def test_units_are_ordered_by_org_type_then_specialty():
("individual_clinician", "GERIATRIC MEDICINE"),
("individual", "NONE"),
]
def test_units_without_money_say_why_not_that_they_bill_under_one_percent():
cs = [
_c("h", "health_system", "NONE", org="Some House-Call Group"),
_c("s", "specialty_society", "PODIATRY", org="A Podiatry Society"),
_c("i", "individual_clinician", "", name="A Clinician"),
]
units = {
(u.org_type, u.specialty): u
for u in stakeholder_units(cs, [], _MONEY, event_key="E1", year=2023)
}
assert units[("health_system", "NONE")].expected == Expectation(
"n/a", "no billing specialty identified"
)
assert units[("individual_clinician", "NONE")].expected.reason == (
"no billing specialty identified"
)
assert units[("specialty_society", "PODIATRY")].expected == Expectation(
"n/a", "no family billing on record"
)
assert units[("specialty_society", "PODIATRY")].share == 0.0