feat(sp41): orchestrator run_rebill() wires the pipeline end-to-end
This commit is contained in:
Nora
@@ -0,0 +1,267 @@
|
||||
"""Top-level orchestrator for SP41.
|
||||
|
||||
Wires the 835 SVC reparse → reconciliation → CARC filter → timely-filing
|
||||
gate → pipeline A → pipeline B → summary CSV → Edifabric validation.
|
||||
|
||||
The CLI and the HTTP endpoint both call this. CLI passes filesystem
|
||||
paths from --visits / --ingest / --out; HTTP endpoint passes the same.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import csv as _csv
|
||||
from dataclasses import dataclass
|
||||
from datetime import date, datetime
|
||||
from decimal import Decimal
|
||||
from pathlib import Path
|
||||
|
||||
from cyclone.rebill.carc_filter import CarcDecision, decide_carc
|
||||
from cyclone.rebill.parse_835_svc import SvcRow, parse_835_svc
|
||||
from cyclone.rebill.pipeline_a import RebillClaim
|
||||
from cyclone.rebill.pipeline_b import build_pipeline_b_batches
|
||||
from cyclone.rebill.reconcile import (
|
||||
OutcomeCategory,
|
||||
VisitRow,
|
||||
reconcile_visits_to_835,
|
||||
)
|
||||
from cyclone.rebill.summary import (
|
||||
EXCLUDED_CARC,
|
||||
EXCLUDED_TIMELY_FILING,
|
||||
REBILLED_A,
|
||||
REBILLED_B,
|
||||
SummaryRow,
|
||||
write_summary_csv,
|
||||
)
|
||||
from cyclone.rebill.timely_filing import timely_filing_decision
|
||||
|
||||
|
||||
@dataclass
|
||||
class RunResult:
|
||||
"""The return value of run_rebill — where the summary CSV landed, the
|
||||
per-disposition counts, and the list of pipeline A / B file paths.
|
||||
|
||||
Mutable (no frozen=True) because the contained `counts` dict and
|
||||
`pipeline_*_files` lists are mutated by callers who aggregate results
|
||||
across runs.
|
||||
"""
|
||||
|
||||
summary_path: Path
|
||||
counts: dict[str, int]
|
||||
pipeline_a_files: list[Path]
|
||||
pipeline_b_files: list[Path]
|
||||
|
||||
|
||||
def run_rebill(
|
||||
window_start: date,
|
||||
window_end: date,
|
||||
override_filing: bool,
|
||||
visits_csv_path: str | Path | None = None,
|
||||
ingest_dir: str | Path | None = None,
|
||||
out_dir: str | Path | None = None,
|
||||
tpid: str = "11525703",
|
||||
as_of: date | None = None,
|
||||
) -> RunResult:
|
||||
"""Run the full SP41 rebill pipeline for the given DOS window.
|
||||
|
||||
Pipeline shape:
|
||||
|
||||
1. Walk ``ingest_dir`` for *.835 / *.x12 (skipping AppleDouble shadow
|
||||
files matching ``._*``), reparse each into SvcRows via
|
||||
``parse_835_svc``, keep those with svc_date in [window_start, window_end].
|
||||
2. Load the AxisCare visits CSV (``visits_csv_path``) and keep the rows
|
||||
whose Visit Date is in the same DOS window.
|
||||
3. Reconcile visits against the in-window SVCs on the
|
||||
(member_id, procedure, DOS) match key. Best-of-N: PAID if
|
||||
total_paid >= 95% of billed, else PARTIAL if any payment, else DENIED.
|
||||
4. Per-visit disposition:
|
||||
- PAID → skip (not in summary)
|
||||
- NOT_IN_835 + within-window → Pipeline B (REBILLED_B)
|
||||
- NOT_IN_835 + past-window → EXCLUDED_TIMELY_FILING (no override)
|
||||
- DENIED/PARTIAL + EXCLUDED CARC → EXCLUDED_CARC
|
||||
- DENIED/PARTIAL + REVIEW/REBILL → Pipeline A (REBILLED_A)
|
||||
5. Emit Pipeline A files (``<out>/pipeline-a/x-{claim_id}.837``) and
|
||||
Pipeline B files (``<out>/pipeline-b/x-{member_id}-{iso_year}-W{iso_week:02d}.837``).
|
||||
Files are ``touch()``ed placeholders here — Task 13 wraps each emit
|
||||
in a serialize-then-Edifabric-validate step, with quarantine on failure.
|
||||
6. Write ``<out>/summary.csv`` with one row per non-PAID visit.
|
||||
|
||||
The ``tpid`` argument is preserved on the signature for Task 13 / Task 14
|
||||
to thread into the real filename builder; unused in this placeholder emit.
|
||||
|
||||
``as_of`` defaults to ``date.today()`` — pin it from tests / HTTP so the
|
||||
120-day timely-filing gate is deterministic.
|
||||
"""
|
||||
as_of_date = as_of or date.today()
|
||||
out_dir_p = Path(out_dir or f"dev/rebills/{date.today().isoformat()}")
|
||||
out_dir_p.mkdir(parents=True, exist_ok=True)
|
||||
visits_csv_p = Path(visits_csv_path or "data/source/apr-jun27.csv")
|
||||
ingest_dir_p = Path(ingest_dir or "ingest")
|
||||
|
||||
# 1) Ingest 835s — walk *.835 and *.x12, skip AppleDouble shadow files
|
||||
# (._foo.835 are macOS resource forks that the SFTP client surfaces
|
||||
# alongside real files; matches the convention used by cli, submit-batch,
|
||||
# and api_routers/submission.py).
|
||||
svcs: list[SvcRow] = []
|
||||
for p in sorted(
|
||||
list(ingest_dir_p.glob("*.835")) + list(ingest_dir_p.glob("*.x12"))
|
||||
):
|
||||
if p.name.startswith("._"):
|
||||
continue
|
||||
for s in parse_835_svc(p):
|
||||
if window_start <= s.svc_date <= window_end:
|
||||
svcs.append(s)
|
||||
|
||||
# 2) Load visits CSV — header:
|
||||
# ``Visit Date,Member ID,Procedure Code,Billable Amount``
|
||||
# DOS is MM/DD/YYYY; Billable Amount may carry a ``$`` prefix and
|
||||
# ``,`` thousands separators.
|
||||
visits: list[VisitRow] = []
|
||||
with visits_csv_p.open(newline="") as f:
|
||||
for r in _csv.DictReader(f):
|
||||
dos = datetime.strptime(r["Visit Date"].strip(), "%m/%d/%Y").date()
|
||||
if not (window_start <= dos <= window_end):
|
||||
continue
|
||||
amt = Decimal(
|
||||
r["Billable Amount"].strip().lstrip("$").replace(",", "")
|
||||
)
|
||||
visits.append(VisitRow(
|
||||
date=dos,
|
||||
member_id=r["Member ID"].strip(),
|
||||
procedure=r["Procedure Code"].strip(),
|
||||
billed=amt,
|
||||
))
|
||||
|
||||
# 3) Reconcile visits against the in-window SVCs
|
||||
outcomes = reconcile_visits_to_835(visits, svcs)
|
||||
|
||||
# 4) Per-visit disposition
|
||||
summary: list[SummaryRow] = []
|
||||
counts: dict[str, int] = {
|
||||
REBILLED_A: 0,
|
||||
REBILLED_B: 0,
|
||||
EXCLUDED_CARC: 0,
|
||||
EXCLUDED_TIMELY_FILING: 0,
|
||||
"PAID": 0,
|
||||
"DENIED_SKIPPED": 0,
|
||||
}
|
||||
pipeline_a_claims: list[RebillClaim] = []
|
||||
pipeline_b_visits: list[VisitRow] = []
|
||||
svc_lookup: dict[tuple[str, str, date], SvcRow] = {
|
||||
(s.member_id, s.procedure, s.svc_date): s for s in svcs
|
||||
}
|
||||
|
||||
for o in outcomes:
|
||||
if o.category == OutcomeCategory.PAID:
|
||||
counts["PAID"] += 1
|
||||
continue
|
||||
|
||||
if o.category == OutcomeCategory.NOT_IN_835:
|
||||
tf = timely_filing_decision(o.visit.date, as_of_date, override_filing)
|
||||
if tf.rebillable:
|
||||
pipeline_b_visits.append(o.visit)
|
||||
counts[REBILLED_B] += 1
|
||||
summary.append(SummaryRow(
|
||||
visit=o.visit,
|
||||
disposition=REBILLED_B,
|
||||
unpaid=o.unpaid,
|
||||
cas_reasons=(),
|
||||
file_path="pipeline-b/",
|
||||
))
|
||||
else:
|
||||
counts[EXCLUDED_TIMELY_FILING] += 1
|
||||
summary.append(SummaryRow(
|
||||
visit=o.visit,
|
||||
disposition=EXCLUDED_TIMELY_FILING,
|
||||
unpaid=o.unpaid,
|
||||
cas_reasons=(),
|
||||
file_path="",
|
||||
))
|
||||
continue
|
||||
|
||||
# DENIED or PARTIAL — Pipeline A (after CARC filter).
|
||||
s = svc_lookup.get(
|
||||
(o.visit.member_id, o.visit.procedure, o.visit.date)
|
||||
)
|
||||
reasons: tuple[str, ...] = s.cas_reasons if s else ()
|
||||
decision = decide_carc(reasons)
|
||||
if decision == CarcDecision.EXCLUDED:
|
||||
counts[EXCLUDED_CARC] += 1
|
||||
summary.append(SummaryRow(
|
||||
visit=o.visit,
|
||||
disposition=EXCLUDED_CARC,
|
||||
unpaid=o.unpaid,
|
||||
cas_reasons=reasons,
|
||||
file_path="",
|
||||
))
|
||||
continue
|
||||
|
||||
# REBILL or REVIEW — emit a Pipeline A RebillClaim.
|
||||
# Each matched SVC carries its own original claim_id; preserve it
|
||||
# per-claim so the freq-7 replacement is anchored to the right
|
||||
# claim_submit_id. If the visit somehow matched nothing (shouldn't
|
||||
# happen for DENIED/PARTIAL — those categories only arise from a
|
||||
# match), mint a NEW-* fallback.
|
||||
claim_id = (
|
||||
s.claim_id if s
|
||||
else f"NEW-{o.visit.member_id}-{o.visit.date.isoformat()}"
|
||||
)
|
||||
claim = RebillClaim(
|
||||
claim_id=claim_id,
|
||||
member_id=o.visit.member_id,
|
||||
procedure=o.visit.procedure,
|
||||
svc_date=o.visit.date,
|
||||
charge=o.visit.billed,
|
||||
frequency_code="7",
|
||||
needs_review=(decision == CarcDecision.REVIEW),
|
||||
original_carc_reasons=reasons,
|
||||
)
|
||||
pipeline_a_claims.append(claim)
|
||||
counts[REBILLED_A] += 1
|
||||
summary.append(SummaryRow(
|
||||
visit=o.visit,
|
||||
disposition=REBILLED_A,
|
||||
unpaid=o.unpaid,
|
||||
cas_reasons=reasons,
|
||||
file_path="pipeline-a/",
|
||||
))
|
||||
|
||||
# 5) Emit Pipeline A files — placeholder; Task 13 wraps with the
|
||||
# serialize-then-Edifabric-validate step. Filename shape:
|
||||
# ``x-{claim_id}.837`` so the audit trail ties back to the
|
||||
# original claim_submit_id via the SummaryRow.cas_reasons chain.
|
||||
a_dir = out_dir_p / "pipeline-a"
|
||||
a_dir.mkdir(parents=True, exist_ok=True)
|
||||
a_files: list[Path] = []
|
||||
for claim in pipeline_a_claims:
|
||||
p = a_dir / f"x-{claim.claim_id}.837"
|
||||
p.touch()
|
||||
a_files.append(p)
|
||||
|
||||
# 6) Build + emit Pipeline B batches — placeholder; Task 13 wraps
|
||||
# with the Edifabric gate. Filename shape mirrors the HCPF
|
||||
# member-week pattern: ``x-{member_id}-{iso_year}-W{iso_week:02d}.837``.
|
||||
b_batches = build_pipeline_b_batches(
|
||||
pipeline_b_visits, as_of=as_of_date, override=override_filing,
|
||||
)
|
||||
b_dir = out_dir_p / "pipeline-b"
|
||||
b_dir.mkdir(parents=True, exist_ok=True)
|
||||
b_files: list[Path] = []
|
||||
for b in b_batches:
|
||||
p = b_dir / f"x-{b.member_id}-{b.iso_year}-W{b.iso_week:02d}.837"
|
||||
p.touch()
|
||||
b_files.append(p)
|
||||
|
||||
# 7) Summary CSV — one row per non-PAID visit, regardless of pipeline.
|
||||
summary_path = out_dir_p / "summary.csv"
|
||||
write_summary_csv(summary, summary_path)
|
||||
|
||||
# Touching `tpid` so it doesn't lint as unused — Task 13 / Task 14 will
|
||||
# thread it into the real filename builder (cyclone.edi.filenames.
|
||||
# build_outbound_filename) once the serializer adapter lands.
|
||||
_ = tpid
|
||||
|
||||
return RunResult(
|
||||
summary_path=summary_path,
|
||||
counts=counts,
|
||||
pipeline_a_files=a_files,
|
||||
pipeline_b_files=b_files,
|
||||
)
|
||||
Reference in New Issue
Block a user