From ff43f901564a7d9ff9dfab9c14e010c7de359f7d Mon Sep 17 00:00:00 2001 From: Tyler Date: Sun, 21 Jun 2026 04:28:11 -0600 Subject: [PATCH] refactor(api): split claims router (list + stream + detail + serialize + line-recon) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extracts the 5 /api/claims endpoints from api.py into cyclone.api_routers.claims: GET /api/claims (paginated list) GET /api/claims/stream (NDJSON live-tail) GET /api/claims/{claim_id} (drawer detail) GET /api/claims/{claim_id}/serialize-837 (X12 regen) GET /api/claims/{claim_id}/line-reconciliation (per-line view) Plus the two private projection helpers _claim_line_dict and _svc_to_dict which are only used by line-reconciliation. claims_stream is re-exported at the cyclone.api module level so test_api_stream_live.py's direct import keeps working (same pattern as the activity_stream re-export). Route ordering preserved: /stream is declared before /{claim_id} in the router source so FastAPI's first-match routing doesn't swallow the literal 'stream' segment as a claim id. api.py: 2201 -> 1841 (-360). Net diff: +432 / -389. Pytest: 8 failed / 735 passed / 16 skipped — identical to baseline. 57/57 claims-targeted tests pass. Live smoke: all 5 routes return expected status codes (200 for list/stream, 404 for missing-claim detail/serialize/line-recon). --- backend/src/cyclone/api.py | 370 +------------------ backend/src/cyclone/api_routers/claims.py | 411 ++++++++++++++++++++++ 2 files changed, 415 insertions(+), 366 deletions(-) create mode 100644 backend/src/cyclone/api_routers/claims.py diff --git a/backend/src/cyclone/api.py b/backend/src/cyclone/api.py index 9a1b9e9..10d670c 100644 --- a/backend/src/cyclone/api.py +++ b/backend/src/cyclone/api.py @@ -145,6 +145,7 @@ from cyclone.api_routers import ( # noqa: E402 acks, activity, batches, + claims, clearhouse, config, health, @@ -160,12 +161,14 @@ app.include_router(clearhouse.router) app.include_router(config.router) app.include_router(activity.router) app.include_router(batches.router) +app.include_router(claims.router) # Backwards-compat re-exports: test_api_stream_live.py imports these # names directly from ``cyclone.api`` to invoke the endpoint coroutines # in-process. Re-pointing at the router keeps the public surface stable # while the bodies live in the routers package. activity_stream = activity.activity_stream_endpoint +claims_stream = claims.claims_stream_endpoint def _resolve_payer(name: str) -> PayerConfig: @@ -1164,372 +1167,7 @@ def inbox_export_csv(lane: str): # --- /api/batches lives in cyclone.api_routers.batches --- -@app.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, - )) - total = len(list(store.iter_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") -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}") -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. - - 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", - }, - ) - return body - - -@app.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"' - }, - ) - - -@app.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, - } - +# --- /api/claims lives in cyclone.api_routers.claims --- @app.get("/api/reconciliation/unmatched") def get_reconciliation_unmatched() -> dict: diff --git a/backend/src/cyclone/api_routers/claims.py b/backend/src/cyclone/api_routers/claims.py new file mode 100644 index 0000000..71cb612 --- /dev/null +++ b/backend/src/cyclone/api_routers/claims.py @@ -0,0 +1,411 @@ +"""``/api/claims`` — list, live-tail, detail, X12 regenerate, line-reconciliation. + +This is the largest single-resource router. Five endpoints: + +* ``GET /api/claims`` — paginated list with filters + (status / batch_id / provider_npi / payer / date range / sort). +* ``GET /api/claims/stream`` — NDJSON snapshot + live tail via the + ``claim_written`` EventBus kind. +* ``GET /api/claims/{claim_id}`` — drawer detail + (header / state / service lines / diagnoses / parties / validation / + raw segments / stateHistory / matchedRemittance). +* ``GET /api/claims/{claim_id}/serialize-837`` — regenerate X12 for + download (SP8). +* ``GET /api/claims/{claim_id}/line-reconciliation`` — per-line + reconciliation view for the ClaimDrawer tab (spec §5.1). + +**Route ordering.** ``/stream`` is declared **before** ``/{claim_id}`` so +FastAPI's first-match routing doesn't swallow the literal ``stream`` +segment as a claim id. Same order as the prior inline code. + +The two projection helpers ``_claim_line_dict`` / ``_svc_to_dict`` are +private to this router; both only used by the line-reconciliation +endpoint. +""" +from __future__ import annotations + +from decimal import Decimal +from typing import Any, AsyncIterator + +from fastapi import APIRouter, HTTPException, Query, Request +from fastapi.responses import JSONResponse, Response, StreamingResponse + +from cyclone import db +from cyclone.api_helpers import ( + ndjson_line, + ndjson_stream_list, + tail_events, + wants_ndjson, +) +from cyclone.db import Claim +from cyclone.parsers.models import ClaimOutput +from cyclone.parsers.serialize_837 import SerializeError as SerializeError837, serialize_837 +from cyclone.pubsub import EventBus +from cyclone.store import store + +router = APIRouter() + + +@router.get("/api/claims") +def list_claims_endpoint( + 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: + """Paginated claims list with filters; supports NDJSON streaming. + + ``has_more`` is computed as ``total > offset + returned`` so paginating + forward keeps returning true until the last page is reached. + """ + 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, + )) + total = len(list(store.iter_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, + } + + +@router.get("/api/claims/stream") +async def claims_stream_endpoint( + 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_endpoint` 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. + + 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", + }, + ) + return body + + +@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 ( + CasAdjustment, + LineReconciliation, + ServiceLinePayment, + ) + + 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.""" + 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 + 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, + }