Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Tyler
|
81aebf54ed | ||
|
Tyler
|
3b5e2af077 | ||
|
Tyler
|
fc73075ef9 |
@@ -418,6 +418,35 @@ was a one-file change (`sftp_paramiko.py` replacing `sftp_stub.py`).
|
||||
the `paramiko` extras aren't installed so the test suite stays green on
|
||||
Linux dev boxes.
|
||||
|
||||
## 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
|
||||
@@ -474,20 +503,31 @@ backup API).
|
||||
.
|
||||
├── backend/
|
||||
│ ├── src/cyclone/
|
||||
│ │ ├── api.py # FastAPI app + parse routes; mounts api_routers/* sub-apps
|
||||
│ │ ├── api.py # FastAPI app + middleware; mounts api_routers/* sub-apps
|
||||
│ │ ├── api_helpers.py # NDJSON / content-negotiation / live-tail helpers
|
||||
│ │ ├── api_routers/ # FastAPI APIRouters extracted from api.py
|
||||
│ │ │ ├── health.py # GET /api/health
|
||||
│ │ │ ├── acks.py # GET /api/acks, /api/acks/{id}
|
||||
│ │ │ ├── ta1_acks.py # GET /api/ta1-acks, /api/ta1-acks/{id}
|
||||
│ │ │ ├── activity.py # GET /api/activity, /api/activity/stream
|
||||
│ │ │ ├── batches.py # GET /api/batches, /api/batches/{id}, /api/batch-diff
|
||||
│ │ │ ├── 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
|
||||
│ │ │ ├── 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)
|
||||
│ │ ├── store.py # CycloneStore, mappers, publish-on-write
|
||||
│ │ ├── db.py # SQLAlchemy engine, session factory, ORM models
|
||||
|
||||
+34
-838
@@ -15,8 +15,6 @@ plus GET/POST with any header.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import csv
|
||||
import io
|
||||
import json
|
||||
import logging
|
||||
import uuid
|
||||
@@ -26,23 +24,16 @@ from typing import Any, AsyncIterator
|
||||
|
||||
from fastapi import FastAPI, File, HTTPException, Query, Request, UploadFile
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from fastapi.responses import JSONResponse, Response, StreamingResponse
|
||||
from fastapi.responses import JSONResponse, StreamingResponse
|
||||
from pydantic import ValidationError
|
||||
|
||||
from cyclone import __version__, db
|
||||
from cyclone.db import Claim, ClaimState, Remittance
|
||||
from cyclone.inbox_state import apply_999_rejections
|
||||
from cyclone.inbox_state_277ca import apply_277ca_rejections
|
||||
from cyclone.audit_log import AuditEvent, append_event, verify_chain
|
||||
from cyclone.audit_log import AuditEvent, append_event
|
||||
from cyclone.parsers.exceptions import CycloneParseError
|
||||
from cyclone.parsers.models import BatchSummary, ClaimOutput, Envelope, ParseResult
|
||||
from cyclone.parsers.models_270 import (
|
||||
EligibilityBenefitInquiry,
|
||||
InformationReceiver270,
|
||||
InformationSource270,
|
||||
ParseResult270,
|
||||
Subscriber270,
|
||||
)
|
||||
from cyclone.parsers.models import BatchSummary, Envelope
|
||||
from cyclone.parsers.models_271 import ParseResult271
|
||||
from cyclone.parsers.models_835 import ParseResult835
|
||||
from cyclone.parsers.payer import PayerConfig, PayerConfig835
|
||||
@@ -53,9 +44,8 @@ from cyclone.parsers.parse_837 import parse
|
||||
from cyclone.parsers.parse_835 import parse as parse_835
|
||||
from cyclone.parsers.parse_999 import parse_999_text
|
||||
from cyclone.parsers.parse_ta1 import parse_ta1_text
|
||||
from cyclone.parsers.serialize_270 import serialize_270
|
||||
from cyclone.parsers.serialize_999 import serialize_999
|
||||
from cyclone.parsers.serialize_837 import SerializeError as SerializeError837, serialize_837, serialize_837_for_resubmit
|
||||
|
||||
from cyclone.parsers.batch_ack_builder import build_ack_for_batch
|
||||
from cyclone.parsers.validator_835 import validate as validate_835
|
||||
from cyclone.pubsub import EventBus
|
||||
@@ -144,12 +134,16 @@ app.add_middleware(
|
||||
from cyclone.api_routers import ( # noqa: E402
|
||||
acks,
|
||||
activity,
|
||||
admin,
|
||||
batches,
|
||||
claims,
|
||||
clearhouse,
|
||||
config,
|
||||
eligibility,
|
||||
health,
|
||||
inbox,
|
||||
providers,
|
||||
reconciliation,
|
||||
remittances,
|
||||
ta1_acks,
|
||||
)
|
||||
@@ -164,6 +158,10 @@ app.include_router(activity.router)
|
||||
app.include_router(batches.router)
|
||||
app.include_router(claims.router)
|
||||
app.include_router(remittances.router)
|
||||
app.include_router(inbox.router)
|
||||
app.include_router(reconciliation.router)
|
||||
app.include_router(eligibility.router)
|
||||
app.include_router(admin.router)
|
||||
|
||||
# Backwards-compat re-exports: test_api_stream_live.py imports these
|
||||
# names directly from ``cyclone.api`` to invoke the endpoint coroutines
|
||||
@@ -173,6 +171,24 @@ activity_stream = activity.activity_stream_endpoint
|
||||
claims_stream = claims.claims_stream_endpoint
|
||||
remittances_stream = remittances.remittances_stream_endpoint
|
||||
|
||||
# Backwards-compat lock re-export: test_api_rotate_key.py does
|
||||
# ``api_mod._db_rotate_lock.acquire()`` to simulate a concurrent
|
||||
# rotation request. The lock itself lives in the admin router (it
|
||||
# guards that router's rotate-key endpoint) but we expose the same
|
||||
# object here so the test holds the same lock the endpoint does.
|
||||
_db_rotate_lock = admin._db_rotate_lock
|
||||
|
||||
# Backwards-compat module re-exports: test_api_rotate_key.py does
|
||||
# ``monkeypatch.setattr("cyclone.api._db_crypto.rotate_db_key", ...)``
|
||||
# and ``monkeypatch.setattr("cyclone.api._secrets.set_secret", ...)``
|
||||
# to stub crypto/keychain calls. Re-exporting here keeps the
|
||||
# ``cyclone.api._db_crypto`` / ``cyclone.api._secrets`` paths alive
|
||||
# so the monkeypatch targets the same module objects the admin
|
||||
# endpoint actually uses (both are the same ``cyclone.db_crypto`` /
|
||||
# ``cyclone.secrets`` module instances).
|
||||
_db_crypto = admin.db_crypto
|
||||
_secrets = admin.secrets
|
||||
|
||||
|
||||
def _resolve_payer(name: str) -> PayerConfig:
|
||||
if name not in PAYER_FACTORIES:
|
||||
@@ -901,416 +917,10 @@ def _serialize_ta1_from_row(row: db.Ta1Ack) -> str:
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
@app.get("/api/inbox/lanes")
|
||||
def inbox_lanes():
|
||||
"""Return all Inbox lanes in one call."""
|
||||
dismissed_pairs = getattr(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,
|
||||
}
|
||||
# --- /api/inbox/* lives in cyclone.api_routers.inbox ---
|
||||
|
||||
# --- /api/reconciliation/* + /api/batch-diff live in cyclone.api_routers.reconciliation ---
|
||||
|
||||
@app.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}
|
||||
|
||||
|
||||
@app.post("/api/inbox/candidates/dismiss")
|
||||
def inbox_dismiss_candidates(body: dict):
|
||||
"""Add candidate pairs to the session-scoped dismissed set."""
|
||||
pairs = body.get("pairs") or []
|
||||
if not hasattr(app.state, "dismissed_pairs"):
|
||||
app.state.dismissed_pairs = set()
|
||||
for p in pairs:
|
||||
cid = p.get("claim_id")
|
||||
rid = p.get("remit_id")
|
||||
if cid and rid:
|
||||
app.state.dismissed_pairs.add(frozenset({cid, rid}))
|
||||
return {"ok": True, "dismissed_count": len(pairs)}
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# SP14: Payer-Rejected acknowledge
|
||||
#
|
||||
# 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".
|
||||
# --------------------------------------------------------------------------- #
|
||||
@app.post("/api/inbox/payer-rejected/acknowledge")
|
||||
def inbox_acknowledge_payer_rejected(body: dict):
|
||||
"""Mark Payer-Rejected claims as acknowledged by the operator."""
|
||||
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:
|
||||
from cyclone.db import Claim
|
||||
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.
|
||||
try:
|
||||
from cyclone.audit_log import append_event, AuditEvent
|
||||
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
|
||||
# Audit append is best-effort; don't block the operator's
|
||||
# acknowledge action on an audit-log failure.
|
||||
pass
|
||||
if transitioned:
|
||||
session.commit()
|
||||
|
||||
return {
|
||||
"ok": True,
|
||||
"transitioned": transitioned,
|
||||
"already_acked": already_acked,
|
||||
"not_found": not_found,
|
||||
"not_rejected": not_rejected,
|
||||
}
|
||||
|
||||
|
||||
@app.post("/api/inbox/rejected/resubmit")
|
||||
def inbox_resubmit_rejected(
|
||||
request: Request,
|
||||
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.
|
||||
import zipfile
|
||||
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,
|
||||
)
|
||||
|
||||
|
||||
@app.get("/api/inbox/export.csv")
|
||||
def inbox_export_csv(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(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"'},
|
||||
)
|
||||
|
||||
|
||||
# --- /api/batches lives in cyclone.api_routers.batches ---
|
||||
|
||||
|
||||
# --- /api/claims lives in cyclone.api_routers.claims ---
|
||||
|
||||
@app.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")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Side-by-side diff between two batches (SP3 P4 / T18)
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
@app.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. Mirrors the same pattern used by other
|
||||
# endpoint-local helpers (e.g. reconciler).
|
||||
from cyclone.batch_diff import diff_batches_to_wire
|
||||
|
||||
return diff_batches_to_wire(a_rec, b_rec)
|
||||
|
||||
|
||||
@app.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",
|
||||
)
|
||||
|
||||
|
||||
@app.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",
|
||||
)
|
||||
|
||||
|
||||
# --- /api/remittances lives in cyclone.api_routers.remittances ---
|
||||
@@ -1330,424 +940,10 @@ def post_reconciliation_unmatch(body: dict) -> dict:
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def _validate_eligibility_request(body: dict) -> tuple[ParseResult270, str]:
|
||||
"""Build a :class:`ParseResult270` from a request body dict.
|
||||
# --- /api/eligibility/* lives in cyclone.api_routers.eligibility ---
|
||||
|
||||
The body shape is the minimum surface needed to build a valid 270
|
||||
inquiry (per spec section 3.4 — operator-driven, ephemeral):
|
||||
# --- /api/admin/* lives in cyclone.api_routers.admin ---
|
||||
|
||||
::
|
||||
|
||||
{
|
||||
"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"},
|
||||
)
|
||||
|
||||
# Build the Pydantic models. The serializer handles all envelope
|
||||
# generation (sender_id/receiver_id/control_number/transaction_date
|
||||
# are filled in by the serializer with sensible defaults).
|
||||
from datetime import date as _date
|
||||
|
||||
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
|
||||
|
||||
|
||||
@app.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()),
|
||||
}
|
||||
|
||||
|
||||
@app.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()),
|
||||
}
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# SP11: tamper-evident audit log (admin)
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
@app.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
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
@app.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,
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# SP15: SQLCipher key rotation
|
||||
#
|
||||
# Re-encrypts the DB in place with a fresh key, then updates the
|
||||
# Keychain so subsequent connections open with the new key. This is
|
||||
# a 1-time operation per rotation; for routine read/write the rest
|
||||
# of the API is unchanged.
|
||||
#
|
||||
# Concurrency: the rotation holds a module-level lock so two
|
||||
# concurrent requests can't race and end up with mismatched Keychain
|
||||
# + DB. The lock is a simple threading.Lock; a process restart
|
||||
# resets it (intentional — the operator's next start-up opens with
|
||||
# whatever key is in the Keychain).
|
||||
# ---------------------------------------------------------------------------
|
||||
import threading as _threading
|
||||
from cyclone import db_crypto as _db_crypto
|
||||
from cyclone import secrets as _secrets
|
||||
|
||||
_db_rotate_lock = _threading.Lock()
|
||||
|
||||
|
||||
@app.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:
|
||||
from cyclone.audit_log import append_event, AuditEvent
|
||||
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()
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
@app.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": []}
|
||||
|
||||
|
||||
__all__ = ["app"]
|
||||
|
||||
@@ -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,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,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",
|
||||
)
|
||||
Reference in New Issue
Block a user