cyclone
831 lines
32 KiB
Python
831 lines
32 KiB
Python
"""Parse endpoints — accept X12 uploads and ingest them.
|
|
|
|
Five routes:
|
|
|
|
* ``POST /api/parse-837`` — 837P professional claim ingest (the
|
|
primary upload path)
|
|
* ``POST /api/parse-835`` — 835 ERA remittance ingest
|
|
* ``POST /api/parse-999`` — 999 ACK ingest + auto-link claims
|
|
* ``POST /api/parse-ta1`` — TA1 envelope ACK ingest + envelope-link batches
|
|
* ``POST /api/parse-277ca`` — 277CA Claim Acknowledgment ingest
|
|
|
|
The 7 cross-router helpers these endpoints need (and the two
|
|
PAYER_FACTORIES dicts they consume) live in
|
|
:mod:`cyclone.api_routers._shared`.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import uuid
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, Depends, File, Query, Request, UploadFile
|
|
from fastapi.responses import JSONResponse, StreamingResponse
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
from cyclone import db
|
|
from cyclone.api_helpers import (
|
|
client_wants_json as _client_wants_json,
|
|
drop_raw_segments_837 as _drop_raw_segments,
|
|
drop_raw_segments_835 as _drop_raw_segments_835,
|
|
has_claim_validation_errors as _has_claim_validation_errors,
|
|
has_835_validation_errors as _has_835_validation_errors,
|
|
ndjson_stream_837 as _ndjson_stream,
|
|
ndjson_stream_835 as _ndjson_stream_835,
|
|
strict_rewrite_837 as _strict_rewrite,
|
|
strict_rewrite_835 as _strict_rewrite_835,
|
|
)
|
|
from cyclone.api_routers._shared import (
|
|
_actor_user_id,
|
|
_build_and_persist_ack,
|
|
_reconciliation_summary_for_batch,
|
|
_resolve_payer,
|
|
_resolve_payer_835,
|
|
_serialize_ta1,
|
|
_ta1_synthetic_source_batch_id,
|
|
_transaction_set_id_from_segments,
|
|
)
|
|
from cyclone.audit_log import AuditEvent, append_event
|
|
from cyclone.auth.deps import matrix_gate
|
|
from cyclone.claim_acks import (
|
|
apply_277ca_acks as _apply_277ca_acks,
|
|
apply_999_acceptances as _apply_999_acceptances,
|
|
apply_ta1_envelope_link as _apply_ta1_envelope_link,
|
|
)
|
|
from cyclone.db import Batch, Claim
|
|
from cyclone.handlers._ack_id import (
|
|
ack_count_summary as _ack_count_summary,
|
|
ack_synthetic_source_batch_id as _ack_synthetic_source_batch_id,
|
|
two77ca_synthetic_source_batch_id as _277ca_synthetic_source_batch_id,
|
|
)
|
|
from cyclone.inbox_state import apply_999_rejections
|
|
from cyclone.inbox_state_277ca import apply_277ca_rejections
|
|
from cyclone.parsers.exceptions import CycloneParseError
|
|
from cyclone.parsers.parse_277ca import parse_277ca_text
|
|
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.segments import tokenize as _tokenize_segments
|
|
from cyclone.parsers.serialize_999 import serialize_999
|
|
from cyclone.parsers.validator_835 import validate as validate_835
|
|
from cyclone.store import BatchRecord, store, utcnow
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(dependencies=[Depends(matrix_gate)])
|
|
|
|
|
|
@router.post("/api/parse-837")
|
|
async def parse_837(
|
|
request: Request,
|
|
file: UploadFile = File(...),
|
|
payer: str = Query("co_medicaid"),
|
|
include_raw_segments: bool = Query(True),
|
|
strict: bool = Query(False),
|
|
ack: bool = Query(False),
|
|
) -> Any:
|
|
# SP35: defense-in-depth input guards. Layer A (UI auto-detect) lives
|
|
# in src/pages/Upload.tsx; the server-side checks below are the
|
|
# authoritative fix because they protect every caller of the API
|
|
# (Upload page, CLI ingestion, any future bulk-import tool). Without
|
|
# these, an 835 file dropped on the Upload page while the dropdown
|
|
# still says "837p" produces a BatchRecord with claims=[] and a bogus
|
|
# row on the History tab. The fix is two checks run BEFORE we persist
|
|
# anything:
|
|
#
|
|
# 1. Envelope check — ST01 must be "837" or "837P". Anything else
|
|
# (an 835, a 999, a 270, garbage that happens to have an ISA)
|
|
# → 400 with error="Mismatched file kind", expected="837p",
|
|
# detected_st=<whatever was there>.
|
|
# 2. Empty-claims check — even with the right envelope, if the
|
|
# parser produced zero CLM segments (truncated file, header-only
|
|
# test fixture) → 400 with error="No claims parsed". A real
|
|
# production 837 batch with zero claims is never valid.
|
|
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)},
|
|
)
|
|
|
|
config = _resolve_payer(payer)
|
|
|
|
# SP35 guard 1: envelope check. Tokenize first so we can return a
|
|
# precise 400 (vs. relying on the parser's "no ISA envelope" error
|
|
# which is correct but doesn't say "you sent an 835 to the 837
|
|
# endpoint"). If tokenization itself fails we fall through to the
|
|
# parser, which raises CycloneParseError → 400 "Parse error" path.
|
|
try:
|
|
_segments = _tokenize_segments(text)
|
|
detected_st = _transaction_set_id_from_segments(_segments) or ""
|
|
except CycloneParseError:
|
|
detected_st = ""
|
|
|
|
if detected_st and not detected_st.upper().startswith("837"):
|
|
return JSONResponse(
|
|
status_code=400,
|
|
content={
|
|
"error": "Mismatched file kind",
|
|
"expected": "837p",
|
|
"detected_st": detected_st,
|
|
"detail": (
|
|
f"File declares ST*{detected_st}* but this endpoint "
|
|
f"expects ST*837*. Pick the matching endpoint on the "
|
|
f"Upload page (or let auto-detect choose for you)."
|
|
),
|
|
},
|
|
)
|
|
|
|
try:
|
|
result = parse(text, config, 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")
|
|
return JSONResponse(
|
|
status_code=500,
|
|
content={"error": "Internal server error", "detail": str(exc)},
|
|
)
|
|
|
|
# SP35 guard 2: empty-claims check. With the envelope validated, the
|
|
# only way to land here is a header-only file (real, but useless)
|
|
# or a file whose CLM loops the parser couldn't extract. Either way
|
|
# we refuse to persist — a BatchRecord with claims=[] is what the
|
|
# original bug produced and is never what the operator wanted.
|
|
if not result.claims:
|
|
return JSONResponse(
|
|
status_code=400,
|
|
content={
|
|
"error": "No claims parsed",
|
|
"detail": (
|
|
"The file passed the envelope check but contained no "
|
|
"CLM segments. Refusing to persist an empty batch."
|
|
),
|
|
},
|
|
)
|
|
|
|
if strict:
|
|
result = _strict_rewrite(result)
|
|
if not include_raw_segments:
|
|
result = _drop_raw_segments(result)
|
|
|
|
if _has_claim_validation_errors(result):
|
|
# Per spec: 422 when claims failed validation.
|
|
# Body still includes the full ParseResult so the client can show errors.
|
|
# Validation-failed parses are NOT persisted (the data is suspect);
|
|
# only parses that survive validation end up in the store.
|
|
return JSONResponse(
|
|
status_code=422,
|
|
content=json.loads(result.model_dump_json()),
|
|
)
|
|
|
|
# Persist the cleaned-up result so the cleaned result is what clients
|
|
# retrieve via /api/batches/{id}.
|
|
rec = BatchRecord(
|
|
id=uuid.uuid4().hex,
|
|
kind="837p",
|
|
input_filename=file.filename or "upload.txt",
|
|
parsed_at=utcnow(),
|
|
result=result,
|
|
)
|
|
try:
|
|
store.add(rec, event_bus=request.app.state.event_bus)
|
|
except IntegrityError as exc:
|
|
# ``(batch_id, patient_control_number)`` is UNIQUE — fires when a
|
|
# single batch file contains the same CLM01 control number twice,
|
|
# or when the same claim id has already been ingested in a prior
|
|
# batch. Surface as 409 with the batch id so the caller can look
|
|
# it up; do NOT 500 (a 500 without CORS headers is misreported by
|
|
# browsers as a CORS error and hides the real cause).
|
|
log.warning("Duplicate claim while persisting batch %s: %s", rec.id, exc)
|
|
return JSONResponse(
|
|
status_code=409,
|
|
content={
|
|
"error": "Duplicate claim",
|
|
"detail": (
|
|
"This file (or one previously ingested with the same "
|
|
"claim control number) collides with an existing "
|
|
"record. Inspect the file for duplicate CLM01 "
|
|
"control numbers, or remove the existing batch "
|
|
"before retrying."
|
|
),
|
|
"batch_id": rec.id,
|
|
},
|
|
)
|
|
|
|
if _client_wants_json(request):
|
|
body = json.loads(result.model_dump_json())
|
|
if ack:
|
|
ack_body = _build_and_persist_ack(rec.id)
|
|
if ack_body is not None:
|
|
body["ack"] = ack_body
|
|
# Surface the server-side batch id so the frontend can call
|
|
# /api/batches/{id}/export-837 (and any other batch-scoped
|
|
# endpoint) without a separate listBatches round-trip.
|
|
body["batch_id"] = rec.id
|
|
return JSONResponse(content=body)
|
|
|
|
# Default: NDJSON stream. Pass the server-side batch id so the
|
|
# streaming client (the React Upload page) can call batch-scoped
|
|
# endpoints like /api/batches/{id}/export-837 without a separate
|
|
# GET /api/batches round-trip.
|
|
return StreamingResponse(
|
|
_ndjson_stream(result, batch_id=rec.id),
|
|
media_type="application/x-ndjson",
|
|
)
|
|
|
|
|
|
@router.post("/api/parse-835")
|
|
async def parse_835_endpoint(
|
|
request: Request,
|
|
file: UploadFile = File(...),
|
|
payer: str = Query("co_medicaid_835"),
|
|
include_raw_segments: bool = Query(True),
|
|
strict: bool = Query(False),
|
|
) -> Any:
|
|
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)},
|
|
)
|
|
|
|
config = _resolve_payer_835(payer)
|
|
|
|
# SP35 guard 1: envelope check. Mirrors the parse-837 path: tokenize,
|
|
# read ST01, reject anything that doesn't start with "835". Same
|
|
# defense-in-depth rationale — the UI auto-detect (src/pages/Upload.tsx)
|
|
# is layer A, but server-side guards protect every API caller.
|
|
try:
|
|
_segments_835 = _tokenize_segments(text)
|
|
detected_st_835 = _transaction_set_id_from_segments(_segments_835) or ""
|
|
except CycloneParseError:
|
|
detected_st_835 = ""
|
|
|
|
if detected_st_835 and not detected_st_835.upper().startswith("835"):
|
|
return JSONResponse(
|
|
status_code=400,
|
|
content={
|
|
"error": "Mismatched file kind",
|
|
"expected": "835",
|
|
"detected_st": detected_st_835,
|
|
"detail": (
|
|
f"File declares ST*{detected_st_835}* but this endpoint "
|
|
f"expects ST*835*. Pick the matching endpoint on the "
|
|
f"Upload page (or let auto-detect choose for you)."
|
|
),
|
|
},
|
|
)
|
|
|
|
try:
|
|
result = parse_835(text, config, 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")
|
|
return JSONResponse(
|
|
status_code=500,
|
|
content={"error": "Internal server error", "detail": str(exc)},
|
|
)
|
|
|
|
# SP35 guard 2: empty-claims check. Same as parse-837: a BatchRecord
|
|
# with claims=[] is never a valid production 835 batch and we refuse
|
|
# to persist it.
|
|
if not result.claims:
|
|
return JSONResponse(
|
|
status_code=400,
|
|
content={
|
|
"error": "No claims parsed",
|
|
"detail": (
|
|
"The file passed the envelope check but contained no "
|
|
"CLP segments. Refusing to persist an empty batch."
|
|
),
|
|
},
|
|
)
|
|
|
|
# Always run the validator; attach the report so the JSON path can
|
|
# surface it and the NDJSON path can fold the counts into the summary.
|
|
# 835 validation is batch-level, so pass/fail applies uniformly to every
|
|
# claim payment in the batch (passed=N or 0, failed=0 or N).
|
|
report = validate_835(result, config)
|
|
n = len(result.claims)
|
|
claim_ids = [c.payer_claim_control_number for c in result.claims]
|
|
if report.passed:
|
|
passed, failed, failed_claim_ids = n, 0, []
|
|
else:
|
|
passed, failed, failed_claim_ids = 0, n, claim_ids
|
|
result = result.model_copy(update={
|
|
"validation": report,
|
|
"summary": result.summary.model_copy(update={
|
|
"passed": passed,
|
|
"failed": failed,
|
|
"failed_claim_ids": failed_claim_ids,
|
|
}),
|
|
})
|
|
|
|
if strict:
|
|
result = _strict_rewrite_835(result)
|
|
if not include_raw_segments:
|
|
result = _drop_raw_segments_835(result)
|
|
|
|
if _has_835_validation_errors(result):
|
|
return JSONResponse(
|
|
status_code=422,
|
|
content=json.loads(result.model_dump_json()),
|
|
)
|
|
|
|
# Persist the cleaned-up result so the cleaned result is what clients
|
|
# retrieve via /api/batches/{id}.
|
|
rec = BatchRecord(
|
|
id=uuid.uuid4().hex,
|
|
kind="835",
|
|
input_filename=file.filename or "upload.txt",
|
|
parsed_at=utcnow(),
|
|
result=result,
|
|
)
|
|
try:
|
|
store.add(rec, event_bus=request.app.state.event_bus)
|
|
except IntegrityError as exc:
|
|
log.warning("Duplicate remittance while persisting batch %s: %s", rec.id, exc)
|
|
return JSONResponse(
|
|
status_code=409,
|
|
content={
|
|
"error": "Duplicate remittance",
|
|
"detail": (
|
|
"This 835 file (or one previously ingested with the "
|
|
"same payer claim control number) collides with an "
|
|
"existing record. Remove the existing remittance "
|
|
"before retrying."
|
|
),
|
|
"batch_id": rec.id,
|
|
},
|
|
)
|
|
|
|
if _client_wants_json(request):
|
|
body = json.loads(result.model_dump_json())
|
|
body["reconciliation"] = _reconciliation_summary_for_batch(rec.id)
|
|
return JSONResponse(content=body)
|
|
|
|
# Default: NDJSON stream. Pass the server-side batch id so the
|
|
# streaming client can call batch-scoped endpoints without a
|
|
# separate GET /api/batches round-trip (see /api/parse-837 for the
|
|
# parallel change).
|
|
return StreamingResponse(
|
|
_ndjson_stream_835(result, batch_id=rec.id),
|
|
media_type="application/x-ndjson",
|
|
)
|
|
|
|
|
|
@router.post("/api/parse-999")
|
|
async def parse_999_endpoint(
|
|
request: Request,
|
|
file: UploadFile = File(...),
|
|
) -> Any:
|
|
"""Parse a 999 ACK file, persist a row, and return JSON.
|
|
|
|
Behavior mirrors ``/api/parse-835``:
|
|
- 400 on empty / undecodable / malformed EDI (never 500).
|
|
- 200 on success with ``{"ack": {id, accepted_count, rejected_count,
|
|
received_count, ack_code, raw_999_text}, "parsed": <ParseResult999>}``.
|
|
|
|
The persisted ``acks.source_batch_id`` is a synthetic id
|
|
(``999-<ISA13>``) because a received 999 has no inbound 837 batch
|
|
to FK to. The dashboard's `/acks` list surfaces these; operators
|
|
can still see which interchange each row came from.
|
|
"""
|
|
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_999_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 999")
|
|
return JSONResponse(
|
|
status_code=500,
|
|
content={"error": "Internal server error", "detail": str(exc)},
|
|
)
|
|
|
|
received, accepted, rejected, ack_code = _ack_count_summary(result)
|
|
icn = result.envelope.control_number
|
|
synthetic_id = _ack_synthetic_source_batch_id(icn)
|
|
|
|
# Build the raw 999 text from the parsed result (round-trip).
|
|
raw_999_text = serialize_999(result, interchange_control_number=icn or "000000001")
|
|
|
|
# SP6 T4: move claims whose 999 set was rejected into ClaimState.REJECTED.
|
|
# The 999's set_control_number (AK202) is the source 837's ST02; in
|
|
# practice we look it up against patient_control_number because that's
|
|
# the field 999 ACKs cross-reference in this product.
|
|
with db.SessionLocal()() as session:
|
|
def _lookup(pcn: str):
|
|
return session.query(Claim).filter_by(patient_control_number=pcn).first()
|
|
_rejection_result = apply_999_rejections(
|
|
session, result, claim_lookup=_lookup,
|
|
)
|
|
if _rejection_result.matched:
|
|
bus = request.app.state.event_bus
|
|
for cid in _rejection_result.matched:
|
|
await bus.publish("claim.rejected", {"claim_id": cid})
|
|
if _rejection_result.orphans:
|
|
log.warning(
|
|
"999 had %d orphan set refs: %s",
|
|
len(_rejection_result.orphans),
|
|
_rejection_result.orphans[:5],
|
|
)
|
|
|
|
row = store.add_ack(
|
|
source_batch_id=synthetic_id,
|
|
accepted_count=accepted,
|
|
rejected_count=rejected,
|
|
received_count=received,
|
|
ack_code=ack_code,
|
|
raw_json=json.loads(result.model_dump_json()),
|
|
event_bus=request.app.state.event_bus,
|
|
)
|
|
|
|
# SP11: append one audit row per rejected claim. Each row chains
|
|
# to the previous one — see cyclone.audit_log.
|
|
if _rejection_result.matched:
|
|
with db.SessionLocal()() as audit_s:
|
|
for cid in _rejection_result.matched:
|
|
append_event(audit_s, AuditEvent(
|
|
event_type="claim.rejected",
|
|
entity_type="claim",
|
|
entity_id=cid,
|
|
payload={"source_batch_id": synthetic_id, "ack_id": row.id},
|
|
actor="999-parser",
|
|
user_id=_actor_user_id(request),
|
|
))
|
|
audit_s.commit()
|
|
|
|
# SP28: auto-link the 999 AK2 set-responses to claims via the
|
|
# D10 two-pass join (ST02 via batch envelope index primary,
|
|
# Claim.patient_control_number fallback). Each created ClaimAck
|
|
# row publishes claim_ack_written so the live-tail subscribers
|
|
# on the claim and ack side see the link immediately.
|
|
claim_ack_links_count = 0
|
|
with db.SessionLocal()() as link_s:
|
|
batch_index = store.batch_envelope_index()
|
|
|
|
def _pcn_lookup(pcn: str):
|
|
return (
|
|
link_s.query(Claim)
|
|
.filter(Claim.patient_control_number == pcn)
|
|
.first()
|
|
)
|
|
|
|
link_result = _apply_999_acceptances(
|
|
link_s, result, ack_id=row.id,
|
|
batch_envelope_index=batch_index,
|
|
pc_claim_lookup=_pcn_lookup,
|
|
)
|
|
for link_row in link_result.linked:
|
|
store.add_claim_ack(
|
|
claim_id=link_row.claim_id,
|
|
batch_id=link_row.batch_id,
|
|
ack_id=row.id,
|
|
ack_kind="999",
|
|
ak2_index=link_row.ak2_index,
|
|
set_control_number=link_row.set_control_number,
|
|
set_accept_reject_code=link_row.set_accept_reject_code,
|
|
linked_by="auto",
|
|
event_bus=request.app.state.event_bus,
|
|
)
|
|
claim_ack_links_count += 1
|
|
|
|
return JSONResponse(content={
|
|
"ack": {
|
|
"id": row.id,
|
|
"accepted_count": accepted,
|
|
"rejected_count": rejected,
|
|
"received_count": received,
|
|
"ack_code": ack_code,
|
|
"source_batch_id": synthetic_id,
|
|
"raw_999_text": raw_999_text,
|
|
"claim_ack_links_count": claim_ack_links_count,
|
|
},
|
|
"parsed": json.loads(result.model_dump_json()),
|
|
})
|
|
|
|
|
|
@router.post("/api/parse-ta1")
|
|
async def parse_ta1_endpoint(
|
|
request: Request,
|
|
file: UploadFile = File(...),
|
|
) -> Any:
|
|
"""Parse a TA1 (Interchange Acknowledgment) file, persist a row, return JSON.
|
|
|
|
Mirrors ``/api/parse-999`` but for the lower-level envelope ack:
|
|
- 400 on empty / undecodable / malformed EDI (never 500).
|
|
- 200 on success with ``{"ta1": {id, control_number, ack_code,
|
|
note_code, interchange_date, interchange_time, sender_id,
|
|
receiver_id, source_batch_id, raw_ta1_text}, "parsed":
|
|
<ParseResultTa1>}``.
|
|
|
|
The persisted ``ta1_acks.source_batch_id`` is a synthetic id
|
|
(``TA1-<ISA13>``) because a received TA1 has no inbound batch to
|
|
FK to. The dashboard's ``/ta1-acks`` list surfaces these.
|
|
|
|
SP25: threads ``event_bus=request.app.state.event_bus`` into
|
|
``store.add_ta1_ack`` so the live-tail ``ta1_ack_received``
|
|
stream fires on manual uploads (not just on the SFTP poller).
|
|
"""
|
|
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_ta1_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 TA1")
|
|
return JSONResponse(
|
|
status_code=500,
|
|
content={"error": "Internal server error", "detail": str(exc)},
|
|
)
|
|
|
|
# Build the raw TA1 text from the parsed result (round-trip).
|
|
raw_ta1_text = _serialize_ta1(result)
|
|
|
|
row = store.add_ta1_ack(
|
|
source_batch_id=result.source_batch_id,
|
|
control_number=result.ta1.control_number,
|
|
interchange_date=result.ta1.interchange_date,
|
|
interchange_time=result.ta1.interchange_time,
|
|
ack_code=result.ta1.ack_code,
|
|
note_code=result.ta1.note_code,
|
|
ack_generated_date=result.ta1.ack_generated_date,
|
|
sender_id=result.envelope.sender_id,
|
|
receiver_id=result.envelope.receiver_id,
|
|
raw_json=json.loads(result.model_dump_json()),
|
|
event_bus=request.app.state.event_bus,
|
|
)
|
|
|
|
# SP28: TA1 envelope-level link to the originating Batch. The
|
|
# closure here matches the most-recent Batch whose envelope
|
|
# sender_id/receiver_id matches the TA1 — see spec §D4.
|
|
claim_ack_links_count = 0
|
|
with db.SessionLocal()() as link_s:
|
|
def _batch_lookup(sender_id, receiver_id):
|
|
rows = (
|
|
link_s.query(Batch)
|
|
.filter(
|
|
Batch.kind == "837p",
|
|
Batch.raw_result_json.isnot(None),
|
|
)
|
|
.order_by(Batch.parsed_at.desc())
|
|
.all()
|
|
)
|
|
for row in rows:
|
|
env = (row.raw_result_json or {}).get("envelope") or {}
|
|
if (
|
|
env.get("sender_id") == sender_id
|
|
and env.get("receiver_id") == receiver_id
|
|
):
|
|
return row
|
|
return None
|
|
|
|
link_result = _apply_ta1_envelope_link(
|
|
link_s, result, ack_id=row.id,
|
|
batch_lookup=_batch_lookup,
|
|
)
|
|
for link_row in link_result.linked:
|
|
store.add_claim_ack(
|
|
claim_id=link_row.claim_id,
|
|
batch_id=link_row.batch_id,
|
|
ack_id=row.id,
|
|
ack_kind="ta1",
|
|
ak2_index=link_row.ak2_index,
|
|
set_control_number=link_row.set_control_number,
|
|
set_accept_reject_code=link_row.set_accept_reject_code,
|
|
linked_by="auto",
|
|
event_bus=request.app.state.event_bus,
|
|
)
|
|
claim_ack_links_count += 1
|
|
|
|
return JSONResponse(content={
|
|
"ta1": {
|
|
"id": row.id,
|
|
"control_number": result.ta1.control_number,
|
|
"ack_code": result.ta1.ack_code,
|
|
"note_code": result.ta1.note_code,
|
|
"interchange_date": result.ta1.interchange_date.isoformat()
|
|
if result.ta1.interchange_date else None,
|
|
"interchange_time": result.ta1.interchange_time,
|
|
"sender_id": result.envelope.sender_id,
|
|
"receiver_id": result.envelope.receiver_id,
|
|
"source_batch_id": result.source_batch_id,
|
|
"raw_ta1_text": raw_ta1_text,
|
|
"claim_ack_links_count": claim_ack_links_count,
|
|
},
|
|
"parsed": json.loads(result.model_dump_json()),
|
|
})
|
|
|
|
|
|
@router.post("/api/parse-277ca")
|
|
async def parse_277ca_endpoint(
|
|
request: Request,
|
|
file: UploadFile = File(...),
|
|
) -> Any:
|
|
"""Parse a 277CA Claim Acknowledgment file, persist a row, and stamp rejections.
|
|
|
|
Behavior mirrors ``/api/parse-999``:
|
|
- 400 on empty / undecodable / malformed EDI (never 500).
|
|
- 200 on success with ``{"ack": {id, control_number, accepted_count,
|
|
rejected_count, payer_claim_control_numbers, raw_277ca_text},
|
|
"parsed": <ParseResult277CA>}``.
|
|
|
|
After parse, runs :func:`apply_277ca_rejections` to stamp the
|
|
payer-rejected fields on each matching claim row. The Inbox
|
|
Payer-Rejected lane lights up as a side-effect of this call.
|
|
"""
|
|
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_277ca_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 277CA")
|
|
return JSONResponse(
|
|
status_code=500,
|
|
content={"error": "Internal server error", "detail": str(exc)},
|
|
)
|
|
|
|
icn = result.envelope.control_number
|
|
synthetic_id = _277ca_synthetic_source_batch_id(icn)
|
|
|
|
accepted = sum(1 for s in result.claim_statuses if s.classification == "accepted")
|
|
paid = sum(1 for s in result.claim_statuses if s.classification == "paid")
|
|
rejected = sum(1 for s in result.claim_statuses if s.classification == "rejected")
|
|
pended = sum(1 for s in result.claim_statuses if s.classification == "pended")
|
|
|
|
# Persist the 277CA row first so we have an id to attach to claims.
|
|
# SP25: thread the event bus so ``two77ca_ack_received`` fires.
|
|
row = store.add_277ca_ack(
|
|
source_batch_id=synthetic_id,
|
|
control_number=icn,
|
|
accepted_count=accepted,
|
|
rejected_count=rejected,
|
|
paid_count=paid,
|
|
pended_count=pended,
|
|
raw_json=json.loads(result.model_dump_json()),
|
|
event_bus=request.app.state.event_bus,
|
|
)
|
|
|
|
# Stamp payer-rejection fields on matching claims. The 277CA's
|
|
# REF*1K carries the patient's claim control number we sent in
|
|
# CLM01 — same convention the 999 ACK uses, so the lookup hits
|
|
# Claim.patient_control_number (mirrors apply_999_rejections).
|
|
with db.SessionLocal()() as session:
|
|
def _lookup(pcn: str):
|
|
return (
|
|
session.query(Claim)
|
|
.filter(Claim.patient_control_number == pcn)
|
|
.first()
|
|
)
|
|
apply_result = apply_277ca_rejections(
|
|
session, result, claim_lookup=_lookup, two77ca_id=row.id,
|
|
)
|
|
if apply_result.matched:
|
|
bus = request.app.state.event_bus
|
|
for cid in apply_result.matched:
|
|
await bus.publish("claim.payer_rejected", {"claim_id": cid})
|
|
# SP11: audit trail for each payer-rejected claim.
|
|
with db.SessionLocal()() as audit_s:
|
|
for cid in apply_result.matched:
|
|
append_event(audit_s, AuditEvent(
|
|
event_type="claim.payer_rejected",
|
|
entity_type="claim",
|
|
entity_id=cid,
|
|
payload={
|
|
"source_batch_id": synthetic_id,
|
|
"277ca_id": row.id,
|
|
},
|
|
actor="277ca-parser",
|
|
user_id=_actor_user_id(request),
|
|
))
|
|
audit_s.commit()
|
|
if apply_result.orphans:
|
|
log.warning(
|
|
"277CA had %d orphan status entries (no matching claim): %s",
|
|
len(apply_result.orphans),
|
|
apply_result.orphans[:5],
|
|
)
|
|
|
|
# SP28: auto-link the 277CA ClaimStatus entries to claims via
|
|
# the D10 two-pass join. Each ClaimAck row publishes
|
|
# claim_ack_written so the live-tail subscribers on the claim
|
|
# and ack side see the link immediately.
|
|
claim_ack_links_count = 0
|
|
with db.SessionLocal()() as link_s:
|
|
batch_index = store.batch_envelope_index()
|
|
|
|
def _pcn_lookup(pcn: str):
|
|
return (
|
|
link_s.query(Claim)
|
|
.filter(Claim.patient_control_number == pcn)
|
|
.first()
|
|
)
|
|
|
|
link_result = _apply_277ca_acks(
|
|
link_s, result, ack_id=row.id,
|
|
batch_envelope_index=batch_index,
|
|
pc_claim_lookup=_pcn_lookup,
|
|
)
|
|
for link_row in link_result.linked:
|
|
store.add_claim_ack(
|
|
claim_id=link_row.claim_id,
|
|
batch_id=link_row.batch_id,
|
|
ack_id=row.id,
|
|
ack_kind="277ca",
|
|
ak2_index=link_row.ak2_index,
|
|
set_control_number=link_row.set_control_number,
|
|
set_accept_reject_code=link_row.set_accept_reject_code,
|
|
linked_by="auto",
|
|
event_bus=request.app.state.event_bus,
|
|
)
|
|
claim_ack_links_count += 1
|
|
|
|
return JSONResponse(content={
|
|
"ack": {
|
|
"id": row.id,
|
|
"control_number": icn,
|
|
"accepted_count": accepted,
|
|
"rejected_count": rejected,
|
|
"paid_count": paid,
|
|
"pended_count": pended,
|
|
"source_batch_id": synthetic_id,
|
|
"matched_claim_ids": apply_result.matched,
|
|
"orphan_status_codes": apply_result.orphans,
|
|
"claim_ack_links_count": claim_ack_links_count,
|
|
},
|
|
"parsed": json.loads(result.model_dump_json()),
|
|
}) |