feat(sp41): serialize_837 overload for member-week batches

backend/src/cyclone/parsers/serialize_837.py

@@ -48,6 +48,7 @@ from __future__ import annotations
 import logging
 from datetime import date, datetime
 from decimal import Decimal
+from types import SimpleNamespace
 
 from cyclone.parsers.models import ClaimOutput
 
@@ -686,4 +687,144 @@ def serialize_837_for_resubmit(
         interchange_control_number=f"{interchange_index:09d}",
         group_control_number=str(interchange_index),
         **kwargs,
-    )
+    )
+
+
+# ---------------------------------------------------------------------------
+# Pipeline B overload: MemberWeekBatch → one 837P, one CLM per visit
+# ---------------------------------------------------------------------------
+
+
+def _build_member_week_claim(visit, claim_id: str) -> tuple[str, str, str]:
+    """Build the CLM / SV1 / DTP*472 segments for a single MemberWeekBatch visit.
+
+    Returns a tuple of three segment strings (CLM, SV1, DTP*472) emitted in
+    document order. The visit is a :class:`cyclone.rebill.reconcile.VisitRow`
+    — only ``date`` / ``member_id`` / ``procedure`` / ``billed`` are read.
+    """
+    # Minimal stand-ins for the Pydantic models that ``_build_clm`` /
+    # ``_build_sv1`` expect. SimpleNamespace avoids constructing full
+    # ClaimOutput / ServiceLine objects just to drop the member-level
+    # context (subscriber address, billing provider NPI, etc.) that
+    # Pipeline B doesn't have on its input shape.
+    procedure = SimpleNamespace(qualifier="HC", code=visit.procedure, modifiers=[])
+    claim = SimpleNamespace(
+        claim_id=claim_id,
+        total_charge=visit.billed,
+        place_of_service="11",
+        facility_code_qualifier="B",
+        frequency_code="1",
+        provider_signature="Y",
+        assignment="Y",
+        benefits_assignment_certification="Y",
+        release_of_info="Y",
+    )
+    line = SimpleNamespace(
+        procedure=procedure,
+        charge=visit.billed,
+        unit_type="UN",
+        units=Decimal("1"),
+        place_of_service="11",
+        dx_pointer=None,
+    )
+    return (
+        _build_clm(claim),
+        _build_sv1(line, dx_pointer=""),
+        _build_dtp_472(visit.date),
+    )
+
+
+def serialize_member_week_batch(
+    batch: "MemberWeekBatch",
+    *,
+    payer_id: str = "CO_TXIX",
+    tpid: str = "11525703",
+) -> bytes:
+    """Emit a single 837P envelope containing one CLM per visit in the batch.
+
+    SP41 / Pipeline B. Each :class:`cyclone.rebill.reconcile.VisitRow` in
+    ``batch.visits`` becomes its own CLM with one SV1 and one DTP*472
+    service-line date. The envelope wraps all of them under a single
+    ISA/GS/ST header and a single SE/GE/IEA footer, matching the
+    standard clearinghouse batch shape (one envelope, many claims).
+
+    Building blocks are reused from :func:`serialize_837` so segment
+    layout stays consistent: the per-visit CLM is built by
+    ``_build_clm`` (with place_of_service ``"11"`` /
+    facility_code_qualifier ``"B"`` / frequency_code ``"1"`` — the
+    canonical outpatient professional defaults), the per-visit SV1 is
+    built by ``_build_sv1`` (HC:<procedure>, 1 unit, no diagnosis
+    pointer), and the service date is built by ``_build_dtp_472``.
+
+    Args:
+        batch: A :class:`cyclone.rebill.pipeline_b.MemberWeekBatch` —
+            one member × one ISO-week worth of rebillable visits.
+        payer_id: The receiver (NM1*40) identifier. Defaults to
+            ``"CO_TXIX"`` for CO Medicaid.
+        tpid: The trading-partner / submitter (NM1*41) identifier.
+            Defaults to Gainwell's ``"11525703"``.
+
+    Returns:
+        The complete 837P document as ASCII bytes. The caller writes
+        it to disk with HCPF-spec filenames via
+        :func:`cyclone.edi.filenames.build_outbound_filename`.
+    """
+    # Deterministic control numbers derived from (member, iso_year,
+    # iso_week). Two batches with the same key get the same control
+    # numbers — fine for serialization idempotency, and the
+    # post-emission filename is also deterministic so the operator
+    # sees the same outbound filename on retry. Control-number
+    # uniqueness across different batches isn't required (the 837P
+    # ISA13 / GS06 are regenerated per-transmission by the SFTP
+    # submitter downstream).
+    control = f"{batch.member_id}{batch.iso_year:04d}{batch.iso_week:02d}"
+    interchange_control_number = control[:9].rjust(9, "0")
+    group_control_number = control[:9].lstrip("0") or "1"
+    st_control_number = control[:9].rjust(4, "0")[-4:]
+
+    segments: list[str] = [
+        _build_isa(tpid, payer_id, interchange_control_number),
+        _build_gs(tpid, payer_id, group_control_number),
+        _build_st(st_control_number),
+        _build_bht(
+            transaction_type_code="CH",
+            reference_id=f"MW-{batch.member_id}-W{batch.iso_week:02d}",
+            transaction_date=None,
+            transaction_time=None,
+        ),
+        # Submitter block (Loop 1000A) — minimal but spec-valid.
+        # Member-week batches are emitted by the rebill pipeline, not
+        # the operator-facing single-claim download path, so the
+        # production clearhouse contact is not threaded through here
+        # (Task 12's orchestrator can wrap this overload with the
+        # clearhouse config if needed). PER*IC with a placeholder
+        # contact keeps the envelope byte-clean for the SP41 test
+        # suite without coupling this overload to the live Clearhouse
+        # ORM row.
+        _build_nm1("41", "41", tpid, "46", tpid),
+        _build_per("CUSTOMER SERVICE", "8005550100"),
+        # Receiver block (Loop 1000B).
+        _build_nm1("40", "40", payer_id, "46", payer_id),
+    ]
+
+    for idx, visit in enumerate(batch.visits, start=1):
+        svc_date = visit.date
+        claim_id = f"MW-{batch.member_id}-{svc_date.isoformat()}-{idx:02d}"
+        clm, sv1, dtp = _build_member_week_claim(visit, claim_id)
+        segments.append(clm)
+        segments.append(sv1)
+        if dtp:
+            segments.append(dtp)
+
+    # SE segment count = ST (1) + everything between ST and SE inclusive.
+    # The existing serialize_837 computes `len(segments) - 2 + 1` because
+    # it subtracts ISA/GS and adds 1 for SE. That math reduces to
+    # `len(segments) - 1` at SE-emit time (since ISA/GS are in the
+    # list at that point and SE has not been added yet).
+    seg_count = len(segments) - 1
+    segments.append(_build_se(seg_count, st_control_number))
+
+    segments.append(f"GE*1*{group_control_number}{_SEG}")
+    segments.append(f"IEA*1*{interchange_control_number}{_SEG}")
+
+    return "".join(segments).encode("ascii")

