feat(sp41): summary CSV with REBILLED_A/B + EXCLUDED_* categories

backend/src/cyclone/rebill/summary.py

@@ -0,0 +1,57 @@
+"""Summary CSV for SP41.
+
+One row per visit, classified into one of:
+  REBILLED_A          — pipeline A emitted a frequency-7 replacement
+  REBILLED_B          — pipeline B emitted a fresh 837P
+  EXCLUDED_CARC       — CO-45/CO-26/CO-129 (or other excluded CARC)
+  EXCLUDED_PAYER      — non-CO_TXIX payer per AxisCare API spot audit
+  EXCLUDED_NO_EVV     — no Authorized=Yes AND no AxisCare EVV ref
+  EXCLUDED_TIMELY_FILING — DOS > 120 days before run date, no override
+"""
+import csv
+from dataclasses import dataclass
+from decimal import Decimal
+from pathlib import Path
+
+from cyclone.rebill.reconcile import VisitRow
+
+REBILLED_A = "REBILLED_A"
+REBILLED_B = "REBILLED_B"
+EXCLUDED_CARC = "EXCLUDED_CARC"
+EXCLUDED_PAYER = "EXCLUDED_PAYER"
+EXCLUDED_NO_EVV = "EXCLUDED_NO_EVV"
+EXCLUDED_TIMELY_FILING = "EXCLUDED_TIMELY_FILING"
+
+
+@dataclass(frozen=True)
+class SummaryRow:
+    """One visit + its post-pipeline disposition for the operator audit trail."""
+
+    visit: VisitRow
+    disposition: str
+    unpaid: Decimal
+    cas_reasons: tuple[str, ...]
+    file_path: str  # relative to dev/rebills/<date>/
+
+
+def write_summary_csv(rows: list[SummaryRow], out_path: Path) -> int:
+    """Write the summary CSV; return the number of rows written."""
+    out_path.parent.mkdir(parents=True, exist_ok=True)
+    with out_path.open("w", newline="") as f:
+        w = csv.writer(f)
+        w.writerow([
+            "dos", "member_id", "procedure", "billed",
+            "disposition", "unpaid", "cas_reasons", "file_path",
+        ])
+        for r in rows:
+            w.writerow([
+                r.visit.date.isoformat(),
+                r.visit.member_id,
+                r.visit.procedure,
+                str(r.visit.billed),
+                r.disposition,
+                str(r.unpaid),
+                "|".join(r.cas_reasons),
+                r.file_path,
+            ])
+    return len(rows)

backend/tests/test_rebill_summary.py

@@ -0,0 +1,42 @@
+"""Summary CSV: one row per visit with its disposition."""
+from datetime import date
+from decimal import Decimal
+from pathlib import Path
+from cyclone.rebill.summary import (
+    SummaryRow,
+    write_summary_csv,
+    REBILLED_A, REBILLED_B, EXCLUDED_CARC, EXCLUDED_TIMELY_FILING,
+    EXCLUDED_PAYER, EXCLUDED_NO_EVV,
+)
+from cyclone.rebill.reconcile import VisitRow
+
+
+def test_summary_csv_emits_one_row_per_visit(tmp_path: Path):
+    out = tmp_path / "summary.csv"
+    rows = [
+        SummaryRow(
+            visit=VisitRow(date=date(2026, 6, 27), member_id="J813715",
+                           procedure="T1019", billed=Decimal("2.32")),
+            disposition=REBILLED_A, unpaid=Decimal("0.00"),
+            cas_reasons=("CO-45",), file_path="pipeline-a/x.837",
+        ),
+        SummaryRow(
+            visit=VisitRow(date=date(2026, 1, 1), member_id="OLD",
+                           procedure="T1019", billed=Decimal("2.32")),
+            disposition=EXCLUDED_TIMELY_FILING, unpaid=Decimal("2.32"),
+            cas_reasons=(), file_path="",
+        ),
+    ]
+    write_summary_csv(rows, out)
+    text = out.read_text()
+    assert "J813715" in text
+    assert "REBILLED_A" in text
+    assert "EXCLUDED_TIMELY_FILING" in text
+    assert "OLD" in text
+
+
+def test_disposition_constants():
+    assert REBILLED_A == "REBILLED_A"
+    assert REBILLED_B == "REBILLED_B"
+    assert EXCLUDED_CARC == "EXCLUDED_CARC"
+    assert EXCLUDED_TIMELY_FILING == "EXCLUDED_TIMELY_FILING"