docs: surface SP14 (Payer-Rejected UI + acknowledge) + SP15 (key rotation) #3
@@ -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/{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/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/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. |
|
| GET | `/api/inbox/export.csv?lane=<lane>` | Streams CSV of the lane's rows. |
|
||||||
|
|
||||||
## Outbound 837 Serializer
|
## 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 `audit_log` table is the canonical record of every state transition
|
||||||
the system has ever observed — claim lifecycle, reconciliation
|
the system has ever observed — claim lifecycle, reconciliation
|
||||||
decisions, config reloads, SFTP submissions, 277CA rejects. SP11 made
|
decisions, config reloads, SFTP submissions, 277CA rejects, Payer-Rejected
|
||||||
it tamper-evident: every row carries a SHA-256 hash of
|
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.
|
`(prev_hash || row_payload)`, forming a chain back to a genesis row.
|
||||||
Any `INSERT`, `UPDATE`, or `DELETE` that breaks the chain is detectable
|
Any `INSERT`, `UPDATE`, or `DELETE` that breaks the chain is detectable
|
||||||
in a single walk.
|
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)
|
for the one-time setup recipe and the HIPAA Security Rule §164.312(a)(2)(iv)
|
||||||
mapping.
|
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)
|
## SFTP Wire-Up (paramiko)
|
||||||
|
|
||||||
The `clearhouse.submit` endpoint uses `paramiko` to push a batch of
|
The `clearhouse.submit` endpoint uses `paramiko` to push a batch of
|
||||||
@@ -382,6 +418,58 @@ 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
|
the `paramiko` extras aren't installed so the test suite stays green on
|
||||||
Linux dev boxes.
|
Linux dev boxes.
|
||||||
|
|
||||||
|
## Inbound MFT Scheduler (SP16)
|
||||||
|
|
||||||
|
The dzinesco SFTP server accepts inbound files (835 remits, 999/TA1
|
||||||
|
acks, 271 eligibility responses, 277/277CA claim acknowledgments) that
|
||||||
|
need to be pulled on a cadence, parsed, and persisted. SP16 ships the
|
||||||
|
**operator-facing control surface** for the polling loop, even though
|
||||||
|
the actual background scheduler is still in flight:
|
||||||
|
|
||||||
|
| Method | Path | Notes |
|
||||||
|
| ------ | --------------------------------------------- | ---------------------------------------------------------------- |
|
||||||
|
| POST | `/api/admin/scheduler/start` | Start the background MFT polling loop (idempotent). |
|
||||||
|
| POST | `/api/admin/scheduler/stop` | Stop the background loop. Waits for the current tick to finish. |
|
||||||
|
| GET | `/api/admin/scheduler/status` | Running state + last tick summary. |
|
||||||
|
| POST | `/api/admin/scheduler/tick` | Force a single poll cycle. Coalesces with an in-flight tick. |
|
||||||
|
| GET | `/api/admin/scheduler/processed-files` | Recently processed inbound files. `?status=ok|error|skipped|pending`, `?limit=` (1–500, default 50). |
|
||||||
|
|
||||||
|
The endpoints read from a `cyclone.scheduler` module and a
|
||||||
|
`cyclone.db.ProcessedInboundFile` ORM model that are being landed
|
||||||
|
alongside SP16 — the admin router's import surface is in place, the
|
||||||
|
implementation is in progress. The control surface is forward-compatible
|
||||||
|
with the eventual scheduler: starting / stopping / ticking / listing
|
||||||
|
processed files will all work as soon as the loop lands.
|
||||||
|
|
||||||
|
The `paramiko` SFTP client from [SP13](#sftp-wire-up-paramiko) is the
|
||||||
|
underlying transport; the scheduler reuses it to list + download files
|
||||||
|
under the inbound SFTP path, then dispatches each downloaded file to
|
||||||
|
the matching parser based on the `<FileType>` token in the filename
|
||||||
|
(see [SP13 file-naming template](#sftp-wire-up-paramiko)).
|
||||||
|
|
||||||
|
## 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
|
## Persistence
|
||||||
|
|
||||||
Parsed batches, claims, remittances, matches, 277CA rejections,
|
Parsed batches, claims, remittances, matches, 277CA rejections,
|
||||||
@@ -415,8 +503,31 @@ backup API).
|
|||||||
.
|
.
|
||||||
├── backend/
|
├── backend/
|
||||||
│ ├── src/cyclone/
|
│ ├── src/cyclone/
|
||||||
│ │ ├── api.py # FastAPI app, GET + parse routes, /api/{resource}/stream
|
│ │ ├── api.py # FastAPI app + middleware; mounts api_routers/* sub-apps
|
||||||
│ │ ├── api_helpers.py # NDJSON / content-negotiation / live-tail helpers
|
│ │ ├── 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}
|
||||||
|
│ │ │ ├── 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
|
||||||
|
│ │ │ ├── inbox.py # GET /api/inbox/lanes, POST /api/inbox/* (match/dismiss/acknowledge/resubmit),
|
||||||
|
│ │ │ │ # GET /api/inbox/export.csv
|
||||||
|
│ │ │ ├── reconciliation.py # GET /api/reconciliation/unmatched, /api/batch-diff,
|
||||||
|
│ │ │ │ # POST /api/reconciliation/match, /api/reconciliation/unmatch
|
||||||
|
│ │ │ ├── eligibility.py # POST /api/eligibility/request, /api/eligibility/parse-271
|
||||||
|
│ │ │ ├── clearhouse.py # GET /api/clearhouse, POST /api/clearhouse/submit
|
||||||
|
│ │ │ ├── config.py # /api/config/providers[/...], /api/config/payers[/...], /api/admin/reload-config
|
||||||
|
│ │ │ ├── parse_835.py # POST /api/parse-835
|
||||||
|
│ │ │ ├── parse_837.py # POST /api/parse-837
|
||||||
|
│ │ │ ├── parse_999.py # POST /api/parse-999
|
||||||
|
│ │ │ ├── parse_ta1.py # POST /api/parse-ta1
|
||||||
|
│ │ │ └── admin.py # /api/admin/audit-log[/verify], /api/admin/db/rotate-key,
|
||||||
|
│ │ │ # /api/admin/scheduler/* (SP16)
|
||||||
│ │ ├── pubsub.py # in-process EventBus (drop-oldest, per-kind fan-out)
|
│ │ ├── pubsub.py # in-process EventBus (drop-oldest, per-kind fan-out)
|
||||||
│ │ ├── store.py # CycloneStore, mappers, publish-on-write
|
│ │ ├── store.py # CycloneStore, mappers, publish-on-write
|
||||||
│ │ ├── db.py # SQLAlchemy engine, session factory, ORM models
|
│ │ ├── db.py # SQLAlchemy engine, session factory, ORM models
|
||||||
@@ -473,7 +584,7 @@ backup API).
|
|||||||
|
|
||||||
## Roadmap
|
## 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
|
review](docs/reviews/2026-06-20-cyclone-completeness-review.md) for
|
||||||
the honest gap analysis against the industry definition of a HIPAA
|
the honest gap analysis against the industry definition of a HIPAA
|
||||||
clearinghouse — the short version is that the local-only,
|
clearinghouse — the short version is that the local-only,
|
||||||
@@ -484,6 +595,19 @@ scope.
|
|||||||
|
|
||||||
Shipped sub-projects (most recent first):
|
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
|
- **Sub-project 13 (shipped) — SFTP wire-up.** `paramiko`-backed
|
||||||
`SftpClient` replaces the SP9 stub. The clearhouse.submit endpoint
|
`SftpClient` replaces the SP9 stub. The clearhouse.submit endpoint
|
||||||
actually pushes to
|
actually pushes to
|
||||||
@@ -752,6 +876,34 @@ the one-time setup recipe.
|
|||||||
`mft.gainwelltechnologies.com:/CO XIX/PROD/coxix_prod_11525703/FromHPE`.
|
`mft.gainwelltechnologies.com:/CO XIX/PROD/coxix_prod_11525703/FromHPE`.
|
||||||
SFTP credentials are fetched from the macOS Keychain at call time.
|
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
|
## License
|
||||||
|
|
||||||
No license file yet; this is internal-use software. Add a `LICENSE` file
|
No license file yet; this is internal-use software. Add a `LICENSE` file
|
||||||
|
|||||||
+62
-1587
File diff suppressed because it is too large
Load Diff
@@ -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,332 @@
|
|||||||
|
"""``/api/admin/*`` — operator-facing admin surface (SP11 + SP15).
|
||||||
|
|
||||||
|
Four endpoints:
|
||||||
|
|
||||||
|
* ``GET /api/admin/audit-log`` — list audit-log rows with optional
|
||||||
|
``entity_type`` / ``entity_id`` / ``event_type`` filters (SP11).
|
||||||
|
* ``GET /api/admin/audit-log/verify`` — walk the audit-log hash chain
|
||||||
|
and verify every row (SP11).
|
||||||
|
* ``POST /api/admin/db/rotate-key`` — SP15 SQLCipher key rotation.
|
||||||
|
Re-encrypts the DB in place with a fresh key, updates the Keychain.
|
||||||
|
Protected by a module-level ``threading.Lock`` so two concurrent
|
||||||
|
rotations can't race (a single process restart resets the lock —
|
||||||
|
intentional, the next start-up opens with whatever key is in the
|
||||||
|
Keychain).
|
||||||
|
* ``POST /api/admin/reload-config`` — re-read ``config/payers.yaml``
|
||||||
|
and revalidate; returns the number of configs loaded.
|
||||||
|
"""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import json
|
||||||
|
import logging
|
||||||
|
import threading
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
from fastapi import APIRouter, HTTPException, Query
|
||||||
|
|
||||||
|
from cyclone import db, db_crypto, secrets
|
||||||
|
from cyclone.audit_log import AuditEvent, append_event, verify_chain
|
||||||
|
|
||||||
|
log = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
router = APIRouter()
|
||||||
|
|
||||||
|
# Module-level lock: protects against concurrent key rotations within
|
||||||
|
# the same process. A process restart resets it (intentional — the
|
||||||
|
# operator's next start-up opens with whatever key is in the Keychain).
|
||||||
|
_db_rotate_lock = threading.Lock()
|
||||||
|
|
||||||
|
|
||||||
|
@router.get("/api/admin/audit-log")
|
||||||
|
def list_audit_log_endpoint(
|
||||||
|
entity_type: str | None = Query(default=None),
|
||||||
|
entity_id: str | None = Query(default=None),
|
||||||
|
event_type: str | None = Query(default=None),
|
||||||
|
limit: int = Query(default=100, ge=1, le=1000),
|
||||||
|
) -> Any:
|
||||||
|
"""List audit-log rows, newest first, with optional filters.
|
||||||
|
|
||||||
|
Filters match the (entity_type, entity_id) pair (typical use:
|
||||||
|
"show me everything that happened to claim C-123") or a single
|
||||||
|
event_type (typical use: "show me all clearhouse.submitted
|
||||||
|
events today").
|
||||||
|
"""
|
||||||
|
with db.SessionLocal()() as s:
|
||||||
|
q = s.query(db.AuditLog)
|
||||||
|
if entity_type:
|
||||||
|
q = q.filter(db.AuditLog.entity_type == entity_type)
|
||||||
|
if entity_id:
|
||||||
|
q = q.filter(db.AuditLog.entity_id == entity_id)
|
||||||
|
if event_type:
|
||||||
|
q = q.filter(db.AuditLog.event_type == event_type)
|
||||||
|
rows = q.order_by(db.AuditLog.id.desc()).limit(limit).all()
|
||||||
|
return {
|
||||||
|
"total": len(rows),
|
||||||
|
"items": [
|
||||||
|
{
|
||||||
|
"id": r.id,
|
||||||
|
"event_type": r.event_type,
|
||||||
|
"entity_type": r.entity_type,
|
||||||
|
"entity_id": r.entity_id,
|
||||||
|
"actor": r.actor,
|
||||||
|
"payload": json.loads(r.payload_json) if r.payload_json else None,
|
||||||
|
"created_at": r.created_at.isoformat() if r.created_at else None,
|
||||||
|
"prev_hash": r.prev_hash,
|
||||||
|
"hash": r.hash,
|
||||||
|
}
|
||||||
|
for r in rows
|
||||||
|
],
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
@router.get("/api/admin/audit-log/verify")
|
||||||
|
def verify_audit_log_endpoint() -> Any:
|
||||||
|
"""Walk the audit-log chain and verify every row's hash.
|
||||||
|
|
||||||
|
Returns ``{"ok": true, "checked": N}`` for a clean chain, or
|
||||||
|
``{"ok": false, "checked": K, "first_bad_id": X, "reason": "..."}``
|
||||||
|
for a broken chain. This is the operator's "did anyone tamper?"
|
||||||
|
endpoint; run it on demand or via a nightly cron job.
|
||||||
|
"""
|
||||||
|
with db.SessionLocal()() as s:
|
||||||
|
result = verify_chain(s)
|
||||||
|
return {
|
||||||
|
"ok": result.ok,
|
||||||
|
"checked": result.checked,
|
||||||
|
"first_bad_id": result.first_bad_id,
|
||||||
|
"reason": result.reason,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
@router.post("/api/admin/db/rotate-key")
|
||||||
|
def rotate_db_key_endpoint(body: dict | None = None) -> Any:
|
||||||
|
"""Generate a fresh DB key, re-encrypt the DB, update the Keychain.
|
||||||
|
|
||||||
|
Request body (optional):
|
||||||
|
actor: who initiated the rotation. Defaults to "operator".
|
||||||
|
reason: human-readable reason. Written to the audit log.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
``{ok, old_fingerprint, new_fingerprint, rotated_at, table_count}``
|
||||||
|
on success. On failure (DB not encrypted, rekey failed,
|
||||||
|
Keychain update failed) returns the same shape with
|
||||||
|
``ok=false`` and a ``reason``. HTTP 503 is returned if the
|
||||||
|
rekey fails or encryption is not enabled.
|
||||||
|
|
||||||
|
The Keychain write happens *after* the rekey succeeds. If the
|
||||||
|
Keychain write fails, the DB has the new key but the Keychain
|
||||||
|
still has the old one — the endpoint returns 503 with a
|
||||||
|
"keychain update failed" reason and the operator must restore
|
||||||
|
the old key manually (``cyclone db restore-key <old_key>``) to
|
||||||
|
avoid being locked out.
|
||||||
|
"""
|
||||||
|
body = body or {}
|
||||||
|
actor = body.get("actor") or "operator"
|
||||||
|
reason = body.get("reason") or ""
|
||||||
|
|
||||||
|
if not db_crypto.is_encryption_enabled():
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=400,
|
||||||
|
detail="encryption not enabled (sqlcipher3 missing or no Keychain key)",
|
||||||
|
)
|
||||||
|
|
||||||
|
# Acquire the lock; non-blocking so a stuck rotation doesn't
|
||||||
|
# silently hold up other requests.
|
||||||
|
if not _db_rotate_lock.acquire(blocking=False):
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=409,
|
||||||
|
detail="another key rotation is in progress",
|
||||||
|
)
|
||||||
|
try:
|
||||||
|
url = db._resolve_url()
|
||||||
|
old_key = db_crypto.get_db_key()
|
||||||
|
if not old_key:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=400,
|
||||||
|
detail="no DB key in Keychain; cannot rotate",
|
||||||
|
)
|
||||||
|
|
||||||
|
new_key = db_crypto.generate_db_key()
|
||||||
|
result = db_crypto.rotate_db_key(
|
||||||
|
url=url, old_key=old_key, new_key=new_key,
|
||||||
|
)
|
||||||
|
if not result.ok:
|
||||||
|
# Rekey failed. The DB still has the old key. The
|
||||||
|
# Keychain is unchanged. Caller should NOT retry with
|
||||||
|
# the same new key (it's lost); generate a fresh one.
|
||||||
|
log.error("SQLCipher rotate failed: %s", result.reason)
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=503,
|
||||||
|
detail={
|
||||||
|
"ok": False,
|
||||||
|
"old_fingerprint": result.old_fingerprint,
|
||||||
|
"new_fingerprint": result.new_fingerprint,
|
||||||
|
"rotated_at": result.rotated_at,
|
||||||
|
"reason": result.reason,
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
# Rekey succeeded. Now update the Keychain. If this fails
|
||||||
|
# the DB is locked behind the new key — operator must
|
||||||
|
# restore the old key manually.
|
||||||
|
if not secrets.set_secret(db_crypto.KEYCHAIN_ACCOUNT, new_key):
|
||||||
|
log.error("Keychain update failed after successful rekey!")
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=503,
|
||||||
|
detail={
|
||||||
|
"ok": False,
|
||||||
|
"old_fingerprint": result.old_fingerprint,
|
||||||
|
"new_fingerprint": result.new_fingerprint,
|
||||||
|
"rotated_at": result.rotated_at,
|
||||||
|
"reason": (
|
||||||
|
"rekey succeeded but Keychain update failed — "
|
||||||
|
"the DB is now encrypted with the new key but "
|
||||||
|
"the Keychain still has the old one. "
|
||||||
|
"Restore the old key to the Keychain to recover."
|
||||||
|
),
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
# Store the old key in the "previous" account for a grace
|
||||||
|
# period so the operator can roll back if they discover the
|
||||||
|
# new key is broken (e.g. the Keychain entry got truncated).
|
||||||
|
secrets.set_secret(db_crypto.KEYCHAIN_ACCOUNT_PREVIOUS, old_key)
|
||||||
|
|
||||||
|
# Rebuild the engine so subsequent connections use the new
|
||||||
|
# key. dispose_engine() closes every pooled connection that
|
||||||
|
# was using the old key; init_db() opens new ones with the
|
||||||
|
# new key from the (now-updated) Keychain.
|
||||||
|
db.reinit_engine()
|
||||||
|
|
||||||
|
# Audit log the rotation. We do this after the engine is
|
||||||
|
# rebuilt so the audit event is written with the new key —
|
||||||
|
# proving that the new key works for new writes.
|
||||||
|
try:
|
||||||
|
with db.SessionLocal()() as s:
|
||||||
|
append_event(s, AuditEvent(
|
||||||
|
event_type="db.key_rotated",
|
||||||
|
entity_type="database",
|
||||||
|
entity_id="cyclone.db",
|
||||||
|
actor=actor,
|
||||||
|
payload={
|
||||||
|
"old_fingerprint": result.old_fingerprint,
|
||||||
|
"new_fingerprint": result.new_fingerprint,
|
||||||
|
"table_count": result.table_count,
|
||||||
|
"reason": reason,
|
||||||
|
},
|
||||||
|
))
|
||||||
|
s.commit()
|
||||||
|
except Exception as exc: # noqa: BLE001
|
||||||
|
# Audit append is best-effort; rotation already succeeded.
|
||||||
|
log.warning("could not write audit event for rotation: %s", exc)
|
||||||
|
|
||||||
|
return {
|
||||||
|
"ok": True,
|
||||||
|
"old_fingerprint": result.old_fingerprint,
|
||||||
|
"new_fingerprint": result.new_fingerprint,
|
||||||
|
"rotated_at": result.rotated_at,
|
||||||
|
"table_count": result.table_count,
|
||||||
|
}
|
||||||
|
finally:
|
||||||
|
_db_rotate_lock.release()
|
||||||
|
|
||||||
|
|
||||||
|
@router.post("/api/admin/reload-config")
|
||||||
|
def reload_config():
|
||||||
|
"""Re-read ``config/payers.yaml`` and revalidate. Returns counts."""
|
||||||
|
from cyclone import payers as payer_loader
|
||||||
|
try:
|
||||||
|
configs = payer_loader.load_payer_configs()
|
||||||
|
except ValueError as e:
|
||||||
|
raise HTTPException(status_code=400, detail=str(e))
|
||||||
|
return {"ok": True, "loaded": len(configs), "errors": []}
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# SP16: inbound MFT scheduler
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
@router.post("/api/admin/scheduler/start")
|
||||||
|
async def scheduler_start():
|
||||||
|
"""Start the background MFT polling loop (idempotent)."""
|
||||||
|
from cyclone import scheduler as sched_mod
|
||||||
|
sched = sched_mod.get_scheduler()
|
||||||
|
await sched.start()
|
||||||
|
return {"ok": True, "status": sched.status().as_dict()}
|
||||||
|
|
||||||
|
|
||||||
|
@router.post("/api/admin/scheduler/stop")
|
||||||
|
async def scheduler_stop():
|
||||||
|
"""Stop the background MFT polling loop. Waits for the current tick."""
|
||||||
|
from cyclone import scheduler as sched_mod
|
||||||
|
sched = sched_mod.get_scheduler()
|
||||||
|
await sched.stop()
|
||||||
|
return {"ok": True, "status": sched.status().as_dict()}
|
||||||
|
|
||||||
|
|
||||||
|
@router.get("/api/admin/scheduler/status")
|
||||||
|
def scheduler_status():
|
||||||
|
"""Return the scheduler's running state + last tick summary."""
|
||||||
|
from cyclone import scheduler as sched_mod
|
||||||
|
sched = sched_mod.get_scheduler()
|
||||||
|
return sched.status().as_dict()
|
||||||
|
|
||||||
|
|
||||||
|
@router.post("/api/admin/scheduler/tick")
|
||||||
|
async def scheduler_tick():
|
||||||
|
"""Force a single poll cycle. Returns the TickResult.
|
||||||
|
|
||||||
|
Useful for the operator's "run now" button or for testing. If a
|
||||||
|
scheduled tick is already in flight, this call coalesces onto it
|
||||||
|
(waits for it to finish, returns its result).
|
||||||
|
"""
|
||||||
|
from cyclone import scheduler as sched_mod
|
||||||
|
sched = sched_mod.get_scheduler()
|
||||||
|
result = await sched.tick()
|
||||||
|
return {"ok": True, "tick": result.as_dict()}
|
||||||
|
|
||||||
|
|
||||||
|
@router.get("/api/admin/scheduler/processed-files")
|
||||||
|
def scheduler_processed_files(
|
||||||
|
limit: int = 50,
|
||||||
|
status: str | None = None,
|
||||||
|
):
|
||||||
|
"""List recently-processed inbound files (newest first).
|
||||||
|
|
||||||
|
The operator uses this to confirm the scheduler is keeping up
|
||||||
|
with MFT traffic. ``status`` filters to one of: ok, error,
|
||||||
|
skipped, pending.
|
||||||
|
"""
|
||||||
|
from sqlalchemy import desc
|
||||||
|
from cyclone import db
|
||||||
|
from cyclone.db import ProcessedInboundFile
|
||||||
|
with db.SessionLocal()() as session:
|
||||||
|
q = session.query(ProcessedInboundFile).order_by(
|
||||||
|
desc(ProcessedInboundFile.processed_at),
|
||||||
|
)
|
||||||
|
if status:
|
||||||
|
q = q.filter(ProcessedInboundFile.status == status)
|
||||||
|
q = q.limit(min(limit, 500))
|
||||||
|
rows = q.all()
|
||||||
|
return {
|
||||||
|
"count": len(rows),
|
||||||
|
"files": [
|
||||||
|
{
|
||||||
|
"id": r.id,
|
||||||
|
"name": r.name,
|
||||||
|
"size": r.size,
|
||||||
|
"file_type": r.file_type,
|
||||||
|
"processed_at": r.processed_at.isoformat() if r.processed_at else None,
|
||||||
|
"parser_used": r.parser_used,
|
||||||
|
"claim_count": r.claim_count,
|
||||||
|
"status": r.status,
|
||||||
|
"error_message": (
|
||||||
|
r.error_message[:200] + "..."
|
||||||
|
if r.error_message and len(r.error_message) > 200
|
||||||
|
else r.error_message
|
||||||
|
),
|
||||||
|
}
|
||||||
|
for r in rows
|
||||||
|
],
|
||||||
|
}
|
||||||
@@ -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,220 @@
|
|||||||
|
"""``/api/eligibility/*`` — 270/271 eligibility surface (SP3 P4 T23–T24).
|
||||||
|
|
||||||
|
API-only — no DB persistence per spec section 3.4. The operator builds
|
||||||
|
a 270 inquiry from a small JSON body, downloads the raw text to paste
|
||||||
|
into a payer portal, then re-pastes the 271 response and parses it
|
||||||
|
back into structured fields on demand.
|
||||||
|
|
||||||
|
Two endpoints:
|
||||||
|
* ``POST /api/eligibility/request`` — build a 270 inquiry from a JSON
|
||||||
|
body and return ``{raw_270_text, parsed}``.
|
||||||
|
* ``POST /api/eligibility/parse-271`` — accept a raw 271 file upload
|
||||||
|
and return the parsed ``coverage_benefits`` / ``subscriber`` /
|
||||||
|
``summary`` / envelope triples.
|
||||||
|
|
||||||
|
The ``_validate_eligibility_request`` helper is private to this router
|
||||||
|
— it builds the Pydantic ``ParseResult270`` and is only used by the
|
||||||
|
``/request`` endpoint.
|
||||||
|
"""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import json
|
||||||
|
import logging
|
||||||
|
from datetime import date as _date
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
from fastapi import APIRouter, File, HTTPException, UploadFile
|
||||||
|
from fastapi.responses import JSONResponse
|
||||||
|
|
||||||
|
from cyclone.parsers.exceptions import CycloneParseError
|
||||||
|
from cyclone.parsers.models import BatchSummary, Envelope
|
||||||
|
from cyclone.parsers.models_270 import (
|
||||||
|
EligibilityBenefitInquiry,
|
||||||
|
InformationReceiver270,
|
||||||
|
InformationSource270,
|
||||||
|
ParseResult270,
|
||||||
|
Subscriber270,
|
||||||
|
)
|
||||||
|
from cyclone.parsers.parse_271 import parse as parse_271_text
|
||||||
|
from cyclone.parsers.serialize_270 import serialize_270
|
||||||
|
|
||||||
|
log = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
router = APIRouter()
|
||||||
|
|
||||||
|
|
||||||
|
def _validate_eligibility_request(body: dict) -> tuple[ParseResult270, str]:
|
||||||
|
"""Build a :class:`ParseResult270` from a request body dict.
|
||||||
|
|
||||||
|
The body shape is the minimum surface needed to build a valid 270
|
||||||
|
inquiry (per spec section 3.4 — operator-driven, ephemeral):
|
||||||
|
|
||||||
|
::
|
||||||
|
|
||||||
|
{
|
||||||
|
"subscriber": {first_name, last_name, member_id, dob},
|
||||||
|
"provider": {npi, name},
|
||||||
|
"payer": {id, name},
|
||||||
|
"service_type_code": "1"
|
||||||
|
}
|
||||||
|
|
||||||
|
Returns ``(ParseResult270, service_type_code)``. Raises
|
||||||
|
:class:`HTTPException` (400) when the body is missing required
|
||||||
|
fields.
|
||||||
|
"""
|
||||||
|
subscriber_in = body.get("subscriber") or {}
|
||||||
|
provider_in = body.get("provider") or {}
|
||||||
|
payer_in = body.get("payer") or {}
|
||||||
|
service_type_code = (body.get("service_type_code") or "").strip()
|
||||||
|
|
||||||
|
# Required-field checks. We surface a single 400 with the first
|
||||||
|
# missing field name to match the rest of the API's error contract.
|
||||||
|
if not service_type_code:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=400,
|
||||||
|
detail={"error": "Bad request", "detail": "service_type_code is required"},
|
||||||
|
)
|
||||||
|
if not subscriber_in.get("member_id"):
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=400,
|
||||||
|
detail={"error": "Bad request", "detail": "subscriber.member_id is required"},
|
||||||
|
)
|
||||||
|
if not provider_in.get("npi"):
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=400,
|
||||||
|
detail={"error": "Bad request", "detail": "provider.npi is required"},
|
||||||
|
)
|
||||||
|
if not payer_in.get("name"):
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=400,
|
||||||
|
detail={"error": "Bad request", "detail": "payer.name is required"},
|
||||||
|
)
|
||||||
|
|
||||||
|
subscriber_dob_raw = subscriber_in.get("dob")
|
||||||
|
subscriber_dob: _date | None = None
|
||||||
|
if subscriber_dob_raw:
|
||||||
|
try:
|
||||||
|
subscriber_dob = _date.fromisoformat(subscriber_dob_raw)
|
||||||
|
except (TypeError, ValueError) as exc:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=400,
|
||||||
|
detail={
|
||||||
|
"error": "Bad request",
|
||||||
|
"detail": f"subscriber.dob must be YYYY-MM-DD: {exc}",
|
||||||
|
},
|
||||||
|
) from exc
|
||||||
|
|
||||||
|
result = ParseResult270(
|
||||||
|
envelope=Envelope(
|
||||||
|
sender_id="SUBMITTERID",
|
||||||
|
receiver_id=str(payer_in.get("id") or "RECEIVERID"),
|
||||||
|
control_number="000000001",
|
||||||
|
transaction_date=_date.today(),
|
||||||
|
implementation_guide="005010X279A1",
|
||||||
|
),
|
||||||
|
information_source=InformationSource270(
|
||||||
|
name=str(payer_in["name"]),
|
||||||
|
id=str(payer_in.get("id") or "") or None,
|
||||||
|
),
|
||||||
|
information_receiver=InformationReceiver270(
|
||||||
|
name=str(provider_in.get("name") or ""),
|
||||||
|
npi=str(provider_in["npi"]),
|
||||||
|
),
|
||||||
|
subscriber=Subscriber270(
|
||||||
|
member_id=str(subscriber_in["member_id"]),
|
||||||
|
first_name=str(subscriber_in.get("first_name") or "") or None,
|
||||||
|
last_name=str(subscriber_in.get("last_name") or "") or None,
|
||||||
|
dob=subscriber_dob,
|
||||||
|
),
|
||||||
|
inquiries=[EligibilityBenefitInquiry(service_type_code=service_type_code)],
|
||||||
|
summary=BatchSummary(
|
||||||
|
input_file="eligibility_request",
|
||||||
|
control_number="000000001",
|
||||||
|
transaction_date=_date.today(),
|
||||||
|
total_claims=1,
|
||||||
|
passed=1,
|
||||||
|
failed=0,
|
||||||
|
),
|
||||||
|
)
|
||||||
|
return result, service_type_code
|
||||||
|
|
||||||
|
|
||||||
|
@router.post("/api/eligibility/request")
|
||||||
|
def post_eligibility_request(body: dict) -> Any:
|
||||||
|
"""Build a 270 eligibility inquiry from a small JSON body.
|
||||||
|
|
||||||
|
Returns ``{"raw_270_text": <X12>, "parsed": <ParseResult270>}``
|
||||||
|
so the operator can either download the raw text (paste into a
|
||||||
|
payer portal) or render the parsed fields directly. Per spec
|
||||||
|
section 3.4, nothing is persisted to the DB.
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
result, _ = _validate_eligibility_request(body)
|
||||||
|
except HTTPException:
|
||||||
|
raise
|
||||||
|
except (KeyError, TypeError, ValueError) as exc:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=400,
|
||||||
|
detail={"error": "Bad request", "detail": f"Malformed body: {exc}"},
|
||||||
|
) from exc
|
||||||
|
|
||||||
|
raw_270_text = serialize_270(result)
|
||||||
|
return {
|
||||||
|
"raw_270_text": raw_270_text,
|
||||||
|
"parsed": json.loads(result.model_dump_json()),
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
@router.post("/api/eligibility/parse-271")
|
||||||
|
async def post_eligibility_parse_271(
|
||||||
|
file: UploadFile = File(...),
|
||||||
|
) -> Any:
|
||||||
|
"""Parse a 271 eligibility response and return the structured summary.
|
||||||
|
|
||||||
|
Accepts the raw 271 text as a file upload (multipart/form-data),
|
||||||
|
mirrors the ``/api/parse-999`` contract. Per spec section 3.4 the
|
||||||
|
result is NOT persisted — the operator re-pastes the 271 each
|
||||||
|
time they need a fresh read.
|
||||||
|
|
||||||
|
The response body is a JSON object with three top-level keys:
|
||||||
|
``coverage_benefits``, ``subscriber``, and ``summary``. 400 is
|
||||||
|
returned on empty / undecodable / malformed EDI; 200 on success.
|
||||||
|
"""
|
||||||
|
raw = await file.read()
|
||||||
|
if not raw:
|
||||||
|
return JSONResponse(
|
||||||
|
status_code=400,
|
||||||
|
content={"error": "Empty file", "detail": "Uploaded file contained no bytes."},
|
||||||
|
)
|
||||||
|
try:
|
||||||
|
text = raw.decode("utf-8")
|
||||||
|
except UnicodeDecodeError as exc:
|
||||||
|
return JSONResponse(
|
||||||
|
status_code=400,
|
||||||
|
content={"error": "Encoding error", "detail": str(exc)},
|
||||||
|
)
|
||||||
|
|
||||||
|
try:
|
||||||
|
result = parse_271_text(text, input_file=file.filename or "")
|
||||||
|
except CycloneParseError as exc:
|
||||||
|
return JSONResponse(
|
||||||
|
status_code=400,
|
||||||
|
content={"error": "Parse error", "detail": str(exc)},
|
||||||
|
)
|
||||||
|
except Exception as exc: # pragma: no cover - safety net
|
||||||
|
log.exception("Unexpected parser failure on 271")
|
||||||
|
return JSONResponse(
|
||||||
|
status_code=500,
|
||||||
|
content={"error": "Internal server error", "detail": str(exc)},
|
||||||
|
)
|
||||||
|
|
||||||
|
return {
|
||||||
|
"coverage_benefits": [
|
||||||
|
json.loads(cb.model_dump_json()) for cb in result.coverage_benefits
|
||||||
|
],
|
||||||
|
"subscriber": json.loads(result.subscriber.model_dump_json()),
|
||||||
|
"summary": json.loads(result.summary.model_dump_json()),
|
||||||
|
"envelope": json.loads(result.envelope.model_dump_json()),
|
||||||
|
"information_source": json.loads(result.information_source.model_dump_json()),
|
||||||
|
"information_receiver": json.loads(result.information_receiver.model_dump_json()),
|
||||||
|
}
|
||||||
@@ -0,0 +1,309 @@
|
|||||||
|
"""``/api/inbox/*`` — operator-facing inbox surface.
|
||||||
|
|
||||||
|
Six endpoints powering the Inbox page (lanes, candidates, payer-rejected
|
||||||
|
acknowledge, rejected resubmit, CSV export). All endpoints run inside a
|
||||||
|
``db.SessionLocal()`` scope; the lanes / dismiss endpoints also read
|
||||||
|
``app.state.dismissed_pairs`` (a session-scoped set so dismissed items
|
||||||
|
don't re-appear until process restart).
|
||||||
|
|
||||||
|
Endpoints:
|
||||||
|
* ``GET /api/inbox/lanes`` — return all four lanes in one call
|
||||||
|
(rejected / payer_rejected / candidates / unmatched / done_today).
|
||||||
|
* ``POST /api/inbox/candidates/{remit_id}/match`` — manually link a
|
||||||
|
remit to a claim from the candidates lane.
|
||||||
|
* ``POST /api/inbox/candidates/dismiss`` — add candidate pairs to the
|
||||||
|
dismissed set so they stop appearing in the candidates lane.
|
||||||
|
* ``POST /api/inbox/payer-rejected/acknowledge`` — clear SP14
|
||||||
|
payer-rejected claims from the working surface without deleting the
|
||||||
|
rejection (idempotent: re-acknowledging is a noop).
|
||||||
|
* ``POST /api/inbox/rejected/resubmit`` — bulk move REJECTED claims
|
||||||
|
back to SUBMITTED; with ``?download=true`` returns a ZIP of
|
||||||
|
regenerated 837s with unique interchange control numbers.
|
||||||
|
* ``GET /api/inbox/export.csv?lane=…`` — stream a CSV for a single lane.
|
||||||
|
"""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import csv
|
||||||
|
import io
|
||||||
|
import json
|
||||||
|
import zipfile
|
||||||
|
from datetime import datetime, timezone
|
||||||
|
|
||||||
|
from fastapi import APIRouter, HTTPException, Query, Request, Response
|
||||||
|
from fastapi.responses import StreamingResponse
|
||||||
|
|
||||||
|
from cyclone import db
|
||||||
|
from cyclone.audit_log import AuditEvent, append_event
|
||||||
|
from cyclone.db import Claim, ClaimState, Remittance
|
||||||
|
from cyclone.parsers.models import ClaimOutput
|
||||||
|
from cyclone.parsers.serialize_837 import (
|
||||||
|
SerializeError as SerializeError837,
|
||||||
|
serialize_837_for_resubmit,
|
||||||
|
)
|
||||||
|
|
||||||
|
router = APIRouter()
|
||||||
|
|
||||||
|
|
||||||
|
@router.get("/api/inbox/lanes")
|
||||||
|
def inbox_lanes(request: Request) -> dict:
|
||||||
|
"""Return all Inbox lanes in one call.
|
||||||
|
|
||||||
|
Reads the session-scoped ``dismissed_pairs`` set so the
|
||||||
|
candidates lane excludes pairs the operator already dismissed in
|
||||||
|
this process lifetime.
|
||||||
|
"""
|
||||||
|
dismissed_pairs = getattr(request.app.state, "dismissed_pairs", set())
|
||||||
|
with db.SessionLocal()() as session:
|
||||||
|
from cyclone.inbox_lanes import compute_lanes
|
||||||
|
lanes = compute_lanes(session, dismissed_pairs=dismissed_pairs)
|
||||||
|
return {
|
||||||
|
"rejected": lanes.rejected,
|
||||||
|
# SP10: payer-rejected lane (277CA STC A4/A6/A7). Distinct from
|
||||||
|
# the 999 envelope rejection in ``rejected`` above.
|
||||||
|
"payer_rejected": lanes.payer_rejected,
|
||||||
|
"candidates": lanes.candidates,
|
||||||
|
"unmatched": lanes.unmatched,
|
||||||
|
"done_today": lanes.done_today,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
@router.post("/api/inbox/candidates/{remit_id}/match")
|
||||||
|
def inbox_match_candidate(remit_id: str, body: dict):
|
||||||
|
"""Manually link a remit to a claim."""
|
||||||
|
claim_id = body.get("claim_id")
|
||||||
|
if not claim_id:
|
||||||
|
raise HTTPException(400, "claim_id required")
|
||||||
|
with db.SessionLocal()() as s:
|
||||||
|
claim = s.get(Claim, claim_id)
|
||||||
|
remit = s.get(Remittance, remit_id)
|
||||||
|
if claim is None or remit is None:
|
||||||
|
raise HTTPException(404, "claim or remit not found")
|
||||||
|
if claim.matched_remittance_id and claim.matched_remittance_id != remit_id:
|
||||||
|
raise HTTPException(
|
||||||
|
409,
|
||||||
|
detail={
|
||||||
|
"error": "claim_already_matched",
|
||||||
|
"current_state": (
|
||||||
|
claim.state.value if hasattr(claim.state, "value")
|
||||||
|
else str(claim.state)
|
||||||
|
),
|
||||||
|
"matched_remittance_id": claim.matched_remittance_id,
|
||||||
|
},
|
||||||
|
)
|
||||||
|
claim.matched_remittance_id = remit_id
|
||||||
|
remit.claim_id = claim_id
|
||||||
|
s.commit()
|
||||||
|
return {"ok": True, "claim_id": claim_id, "remit_id": remit_id}
|
||||||
|
|
||||||
|
|
||||||
|
@router.post("/api/inbox/candidates/dismiss")
|
||||||
|
def inbox_dismiss_candidates(request: Request, body: dict):
|
||||||
|
"""Add candidate pairs to the session-scoped dismissed set."""
|
||||||
|
pairs = body.get("pairs") or []
|
||||||
|
if not hasattr(request.app.state, "dismissed_pairs"):
|
||||||
|
request.app.state.dismissed_pairs = set()
|
||||||
|
for p in pairs:
|
||||||
|
cid = p.get("claim_id")
|
||||||
|
rid = p.get("remit_id")
|
||||||
|
if cid and rid:
|
||||||
|
request.app.state.dismissed_pairs.add(frozenset({cid, rid}))
|
||||||
|
return {"ok": True, "dismissed_count": len(pairs)}
|
||||||
|
|
||||||
|
|
||||||
|
@router.post("/api/inbox/payer-rejected/acknowledge")
|
||||||
|
def inbox_acknowledge_payer_rejected(body: dict):
|
||||||
|
"""Mark Payer-Rejected claims as acknowledged by the operator.
|
||||||
|
|
||||||
|
Operator hits "Acknowledge" on the Payer-Rejected Inbox lane to clear
|
||||||
|
the claim from the working surface. We don't delete the rejection
|
||||||
|
(the original payer_rejected_* fields stay for SP11 audit), we just
|
||||||
|
set ``payer_rejected_acknowledged_at`` so the lane query filters it
|
||||||
|
out.
|
||||||
|
|
||||||
|
Idempotent: re-acknowledging an already-acknowledged claim is a noop
|
||||||
|
(the timestamp is not bumped). Returns the count actually transitioned
|
||||||
|
so the UI can show "3 of 5 were already acknowledged".
|
||||||
|
"""
|
||||||
|
claim_ids = body.get("claim_ids") or []
|
||||||
|
actor = body.get("actor") or "operator"
|
||||||
|
if not isinstance(claim_ids, list) or not claim_ids:
|
||||||
|
raise HTTPException(400, "claim_ids must be a non-empty list")
|
||||||
|
if not all(isinstance(c, str) for c in claim_ids):
|
||||||
|
raise HTTPException(400, "claim_ids must be a list of strings")
|
||||||
|
|
||||||
|
with db.SessionLocal()() as session:
|
||||||
|
now = datetime.now(timezone.utc)
|
||||||
|
transitioned = 0
|
||||||
|
already_acked = 0
|
||||||
|
not_found = 0
|
||||||
|
not_rejected = 0
|
||||||
|
for cid in claim_ids:
|
||||||
|
claim = session.get(Claim, cid)
|
||||||
|
if claim is None:
|
||||||
|
not_found += 1
|
||||||
|
continue
|
||||||
|
if claim.payer_rejected_at is None:
|
||||||
|
not_rejected += 1
|
||||||
|
continue
|
||||||
|
if claim.payer_rejected_acknowledged_at is not None:
|
||||||
|
already_acked += 1
|
||||||
|
continue
|
||||||
|
claim.payer_rejected_acknowledged_at = now
|
||||||
|
claim.payer_rejected_acknowledged_actor = actor
|
||||||
|
transitioned += 1
|
||||||
|
# SP11: audit event for the acknowledge action. Best-effort:
|
||||||
|
# an audit-log failure must not block the operator's action.
|
||||||
|
try:
|
||||||
|
append_event(session, AuditEvent(
|
||||||
|
event_type="claim.payer_rejected_acknowledged",
|
||||||
|
entity_type="claim",
|
||||||
|
entity_id=claim.id,
|
||||||
|
actor=actor,
|
||||||
|
payload={
|
||||||
|
"payer_rejected_status_code": claim.payer_rejected_status_code,
|
||||||
|
"payer_rejected_by_277ca_id": claim.payer_rejected_by_277ca_id,
|
||||||
|
},
|
||||||
|
))
|
||||||
|
except Exception: # noqa: BLE001
|
||||||
|
pass
|
||||||
|
if transitioned:
|
||||||
|
session.commit()
|
||||||
|
|
||||||
|
return {
|
||||||
|
"ok": True,
|
||||||
|
"transitioned": transitioned,
|
||||||
|
"already_acked": already_acked,
|
||||||
|
"not_found": not_found,
|
||||||
|
"not_rejected": not_rejected,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
@router.post("/api/inbox/rejected/resubmit")
|
||||||
|
def inbox_resubmit_rejected(
|
||||||
|
body: dict,
|
||||||
|
download: bool = Query(False, description="When true, return a ZIP of regenerated 837 files for the resubmitted claims (instead of JSON)."),
|
||||||
|
):
|
||||||
|
"""Bulk move REJECTED claims back to SUBMITTED.
|
||||||
|
|
||||||
|
With ``?download=true``, the response is a ``application/zip`` archive
|
||||||
|
containing one ``claim-{id}.x12`` per successfully resubmitted claim
|
||||||
|
(regenerated via ``serialize_837_for_resubmit`` so each file gets a
|
||||||
|
unique interchange/group control number). Conflicts are omitted from
|
||||||
|
the ZIP — they remain visible to the caller via the JSON shape of the
|
||||||
|
non-download path. Empty resubmit + download → 200 with an empty zip
|
||||||
|
so the UI can still hand the user a downloadable artifact.
|
||||||
|
"""
|
||||||
|
ids = body.get("claim_ids") or []
|
||||||
|
if not ids:
|
||||||
|
raise HTTPException(400, "claim_ids required")
|
||||||
|
accepted: list[str] = []
|
||||||
|
conflicts: list[dict] = []
|
||||||
|
# Track which claims are about to be resubmitted (and their index in
|
||||||
|
# the bundle) so the download path can serialize them with unique
|
||||||
|
# control numbers — back-to-back resubmits in the same file would
|
||||||
|
# otherwise all share ISA13/GS06 = "000000001".
|
||||||
|
accepted_with_rows: list[tuple[str, Claim]] = []
|
||||||
|
with db.SessionLocal()() as s:
|
||||||
|
for cid in ids:
|
||||||
|
c = s.get(Claim, cid)
|
||||||
|
if c is None:
|
||||||
|
continue
|
||||||
|
if c.state != ClaimState.REJECTED:
|
||||||
|
conflicts.append({
|
||||||
|
"claim_id": cid,
|
||||||
|
"current_state": (
|
||||||
|
c.state.value if hasattr(c.state, "value")
|
||||||
|
else str(c.state)
|
||||||
|
),
|
||||||
|
})
|
||||||
|
continue
|
||||||
|
c.state = ClaimState.SUBMITTED
|
||||||
|
c.state_changed_at = datetime.now(timezone.utc)
|
||||||
|
c.rejection_reason = None
|
||||||
|
c.rejected_at = None
|
||||||
|
c.resubmit_count = (c.resubmit_count or 0) + 1
|
||||||
|
accepted.append(cid)
|
||||||
|
accepted_with_rows.append((cid, c))
|
||||||
|
s.commit()
|
||||||
|
|
||||||
|
if not download:
|
||||||
|
return {"ok": True, "resubmitted": accepted, "conflicts": conflicts}
|
||||||
|
|
||||||
|
# Build a ZIP of regenerated 837s for the accepted claims. Conflicts
|
||||||
|
# and missing ids are deliberately excluded — the user already saw
|
||||||
|
# them in the JSON response on prior actions; the download is the
|
||||||
|
# "give me the files I asked for" payload.
|
||||||
|
buf = io.BytesIO()
|
||||||
|
serialize_errors: list[dict] = []
|
||||||
|
with zipfile.ZipFile(buf, mode="w", compression=zipfile.ZIP_DEFLATED) as zf:
|
||||||
|
for idx, (cid, c) in enumerate(accepted_with_rows, start=1):
|
||||||
|
if not c.raw_json:
|
||||||
|
serialize_errors.append({"claim_id": cid, "reason": "no raw_json"})
|
||||||
|
continue
|
||||||
|
try:
|
||||||
|
claim_obj = ClaimOutput.model_validate(c.raw_json)
|
||||||
|
except Exception as exc:
|
||||||
|
serialize_errors.append({"claim_id": cid, "reason": f"raw_json invalid: {exc}"})
|
||||||
|
continue
|
||||||
|
try:
|
||||||
|
text = serialize_837_for_resubmit(claim_obj, interchange_index=idx)
|
||||||
|
except SerializeError837 as exc:
|
||||||
|
serialize_errors.append({"claim_id": cid, "reason": str(exc)})
|
||||||
|
continue
|
||||||
|
zf.writestr(f"claim-{cid}.x12", text)
|
||||||
|
buf.seek(0)
|
||||||
|
headers = {
|
||||||
|
"Content-Disposition": (
|
||||||
|
f'attachment; filename="resubmit-{len(accepted)}-claims.zip"'
|
||||||
|
),
|
||||||
|
}
|
||||||
|
# Surface per-claim serialization failures as a custom response header
|
||||||
|
# so the UI can show "10 resubmitted, 2 couldn't be regenerated" without
|
||||||
|
# parsing the binary. The header value is JSON-encoded; the UI is
|
||||||
|
# expected to JSON.parse it after a fetch with response.ok.
|
||||||
|
if serialize_errors:
|
||||||
|
headers["X-Cyclone-Serialize-Errors"] = json.dumps(serialize_errors)
|
||||||
|
return Response(
|
||||||
|
content=buf.getvalue(),
|
||||||
|
media_type="application/zip",
|
||||||
|
headers=headers,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@router.get("/api/inbox/export.csv")
|
||||||
|
def inbox_export_csv(request: Request, lane: str):
|
||||||
|
"""Stream a CSV for a single lane."""
|
||||||
|
if lane not in {"rejected", "candidates", "unmatched", "done_today"}:
|
||||||
|
raise HTTPException(400, f"unknown lane: {lane}")
|
||||||
|
dismissed_pairs = getattr(request.app.state, "dismissed_pairs", set())
|
||||||
|
with db.SessionLocal()() as session:
|
||||||
|
from cyclone.inbox_lanes import compute_lanes
|
||||||
|
lanes = compute_lanes(session, dismissed_pairs=dismissed_pairs)
|
||||||
|
rows = getattr(lanes, lane)
|
||||||
|
|
||||||
|
buf = io.StringIO()
|
||||||
|
writer = csv.writer(buf)
|
||||||
|
writer.writerow([
|
||||||
|
"id", "kind", "patient_control_number", "charge_amount",
|
||||||
|
"payer_id", "provider_npi", "state", "rejection_reason",
|
||||||
|
"service_date", "score",
|
||||||
|
])
|
||||||
|
for r in rows:
|
||||||
|
writer.writerow([
|
||||||
|
r.get("id") or r.get("payer_claim_control_number"),
|
||||||
|
r.get("kind"),
|
||||||
|
r.get("patient_control_number"),
|
||||||
|
r.get("charge_amount"),
|
||||||
|
r.get("payer_id"),
|
||||||
|
r.get("provider_npi") or r.get("rendering_provider_npi"),
|
||||||
|
r.get("state"),
|
||||||
|
r.get("rejection_reason"),
|
||||||
|
r.get("service_date_from") or r.get("service_date"),
|
||||||
|
r.get("score"),
|
||||||
|
])
|
||||||
|
buf.seek(0)
|
||||||
|
return StreamingResponse(
|
||||||
|
iter([buf.getvalue()]),
|
||||||
|
media_type="text/csv",
|
||||||
|
headers={"Content-Disposition": f'attachment; filename="inbox-{lane}.csv"'},
|
||||||
|
)
|
||||||
@@ -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,164 @@
|
|||||||
|
"""``/api/reconciliation/*`` + ``/api/batch-diff`` — operator review surface.
|
||||||
|
|
||||||
|
Four endpoints powering the reconciliation page:
|
||||||
|
|
||||||
|
* ``GET /api/reconciliation/unmatched`` — return unmatched Claims
|
||||||
|
(left) and unmatched Remittances (right).
|
||||||
|
* ``POST /api/reconciliation/match`` — manually pair a Claim with a
|
||||||
|
Remittance (operator override).
|
||||||
|
* ``POST /api/reconciliation/unmatch`` — remove the current match and
|
||||||
|
reset the Claim to submitted.
|
||||||
|
* ``GET /api/batch-diff`` — side-by-side diff of two batches (SP3 P4 /
|
||||||
|
T18).
|
||||||
|
|
||||||
|
The two POST endpoints delegate to ``store.manual_match`` /
|
||||||
|
``store.manual_unmatch``; the router is responsible for translating the
|
||||||
|
store's exception hierarchy (``AlreadyMatchedError``, ``InvalidStateError``,
|
||||||
|
``NotMatchedError``, ``LookupError``) into the HTTP error contract.
|
||||||
|
"""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from fastapi import APIRouter, HTTPException, Query
|
||||||
|
|
||||||
|
from cyclone.store import (
|
||||||
|
AlreadyMatchedError,
|
||||||
|
InvalidStateError,
|
||||||
|
store,
|
||||||
|
)
|
||||||
|
|
||||||
|
router = APIRouter()
|
||||||
|
|
||||||
|
|
||||||
|
@router.get("/api/reconciliation/unmatched")
|
||||||
|
def get_reconciliation_unmatched() -> dict:
|
||||||
|
"""Return unmatched Claims (left) and unmatched Remittances (right).
|
||||||
|
|
||||||
|
Powers the reconciliation review surface: every Claim with no
|
||||||
|
paired Remittance appears on the left, every Remittance with no
|
||||||
|
paired Claim appears on the right. The two lists are always present
|
||||||
|
(empty list, never absent) so the UI can index unconditionally.
|
||||||
|
"""
|
||||||
|
return store.list_unmatched(kind="both")
|
||||||
|
|
||||||
|
|
||||||
|
@router.get("/api/batch-diff")
|
||||||
|
def get_batch_diff(
|
||||||
|
a: str | None = Query(None),
|
||||||
|
b: str | None = Query(None),
|
||||||
|
) -> dict:
|
||||||
|
"""Return a side-by-side diff of two batches identified by id.
|
||||||
|
|
||||||
|
Query params: ``a=<batch_id>``, ``b=<batch_id>`` (both required).
|
||||||
|
|
||||||
|
Response body (snake_case keys, see :mod:`cyclone.batch_diff` for the
|
||||||
|
projector shapes):
|
||||||
|
- ``a`` / ``b`` — small metadata blocks (id, kind, parsedAt,
|
||||||
|
inputFilename, claimCount)
|
||||||
|
- ``added`` — claims present in B but not A
|
||||||
|
- ``removed`` — claims present in A but not B
|
||||||
|
- ``changed`` — claims present in both, with field deltas
|
||||||
|
- ``summary`` — precomputed counts
|
||||||
|
|
||||||
|
Errors:
|
||||||
|
- 400 — missing ``a`` or ``b``
|
||||||
|
- 404 — either batch id is unknown
|
||||||
|
|
||||||
|
Pure read endpoint — never mutates the store. Both 837P and 835
|
||||||
|
batches are accepted (mixed-kind diffs are valid: comparing the
|
||||||
|
submitted claims against the matching remittances).
|
||||||
|
"""
|
||||||
|
if not a or not b:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=400,
|
||||||
|
detail={"error": "Missing param", "detail": "Both ?a=<batch_id> and ?b=<batch_id> are required."},
|
||||||
|
)
|
||||||
|
try:
|
||||||
|
a_rec, b_rec = store.load_two_for_diff(a, b)
|
||||||
|
except LookupError as exc:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=404,
|
||||||
|
detail={"error": "Not found", "detail": str(exc)},
|
||||||
|
)
|
||||||
|
|
||||||
|
# Lazy import — keeps the module's import surface small until the
|
||||||
|
# endpoint is actually hit.
|
||||||
|
from cyclone.batch_diff import diff_batches_to_wire
|
||||||
|
|
||||||
|
return diff_batches_to_wire(a_rec, b_rec)
|
||||||
|
|
||||||
|
|
||||||
|
@router.post("/api/reconciliation/match")
|
||||||
|
def post_reconciliation_match(body: dict) -> dict:
|
||||||
|
"""Manually pair a Claim with a Remittance (operator override).
|
||||||
|
|
||||||
|
Body: ``{"claim_id": ..., "remit_id": ...}``. Returns
|
||||||
|
``{"claim": <ui>, "match": <ui>}`` on success. Errors:
|
||||||
|
- 400: missing ``claim_id`` or ``remit_id``
|
||||||
|
- 404: claim or remittance not found
|
||||||
|
- 409: claim already matched, or apply_* returned a noop
|
||||||
|
(claim in terminal state) — detail echoes ``current_state``
|
||||||
|
and ``activity_kind`` so the UI can render a precise message.
|
||||||
|
"""
|
||||||
|
claim_id = body.get("claim_id")
|
||||||
|
remit_id = body.get("remit_id")
|
||||||
|
if not claim_id or not remit_id:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=400,
|
||||||
|
detail="claim_id and remit_id required",
|
||||||
|
)
|
||||||
|
try:
|
||||||
|
return store.manual_match(claim_id, remit_id)
|
||||||
|
except AlreadyMatchedError as e:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=409,
|
||||||
|
detail={"error": "already_matched", "message": str(e)},
|
||||||
|
)
|
||||||
|
except InvalidStateError as e:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=409,
|
||||||
|
detail={
|
||||||
|
"error": "invalid_state",
|
||||||
|
"current_state": e.current_state,
|
||||||
|
"activity_kind": e.activity_kind,
|
||||||
|
},
|
||||||
|
)
|
||||||
|
except LookupError:
|
||||||
|
# manual_match raises LookupError when the claim or remittance
|
||||||
|
# row is missing (we catch the parent class so any future
|
||||||
|
# KeyError subclasses in the store get the same treatment).
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=404,
|
||||||
|
detail="claim_or_remit_not_found",
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@router.post("/api/reconciliation/unmatch")
|
||||||
|
def post_reconciliation_unmatch(body: dict) -> dict:
|
||||||
|
"""Remove the current match for a Claim; reset Claim to submitted.
|
||||||
|
|
||||||
|
Body: ``{"claim_id": ...}``. Returns
|
||||||
|
``{"claim": <ui>, "deletedMatches": <count>}``. Errors:
|
||||||
|
- 400: missing ``claim_id``
|
||||||
|
- 404: claim not found
|
||||||
|
- 409: claim has no current match (NotMatchedError is mapped
|
||||||
|
by the store; we surface 409 to match the manual_match contract)
|
||||||
|
"""
|
||||||
|
from cyclone.store import NotMatchedError
|
||||||
|
claim_id = body.get("claim_id")
|
||||||
|
if not claim_id:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=400,
|
||||||
|
detail="claim_id required",
|
||||||
|
)
|
||||||
|
try:
|
||||||
|
return store.manual_unmatch(claim_id)
|
||||||
|
except NotMatchedError as e:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=409,
|
||||||
|
detail={"error": "not_matched", "message": str(e)},
|
||||||
|
)
|
||||||
|
except LookupError:
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=404,
|
||||||
|
detail="claim_not_found",
|
||||||
|
)
|
||||||
@@ -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