diff --git a/backend/src/cyclone/api.py b/backend/src/cyclone/api.py index 5a3255c..ac1c08a 100644 --- a/backend/src/cyclone/api.py +++ b/backend/src/cyclone/api.py @@ -1272,421 +1272,6 @@ def _serialize_ta1_from_row(row: db.Ta1Ack) -> str: ) -# --------------------------------------------------------------------------- # - - -@app.get("/api/claims", dependencies=[Depends(matrix_gate)]) -def list_claims( - request: Request, - batch_id: str | None = Query(None), - status: str | None = Query(None), - provider_npi: str | None = Query(None), - payer: str | None = Query(None), - date_from: str | None = Query(None), - date_to: str | None = Query(None), - sort: str | None = Query(None), - order: str = Query("desc"), - limit: int = Query(100, ge=1, le=1000), - offset: int = Query(0, ge=0), -) -> Any: - common = dict( - batch_id=batch_id, - status=status, - provider_npi=provider_npi, - payer=payer, - date_from=date_from, - date_to=date_to, - ) - items = list(store.iter_claims( - sort=sort, order=order, limit=limit, offset=offset, **common, - )) - # SP27 Task 13b: count the full population, not a 100-row sample. - # `iter_claims` defaults to limit=100; counting its output silently - # capped the reported total at 100 even when the DB held 60k rows. - total = store.count_claims(**common) - returned = len(items) - has_more = total > offset + returned - if _wants_ndjson(request): - return StreamingResponse( - _ndjson_stream_list(items, total, returned, has_more), - media_type="application/x-ndjson", - ) - return { - "items": items, - "total": total, - "returned": returned, - "has_more": has_more, - } - - -# --------------------------------------------------------------------------- # -# Live-tail NDJSON streaming endpoints (Phase 3 — SP5) -# --------------------------------------------------------------------------- # - - -@app.get("/api/claims/stream", dependencies=[Depends(matrix_gate)]) -async def claims_stream( - request: Request, - status: str | None = Query(None), - provider_npi: str | None = Query(None), - payer: str | None = Query(None), - date_from: str | None = Query(None), - date_to: str | None = Query(None), - sort: str | None = Query(None), - order: str = Query("desc"), - limit: int = Query(100, ge=1, le=1000), -) -> StreamingResponse: - """Stream Claims as NDJSON: snapshot first, then live events. - - Wire format: - * ``{"type":"item","data":}`` per snapshot row, then per - new ``claim_written`` event - * ``{"type":"snapshot_end","data":{"count":N}}`` after the snapshot - * ``{"type":"heartbeat","data":{"ts":}}`` every - ``CYCLONE_TAIL_HEARTBEAT_S`` seconds when idle - - Query params mirror :func:`list_claims` so a frontend can swap a - one-shot fetch for a tail with no URL surgery. - - NOTE: registered before ``/api/claims/{claim_id}`` so the literal - ``stream`` path segment doesn't get matched as a claim id. - """ - bus: EventBus = request.app.state.event_bus - - async def gen() -> AsyncIterator[bytes]: - # 1. Snapshot (eager — iter_claims returns a list already). - rows = store.iter_claims( - status=status, provider_npi=provider_npi, payer=payer, - date_from=date_from, date_to=date_to, - sort=sort or "-submission_date", order=order, limit=limit, - ) - for row in rows: - yield _ndjson_line({"type": "item", "data": row}) - yield _ndjson_line({"type": "snapshot_end", "data": {"count": len(rows)}}) - - # 2. Subscribe + heartbeats. - async for chunk in _tail_events(request, bus, ["claim_written"]): - yield chunk - - return StreamingResponse(gen(), media_type="application/x-ndjson") - - -@app.get("/api/claims/{claim_id}", dependencies=[Depends(matrix_gate)]) -def get_claim_detail_endpoint(claim_id: str) -> dict: - """Return one claim with full drawer context (SP4). - - Body shape is produced by :meth:`CycloneStore.get_claim_detail`: - header, state, service lines, diagnoses, parties, validation, - raw segments, ``stateHistory`` (most-recent-first, capped at 50), - and a populated ``matchedRemittance`` block when paired. - - SP28: response gains ``ack_links: list[dict]`` (compact form: - ``[{ack_id, ack_kind, set_accept_reject_code, parsed_at}]``) - so the ``ClaimDrawer`` Acknowledgments panel can render on - initial load. TA1 batch-level rows (``claim_id IS NULL``) are - excluded — those don't belong to a specific claim. - - Path param is ``claim_id`` (matches the SP3 ``/api/acks/{ack_id}`` - convention). Returns 404 — never 500 — on a missing claim so the - UI can distinguish "doesn't exist" from a transient fetch error. - """ - body = store.get_claim_detail(claim_id) - if body is None: - raise HTTPException( - status_code=404, - detail={ - "error": "Not found", - "detail": f"Claim {claim_id} not found", - }, - ) - # SP28: attach ack_links (compact form for the drawer panel). - body["ack_links"] = _compact_ack_links_for_claim(claim_id) - return body - - -def _compact_ack_links_for_claim(claim_id: str) -> list[dict]: - """Return compact ack_links for one claim, newest first. - - TA1 batch-level rows (claim_id IS NULL) are filtered out — those - hang off the originating 837 batch, not a specific claim. The - shape is the slimmer ``{ack_id, ack_kind, - set_accept_reject_code, parsed_at, ak2_index}`` form so the - ClaimDrawer can render without an N+1 round-trip per row. - """ - rows = store.list_acks_for_claim(claim_id) - out: list[dict] = [] - for row in rows: - if row.claim_id is None: - continue - out.append({ - "id": row.id, - "ack_id": row.ack_id, - "ack_kind": row.ack_kind, - "ak2_index": row.ak2_index, - "set_control_number": row.set_control_number, - "set_accept_reject_code": row.set_accept_reject_code, - "linked_at": ( - row.linked_at.isoformat().replace("+00:00", "Z") - if row.linked_at is not None else "" - ), - "linked_by": row.linked_by, - }) - return out - - -@app.get("/api/claims/{claim_id}/serialize-837", dependencies=[Depends(matrix_gate)]) -def serialize_claim_as_837(claim_id: str): - """Return the claim as a regenerated X12 837P file (SP8). - - Loads the ClaimOutput from the persisted ``raw_json`` and runs the - outbound serializer. Returns 404 if the claim doesn't exist, 422 if - the stored payload has no parseable ClaimOutput (data integrity - issue, not a transient failure). - """ - with db.SessionLocal()() as s: - row = s.get(Claim, claim_id) - if row is None: - return JSONResponse( - {"error": "Not found", "detail": f"Claim {claim_id} not found"}, - status_code=404, - ) - if not row.raw_json: - return JSONResponse( - { - "error": "Unprocessable", - "detail": f"Claim {claim_id} has no raw_json; cannot serialize", - }, - status_code=422, - ) - try: - claim_obj = ClaimOutput.model_validate(row.raw_json) - except Exception as exc: - return JSONResponse( - { - "error": "Unprocessable", - "detail": f"Claim {claim_id} raw_json is malformed: {exc}", - }, - status_code=422, - ) - - try: - text = serialize_837(claim_obj) - except SerializeError837 as exc: - return JSONResponse( - {"error": "Unprocessable", "detail": str(exc)}, - status_code=422, - ) - - return Response( - content=text, - media_type="text/x12", - headers={ - "Content-Disposition": f'attachment; filename="claim-{claim_id}.x12"' - }, - ) - - -@app.get("/api/claims/{claim_id}/line-reconciliation", dependencies=[Depends(matrix_gate)]) -def get_claim_line_reconciliation(claim_id: str) -> dict: - """Per-line reconciliation view for the ClaimDrawer tab. - - Spec §5.1. Returns the 837 service lines and 835 SVC composites - side-by-side, with per-line CAS adjustments and a summary block. - - Architecture note: 837 service lines live in ``Claim.raw_json`` - (not a separate ORM table), so the 837-side rows are read from the - JSON blob; the 835-side rows come from ``ServiceLinePayment`` ORM. - ``LineReconciliation.claim_service_line_number`` stores the 1-based - line number to join them. - """ - from sqlalchemy import select - from cyclone.db import ( - LineReconciliation, ServiceLinePayment, CasAdjustment, - ) - import json as _json - from decimal import Decimal - - with db.SessionLocal()() as s: - claim = s.get(db.Claim, claim_id) - if claim is None: - raise HTTPException( - status_code=404, - detail={"error": "Not found", "detail": f"Claim {claim_id} not found"}, - ) - - # 837 service lines: from raw_json. - raw = claim.raw_json or {} - claim_lines_raw = raw.get("service_lines") or [] - # Normalize to dicts for the response. - claim_lines = [_claim_line_dict(d) for d in claim_lines_raw] - - # 835 service payments: ORM rows from the matched remit. - remits = list( - s.execute( - select(db.Remittance).where(db.Remittance.claim_id == claim_id) - ).scalars().all() - ) - svc_payments: list[dict] = [] - svc_ids: list[int] = [] - if remits: - svc_rows = list( - s.execute( - select(ServiceLinePayment).where( - ServiceLinePayment.remittance_id.in_([r.id for r in remits]) - ).order_by(ServiceLinePayment.line_number) - ).scalars().all() - ) - for svc in svc_rows: - d = _svc_to_dict(svc) - svc_payments.append(d) - svc_ids.append(svc.id) - - # LineReconciliation rows. - lrs = list( - s.execute( - select(LineReconciliation).where(LineReconciliation.claim_id == claim_id) - ).scalars().all() - ) - # Index by claim_service_line_number and service_line_payment_id. - lr_by_claim_num: dict[int, LineReconciliation] = { - lr.claim_service_line_number: lr for lr in lrs if lr.claim_service_line_number is not None - } - lr_by_svc: dict[int, LineReconciliation] = { - lr.service_line_payment_id: lr for lr in lrs if lr.service_line_payment_id is not None - } - - # CAS rows grouped by svc id. - cas_by_svc: dict[int, list[CasAdjustment]] = {} - if svc_ids: - cas_rows = list( - s.execute( - select(CasAdjustment).where(CasAdjustment.service_line_payment_id.in_(svc_ids)) - ).scalars().all() - ) - for c in cas_rows: - cas_by_svc.setdefault(c.service_line_payment_id, []).append(c) - - # Build output lines array, preserving 837 order then 835-only. - svc_by_id: dict[int, dict] = {d["id"]: d for d in svc_payments} - lines_out: list[dict] = [] - billed_total = Decimal("0") - paid_total = Decimal("0") - adjustment_total = Decimal("0") - matched_count = 0 - used_svc_ids: set[int] = set() - - for cl in claim_lines: - billed_total += Decimal(str(cl["charge"])) - lr = lr_by_claim_num.get(cl["line_number"]) - if lr is None: - lines_out.append({ - "claim_service_line": cl, - "service_line_payment": None, - "status": "unmatched_837_only", - "adjustments": [], - }) - continue - svc_id = lr.service_line_payment_id - svc = svc_by_id.get(svc_id) if svc_id else None - if svc_id is not None: - used_svc_ids.add(svc_id) - cas_list = cas_by_svc.get(svc_id, []) if svc_id is not None else [] - cas_total = sum((Decimal(str(c.amount)) for c in cas_list), Decimal("0")) - if svc: - paid_total += Decimal(str(svc["payment"])) - adjustment_total += cas_total - if lr.status == "matched": - matched_count += 1 - lines_out.append({ - "claim_service_line": cl, - "service_line_payment": svc, - "status": lr.status, - "adjustments": [ - {"group_code": c.group_code, "reason_code": c.reason_code, - "amount": str(Decimal(str(c.amount)))} - for c in cas_list - ], - }) - - # 835-only lines (no claim match). - for lr in lrs: - if lr.claim_service_line_number is not None: - continue - svc_id = lr.service_line_payment_id - if svc_id is None: - continue - if svc_id in used_svc_ids: - continue - svc = svc_by_id.get(svc_id) - cas_list = cas_by_svc.get(svc_id, []) - cas_total = sum((Decimal(str(c.amount)) for c in cas_list), Decimal("0")) - if svc: - paid_total += Decimal(str(svc["payment"])) - adjustment_total += cas_total - lines_out.append({ - "claim_service_line": None, - "service_line_payment": svc, - "status": lr.status, - "adjustments": [ - {"group_code": c.group_code, "reason_code": c.reason_code, - "amount": str(Decimal(str(c.amount)))} - for c in cas_list - ], - }) - - return { - "claim_id": claim_id, - "summary": { - "billed_total": str(billed_total), - "paid_total": str(paid_total), - "adjustment_total": str(adjustment_total), - "matched_lines": matched_count, - "total_lines": len(claim_lines), - }, - "lines": lines_out, - } - - -def _claim_line_dict(d: dict) -> dict: - """Project an 837 service-line dict from ``Claim.raw_json`` to wire shape.""" - from decimal import Decimal - proc = d.get("procedure") or {} - charge = d.get("charge") - units = d.get("units") - return { - "line_number": d.get("line_number"), - "procedure_qualifier": proc.get("qualifier", "HC"), - "procedure_code": proc.get("code", ""), - "modifiers": proc.get("modifiers") or [], - "charge": str(Decimal(str(charge))) if charge is not None else "0", - "units": str(Decimal(str(units))) if units is not None else None, - "unit_type": d.get("unit_type"), - "service_date": d.get("service_date"), - } - - -def _svc_to_dict(svc) -> dict: - """Project an ORM ``ServiceLinePayment`` to wire shape.""" - import json as _json - from decimal import Decimal - return { - "id": svc.id, - "line_number": svc.line_number, - "procedure_qualifier": svc.procedure_qualifier, - "procedure_code": svc.procedure_code, - "modifiers": _json.loads(svc.modifiers_json or "[]"), - "charge": str(Decimal(str(svc.charge))), - "payment": str(Decimal(str(svc.payment))), - "units": str(Decimal(str(svc.units))) if svc.units is not None else None, - "unit_type": svc.unit_type, - "service_date": svc.service_date.isoformat() if svc.service_date else None, - } - - - - -# --------------------------------------------------------------------------- # - from cyclone.auth.routes import router as auth_router from cyclone.auth.admin import router as admin_users_router @@ -1703,6 +1288,17 @@ app.include_router(admin_users_router, dependencies=[Depends(matrix_gate)]) # line once the test is updated to import from the new home. from cyclone.api_routers.payers import _clear_summary_cache # noqa: F401 # SP36 backward-compat shim +# SP36 Task 15 backward-compat: ``claims_stream`` moved into +# ``cyclone.api_routers.claims`` (it's a single-router handler per +# spec D4). The live-tail integration test in +# ``tests/test_api_stream_live.py`` imports it by name from this +# module to drive ``_call_endpoint(stream_fn, path)``. Per the +# SP-N invariant that we don't rewrite tests for a structural +# refactor, expose it on this module's namespace via a one-line +# re-import. Delete this line once the test is updated to +# import from the new home. +from cyclone.api_routers.claims import claims_stream # noqa: F401 # SP36 backward-compat shim + # SP36 Tasks 9+10 backward-compat: ``remittances_stream`` and # ``activity_stream`` moved into ``cyclone.api_routers.remittances`` # and ``cyclone.api_routers.activity`` respectively (they are diff --git a/backend/src/cyclone/api_routers/__init__.py b/backend/src/cyclone/api_routers/__init__.py index b27be9d..1e05cc4 100644 --- a/backend/src/cyclone/api_routers/__init__.py +++ b/backend/src/cyclone/api_routers/__init__.py @@ -16,6 +16,7 @@ from cyclone.api_routers import ( admin, batches, claim_acks, + claims, clearhouse, config, dashboard, @@ -35,6 +36,7 @@ routers: list[APIRouter] = [ admin.router, # gated batches.router, # gated claim_acks.router, # gated + claims.router, # gated clearhouse.router, # gated config.router, # gated dashboard.router, # gated diff --git a/backend/src/cyclone/api_routers/claims.py b/backend/src/cyclone/api_routers/claims.py new file mode 100644 index 0000000..abfa5c5 --- /dev/null +++ b/backend/src/cyclone/api_routers/claims.py @@ -0,0 +1,472 @@ +"""``/api/claims*`` — Claims list / detail / streaming / serialize / line-reconciliation. + +Five endpoints, all gated by ``matrix_gate``: + +- ``GET /api/claims`` — paginated list + with filter+sort, plus an NDJSON variant when the caller sends + ``Accept: application/x-ndjson``. SP27: counts the full filtered + population, not a page-limited sample. +- ``GET /api/claims/stream`` — NDJSON live-tail + on ``claim_written``. Snapshot first (eager + ``store.iter_claims``), then ``tail_events`` subscribes + emits + heartbeats. Registered before ``/api/claims/{claim_id}`` so the + literal ``stream`` segment isn't captured as a claim id. +- ``GET /api/claims/{claim_id}`` — full drawer + context (SP4) with the SP28 ``ack_links`` block pre-attached. + 404 on missing id — never 500. +- ``GET /api/claims/{claim_id}/serialize-837`` — regenerate X12 + 837P from the stored ``raw_json`` payload. 404 unknown claim, 422 + no-``raw_json`` / unparseable / serializer failure. +- ``GET /api/claims/{claim_id}/line-reconciliation`` — per-line 837 + vs 835 side-by-side with CAS adjustments and a summary block. + +Three single-router helpers stay in this file (per spec D4): + +- :func:`_compact_ack_links_for_claim` — slim form + ``{ack_id, ack_kind, set_accept_reject_code, …}`` for the drawer + Acknowledgments panel. +- :func:`_claim_line_dict` — project an 837 service-line + dict from ``Claim.raw_json`` to wire shape. +- :func:`_svc_to_dict` — project an ORM + ``ServiceLinePayment`` to wire shape. + +Inline imports inside handlers (preserved verbatim per spec D5): +``select`` (sqlalchemy), ``LineReconciliation``/``ServiceLinePayment``/ +``CasAdjustment`` (cyclone.db), ``json as _json``, ``Decimal``. + +SP36 Task 15: this block moved here from ``api.py:1278`` (the 5 +``/api/claims*`` routes + 3 single-router helpers). +""" +from __future__ import annotations + +import json +from decimal import Decimal +from typing import Any, AsyncIterator + +from fastapi import APIRouter, Depends, HTTPException, Query, Request +from fastapi.responses import JSONResponse, Response, StreamingResponse + +from cyclone import db +from cyclone.api_helpers import ( + ndjson_line as _ndjson_line, + ndjson_stream_list as _ndjson_stream_list, + tail_events as _tail_events, + wants_ndjson as _wants_ndjson, +) +from cyclone.auth.deps import matrix_gate +from cyclone.db import Claim +from cyclone.parsers.models import ClaimOutput +from cyclone.parsers.serialize_837 import SerializeError as SerializeError837 +from cyclone.parsers.serialize_837 import serialize_837 +from cyclone.pubsub import EventBus +from cyclone.store import store + +router = APIRouter(dependencies=[Depends(matrix_gate)]) + + +@router.get("/api/claims") +def list_claims( + request: Request, + batch_id: str | None = Query(None), + status: str | None = Query(None), + provider_npi: str | None = Query(None), + payer: str | None = Query(None), + date_from: str | None = Query(None), + date_to: str | None = Query(None), + sort: str | None = Query(None), + order: str = Query("desc"), + limit: int = Query(100, ge=1, le=1000), + offset: int = Query(0, ge=0), +) -> Any: + common = dict( + batch_id=batch_id, + status=status, + provider_npi=provider_npi, + payer=payer, + date_from=date_from, + date_to=date_to, + ) + items = list(store.iter_claims( + sort=sort, order=order, limit=limit, offset=offset, **common, + )) + # SP27 Task 13b: count the full population, not a 100-row sample. + # `iter_claims` defaults to limit=100; counting its output silently + # capped the reported total at 100 even when the DB held 60k rows. + total = store.count_claims(**common) + returned = len(items) + has_more = total > offset + returned + if _wants_ndjson(request): + return StreamingResponse( + _ndjson_stream_list(items, total, returned, has_more), + media_type="application/x-ndjson", + ) + return { + "items": items, + "total": total, + "returned": returned, + "has_more": has_more, + } + + +# --------------------------------------------------------------------------- # +# Live-tail NDJSON streaming endpoints (Phase 3 — SP5) +# --------------------------------------------------------------------------- # + + +@router.get("/api/claims/stream") +async def claims_stream( + request: Request, + status: str | None = Query(None), + provider_npi: str | None = Query(None), + payer: str | None = Query(None), + date_from: str | None = Query(None), + date_to: str | None = Query(None), + sort: str | None = Query(None), + order: str = Query("desc"), + limit: int = Query(100, ge=1, le=1000), +) -> StreamingResponse: + """Stream Claims as NDJSON: snapshot first, then live events. + + Wire format: + * ``{"type":"item","data":}`` per snapshot row, then per + new ``claim_written`` event + * ``{"type":"snapshot_end","data":{"count":N}}`` after the snapshot + * ``{"type":"heartbeat","data":{"ts":}}`` every + ``CYCLONE_TAIL_HEARTBEAT_S`` seconds when idle + + Query params mirror :func:`list_claims` so a frontend can swap a + one-shot fetch for a tail with no URL surgery. + + NOTE: registered before ``/api/claims/{claim_id}`` so the literal + ``stream`` path segment doesn't get matched as a claim id. + """ + bus: EventBus = request.app.state.event_bus + + async def gen() -> AsyncIterator[bytes]: + # 1. Snapshot (eager — iter_claims returns a list already). + rows = store.iter_claims( + status=status, provider_npi=provider_npi, payer=payer, + date_from=date_from, date_to=date_to, + sort=sort or "-submission_date", order=order, limit=limit, + ) + for row in rows: + yield _ndjson_line({"type": "item", "data": row}) + yield _ndjson_line({"type": "snapshot_end", "data": {"count": len(rows)}}) + + # 2. Subscribe + heartbeats. + async for chunk in _tail_events(request, bus, ["claim_written"]): + yield chunk + + return StreamingResponse(gen(), media_type="application/x-ndjson") + + +@router.get("/api/claims/{claim_id}") +def get_claim_detail_endpoint(claim_id: str) -> dict: + """Return one claim with full drawer context (SP4). + + Body shape is produced by :meth:`CycloneStore.get_claim_detail`: + header, state, service lines, diagnoses, parties, validation, + raw segments, ``stateHistory`` (most-recent-first, capped at 50), + and a populated ``matchedRemittance`` block when paired. + + SP28: response gains ``ack_links: list[dict]`` (compact form: + ``[{ack_id, ack_kind, set_accept_reject_code, parsed_at}]``) + so the ``ClaimDrawer`` Acknowledgments panel can render on + initial load. TA1 batch-level rows (``claim_id IS NULL``) are + excluded — those don't belong to a specific claim. + + Path param is ``claim_id`` (matches the SP3 ``/api/acks/{ack_id}`` + convention). Returns 404 — never 500 — on a missing claim so the + UI can distinguish "doesn't exist" from a transient fetch error. + """ + body = store.get_claim_detail(claim_id) + if body is None: + raise HTTPException( + status_code=404, + detail={ + "error": "Not found", + "detail": f"Claim {claim_id} not found", + }, + ) + # SP28: attach ack_links (compact form for the drawer panel). + body["ack_links"] = _compact_ack_links_for_claim(claim_id) + return body + + +def _compact_ack_links_for_claim(claim_id: str) -> list[dict]: + """Return compact ack_links for one claim, newest first. + + TA1 batch-level rows (claim_id IS NULL) are filtered out — those + hang off the originating 837 batch, not a specific claim. The + shape is the slimmer ``{ack_id, ack_kind, + set_accept_reject_code, parsed_at, ak2_index}`` form so the + ClaimDrawer can render without an N+1 round-trip per row. + """ + rows = store.list_acks_for_claim(claim_id) + out: list[dict] = [] + for row in rows: + if row.claim_id is None: + continue + out.append({ + "id": row.id, + "ack_id": row.ack_id, + "ack_kind": row.ack_kind, + "ak2_index": row.ak2_index, + "set_control_number": row.set_control_number, + "set_accept_reject_code": row.set_accept_reject_code, + "linked_at": ( + row.linked_at.isoformat().replace("+00:00", "Z") + if row.linked_at is not None else "" + ), + "linked_by": row.linked_by, + }) + return out + + +@router.get("/api/claims/{claim_id}/serialize-837") +def serialize_claim_as_837(claim_id: str): + """Return the claim as a regenerated X12 837P file (SP8). + + Loads the ClaimOutput from the persisted ``raw_json`` and runs the + outbound serializer. Returns 404 if the claim doesn't exist, 422 if + the stored payload has no parseable ClaimOutput (data integrity + issue, not a transient failure). + """ + with db.SessionLocal()() as s: + row = s.get(Claim, claim_id) + if row is None: + return JSONResponse( + {"error": "Not found", "detail": f"Claim {claim_id} not found"}, + status_code=404, + ) + if not row.raw_json: + return JSONResponse( + { + "error": "Unprocessable", + "detail": f"Claim {claim_id} has no raw_json; cannot serialize", + }, + status_code=422, + ) + try: + claim_obj = ClaimOutput.model_validate(row.raw_json) + except Exception as exc: + return JSONResponse( + { + "error": "Unprocessable", + "detail": f"Claim {claim_id} raw_json is malformed: {exc}", + }, + status_code=422, + ) + + try: + text = serialize_837(claim_obj) + except SerializeError837 as exc: + return JSONResponse( + {"error": "Unprocessable", "detail": str(exc)}, + status_code=422, + ) + + return Response( + content=text, + media_type="text/x12", + headers={ + "Content-Disposition": f'attachment; filename="claim-{claim_id}.x12"' + }, + ) + + +@router.get("/api/claims/{claim_id}/line-reconciliation") +def get_claim_line_reconciliation(claim_id: str) -> dict: + """Per-line reconciliation view for the ClaimDrawer tab. + + Spec §5.1. Returns the 837 service lines and 835 SVC composites + side-by-side, with per-line CAS adjustments and a summary block. + + Architecture note: 837 service lines live in ``Claim.raw_json`` + (not a separate ORM table), so the 837-side rows are read from the + JSON blob; the 835-side rows come from ``ServiceLinePayment`` ORM. + ``LineReconciliation.claim_service_line_number`` stores the 1-based + line number to join them. + """ + from sqlalchemy import select + from cyclone.db import ( + LineReconciliation, ServiceLinePayment, CasAdjustment, + ) + import json as _json + from decimal import Decimal + + with db.SessionLocal()() as s: + claim = s.get(db.Claim, claim_id) + if claim is None: + raise HTTPException( + status_code=404, + detail={"error": "Not found", "detail": f"Claim {claim_id} not found"}, + ) + + # 837 service lines: from raw_json. + raw = claim.raw_json or {} + claim_lines_raw = raw.get("service_lines") or [] + # Normalize to dicts for the response. + claim_lines = [_claim_line_dict(d) for d in claim_lines_raw] + + # 835 service payments: ORM rows from the matched remit. + remits = list( + s.execute( + select(db.Remittance).where(db.Remittance.claim_id == claim_id) + ).scalars().all() + ) + svc_payments: list[dict] = [] + svc_ids: list[int] = [] + if remits: + svc_rows = list( + s.execute( + select(ServiceLinePayment).where( + ServiceLinePayment.remittance_id.in_([r.id for r in remits]) + ).order_by(ServiceLinePayment.line_number) + ).scalars().all() + ) + for svc in svc_rows: + d = _svc_to_dict(svc) + svc_payments.append(d) + svc_ids.append(svc.id) + + # LineReconciliation rows. + lrs = list( + s.execute( + select(LineReconciliation).where(LineReconciliation.claim_id == claim_id) + ).scalars().all() + ) + # Index by claim_service_line_number and service_line_payment_id. + lr_by_claim_num: dict[int, LineReconciliation] = { + lr.claim_service_line_number: lr for lr in lrs if lr.claim_service_line_number is not None + } + lr_by_svc: dict[int, LineReconciliation] = { + lr.service_line_payment_id: lr for lr in lrs if lr.service_line_payment_id is not None + } + + # CAS rows grouped by svc id. + cas_by_svc: dict[int, list[CasAdjustment]] = {} + if svc_ids: + cas_rows = list( + s.execute( + select(CasAdjustment).where(CasAdjustment.service_line_payment_id.in_(svc_ids)) + ).scalars().all() + ) + for c in cas_rows: + cas_by_svc.setdefault(c.service_line_payment_id, []).append(c) + + # Build output lines array, preserving 837 order then 835-only. + svc_by_id: dict[int, dict] = {d["id"]: d for d in svc_payments} + lines_out: list[dict] = [] + billed_total = Decimal("0") + paid_total = Decimal("0") + adjustment_total = Decimal("0") + matched_count = 0 + used_svc_ids: set[int] = set() + + for cl in claim_lines: + billed_total += Decimal(str(cl["charge"])) + lr = lr_by_claim_num.get(cl["line_number"]) + if lr is None: + lines_out.append({ + "claim_service_line": cl, + "service_line_payment": None, + "status": "unmatched_837_only", + "adjustments": [], + }) + continue + svc_id = lr.service_line_payment_id + svc = svc_by_id.get(svc_id) if svc_id else None + if svc_id is not None: + used_svc_ids.add(svc_id) + cas_list = cas_by_svc.get(svc_id, []) if svc_id is not None else [] + cas_total = sum((Decimal(str(c.amount)) for c in cas_list), Decimal("0")) + if svc: + paid_total += Decimal(str(svc["payment"])) + adjustment_total += cas_total + if lr.status == "matched": + matched_count += 1 + lines_out.append({ + "claim_service_line": cl, + "service_line_payment": svc, + "status": lr.status, + "adjustments": [ + {"group_code": c.group_code, "reason_code": c.reason_code, + "amount": str(Decimal(str(c.amount)))} + for c in cas_list + ], + }) + + # 835-only lines (no claim match). + for lr in lrs: + if lr.claim_service_line_number is not None: + continue + svc_id = lr.service_line_payment_id + if svc_id is None: + continue + if svc_id in used_svc_ids: + continue + svc = svc_by_id.get(svc_id) + cas_list = cas_by_svc.get(svc_id, []) + cas_total = sum((Decimal(str(c.amount)) for c in cas_list), Decimal("0")) + if svc: + paid_total += Decimal(str(svc["payment"])) + adjustment_total += cas_total + lines_out.append({ + "claim_service_line": None, + "service_line_payment": svc, + "status": lr.status, + "adjustments": [ + {"group_code": c.group_code, "reason_code": c.reason_code, + "amount": str(Decimal(str(c.amount)))} + for c in cas_list + ], + }) + + return { + "claim_id": claim_id, + "summary": { + "billed_total": str(billed_total), + "paid_total": str(paid_total), + "adjustment_total": str(adjustment_total), + "matched_lines": matched_count, + "total_lines": len(claim_lines), + }, + "lines": lines_out, + } + + +def _claim_line_dict(d: dict) -> dict: + """Project an 837 service-line dict from ``Claim.raw_json`` to wire shape.""" + from decimal import Decimal + proc = d.get("procedure") or {} + charge = d.get("charge") + units = d.get("units") + return { + "line_number": d.get("line_number"), + "procedure_qualifier": proc.get("qualifier", "HC"), + "procedure_code": proc.get("code", ""), + "modifiers": proc.get("modifiers") or [], + "charge": str(Decimal(str(charge))) if charge is not None else "0", + "units": str(Decimal(str(units))) if units is not None else None, + "unit_type": d.get("unit_type"), + "service_date": d.get("service_date"), + } + + +def _svc_to_dict(svc) -> dict: + """Project an ORM ``ServiceLinePayment`` to wire shape.""" + import json as _json + from decimal import Decimal + return { + "id": svc.id, + "line_number": svc.line_number, + "procedure_qualifier": svc.procedure_qualifier, + "procedure_code": svc.procedure_code, + "modifiers": _json.loads(svc.modifiers_json or "[]"), + "charge": str(Decimal(str(svc.charge))), + "payment": str(Decimal(str(svc.payment))), + "units": str(Decimal(str(svc.units))) if svc.units is not None else None, + "unit_type": svc.unit_type, + "service_date": svc.service_date.isoformat() if svc.service_date else None, + }