diff --git a/backend/src/cyclone/cli.py b/backend/src/cyclone/cli.py index 0d00a36..f3cb0b6 100644 --- a/backend/src/cyclone/cli.py +++ b/backend/src/cyclone/cli.py @@ -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 + ``/dev/rebills//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 /dev/rebills//).", +) +@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() diff --git a/backend/tests/test_cli_rebill.py b/backend/tests/test_cli_rebill.py new file mode 100644 index 0000000..b65a096 --- /dev/null +++ b/backend/tests/test_cli_rebill.py @@ -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 \ No newline at end of file