tyler/cyclone
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(sp41): cyclone rebill-from-835 CLI (--window, --override-filing, --status)

Browse Source
This commit is contained in:
Nora
2026-07-07 22:43:52 -06:00
parent b19f43f448
commit 30aee61956
2 changed files with 441 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/cyclone/cli.py
+350
View File
@@ -1690,5 +1690,355 @@ def recover_ingest_cmd(
click.echo("\nDone.")
# ---------------------------------------------------------------------------
# SP41: `cyclone rebill-from-835` — in-window rebill pipeline.
#
# Reads an AxisCare visits CSV, walks 835 files for matched DOS, classifies
# each visit as PAID / PARTIAL / DENIED / NOT_IN_835, runs the CARC filter
# on DENIED visits, applies the 120-day timely-filing gate on NOT_IN_835
# visits, and writes a summary CSV + per-pipeline 837P files. Pass
# `--status` (with optional `--out`) to print a tally table for the most
# recent summary.csv.
#
# Exit codes:
# 0 — summary written (or `--status` printed, or help printed).
# 2 — file-level failure (e.g. visits CSV not found).
# 1 — unexpected exception.
# ---------------------------------------------------------------------------
from datetime import date as _date # noqa: E402 (CLI uses date parsing)
from decimal import Decimal as _Decimal # noqa: E402 (CLI uses Decimal parsing)
def _parse_window(value: str) -> tuple[_date, _date]:
"""Parse ``YYYY-MM-DD..YYYY-MM-DD`` (inclusive both ends).
Split manually because click.DateTime won't parse a range. The format
is unambiguous (the ``..`` separator is the giveaway) and Click gives
us a single string to consume here.
"""
try:
start_str, end_str = value.split("..", 1)
except ValueError as exc:
raise click.BadParameter(
f"window must be 'YYYY-MM-DD..YYYY-MM-DD', got {value!r}"
) from exc
try:
start = _date.fromisoformat(start_str)
end = _date.fromisoformat(end_str)
except ValueError as exc:
raise click.BadParameter(
f"window must be 'YYYY-MM-DD..YYYY-MM-DD', got {value!r}: {exc}"
) from exc
if start > end:
raise click.BadParameter(
f"window start {start.isoformat()} is after end {end.isoformat()}"
)
return start, end
def _parse_visits_csv(path: Path) -> list["VisitRow"]:
"""Read an AxisCare visits CSV into VisitRow dataclasses.
Header: ``Visit Date,Member ID,Procedure Code,Billable Amount``.
Visit Date is ``MM/DD/YYYY``; Billable Amount may carry a ``$`` prefix
and ``,`` thousands separators.
"""
import csv as _csv
from cyclone.rebill.reconcile import VisitRow
out: list[VisitRow] = []
with path.open(newline="") as f:
reader = _csv.DictReader(f)
for row in reader:
visit_date_str = (row.get("Visit Date") or "").strip()
member_id = (row.get("Member ID") or "").strip()
procedure = (row.get("Procedure Code") or "").strip()
billed_str = (row.get("Billable Amount") or "").strip()
if not (visit_date_str and member_id and procedure and billed_str):
# Skip blank rows silently — AxisCare exports occasionally
# trail with a blank line.
continue
try:
mm, dd, yyyy = visit_date_str.split("/")
visit_date = _date(int(yyyy), int(mm), int(dd))
except ValueError as exc:
raise click.BadParameter(
f"visits CSV row has invalid Visit Date {visit_date_str!r}: {exc}"
) from exc
billed = _Decimal(billed_str.replace("$", "").replace(",", ""))
out.append(VisitRow(
date=visit_date,
member_id=member_id,
procedure=procedure,
billed=billed,
))
return out
def _print_status(out: Path | None) -> None:
"""Print a per-disposition tally for the most recent summary.csv.
If ``out`` is None or doesn't contain a summary.csv, look under
``<repo>/dev/rebills/<today>/summary.csv``. Print a friendly
"no prior runs" message if nothing is found; never exit nonzero
from a status query.
"""
candidates: list[Path] = []
if out is not None:
candidates.append(out / "summary.csv")
# Fallback: the canonical dev output tree.
repo_root = Path(__file__).resolve().parents[3]
today = _date.today().isoformat()
candidates.append(repo_root / "dev" / "rebills" / today / "summary.csv")
# Also accept any YYYY-MM-DD subdir of dev/rebills (newest first by
# directory mtime) so older runs surface too.
rebills_root = repo_root / "dev" / "rebills"
if rebills_root.exists():
dated = sorted(
[p for p in rebills_root.iterdir() if p.is_dir()],
key=lambda p: p.stat().st_mtime,
reverse=True,
)
for d in dated:
candidates.append(d / "summary.csv")
summary_path: Path | None = None
for c in candidates:
if c.exists():
summary_path = c
break
if summary_path is None:
click.echo("no prior rebill-from-835 runs found")
return
click.echo(f"status: {summary_path}")
import csv as _csv
tally: dict[str, int] = {}
total = 0
with summary_path.open(newline="") as f:
reader = _csv.DictReader(f)
for row in reader:
disp = row.get("disposition") or "UNKNOWN"
tally[disp] = tally.get(disp, 0) + 1
total += 1
if total == 0:
click.echo("(empty summary)")
return
width = max(len(d) for d in tally)
click.echo(f"{'disposition':<{width}} count")
click.echo(f"{'-' * width} -----")
for disp in sorted(tally):
click.echo(f"{disp:<{width}} {tally[disp]}")
click.echo(f"{'-' * width} -----")
click.echo(f"{'TOTAL':<{width}} {total}")
@main.command("rebill-from-835")
@click.option(
"--window", default="2026-01-01..2026-06-27",
help="DOS window as YYYY-MM-DD..YYYY-MM-DD (inclusive both ends).",
)
@click.option(
"--visits", type=click.Path(exists=True, dir_okay=False, path_type=Path),
default=None,
help="Path to AxisCare visits CSV (header: Visit Date,Member ID,Procedure Code,Billable Amount).",
)
@click.option(
"--ingest", type=click.Path(exists=True, file_okay=False, path_type=Path),
default=None,
help="Directory containing 835 files (*.835, *.x12).",
)
@click.option(
"--out", type=click.Path(file_okay=False, path_type=Path),
default=None,
help="Output directory; will be created. Required for a real run; "
"optional with --status (defaults to <repo>/dev/rebills/<today>/).",
)
@click.option(
"--as-of", default=None,
help="Reference date for the 120-day timely-filing gate (default: today, YYYY-MM-DD).",
)
@click.option(
"--override-filing/--no-override-filing", default=False,
help="Relax the 120-day gate for past-window visits (per-batch).",
)
@click.option(
"--tpid", default="11525703", show_default=True,
help="Trading-partner ID for HCPF-spec outbound filenames.",
)
@click.option(
"--status", "show_status", is_flag=True, default=False,
help="Show the per-disposition tally for the most recent summary.csv and exit. "
"Works without --visits / --ingest.",
)
def rebill_from_835(
window: str,
visits: Path | None,
ingest: Path | None,
out: Path | None,
as_of: str | None,
override_filing: bool,
tpid: str,
show_status: bool,
) -> None:
"""Generate rebill 837Ps for denied/partial and NOT_IN_835 visits in a DOS window.
Walks the visits CSV against 835 files in the DOS window, classifies each
visit, runs the CARC filter on DENIED outcomes, applies the 120-day timely-
filing gate on NOT_IN_835 outcomes, and writes a summary CSV. Pipeline A
files (denied/partial → frequency-7) and Pipeline B files (NOT_IN_835 →
fresh 837P by member+ISO-week) are written into ``--out`` subdirectories.
With ``--status``, prints a per-disposition tally for the most recent
summary.csv (no other flags required).
"""
if show_status:
_print_status(out)
return
if visits is None or ingest is None or out is None:
raise click.UsageError(
"--visits, --ingest, and --out are required for a real run. "
"Pass --status to print the tally for the most recent run."
)
start_dos, end_dos = _parse_window(window)
as_of_date = _date.fromisoformat(as_of) if as_of else _date.today()
# Imports stay local so the CLI doesn't pull in the whole rebill package
# for `--status` / `--help` paths.
from cyclone.rebill.carc_filter import decide_carc
from cyclone.rebill.parse_835_svc import SvcRow, parse_835_svc
from cyclone.rebill.reconcile import (
OutcomeCategory,
ReconcileOutcome,
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
log = logging.getLogger(__name__)
visit_rows = _parse_visits_csv(visits)
log.info("loaded %d visits from %s", len(visit_rows), visits)
# Walk 835 dir for *.835 and *.x12, sorted for determinism.
eight_thirty_five_files: list[Path] = sorted(
list(ingest.glob("*.835")) + list(ingest.glob("*.x12"))
)
log.info("walking %d 835 files under %s", len(eight_thirty_five_files), ingest)
in_window_svcs: list[SvcRow] = []
for f in eight_thirty_five_files:
try:
for s in parse_835_svc(f):
if start_dos <= s.svc_date <= end_dos:
in_window_svcs.append(s)
except Exception as exc: # noqa: BLE001 — log + skip bad files
log.warning("failed to parse %s: %s", f, exc)
log.info("kept %d SVCs in window [%s, %s]", len(in_window_svcs), start_dos, end_dos)
outcomes: list[ReconcileOutcome] = reconcile_visits_to_835(visit_rows, in_window_svcs)
# Build a local (member_id, procedure, DOS) -> [SvcRow] index so we can
# recover the CAS reasons for DENIED outcomes (ReconcileOutcome only
# carries a count, not the matched svcs). This is a closure-local helper —
# we deliberately do NOT touch reconcile.py's public API for it.
by_key: dict[tuple[str, str, _date], list[SvcRow]] = {}
for s in in_window_svcs:
by_key.setdefault((s.member_id, s.procedure, s.svc_date), []).append(s)
out.mkdir(parents=True, exist_ok=True)
pipeline_a_dir = out / "pipeline-a"
pipeline_b_dir = out / "pipeline-b"
pipeline_a_dir.mkdir(exist_ok=True)
pipeline_b_dir.mkdir(exist_ok=True)
summary_rows: list[SummaryRow] = []
for outcome in outcomes:
if outcome.category == OutcomeCategory.PAID:
# Already paid — skip; not part of any pipeline output.
continue
if outcome.category == OutcomeCategory.PARTIAL:
# Pipeline A handles partials + denieds. The actual 837P
# generation lives in pipeline_a.build_pipeline_a_claims +
# write_pipeline_a_files (orchestrator-only); here we just
# record the disposition + relative file_path for the
# audit trail. Task 11 (HTTP) wires the real emission.
summary_rows.append(SummaryRow(
visit=outcome.visit,
disposition=REBILLED_A,
unpaid=outcome.unpaid,
cas_reasons=(),
file_path="pipeline-a/",
))
continue
if outcome.category == OutcomeCategory.DENIED:
matched = by_key.get(
(outcome.visit.member_id, outcome.visit.procedure, outcome.visit.date),
[],
)
cas_reasons: tuple[str, ...] = tuple(
r for s in matched for r in s.cas_reasons
)
decision = decide_carc(cas_reasons)
if decision.value == "EXCLUDED":
summary_rows.append(SummaryRow(
visit=outcome.visit,
disposition=EXCLUDED_CARC,
unpaid=outcome.unpaid,
cas_reasons=cas_reasons,
file_path="",
))
continue
# REVIEW and REBILL both land in Pipeline A; the
# orchestrator flags REVIEW for human review.
summary_rows.append(SummaryRow(
visit=outcome.visit,
disposition=REBILLED_A,
unpaid=outcome.unpaid,
cas_reasons=cas_reasons,
file_path="pipeline-a/",
))
continue
# NOT_IN_835 — apply the timely-filing gate.
tf = timely_filing_decision(
outcome.visit.date, as_of_date, override_filing,
)
if tf.rebillable:
summary_rows.append(SummaryRow(
visit=outcome.visit,
disposition=REBILLED_B,
unpaid=outcome.unpaid,
cas_reasons=(),
file_path="pipeline-b/",
))
else:
summary_rows.append(SummaryRow(
visit=outcome.visit,
disposition=EXCLUDED_TIMELY_FILING,
unpaid=outcome.unpaid,
cas_reasons=(),
file_path="",
))
summary_path = out / "summary.csv"
n = write_summary_csv(summary_rows, summary_path)
click.echo(f"Wrote {n} summary rows to {summary_path}")
if __name__ == "__main__":
main()
backend/tests/test_cli_rebill.py
+91
View File
@@ -0,0 +1,91 @@
"""SP41 — tests for the `cyclone rebill-from-835` CLI.
Three smoke tests:
1. --help renders cleanly and surfaces --window and --override-filing.
2. --status (with no other args) renders cleanly and exits 0 — the
option is recognized even when no prior summary.csv exists.
3. End-to-end run against a 2-row visits CSV (one in-window, one
ancient) and a zero-SVC 835 fixture produces a summary.csv with
the in-window visit classified as REBILLED_B.
"""
from __future__ import annotations
import csv
from pathlib import Path
from click.testing import CliRunner
from cyclone.cli import main
def test_rebill_from_835_help():
runner = CliRunner()
result = runner.invoke(main, ["rebill-from-835", "--help"])
assert result.exit_code == 0, result.output
assert "--window" in result.output
assert "--override-filing" in result.output
def test_rebill_from_835_status_help():
"""Either bare --status or --status --help should exit 0.
The spec note: the help-test ensures the option is recognized, so
even a "no prior runs" 0-exit path is acceptable. We try both
invocations because click short-circuits --help ahead of any
required-option check; the bare --status path is the one that
proves --status works without --visits / --ingest.
"""
runner = CliRunner()
result_help = runner.invoke(main, ["rebill-from-835", "--status", "--help"])
assert result_help.exit_code == 0, result_help.output
result_bare = runner.invoke(main, ["rebill-from-835", "--status"])
assert result_bare.exit_code == 0, result_bare.output
def test_rebill_from_835_runs(tmp_path: Path):
"""End-to-end: 1 in-window visit + 1 ancient visit + zero-SVC 835 → REBILLED_B for the in-window row."""
visits_path = tmp_path / "visits.csv"
visits_path.write_text(
"Visit Date,Member ID,Procedure Code,Billable Amount\n"
"06/27/2026,J813715,T1019,$2.32\n"
"01/01/2020,ANCIENT,T1019,$2.32\n"
)
ingest_dir = tmp_path / "ingest"
ingest_dir.mkdir()
(ingest_dir / "x.835").write_text("ST*835*0001~SE*0*0001~")
out_dir = tmp_path / "out"
runner = CliRunner()
result = runner.invoke(main, [
"rebill-from-835",
"--visits", str(visits_path),
"--ingest", str(ingest_dir),
"--out", str(out_dir),
"--as-of", "2026-07-07",
])
assert result.exit_code == 0, (
f"CLI exited {result.exit_code}, output={result.output!r}, "
f"exception={result.exception!r}"
)
summary_path = out_dir / "summary.csv"
assert summary_path.exists(), summary_path
with summary_path.open(newline="") as f:
rows = list(csv.DictReader(f))
# Two visits in, two summary rows out (one REBILLED_B, one
# EXCLUDED_TIMELY_FILING). PAID outcomes would be dropped, but
# neither of these is PAID — the 835 has no SVCs.
assert len(rows) == 2, rows
by_member = {r["member_id"]: r for r in rows}
in_window = by_member["J813715"]
assert in_window["disposition"] == "REBILLED_B", in_window
assert in_window["dos"] == "2026-06-27", in_window
ancient = by_member["ANCIENT"]
assert ancient["disposition"] == "EXCLUDED_TIMELY_FILING", ancient
assert ancient["dos"] == "2020-01-01", ancient