Compare commits
11 Commits
v0.2.0
...
27181144f2
| Author | SHA1 | Date | |
|---|---|---|---|
| 27181144f2 | |||
| eb674f890f | |||
| 804e557a49 | |||
| ff43f90156 | |||
| ea64e6e0f0 | |||
| 6ce638553b | |||
| e63be87ba9 | |||
| 6aa440d3e4 | |||
| 45cafbdb32 | |||
| bbf89c9dd8 | |||
| d25f00ac58 |
@@ -155,6 +155,7 @@ parses and rejects claims, the inbox reflects the new
|
||||
| POST | `/api/inbox/candidates/{remit_id}/match` | Manual match. `409` if the claim state moved out from under us. |
|
||||
| POST | `/api/inbox/candidates/dismiss` | `{pairs: [{claim_id, remit_id}]}`. Session-scoped. |
|
||||
| POST | `/api/inbox/rejected/resubmit` | `{claim_ids: [...]}`. `200` with `conflicts` for non-rejected. |
|
||||
| POST | `/api/inbox/payer-rejected/acknowledge` | `{claim_ids: [...], actor: "..."}`. Bulk-acknowledge payer rejections. Idempotent. `200` with `transitioned` / `already_acked` / `not_found` / `not_rejected` counts. Writes an audit event per transition; never overwrites the underlying 277CA rejection. |
|
||||
| GET | `/api/inbox/export.csv?lane=<lane>` | Streams CSV of the lane's rows. |
|
||||
|
||||
## Outbound 837 Serializer
|
||||
@@ -321,8 +322,9 @@ both be true for the same claim.
|
||||
|
||||
The `audit_log` table is the canonical record of every state transition
|
||||
the system has ever observed — claim lifecycle, reconciliation
|
||||
decisions, config reloads, SFTP submissions, 277CA rejects. SP11 made
|
||||
it tamper-evident: every row carries a SHA-256 hash of
|
||||
decisions, config reloads, SFTP submissions, 277CA rejects, Payer-Rejected
|
||||
acknowledgements, SQLCipher key rotations. SP11 made it tamper-evident:
|
||||
every row carries a SHA-256 hash of
|
||||
`(prev_hash || row_payload)`, forming a chain back to a genesis row.
|
||||
Any `INSERT`, `UPDATE`, or `DELETE` that breaks the chain is detectable
|
||||
in a single walk.
|
||||
@@ -357,6 +359,40 @@ operator has created the Keychain entry on first run. See
|
||||
for the one-time setup recipe and the HIPAA Security Rule §164.312(a)(2)(iv)
|
||||
mapping.
|
||||
|
||||
### Key rotation
|
||||
|
||||
The DB encryption key is rotated in place via `POST /api/admin/db/rotate-key`.
|
||||
The handler:
|
||||
|
||||
1. Generates a fresh 256-bit CSPRNG key (`db_crypto.generate_db_key()`).
|
||||
2. Persists the new key to the Keychain under the same
|
||||
`cyclone.db.key` account (overwriting the old one).
|
||||
3. Disposes the SQLAlchemy engine so pooled connections release the
|
||||
file (SQLCipher refuses to `PRAGMA rekey` while another connection
|
||||
holds the DB).
|
||||
4. Opens with the old key, issues `PRAGMA rekey`, and verifies the
|
||||
schema survived (table-count sanity check).
|
||||
5. Rebuilds the engine with the new key.
|
||||
6. Writes a `db.key_rotated` audit event carrying the SHA-256
|
||||
fingerprints of the old and new keys (first 8 hex chars) plus the
|
||||
post-rotation `table_count`.
|
||||
|
||||
When SQLCipher is enabled the engine uses SQLAlchemy's `NullPool`
|
||||
instead of the default `QueuePool`. `QueuePool` returns connections to
|
||||
a shared queue that any thread can pull from, which breaks SQLCipher's
|
||||
thread affinity. `NullPool` trades connection reuse for thread safety
|
||||
— the only correct behavior under FastAPI's per-request threadpool.
|
||||
|
||||
The rotation endpoint is serialized with a module-level
|
||||
`threading.Lock` (one rotation in flight at a time), returns `409` if
|
||||
a rotation is already running, `400` if encryption is not enabled,
|
||||
and `503` with a `reason` on `PRAGMA rekey` or Keychain failure so the
|
||||
operator can take the next step without parsing the traceback.
|
||||
|
||||
| Method | Path | Notes |
|
||||
| ------ | --------------------------------- | -------------------------------------------------------------- |
|
||||
| POST | `/api/admin/db/rotate-key` | Rotate the SQLCipher key in place. Audit-logged. |
|
||||
|
||||
## SFTP Wire-Up (paramiko)
|
||||
|
||||
The `clearhouse.submit` endpoint uses `paramiko` to push a batch of
|
||||
@@ -382,6 +418,29 @@ was a one-file change (`sftp_paramiko.py` replacing `sftp_stub.py`).
|
||||
the `paramiko` extras aren't installed so the test suite stays green on
|
||||
Linux dev boxes.
|
||||
|
||||
## Batches, Reconciliation, and Activity
|
||||
|
||||
These read/write endpoints are the core operator surface for browsing
|
||||
the parsed record and acting on matches. They predate the per-SP
|
||||
endpoint reference sections in the **Roadmap** and are listed here in
|
||||
one place so the route inventory stays discoverable.
|
||||
|
||||
| Method | Path | Notes |
|
||||
| ------ | --------------------------------------------- | ---------------------------------------------------------------- |
|
||||
| GET | `/api/batches` | All parsed batches, newest first. `?limit=` (1–1000, default 100). |
|
||||
| GET | `/api/batches/{batch_id}` | One batch detail (envelope, claims / remits, validation summary). |
|
||||
| GET | `/api/batch-diff?a=<id>&b=<id>` | Side-by-side diff of two batches: `added` / `removed` / `changed` claims + envelope metadata for each. Both query params required. |
|
||||
| GET | `/api/reconciliation/unmatched` | `{"claims": [...], "remittances": [...]}` — every Claim with no paired Remittance and vice versa. The two lists are always present (empty list, never absent) so the UI can index unconditionally. |
|
||||
| POST | `/api/reconciliation/match` | Body `{claim_id, remit_id}`. Manually pair a claim with a remit. `400` on missing ids, `404` on unknown id, `409` on already-matched or terminal-state claim. |
|
||||
| POST | `/api/reconciliation/unmatch` | Body `{claim_id}`. Remove the current match and reset the claim to `submitted`. `404` on unknown id, `409` on no current match. |
|
||||
| GET | `/api/providers` | Distinct providers across parsed claims. `?npi=`, `?state=`, `?limit=`, `?offset=`. Distinct from `/api/config/providers/{npi}` (SP9 config table) — this endpoint surfaces providers derived from the parsed claim stream. |
|
||||
| GET | `/api/activity` | Recent activity events. `?kind=`, `?since=`, `?limit=` (1–500, default 200). Powers the Activity page; the streaming counterpart `/api/activity/stream` is documented under **Live updates**. |
|
||||
|
||||
The per-SP endpoint blocks at the bottom of the **Roadmap** cover the
|
||||
SP-specific routes (parse, 999/TA1/277CA, Inbox, claim drawer, line
|
||||
reconciliation, outbound 837, multi-payer config, audit log, 277CA,
|
||||
key rotation, payer-rejected acknowledge).
|
||||
|
||||
## Persistence
|
||||
|
||||
Parsed batches, claims, remittances, matches, 277CA rejections,
|
||||
@@ -415,8 +474,20 @@ backup API).
|
||||
.
|
||||
├── backend/
|
||||
│ ├── src/cyclone/
|
||||
│ │ ├── api.py # FastAPI app, GET + parse routes, /api/{resource}/stream
|
||||
│ │ ├── api.py # FastAPI app + parse routes; mounts api_routers/* sub-apps
|
||||
│ │ ├── api_helpers.py # NDJSON / content-negotiation / live-tail helpers
|
||||
│ │ ├── api_routers/ # FastAPI APIRouters extracted from api.py
|
||||
│ │ │ ├── health.py # GET /api/health
|
||||
│ │ │ ├── acks.py # GET /api/acks, /api/acks/{id}
|
||||
│ │ │ ├── ta1_acks.py # GET /api/ta1-acks, /api/ta1-acks/{id}
|
||||
│ │ │ ├── activity.py # GET /api/activity, /api/activity/stream
|
||||
│ │ │ ├── batches.py # GET /api/batches, /api/batches/{id}, /api/batch-diff
|
||||
│ │ │ ├── claims.py # GET /api/claims, /api/claims/stream, /api/claims/{id},
|
||||
│ │ │ │ # /api/claims/{id}/serialize-837, /api/claims/{id}/line-reconciliation
|
||||
│ │ │ ├── remittances.py # GET /api/remittances, /api/remittances/stream, /api/remittances/{id}
|
||||
│ │ │ ├── providers.py # GET /api/providers
|
||||
│ │ │ ├── clearhouse.py # GET /api/clearhouse, POST /api/clearhouse/submit
|
||||
│ │ │ └── config.py # /api/config/providers[/...], /api/config/payers[/...], /api/admin/reload-config
|
||||
│ │ ├── pubsub.py # in-process EventBus (drop-oldest, per-kind fan-out)
|
||||
│ │ ├── store.py # CycloneStore, mappers, publish-on-write
|
||||
│ │ ├── db.py # SQLAlchemy engine, session factory, ORM models
|
||||
@@ -473,7 +544,7 @@ backup API).
|
||||
|
||||
## Roadmap
|
||||
|
||||
Sub-projects 2 through 13 are **shipped**. See the [completeness
|
||||
Sub-projects 2 through 15 are **shipped**. See the [completeness
|
||||
review](docs/reviews/2026-06-20-cyclone-completeness-review.md) for
|
||||
the honest gap analysis against the industry definition of a HIPAA
|
||||
clearinghouse — the short version is that the local-only,
|
||||
@@ -484,6 +555,19 @@ scope.
|
||||
|
||||
Shipped sub-projects (most recent first):
|
||||
|
||||
- **Sub-project 15 (shipped) — SQLCipher key rotation.** In-place
|
||||
rotation via `PRAGMA rekey`, serialized through a module-level
|
||||
`threading.Lock` and a SQLAlchemy `NullPool` to keep SQLCipher
|
||||
thread-affine under FastAPI's per-request threadpool. Writes a
|
||||
`db.key_rotated` audit event with old + new key fingerprints and
|
||||
post-rotation `table_count`. See
|
||||
[Encryption at Rest — Key rotation](#key-rotation).
|
||||
- **Sub-project 14 (shipped) — 5-lane Inbox UI.** The Payer-Rejected
|
||||
lane is now rendered in the Inbox alongside Rejected / Candidates /
|
||||
Unmatched / Done today. New bulk action
|
||||
`POST /api/inbox/payer-rejected/acknowledge` drops claims from the
|
||||
working surface without erasing the original 277CA rejection event
|
||||
(audit log stays intact, SP11).
|
||||
- **Sub-project 13 (shipped) — SFTP wire-up.** `paramiko`-backed
|
||||
`SftpClient` replaces the SP9 stub. The clearhouse.submit endpoint
|
||||
actually pushes to
|
||||
@@ -752,6 +836,34 @@ the one-time setup recipe.
|
||||
`mft.gainwelltechnologies.com:/CO XIX/PROD/coxix_prod_11525703/FromHPE`.
|
||||
SFTP credentials are fetched from the macOS Keychain at call time.
|
||||
|
||||
### SP14 endpoints (5-lane Inbox UI + acknowledge)
|
||||
|
||||
- `GET /api/inbox/lanes` — same shape as SP6; the `payer_rejected`
|
||||
lane payload now also includes
|
||||
`payer_rejected_acknowledged_at` + `payer_rejected_acknowledged_actor`
|
||||
per row so the UI can badge acknowledged claims (forward-compat for
|
||||
a future "Recently acknowledged" view).
|
||||
- `POST /api/inbox/payer-rejected/acknowledge` — bulk-acknowledge
|
||||
Payer-Rejected claims. Body: `{claim_ids: [...], actor: "..."}`.
|
||||
Idempotent. Response: `200` with
|
||||
`{transitioned, already_acked, not_found, not_rejected}`. Writes an
|
||||
`inbox.payer_rejected_acknowledged` audit event per transition.
|
||||
Acknowledgement hides a claim from the lane but never overwrites
|
||||
`payer_rejected_at` / `payer_rejected_reason` /
|
||||
`payer_rejected_by_277ca_id` — the original 277CA evidence stays
|
||||
intact in the audit log.
|
||||
|
||||
### SP15 endpoints (SQLCipher key rotation)
|
||||
|
||||
- `POST /api/admin/db/rotate-key` — rotate the SQLCipher DB key in
|
||||
place. Generates a fresh 256-bit key, writes it to the Keychain
|
||||
(overwriting `cyclone.db.key`), disposes the engine, issues
|
||||
`PRAGMA rekey`, verifies the schema, rebuilds the engine. Writes
|
||||
a `db.key_rotated` audit event with old + new key fingerprints
|
||||
and `table_count`. Returns `409` when a rotation is already in
|
||||
flight, `400` when encryption is not enabled, `503` with a
|
||||
`reason` on `PRAGMA rekey` or Keychain failure.
|
||||
|
||||
## License
|
||||
|
||||
No license file yet; this is internal-use software. Add a `LICENSE` file
|
||||
|
||||
+30
-751
@@ -141,11 +141,37 @@ app.add_middleware(
|
||||
# and are wired in here. (Kept as a top-level package rather than nested
|
||||
# under `cyclone.api` so the existing ``cyclone.api`` module path keeps
|
||||
# working — Python prefers packages over same-named modules.)
|
||||
from cyclone.api_routers import acks, health, ta1_acks # noqa: E402
|
||||
from cyclone.api_routers import ( # noqa: E402
|
||||
acks,
|
||||
activity,
|
||||
batches,
|
||||
claims,
|
||||
clearhouse,
|
||||
config,
|
||||
health,
|
||||
providers,
|
||||
remittances,
|
||||
ta1_acks,
|
||||
)
|
||||
|
||||
app.include_router(health.router)
|
||||
app.include_router(acks.router)
|
||||
app.include_router(ta1_acks.router)
|
||||
app.include_router(providers.router)
|
||||
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)
|
||||
app.include_router(remittances.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
|
||||
remittances_stream = remittances.remittances_stream_endpoint
|
||||
|
||||
|
||||
def _resolve_payer(name: str) -> PayerConfig:
|
||||
@@ -1141,431 +1167,10 @@ def inbox_export_csv(lane: str):
|
||||
)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# GET endpoints (read views over the in-memory store)
|
||||
# --------------------------------------------------------------------------- #
|
||||
# --- /api/batches lives in cyclone.api_routers.batches ---
|
||||
|
||||
|
||||
def _batch_summary_claim_count(rec: BatchRecord) -> int:
|
||||
"""Return the number of claims on a batch, handling both 837P and 835."""
|
||||
if rec.kind == "837p":
|
||||
return len(rec.result.claims) # type: ignore[attr-defined]
|
||||
if rec.kind == "835":
|
||||
return len(rec.result.claims) # type: ignore[attr-defined]
|
||||
return 0
|
||||
|
||||
|
||||
@app.get("/api/batches")
|
||||
def list_batches(
|
||||
request: Request,
|
||||
limit: int = Query(100, ge=1, le=1000),
|
||||
) -> Any:
|
||||
"""Summary of all parsed batches, newest first."""
|
||||
records = store.list(limit=limit)
|
||||
items = [
|
||||
{
|
||||
"id": r.id,
|
||||
"kind": r.kind,
|
||||
"inputFilename": r.input_filename,
|
||||
"parsedAt": r.parsed_at.isoformat().replace("+00:00", "Z"),
|
||||
"claimCount": _batch_summary_claim_count(r),
|
||||
}
|
||||
for r in records
|
||||
]
|
||||
all_records = store.all()
|
||||
total = len(all_records)
|
||||
returned = len(items)
|
||||
has_more = total > 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,
|
||||
}
|
||||
|
||||
|
||||
@app.get("/api/batches/{batch_id}")
|
||||
def get_batch(batch_id: str) -> Any:
|
||||
rec = store.get(batch_id)
|
||||
if rec is None:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail={"error": "Not found", "detail": f"Batch {batch_id} not found"},
|
||||
)
|
||||
return json.loads(rec.result.model_dump_json())
|
||||
|
||||
|
||||
@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":<claim>}`` per snapshot row, then per
|
||||
new ``claim_written`` event
|
||||
* ``{"type":"snapshot_end","data":{"count":N}}`` after the snapshot
|
||||
* ``{"type":"heartbeat","data":{"ts":<iso>}}`` 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:
|
||||
@@ -1708,193 +1313,11 @@ def post_reconciliation_unmatch(body: dict) -> dict:
|
||||
)
|
||||
|
||||
|
||||
@app.get("/api/remittances")
|
||||
def list_remittances(
|
||||
request: Request,
|
||||
batch_id: str | None = Query(None),
|
||||
payer: str | None = Query(None),
|
||||
claim_id: 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,
|
||||
payer=payer,
|
||||
claim_id=claim_id,
|
||||
date_from=date_from,
|
||||
date_to=date_to,
|
||||
)
|
||||
items = list(store.iter_remittances(
|
||||
sort=sort, order=order, limit=limit, offset=offset, **common,
|
||||
))
|
||||
total = len(list(store.iter_remittances(**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,
|
||||
}
|
||||
# --- /api/remittances lives in cyclone.api_routers.remittances ---
|
||||
|
||||
|
||||
@app.get("/api/remittances/stream")
|
||||
async def remittances_stream(
|
||||
request: Request,
|
||||
payer: str | None = Query(None),
|
||||
claim_id: 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 Remittances as NDJSON: snapshot first, then live events.
|
||||
|
||||
Subscribes to ``remittance_written``. Default sort is
|
||||
``-received_date`` (newest-first), matching the list endpoint's
|
||||
most common sort.
|
||||
|
||||
NOTE: registered before ``/api/remittances/{remittance_id}`` so
|
||||
the literal ``stream`` path segment doesn't get matched as a
|
||||
remittance id.
|
||||
"""
|
||||
bus: EventBus = request.app.state.event_bus
|
||||
|
||||
async def gen() -> AsyncIterator[bytes]:
|
||||
rows = store.iter_remittances(
|
||||
payer=payer, claim_id=claim_id,
|
||||
date_from=date_from, date_to=date_to,
|
||||
sort=sort or "-received_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)}})
|
||||
|
||||
async for chunk in _tail_events(request, bus, ["remittance_written"]):
|
||||
yield chunk
|
||||
|
||||
return StreamingResponse(gen(), media_type="application/x-ndjson")
|
||||
|
||||
|
||||
@app.get("/api/remittances/{remittance_id}")
|
||||
def get_remittance(remittance_id: str) -> dict:
|
||||
"""Return one remittance with its labeled CAS ``adjustments`` array.
|
||||
|
||||
Path param is ``remittance_id`` (not ``id``) to avoid shadowing
|
||||
FastAPI's internal ``id`` name and to keep OpenAPI docs self-
|
||||
describing. Returns 404 when the remittance is missing — never 500.
|
||||
"""
|
||||
body = store.get_remittance(remittance_id)
|
||||
if body is None:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail={"error": "Not found", "detail": f"Remittance {remittance_id} not found"},
|
||||
)
|
||||
return body
|
||||
|
||||
|
||||
@app.get("/api/providers")
|
||||
def list_providers(
|
||||
request: Request,
|
||||
npi: str | None = Query(None),
|
||||
state: str | None = Query(None),
|
||||
limit: int = Query(100, ge=1, le=1000),
|
||||
offset: int = Query(0, ge=0),
|
||||
) -> Any:
|
||||
items = store.distinct_providers()
|
||||
if npi is not None:
|
||||
items = [p for p in items if p["npi"] == npi]
|
||||
if state is not None:
|
||||
items = [p for p in items if p.get("state") == state]
|
||||
paged = items[offset:offset + limit]
|
||||
total = len(items)
|
||||
returned = len(paged)
|
||||
has_more = total > offset + returned
|
||||
if _wants_ndjson(request):
|
||||
return StreamingResponse(
|
||||
_ndjson_stream_list(paged, total, returned, has_more),
|
||||
media_type="application/x-ndjson",
|
||||
)
|
||||
return {
|
||||
"items": paged,
|
||||
"total": total,
|
||||
"returned": returned,
|
||||
"has_more": has_more,
|
||||
}
|
||||
|
||||
|
||||
@app.get("/api/activity")
|
||||
def list_activity(
|
||||
request: Request,
|
||||
kind: str | None = Query(None),
|
||||
since: str | None = Query(None),
|
||||
limit: int = Query(200, ge=1, le=500),
|
||||
) -> Any:
|
||||
events = store.recent_activity(limit=limit)
|
||||
if kind is not None:
|
||||
events = [e for e in events if e["kind"] == kind]
|
||||
if since is not None:
|
||||
events = [e for e in events if e["timestamp"] >= since]
|
||||
total = len(events)
|
||||
has_more = False
|
||||
if _wants_ndjson(request):
|
||||
return StreamingResponse(
|
||||
_ndjson_stream_list(events, total, total, has_more),
|
||||
media_type="application/x-ndjson",
|
||||
)
|
||||
return {
|
||||
"items": events,
|
||||
"total": total,
|
||||
"returned": total,
|
||||
"has_more": has_more,
|
||||
}
|
||||
|
||||
|
||||
@app.get("/api/activity/stream")
|
||||
async def activity_stream(
|
||||
request: Request,
|
||||
kind: str | None = Query(None),
|
||||
since: str | None = Query(None),
|
||||
limit: int = Query(50, ge=1, le=500),
|
||||
) -> StreamingResponse:
|
||||
"""Stream Activity events as NDJSON: snapshot first, then live events.
|
||||
|
||||
Subscribes to ``activity_recorded``. Default ``limit`` is 50
|
||||
(smaller than the list endpoint's 200) because activity is
|
||||
high-volume — callers usually want the most recent handful, not a
|
||||
full replay.
|
||||
"""
|
||||
bus: EventBus = request.app.state.event_bus
|
||||
|
||||
async def gen() -> AsyncIterator[bytes]:
|
||||
# Snapshot reuses the same in-memory filter as ``list_activity``
|
||||
# so the two endpoints are interchangeable for the snapshot
|
||||
# half.
|
||||
events = store.recent_activity(limit=limit)
|
||||
if kind is not None:
|
||||
events = [e for e in events if e["kind"] == kind]
|
||||
if since is not None:
|
||||
events = [e for e in events if e["timestamp"] >= since]
|
||||
for ev in events:
|
||||
yield _ndjson_line({"type": "item", "data": ev})
|
||||
yield _ndjson_line({
|
||||
"type": "snapshot_end", "data": {"count": len(events)},
|
||||
})
|
||||
|
||||
async for chunk in _tail_events(request, bus, ["activity_recorded"]):
|
||||
yield chunk
|
||||
|
||||
return StreamingResponse(gen(), media_type="application/x-ndjson")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
@@ -2089,125 +1512,8 @@ async def post_eligibility_parse_271(
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# SP9: providers / payers / clearhouse endpoints
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@app.get("/api/clearhouse")
|
||||
def get_clearhouse():
|
||||
"""Return the singleton clearhouse config (dzinesco's identity, SFTP block, filename block)."""
|
||||
ch = store.get_clearhouse()
|
||||
if ch is None:
|
||||
raise HTTPException(status_code=404, detail="clearhouse not seeded")
|
||||
return json.loads(ch.model_dump_json())
|
||||
|
||||
|
||||
@app.post("/api/clearhouse/submit")
|
||||
def submit_to_clearhouse(body: dict):
|
||||
"""Submit a batch of claims to the clearhouse (SFTP). SP9: stub.
|
||||
|
||||
Body: ``{"claim_ids": [...], "payer_id": "CO_TXIX"}``
|
||||
|
||||
Stub behavior: serializes each claim via the SP7 serializer, builds
|
||||
an HCPF-compliant outbound filename, and copies the result to
|
||||
``{staging_dir}/{outbound_path}/{filename}`` instead of opening a
|
||||
real SFTP connection. Returns a receipt per claim.
|
||||
"""
|
||||
from cyclone.clearhouse import make_client
|
||||
from cyclone.edi.filenames import build_outbound_filename
|
||||
|
||||
claim_ids = body.get("claim_ids", [])
|
||||
payer_id = body.get("payer_id")
|
||||
if not claim_ids:
|
||||
raise HTTPException(status_code=400, detail="claim_ids required")
|
||||
if not payer_id:
|
||||
raise HTTPException(status_code=400, detail="payer_id required")
|
||||
|
||||
ch = store.get_clearhouse()
|
||||
if ch is None:
|
||||
raise HTTPException(status_code=500, detail="clearhouse not seeded")
|
||||
|
||||
client = make_client(ch.sftp_block)
|
||||
results = []
|
||||
for cid in claim_ids:
|
||||
try:
|
||||
x12_text = _serialize_claim_for_submit(cid)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
results.append({"claim_id": cid, "ok": False, "error": str(exc)})
|
||||
continue
|
||||
filename = build_outbound_filename(ch.tpid, "837P")
|
||||
remote = f"{ch.sftp_block.paths['outbound']}/{filename}"
|
||||
staging_path = client.write_file(remote, x12_text.encode("utf-8"))
|
||||
results.append({
|
||||
"claim_id": cid,
|
||||
"ok": True,
|
||||
"filename": filename,
|
||||
"staging_path": str(staging_path),
|
||||
"remote_path": remote,
|
||||
})
|
||||
# SP11: audit trail for each successful clearhouse submission.
|
||||
with db.SessionLocal()() as audit_s:
|
||||
append_event(audit_s, AuditEvent(
|
||||
event_type="clearhouse.submitted",
|
||||
entity_type="claim",
|
||||
entity_id=cid,
|
||||
payload={
|
||||
"filename": filename,
|
||||
"remote_path": remote,
|
||||
"tpid": ch.tpid,
|
||||
"stub": ch.sftp_block.stub,
|
||||
},
|
||||
actor="clearhouse-submit",
|
||||
))
|
||||
audit_s.commit()
|
||||
return {"ok": True, "submitted": results, "stub": ch.sftp_block.stub}
|
||||
|
||||
|
||||
def _serialize_claim_for_submit(claim_id: str) -> str:
|
||||
"""Serialize a claim to X12 for SFTP submission. Lazy import of the
|
||||
serializer to avoid pulling FastAPI machinery at module import time."""
|
||||
from cyclone.parsers.serialize_837 import serialize_837
|
||||
from cyclone import db
|
||||
with db.SessionLocal()() as s:
|
||||
row = s.get(db.Claim, claim_id)
|
||||
if row is None:
|
||||
raise ValueError(f"claim {claim_id!r} not found")
|
||||
# Re-parse the stored raw_json to get a ClaimOutput
|
||||
from cyclone.parsers.models import ClaimOutput, Envelope, Subscriber, Payer, BillingProvider
|
||||
from cyclone.parsers.parse_837 import parse
|
||||
raw = row.raw_json or {}
|
||||
# Reconstruct minimal ClaimOutput from raw_json; this is best-effort.
|
||||
return _serialize_claim_from_raw(row, raw)
|
||||
|
||||
|
||||
def _serialize_claim_from_raw(claim_row, raw: dict) -> str:
|
||||
"""Best-effort serializer that uses the stored raw_json to emit a fresh 837.
|
||||
|
||||
For SP9 this delegates to the existing serialize_837 helper if the
|
||||
claim has a complete raw_segments array. Otherwise it returns a
|
||||
minimal placeholder.
|
||||
"""
|
||||
from cyclone.parsers.serialize_837 import serialize_837
|
||||
from cyclone.parsers.parse_837 import parse
|
||||
|
||||
# Re-parse the original batch text (need to re-derive from store).
|
||||
# SP9 stub: if the claim has a `raw_json` with `x12_text`, use that.
|
||||
if isinstance(raw, dict) and raw.get("x12_text"):
|
||||
result = parse(raw["x12_text"])
|
||||
if result.claims:
|
||||
return serialize_837(result.claims[0])
|
||||
# Fallback: raise so the caller sees an error.
|
||||
raise RuntimeError(
|
||||
f"claim {claim_row.id!r} cannot be re-serialized: no stored x12_text"
|
||||
)
|
||||
|
||||
|
||||
@app.get("/api/config/providers")
|
||||
def list_configured_providers(is_active: bool | None = Query(default=True)):
|
||||
"""List the configured provider rows (3 NPIs for SP9)."""
|
||||
return [json.loads(p.model_dump_json()) for p in store.list_providers(is_active=is_active)]
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# SP11: tamper-evident audit log (admin)
|
||||
@@ -2430,35 +1736,8 @@ def rotate_db_key_endpoint(body: dict | None = None) -> Any:
|
||||
_db_rotate_lock.release()
|
||||
|
||||
|
||||
@app.get("/api/config/providers/{npi}")
|
||||
def get_configured_provider(npi: str):
|
||||
p = store.get_provider(npi)
|
||||
if p is None:
|
||||
raise HTTPException(status_code=404, detail=f"provider {npi!r} not found")
|
||||
return json.loads(p.model_dump_json())
|
||||
|
||||
|
||||
@app.get("/api/config/payers")
|
||||
def list_configured_payers(is_active: bool | None = Query(default=True)):
|
||||
return [json.loads(p.model_dump_json()) for p in store.list_payers(is_active=is_active)]
|
||||
|
||||
|
||||
@app.get("/api/config/payers/{payer_id}/configs")
|
||||
def list_payer_configs(payer_id: str):
|
||||
"""List all (transaction_type, config_json) blocks for a payer."""
|
||||
from cyclone import payers as payer_loader
|
||||
configs = [
|
||||
{"transaction_type": tx, "config_json": block, "source": "yaml"}
|
||||
for (pid, tx), block in payer_loader.all_configs().items()
|
||||
if pid == payer_id
|
||||
]
|
||||
# Also check the DB for runtime-overridden configs
|
||||
for tx in ("837P", "835", "277CA", "999", "TA1"):
|
||||
live = store.get_payer_config(payer_id, tx)
|
||||
if live is not None:
|
||||
configs.append({"transaction_type": tx, "config_json": live, "source": "db"})
|
||||
return configs
|
||||
|
||||
|
||||
@app.post("/api/admin/reload-config")
|
||||
def reload_config():
|
||||
|
||||
@@ -0,0 +1,97 @@
|
||||
"""``/api/activity`` — list & live-tail of recorded Activity events.
|
||||
|
||||
The Activity log is the cross-resource audit feed: every parser write,
|
||||
inbox state change, match/dismiss/resubmit, clearhouse submission, etc.
|
||||
records one row keyed by ``kind``. The list endpoint returns the most
|
||||
recent ``limit`` events with optional ``kind`` / ``since`` filters; the
|
||||
stream endpoint emits a snapshot followed by live updates from the
|
||||
``EventBus``.
|
||||
|
||||
The default ``limit`` on the list endpoint is 200 (matches the prior
|
||||
inline behavior); the stream endpoint defaults to 50 because activity is
|
||||
high-volume — callers usually want the most recent handful, not a full
|
||||
replay.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, AsyncIterator
|
||||
|
||||
from fastapi import APIRouter, Query, Request
|
||||
from fastapi.responses import StreamingResponse
|
||||
|
||||
from cyclone.api_helpers import (
|
||||
ndjson_line,
|
||||
ndjson_stream_list,
|
||||
tail_events,
|
||||
wants_ndjson,
|
||||
)
|
||||
from cyclone.pubsub import EventBus
|
||||
from cyclone.store import store
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
@router.get("/api/activity")
|
||||
def list_activity_endpoint(
|
||||
request: Request,
|
||||
kind: str | None = Query(None),
|
||||
since: str | None = Query(None),
|
||||
limit: int = Query(200, ge=1, le=500),
|
||||
) -> Any:
|
||||
"""Return the recent Activity log, newest first.
|
||||
|
||||
Optional ``kind`` filters by ``event["kind"]`` (e.g. ``claim_submitted``).
|
||||
Optional ``since`` is an inclusive ISO timestamp lower bound applied
|
||||
to ``event["timestamp"]``.
|
||||
"""
|
||||
events = store.recent_activity(limit=limit)
|
||||
if kind is not None:
|
||||
events = [e for e in events if e["kind"] == kind]
|
||||
if since is not None:
|
||||
events = [e for e in events if e["timestamp"] >= since]
|
||||
total = len(events)
|
||||
has_more = False
|
||||
if wants_ndjson(request):
|
||||
return StreamingResponse(
|
||||
ndjson_stream_list(events, total, total, has_more),
|
||||
media_type="application/x-ndjson",
|
||||
)
|
||||
return {
|
||||
"items": events,
|
||||
"total": total,
|
||||
"returned": total,
|
||||
"has_more": has_more,
|
||||
}
|
||||
|
||||
|
||||
@router.get("/api/activity/stream")
|
||||
async def activity_stream_endpoint(
|
||||
request: Request,
|
||||
kind: str | None = Query(None),
|
||||
since: str | None = Query(None),
|
||||
limit: int = Query(50, ge=1, le=500),
|
||||
) -> StreamingResponse:
|
||||
"""Stream Activity events as NDJSON: snapshot first, then live events.
|
||||
|
||||
Subscribes to ``activity_recorded``. The snapshot half reuses the
|
||||
same in-memory filter as ``list_activity`` so the two endpoints are
|
||||
interchangeable for the snapshot half.
|
||||
"""
|
||||
bus: EventBus = request.app.state.event_bus
|
||||
|
||||
async def gen() -> AsyncIterator[bytes]:
|
||||
events = store.recent_activity(limit=limit)
|
||||
if kind is not None:
|
||||
events = [e for e in events if e["kind"] == kind]
|
||||
if since is not None:
|
||||
events = [e for e in events if e["timestamp"] >= since]
|
||||
for ev in events:
|
||||
yield ndjson_line({"type": "item", "data": ev})
|
||||
yield ndjson_line({
|
||||
"type": "snapshot_end", "data": {"count": len(events)},
|
||||
})
|
||||
|
||||
async for chunk in tail_events(request, bus, ["activity_recorded"]):
|
||||
yield chunk
|
||||
|
||||
return StreamingResponse(gen(), media_type="application/x-ndjson")
|
||||
@@ -0,0 +1,75 @@
|
||||
"""``/api/batches`` — list & detail endpoints for parsed batches.
|
||||
|
||||
Each ``store.add(...)`` call (837P, 835, 999, TA1, 277CA) produces a
|
||||
``BatchRecord`` kept in memory by :class:`CycloneStore`. The list
|
||||
endpoint returns a lightweight summary (id, kind, filename, parsedAt,
|
||||
claim count) for the recent batches; the detail endpoint returns the
|
||||
full parsed payload for one batch.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from typing import Any
|
||||
|
||||
from fastapi import APIRouter, HTTPException, Query, Request
|
||||
from fastapi.responses import StreamingResponse
|
||||
|
||||
from cyclone.api_helpers import ndjson_stream_list, wants_ndjson
|
||||
from cyclone.store import BatchRecord, store
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
def _batch_summary_claim_count(rec: BatchRecord) -> int:
|
||||
"""Return the number of claims on a batch, handling both 837P and 835."""
|
||||
if rec.kind == "837p":
|
||||
return len(rec.result.claims) # type: ignore[attr-defined]
|
||||
if rec.kind == "835":
|
||||
return len(rec.result.claims) # type: ignore[attr-defined]
|
||||
return 0
|
||||
|
||||
|
||||
@router.get("/api/batches")
|
||||
def list_batches_endpoint(
|
||||
request: Request,
|
||||
limit: int = Query(100, ge=1, le=1000),
|
||||
) -> Any:
|
||||
"""Summary of all parsed batches, newest first."""
|
||||
records = store.list(limit=limit)
|
||||
items = [
|
||||
{
|
||||
"id": r.id,
|
||||
"kind": r.kind,
|
||||
"inputFilename": r.input_filename,
|
||||
"parsedAt": r.parsed_at.isoformat().replace("+00:00", "Z"),
|
||||
"claimCount": _batch_summary_claim_count(r),
|
||||
}
|
||||
for r in records
|
||||
]
|
||||
all_records = store.all()
|
||||
total = len(all_records)
|
||||
returned = len(items)
|
||||
has_more = total > 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/batches/{batch_id}")
|
||||
def get_batch_endpoint(batch_id: str) -> Any:
|
||||
"""Return the full parsed payload for one batch."""
|
||||
rec = store.get(batch_id)
|
||||
if rec is None:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail={"error": "Not found", "detail": f"Batch {batch_id} not found"},
|
||||
)
|
||||
return json.loads(rec.result.model_dump_json())
|
||||
@@ -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":<claim>}`` per snapshot row, then per
|
||||
new ``claim_written`` event
|
||||
* ``{"type":"snapshot_end","data":{"count":N}}`` after the snapshot
|
||||
* ``{"type":"heartbeat","data":{"ts":<iso>}}`` 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,
|
||||
}
|
||||
@@ -0,0 +1,126 @@
|
||||
"""``/api/clearhouse`` — SP9 single-payer SFTP submit surface.
|
||||
|
||||
Two endpoints:
|
||||
- ``GET /api/clearhouse`` — return the singleton clearhouse config
|
||||
(dzinesco's identity, SFTP block, filename block). 404 when the
|
||||
config hasn't been seeded yet.
|
||||
- ``POST /api/clearhouse/submit`` — submit a batch of claims to the
|
||||
clearhouse (SFTP). SP9 ships a stub that copies to ``staging_dir``
|
||||
instead of opening a real SFTP connection; SP13 wires the real
|
||||
paramiko-backed client behind the same ``make_client()`` factory.
|
||||
|
||||
For each claim the endpoint serializes a fresh 837 via
|
||||
:func:`_serialize_claim_for_submit`, builds an HCPF-compliant outbound
|
||||
filename, and records an audit-log row on success. Failures are
|
||||
captured per-claim so the operator sees exactly which claim_ids
|
||||
didn't go through.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
|
||||
from fastapi import APIRouter, HTTPException
|
||||
|
||||
from cyclone import db
|
||||
from cyclone.audit_log import AuditEvent, append_event
|
||||
from cyclone.parsers.parse_837 import parse
|
||||
from cyclone.parsers.serialize_837 import serialize_837
|
||||
from cyclone.store import store
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
@router.get("/api/clearhouse")
|
||||
def get_clearhouse():
|
||||
"""Return the singleton clearhouse config (dzinesco's identity, SFTP block, filename block)."""
|
||||
ch = store.get_clearhouse()
|
||||
if ch is None:
|
||||
raise HTTPException(status_code=404, detail="clearhouse not seeded")
|
||||
return json.loads(ch.model_dump_json())
|
||||
|
||||
|
||||
@router.post("/api/clearhouse/submit")
|
||||
def submit_to_clearhouse(body: dict):
|
||||
"""Submit a batch of claims to the clearhouse (SFTP). SP9: stub.
|
||||
|
||||
Body: ``{"claim_ids": [...], "payer_id": "CO_TXIX"}``
|
||||
|
||||
Stub behavior: serializes each claim via the SP7 serializer, builds
|
||||
an HCPF-compliant outbound filename, and copies the result to
|
||||
``{staging_dir}/{outbound_path}/{filename}`` instead of opening a
|
||||
real SFTP connection. Returns a receipt per claim.
|
||||
"""
|
||||
from cyclone.clearhouse import make_client
|
||||
from cyclone.edi.filenames import build_outbound_filename
|
||||
|
||||
claim_ids = body.get("claim_ids", [])
|
||||
payer_id = body.get("payer_id")
|
||||
if not claim_ids:
|
||||
raise HTTPException(status_code=400, detail="claim_ids required")
|
||||
if not payer_id:
|
||||
raise HTTPException(status_code=400, detail="payer_id required")
|
||||
|
||||
ch = store.get_clearhouse()
|
||||
if ch is None:
|
||||
raise HTTPException(status_code=500, detail="clearhouse not seeded")
|
||||
|
||||
client = make_client(ch.sftp_block)
|
||||
results = []
|
||||
for cid in claim_ids:
|
||||
try:
|
||||
x12_text = _serialize_claim_for_submit(cid)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
results.append({"claim_id": cid, "ok": False, "error": str(exc)})
|
||||
continue
|
||||
filename = build_outbound_filename(ch.tpid, "837P")
|
||||
remote = f"{ch.sftp_block.paths['outbound']}/{filename}"
|
||||
staging_path = client.write_file(remote, x12_text.encode("utf-8"))
|
||||
results.append({
|
||||
"claim_id": cid,
|
||||
"ok": True,
|
||||
"filename": filename,
|
||||
"staging_path": str(staging_path),
|
||||
"remote_path": remote,
|
||||
})
|
||||
# SP11: audit trail for each successful clearhouse submission.
|
||||
with db.SessionLocal()() as audit_s:
|
||||
append_event(audit_s, AuditEvent(
|
||||
event_type="clearhouse.submitted",
|
||||
entity_type="claim",
|
||||
entity_id=cid,
|
||||
payload={
|
||||
"filename": filename,
|
||||
"remote_path": remote,
|
||||
"tpid": ch.tpid,
|
||||
"stub": ch.sftp_block.stub,
|
||||
},
|
||||
actor="clearhouse-submit",
|
||||
))
|
||||
audit_s.commit()
|
||||
return {"ok": True, "submitted": results, "stub": ch.sftp_block.stub}
|
||||
|
||||
|
||||
def _serialize_claim_for_submit(claim_id: str) -> str:
|
||||
"""Serialize a claim to X12 for SFTP submission."""
|
||||
with db.SessionLocal()() as s:
|
||||
row = s.get(db.Claim, claim_id)
|
||||
if row is None:
|
||||
raise ValueError(f"claim {claim_id!r} not found")
|
||||
raw = row.raw_json or {}
|
||||
return _serialize_claim_from_raw(row, raw)
|
||||
|
||||
|
||||
def _serialize_claim_from_raw(claim_row, raw: dict) -> str:
|
||||
"""Best-effort serializer that uses the stored raw_json to emit a fresh 837.
|
||||
|
||||
For SP9 this delegates to the existing serialize_837 helper if the
|
||||
claim has a complete raw_segments array. Otherwise it raises.
|
||||
"""
|
||||
# SP9 stub: if the claim has a `raw_json` with `x12_text`, use that.
|
||||
if isinstance(raw, dict) and raw.get("x12_text"):
|
||||
result = parse(raw["x12_text"])
|
||||
if result.claims:
|
||||
return serialize_837(result.claims[0])
|
||||
raise RuntimeError(
|
||||
f"claim {claim_row.id!r} cannot be re-serialized: no stored x12_text"
|
||||
)
|
||||
@@ -0,0 +1,56 @@
|
||||
"""``/api/config/*`` — configured-provider / configured-payer lookups.
|
||||
|
||||
The ``config`` prefix is the operator-facing read-only window into the
|
||||
``config/payers.yaml`` static config and the seeded providers table.
|
||||
Used by the frontend Settings page to show what's available.
|
||||
|
||||
The payers endpoint at ``/api/config/payers/{payer_id}/configs`` is
|
||||
slightly different: it returns both the YAML-loaded config and any
|
||||
DB-overridden config (so a runtime override wins over the static
|
||||
fallback). 404 / 200 / array shape match the rest of the API.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
|
||||
from fastapi import APIRouter, HTTPException, Query
|
||||
|
||||
from cyclone import payers as payer_loader
|
||||
from cyclone.store import store
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
@router.get("/api/config/providers")
|
||||
def list_configured_providers(is_active: bool | None = Query(default=True)):
|
||||
"""List the configured provider rows (3 NPIs for SP9)."""
|
||||
return [json.loads(p.model_dump_json()) for p in store.list_providers(is_active=is_active)]
|
||||
|
||||
|
||||
@router.get("/api/config/providers/{npi}")
|
||||
def get_configured_provider(npi: str):
|
||||
p = store.get_provider(npi)
|
||||
if p is None:
|
||||
raise HTTPException(status_code=404, detail=f"provider {npi!r} not found")
|
||||
return json.loads(p.model_dump_json())
|
||||
|
||||
|
||||
@router.get("/api/config/payers")
|
||||
def list_configured_payers(is_active: bool | None = Query(default=True)):
|
||||
return [json.loads(p.model_dump_json()) for p in store.list_payers(is_active=is_active)]
|
||||
|
||||
|
||||
@router.get("/api/config/payers/{payer_id}/configs")
|
||||
def list_payer_configs(payer_id: str):
|
||||
"""List all (transaction_type, config_json) blocks for a payer."""
|
||||
configs = [
|
||||
{"transaction_type": tx, "config_json": block, "source": "yaml"}
|
||||
for (pid, tx), block in payer_loader.all_configs().items()
|
||||
if pid == payer_id
|
||||
]
|
||||
# Also check the DB for runtime-overridden configs
|
||||
for tx in ("837P", "835", "277CA", "999", "TA1"):
|
||||
live = store.get_payer_config(payer_id, tx)
|
||||
if live is not None:
|
||||
configs.append({"transaction_type": tx, "config_json": live, "source": "db"})
|
||||
return configs
|
||||
@@ -0,0 +1,49 @@
|
||||
"""``GET /api/providers`` — distinct providers observed across claims.
|
||||
|
||||
The list is built by ``store.distinct_providers()`` which scans claim
|
||||
rows and returns one entry per unique ``billing_provider_npi``. The
|
||||
``npi`` and ``state`` query params filter client-side. ``limit`` /
|
||||
``offset`` paginate. Accepts ``application/x-ndjson`` like the other
|
||||
list endpoints — see ``api_helpers.ndjson_stream_list``.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from fastapi import APIRouter, Query, Request
|
||||
from fastapi.responses import StreamingResponse
|
||||
|
||||
from cyclone.api_helpers import ndjson_stream_list, wants_ndjson
|
||||
from cyclone.store import store
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
@router.get("/api/providers")
|
||||
def list_providers(
|
||||
request: Request,
|
||||
npi: str | None = Query(None),
|
||||
state: str | None = Query(None),
|
||||
limit: int = Query(100, ge=1, le=1000),
|
||||
offset: int = Query(0, ge=0),
|
||||
) -> Any:
|
||||
items = store.distinct_providers()
|
||||
if npi is not None:
|
||||
items = [p for p in items if p["npi"] == npi]
|
||||
if state is not None:
|
||||
items = [p for p in items if p.get("state") == state]
|
||||
paged = items[offset:offset + limit]
|
||||
total = len(items)
|
||||
returned = len(paged)
|
||||
has_more = total > offset + returned
|
||||
if wants_ndjson(request):
|
||||
return StreamingResponse(
|
||||
ndjson_stream_list(paged, total, returned, has_more),
|
||||
media_type="application/x-ndjson",
|
||||
)
|
||||
return {
|
||||
"items": paged,
|
||||
"total": total,
|
||||
"returned": returned,
|
||||
"has_more": has_more,
|
||||
}
|
||||
@@ -0,0 +1,130 @@
|
||||
"""``/api/remittances`` — list, live-tail, detail endpoints for 835 ERAs.
|
||||
|
||||
Mirror of the ``claims`` router for the 835 / Remittance resource:
|
||||
|
||||
* ``GET /api/remittances`` — paginated list with filters
|
||||
(batch_id / payer / claim_id / date range / sort).
|
||||
* ``GET /api/remittances/stream`` — NDJSON snapshot + live tail via the
|
||||
``remittance_written`` EventBus kind.
|
||||
* ``GET /api/remittances/{remittance_id}`` — detail with the labeled
|
||||
CAS ``adjustments`` array.
|
||||
|
||||
**Route ordering.** ``/stream`` is declared **before** ``/{remittance_id}``
|
||||
so FastAPI's first-match routing doesn't swallow the literal ``stream``
|
||||
segment as a remittance id. Same order as the prior inline code.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, AsyncIterator
|
||||
|
||||
from fastapi import APIRouter, HTTPException, Query, Request
|
||||
from fastapi.responses import StreamingResponse
|
||||
|
||||
from cyclone.api_helpers import (
|
||||
ndjson_line,
|
||||
ndjson_stream_list,
|
||||
tail_events,
|
||||
wants_ndjson,
|
||||
)
|
||||
from cyclone.pubsub import EventBus
|
||||
from cyclone.store import store
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
@router.get("/api/remittances")
|
||||
def list_remittances_endpoint(
|
||||
request: Request,
|
||||
batch_id: str | None = Query(None),
|
||||
payer: str | None = Query(None),
|
||||
claim_id: 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 remittances 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,
|
||||
payer=payer,
|
||||
claim_id=claim_id,
|
||||
date_from=date_from,
|
||||
date_to=date_to,
|
||||
)
|
||||
items = list(store.iter_remittances(
|
||||
sort=sort, order=order, limit=limit, offset=offset, **common,
|
||||
))
|
||||
total = len(list(store.iter_remittances(**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/remittances/stream")
|
||||
async def remittances_stream_endpoint(
|
||||
request: Request,
|
||||
payer: str | None = Query(None),
|
||||
claim_id: 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 Remittances as NDJSON: snapshot first, then live events.
|
||||
|
||||
Subscribes to ``remittance_written``. Default sort is
|
||||
``-received_date`` (newest-first), matching the list endpoint's
|
||||
most common sort.
|
||||
|
||||
NOTE: registered before ``/api/remittances/{remittance_id}`` so
|
||||
the literal ``stream`` path segment doesn't get matched as a
|
||||
remittance id.
|
||||
"""
|
||||
bus: EventBus = request.app.state.event_bus
|
||||
|
||||
async def gen() -> AsyncIterator[bytes]:
|
||||
rows = store.iter_remittances(
|
||||
payer=payer, claim_id=claim_id,
|
||||
date_from=date_from, date_to=date_to,
|
||||
sort=sort or "-received_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)}})
|
||||
|
||||
async for chunk in tail_events(request, bus, ["remittance_written"]):
|
||||
yield chunk
|
||||
|
||||
return StreamingResponse(gen(), media_type="application/x-ndjson")
|
||||
|
||||
|
||||
@router.get("/api/remittances/{remittance_id}")
|
||||
def get_remittance_endpoint(remittance_id: str) -> dict:
|
||||
"""Return one remittance with its labeled CAS ``adjustments`` array.
|
||||
|
||||
Returns 404 when the remittance is missing — never 500.
|
||||
"""
|
||||
body = store.get_remittance(remittance_id)
|
||||
if body is None:
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail={"error": "Not found", "detail": f"Remittance {remittance_id} not found"},
|
||||
)
|
||||
return body
|
||||
Reference in New Issue
Block a user