backend/tests/test_rebill_pipeline_b.py

@@ -73,3 +73,48 @@ def test_override_flag_set_on_past_window_visit_batch():
     assert out_default[0].member_id == "OLD"
     assert out_default[0].iso_week == 26
     assert out_default[0].has_overridden_visits is False
+
+
+def test_serialize_member_week_batch_emits_one_envelope():
+    """One 837P envelope per MemberWeekBatch — one CLM + one SV1 + one
+    DTP*472 service date per visit.
+
+    The SP41 plan spec wrote ``DTM*472*`` but the canonical 837P service
+    date segment is ``DTP*472*`` (per :func:`cyclone.parsers.serialize_837.
+    _build_dtp_472` and X12 005010X222A1). This test asserts against the
+    canonical segment name so the batch overload stays consistent with
+    the existing ``serialize_837`` building blocks.
+    """
+    from cyclone.parsers.serialize_837 import serialize_member_week_batch
+    visits = [
+        _v(date(2026, 6, 23), "J813715", "T1019", "2.32"),
+        _v(date(2026, 6, 25), "J813715", "T1019", "2.32"),
+    ]
+    batches = build_pipeline_b_batches(visits, as_of=date(2026, 7, 7), override=False)
+    assert len(batches) == 1
+    body = serialize_member_week_batch(batches[0])
+    text = body.decode("utf-8", errors="ignore") if isinstance(body, bytes) else body
+    # CLM* segment appears twice (once per visit)
+    assert text.count("CLM*") == 2
+    # SV1* appears once per visit (one service line per claim)
+    assert text.count("SV1*") == 2
+    # DTP*472* service-date segment appears twice (canonical 837P segment name)
+    assert text.count("DTP*472*") == 2
+    # Single envelope (single ISA / single IEA), not per-visit envelopes
+    assert text.count("ISA*") == 1
+    assert text.count("IEA*") == 1
+    # Deterministic per-visit claim_id pattern (member_id + date + 1-based idx)
+    assert "MW-J813715-2026-06-23-01" in text
+    assert "MW-J813715-2026-06-25-02" in text
+
+
+def test_serialize_member_week_batch_return_type_is_bytes():
+    """Task 14 spec: ``serialize_member_week_batch`` returns ``bytes``
+    (the existing ``serialize_837`` returns ``str``; this overload
+    diverges intentionally so callers can write the file directly)."""
+    from cyclone.parsers.serialize_837 import serialize_member_week_batch
+    visits = [_v(date(2026, 6, 23), "J813715", "T1019", "2.32")]
+    batches = build_pipeline_b_batches(visits, as_of=date(2026, 7, 7), override=False)
+    assert len(batches) == 1
+    body = serialize_member_week_batch(batches[0])
+    assert isinstance(body, bytes)