feat(backend): 835 ERA parser with BPR/CLP/SVC balancing, CO Medicaid config, CLI, and FastAPI route
This commit is contained in:
Tyler
+142
-10
@@ -1,13 +1,13 @@
|
||||
"""FastAPI HTTP surface for the Cyclone 837P parser.
|
||||
"""FastAPI HTTP surface for the Cyclone 837P and 835 ERA parsers.
|
||||
|
||||
The frontend (Vite/React on http://localhost:5173) uploads an X12 file via
|
||||
``POST /api/parse-837`` and receives either:
|
||||
``POST /api/parse-837`` or ``POST /api/parse-835`` and receives either:
|
||||
|
||||
* a single ``ParseResult`` JSON object when it sends ``Accept: application/json``,
|
||||
or
|
||||
* a stream of newline-delimited JSON objects (``application/x-ndjson``) — one
|
||||
envelope, one line per claim, then one summary — so the UI can render
|
||||
claims incrementally as they're produced.
|
||||
* a single ``ParseResult`` (or ``ParseResult835``) JSON object when it
|
||||
sends ``Accept: application/json``, or
|
||||
* a stream of newline-delimited JSON objects (``application/x-ndjson``)
|
||||
— one envelope, one line per claim/payout, then one summary — so the
|
||||
UI can render records incrementally as they're produced.
|
||||
|
||||
CORS is configured to allow the Vite dev origin (``http://localhost:5173``)
|
||||
plus GET/POST with any header.
|
||||
@@ -27,8 +27,11 @@ from pydantic import ValidationError
|
||||
from cyclone import __version__
|
||||
from cyclone.parsers.exceptions import CycloneParseError
|
||||
from cyclone.parsers.models import ClaimOutput, ParseResult
|
||||
from cyclone.parsers.payer import PayerConfig
|
||||
from cyclone.parsers.models_835 import ParseResult835
|
||||
from cyclone.parsers.payer import PayerConfig, PayerConfig835
|
||||
from cyclone.parsers.parse_837 import parse
|
||||
from cyclone.parsers.parse_835 import parse as parse_835
|
||||
from cyclone.parsers.validator_835 import validate as validate_835
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
@@ -40,12 +43,17 @@ PAYER_FACTORIES: dict[str, Any] = {
|
||||
"generic_837p": PayerConfig.generic_837p,
|
||||
}
|
||||
|
||||
PAYER_FACTORIES_835: dict[str, Any] = {
|
||||
"co_medicaid_835": PayerConfig835.co_medicaid_835,
|
||||
"generic_835": PayerConfig835.generic_835,
|
||||
}
|
||||
|
||||
VITE_DEV_ORIGIN = "http://localhost:5173"
|
||||
|
||||
app = FastAPI(
|
||||
title="Cyclone 837P Parser API",
|
||||
title="Cyclone 837P / 835 Parser API",
|
||||
version=__version__,
|
||||
description="Upload X12 837P files and receive parsed claims as JSON or NDJSON.",
|
||||
description="Upload X12 837P and 835 files and receive parsed records as JSON or NDJSON.",
|
||||
)
|
||||
|
||||
app.add_middleware(
|
||||
@@ -69,6 +77,18 @@ def _resolve_payer(name: str) -> PayerConfig:
|
||||
return PAYER_FACTORIES[name]()
|
||||
|
||||
|
||||
def _resolve_payer_835(name: str) -> PayerConfig835:
|
||||
if name not in PAYER_FACTORIES_835:
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail={
|
||||
"error": "Unknown payer",
|
||||
"detail": f"Unknown payer {name!r}. Choose from: {', '.join(PAYER_FACTORIES_835)}",
|
||||
},
|
||||
)
|
||||
return PAYER_FACTORIES_835[name]()
|
||||
|
||||
|
||||
def _strict_rewrite(result: ParseResult) -> ParseResult:
|
||||
"""Promote warnings to errors (mirrors the CLI's --strict)."""
|
||||
claims: list[ClaimOutput] = []
|
||||
@@ -201,4 +221,116 @@ def _ndjson_stream(result: ParseResult) -> Iterator[bytes]:
|
||||
yield (json.dumps({"type": "summary", "data": json.loads(result.summary.model_dump_json())}) + "\n").encode("utf-8")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# 835 ERA (Health Care Claim Payment/Advice)
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def _strict_rewrite_835(result: ParseResult835) -> ParseResult835:
|
||||
"""Promote warnings to errors (mirrors the CLI's --strict)."""
|
||||
if result.validation is None:
|
||||
return result
|
||||
report = result.validation
|
||||
promoted = [i.model_copy(update={"severity": "error"}) for i in report.warnings]
|
||||
new_errors = report.errors + promoted
|
||||
new_report = report.model_copy(update={"errors": new_errors, "passed": not new_errors})
|
||||
passed = 1 if new_report.passed else 0
|
||||
failed = 1 if not new_report.passed else 0
|
||||
new_summary = result.summary.model_copy(
|
||||
update={"passed": passed, "failed": failed}
|
||||
)
|
||||
return result.model_copy(update={"validation": new_report, "summary": new_summary})
|
||||
|
||||
|
||||
def _drop_raw_segments_835(result: ParseResult835) -> ParseResult835:
|
||||
"""Return a copy of ``result`` with ``raw_segments`` cleared on every claim."""
|
||||
claims = [c.model_copy(update={"raw_segments": []}) for c in result.claims]
|
||||
return result.model_copy(update={"claims": claims})
|
||||
|
||||
|
||||
def _has_835_validation_errors(result: ParseResult835) -> bool:
|
||||
return result.validation is not None and not result.validation.passed
|
||||
|
||||
|
||||
@app.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)
|
||||
|
||||
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)},
|
||||
)
|
||||
|
||||
# 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.
|
||||
report = validate_835(result, config)
|
||||
passed = 1 if report.passed else 0
|
||||
failed = 1 if not report.passed else 0
|
||||
result = result.model_copy(update={
|
||||
"validation": report,
|
||||
"summary": result.summary.model_copy(update={"passed": passed, "failed": failed}),
|
||||
})
|
||||
|
||||
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()),
|
||||
)
|
||||
|
||||
if _client_wants_json(request):
|
||||
return JSONResponse(content=json.loads(result.model_dump_json()))
|
||||
|
||||
# Default: NDJSON stream.
|
||||
return StreamingResponse(
|
||||
_ndjson_stream_835(result),
|
||||
media_type="application/x-ndjson",
|
||||
)
|
||||
|
||||
|
||||
def _ndjson_stream_835(result: ParseResult835) -> Iterator[bytes]:
|
||||
"""Yield one JSON object per line: envelope → financial → trace → payer → payee → claim_payments → summary."""
|
||||
yield (json.dumps({"type": "envelope", "data": json.loads(result.envelope.model_dump_json())}) + "\n").encode("utf-8")
|
||||
yield (json.dumps({"type": "financial_info", "data": json.loads(result.financial_info.model_dump_json())}) + "\n").encode("utf-8")
|
||||
yield (json.dumps({"type": "trace", "data": json.loads(result.trace.model_dump_json())}) + "\n").encode("utf-8")
|
||||
yield (json.dumps({"type": "payer", "data": json.loads(result.payer.model_dump_json())}) + "\n").encode("utf-8")
|
||||
yield (json.dumps({"type": "payee", "data": json.loads(result.payee.model_dump_json())}) + "\n").encode("utf-8")
|
||||
for claim in result.claims:
|
||||
yield (json.dumps({"type": "claim_payment", "data": json.loads(claim.model_dump_json())}) + "\n").encode("utf-8")
|
||||
yield (json.dumps({"type": "summary", "data": json.loads(result.summary.model_dump_json())}) + "\n").encode("utf-8")
|
||||
|
||||
|
||||
__all__ = ["app"]
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
"""Click-based CLI for the Cyclone 837P parser."""
|
||||
"""Click-based CLI for the Cyclone 837P and 835 ERA parsers."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
@@ -9,9 +9,12 @@ from pathlib import Path
|
||||
import click
|
||||
|
||||
from cyclone.parsers.exceptions import CycloneParseError, CycloneValidationError
|
||||
from cyclone.parsers.payer import PayerConfig
|
||||
from cyclone.parsers.parse_837 import parse
|
||||
from cyclone.parsers.payer import PayerConfig, PayerConfig835
|
||||
from cyclone.parsers.parse_837 import parse as parse_837_text
|
||||
from cyclone.parsers.parse_835 import parse as parse_835_text
|
||||
from cyclone.parsers.validator_835 import validate as validate_835
|
||||
from cyclone.parsers.writer import write_outputs
|
||||
from cyclone.parsers.writer_835 import write_outputs_835
|
||||
|
||||
|
||||
_PAYER_FACTORIES = {
|
||||
@@ -19,6 +22,11 @@ _PAYER_FACTORIES = {
|
||||
"generic_837p": PayerConfig.generic_837p,
|
||||
}
|
||||
|
||||
_PAYER_FACTORIES_835 = {
|
||||
"co_medicaid_835": PayerConfig835.co_medicaid_835,
|
||||
"generic_835": PayerConfig835.generic_835,
|
||||
}
|
||||
|
||||
|
||||
def _payer(name: str) -> PayerConfig:
|
||||
if name not in _PAYER_FACTORIES:
|
||||
@@ -26,6 +34,12 @@ def _payer(name: str) -> PayerConfig:
|
||||
return _PAYER_FACTORIES[name]()
|
||||
|
||||
|
||||
def _payer_835(name: str) -> PayerConfig835:
|
||||
if name not in _PAYER_FACTORIES_835:
|
||||
raise click.BadParameter(f"Unknown payer {name!r}. Choose from: {', '.join(_PAYER_FACTORIES_835)}")
|
||||
return _PAYER_FACTORIES_835[name]()
|
||||
|
||||
|
||||
@click.group()
|
||||
def main() -> None:
|
||||
"""Cyclone EDI suite — X12 parser."""
|
||||
@@ -55,7 +69,7 @@ def parse_837(
|
||||
config = _payer(payer)
|
||||
|
||||
try:
|
||||
result = parse(text, config, input_file=str(input_file))
|
||||
result = parse_837_text(text, config, input_file=str(input_file))
|
||||
except CycloneParseError as exc:
|
||||
click.echo(f"PARSE ERROR: {exc}", err=True)
|
||||
sys.exit(2)
|
||||
@@ -103,5 +117,74 @@ def parse_837(
|
||||
)
|
||||
|
||||
|
||||
@main.command("parse-835")
|
||||
@click.argument("input_file", type=click.Path(exists=True, dir_okay=False, path_type=Path))
|
||||
@click.option("--output-dir", required=True, type=click.Path(file_okay=False, path_type=Path))
|
||||
@click.option("--payer", default="co_medicaid_835", show_default=True, help="Payer configuration preset")
|
||||
@click.option("--strict", is_flag=True, help="Promote warnings to errors")
|
||||
@click.option("--include-raw-segments / --no-raw-segments", default=True, help="Embed raw segments in output JSON")
|
||||
@click.option("--log-level", default="INFO", show_default=True, type=click.Choice(["DEBUG", "INFO", "WARNING", "ERROR"]))
|
||||
def parse_835(
|
||||
input_file: Path,
|
||||
output_dir: Path,
|
||||
payer: str,
|
||||
strict: bool,
|
||||
include_raw_segments: bool,
|
||||
log_level: str,
|
||||
) -> None:
|
||||
"""Parse an X12 835 ERA file into one JSON per claim payment."""
|
||||
logging.basicConfig(level=getattr(logging, log_level))
|
||||
|
||||
text = input_file.read_text()
|
||||
config = _payer_835(payer)
|
||||
|
||||
try:
|
||||
result = parse_835_text(text, config, input_file=str(input_file))
|
||||
except CycloneParseError as exc:
|
||||
click.echo(f"PARSE ERROR: {exc}", err=True)
|
||||
sys.exit(2)
|
||||
|
||||
if not result.claims:
|
||||
raise click.UsageError("No claim payments found in file") from None
|
||||
|
||||
# Run validation; populate summary.passed/failed from the report.
|
||||
report = validate_835(result, config)
|
||||
if strict:
|
||||
promoted = [i.model_copy(update={"severity": "error"}) for i in report.warnings]
|
||||
new_errors = report.errors + promoted
|
||||
report = report.model_copy(update={"errors": new_errors, "passed": not new_errors})
|
||||
passed = 1 if report.passed else 0
|
||||
failed = 1 if not report.passed else 0
|
||||
result = result.model_copy(update={"validation": report})
|
||||
result.summary = result.summary.model_copy(update={
|
||||
"passed": passed,
|
||||
"failed": failed,
|
||||
"issues_by_rule": _count_issues(report),
|
||||
})
|
||||
|
||||
# Drop raw_segments if disabled
|
||||
if not include_raw_segments:
|
||||
result = result.model_copy(update={"claims": [c.model_copy(update={"raw_segments": []}) for c in result.claims]})
|
||||
|
||||
write_outputs_835(result, output_dir)
|
||||
result.summary = result.summary.model_copy(update={"output_dir": str(output_dir)})
|
||||
(output_dir / "summary.json").write_text(result.summary.model_dump_json(indent=2))
|
||||
|
||||
click.echo(
|
||||
f"parsed={result.summary.total_claims} "
|
||||
f"passed={result.summary.passed} "
|
||||
f"failed={result.summary.failed} "
|
||||
f"output={output_dir}"
|
||||
)
|
||||
|
||||
|
||||
def _count_issues(report) -> dict[str, int]:
|
||||
"""Build an issues_by_rule dict from a ValidationReport (for the summary)."""
|
||||
counts: dict[str, int] = {}
|
||||
for issue in report.errors + report.warnings:
|
||||
counts[issue.rule] = counts.get(issue.rule, 0) + 1
|
||||
return counts
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
@@ -15,7 +15,7 @@ _LAZY_EXPORTS: dict[str, str] = {
|
||||
# exceptions
|
||||
"CycloneParseError": "cyclone.parsers.exceptions",
|
||||
"CycloneValidationError": "cyclone.parsers.exceptions",
|
||||
# models
|
||||
# models (837P)
|
||||
"Address": "cyclone.parsers.models",
|
||||
"BatchSummary": "cyclone.parsers.models",
|
||||
"BillingProvider": "cyclone.parsers.models",
|
||||
@@ -30,14 +30,28 @@ _LAZY_EXPORTS: dict[str, str] = {
|
||||
"Subscriber": "cyclone.parsers.models",
|
||||
"ValidationIssue": "cyclone.parsers.models",
|
||||
"ValidationReport": "cyclone.parsers.models",
|
||||
# models (835)
|
||||
"ClaimAdjustment": "cyclone.parsers.models_835",
|
||||
"ClaimPayment": "cyclone.parsers.models_835",
|
||||
"FinancialInfo": "cyclone.parsers.models_835",
|
||||
"ParseResult835": "cyclone.parsers.models_835",
|
||||
"Payer835": "cyclone.parsers.models_835",
|
||||
"Payee835": "cyclone.parsers.models_835",
|
||||
"ReassociationTrace": "cyclone.parsers.models_835",
|
||||
"ServicePayment": "cyclone.parsers.models_835",
|
||||
"claim_status_label": "cyclone.parsers.models_835",
|
||||
# payer
|
||||
"PayerConfig": "cyclone.parsers.payer",
|
||||
"PayerConfig835": "cyclone.parsers.payer",
|
||||
# segments
|
||||
"Delimiters": "cyclone.parsers.segments",
|
||||
"tokenize": "cyclone.parsers.segments",
|
||||
# orchestrator
|
||||
# orchestrators
|
||||
"parse": "cyclone.parsers.parse_837",
|
||||
"parse_835": "cyclone.parsers.parse_835",
|
||||
# writers
|
||||
"write_outputs": "cyclone.parsers.writer",
|
||||
"write_outputs_835": "cyclone.parsers.writer_835",
|
||||
}
|
||||
|
||||
_cache: dict[str, object] = {}
|
||||
|
||||
@@ -0,0 +1,218 @@
|
||||
"""Pydantic v2 models for parsed 835 ERA (Health Care Claim Payment/Advice) files.
|
||||
|
||||
Models mirror the X12 835 005010X221A1 structure:
|
||||
|
||||
- ``Envelope`` (ISA/GS/ST) — reused from :mod:`cyclone.parsers.models`
|
||||
- ``FinancialInfo`` (BPR)
|
||||
- ``ReassociationTrace`` (TRN)
|
||||
- ``Payer835`` / ``Payee835`` (Loop 1000A / 1000B)
|
||||
- ``ClaimPayment`` (Loop 2100) with embedded ``ServicePayment`` (Loop 2110)
|
||||
- ``ParseResult835`` (batch)
|
||||
|
||||
We copy the ``_Base`` model from :mod:`cyclone.parsers.models` because it
|
||||
carries the custom ``@model_serializer`` that converts ``date`` to ``str``
|
||||
on JSON output — a behavior the 835 models must match so the FastAPI
|
||||
output is consistent across transactions.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import date
|
||||
from decimal import Decimal
|
||||
from typing import Any, Literal
|
||||
|
||||
from pydantic import BaseModel, ConfigDict, Field, model_serializer, model_validator
|
||||
|
||||
from cyclone.parsers.models import Address, BatchSummary, Envelope, ValidationIssue, ValidationReport
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Shared base
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
class _Base(BaseModel):
|
||||
"""Shared Pydantic base; matches the 837P models for JSON consistency."""
|
||||
|
||||
model_config = ConfigDict(extra="ignore", str_strip_whitespace=True)
|
||||
|
||||
@model_serializer(mode="wrap")
|
||||
def _serialize(self, handler): # type: ignore[no-untyped-def]
|
||||
data = handler(self)
|
||||
for key, value in data.items():
|
||||
if isinstance(value, date):
|
||||
data[key] = value.isoformat()
|
||||
return data
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# CLP02 Claim Status Code → human-readable label
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
# Source: X12 005010X221A1 element 1029 (Claim Status Code).
|
||||
# In v1 we keep the subset a 835 ERA actually carries; status_code "25"
|
||||
# (predetermination pricing only — no payment) is included per the spec.
|
||||
_CLP_STATUS_LABELS: dict[str, str] = {
|
||||
"1": "Processed as Primary",
|
||||
"2": "Processed as Secondary",
|
||||
"3": "Pending",
|
||||
"4": "Denied",
|
||||
"19": "Processed Primary, Forwarded",
|
||||
"20": "Processed Secondary, Forwarded",
|
||||
"21": "Reversal of Previous Payment",
|
||||
"22": "Reversal of Previous",
|
||||
"23": "Not Our Claim, Forwarded",
|
||||
"25": "Predetermination Pricing Only",
|
||||
}
|
||||
|
||||
|
||||
def claim_status_label(code: str) -> str | None:
|
||||
"""Return a human-readable label for a CLP02 code, or ``None`` if unknown."""
|
||||
return _CLP_STATUS_LABELS.get(code)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Header
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
class FinancialInfo(_Base):
|
||||
"""BPR — Financial Information."""
|
||||
|
||||
handling_code: str
|
||||
paid_amount: Decimal
|
||||
credit_debit_flag: str
|
||||
payment_method: str | None = None
|
||||
payer_tax_id: str | None = None
|
||||
payment_date: date | None = None
|
||||
|
||||
|
||||
class ReassociationTrace(_Base):
|
||||
"""TRN — Reassociation Trace Number."""
|
||||
|
||||
trace_type_code: str
|
||||
trace_number: str
|
||||
originating_company_id: str
|
||||
payer_id: str | None = None
|
||||
|
||||
|
||||
class Payer835(_Base):
|
||||
"""Loop 1000A — Payer Identification (NM1*PR)."""
|
||||
|
||||
name: str
|
||||
id: str | None = None
|
||||
address: Address | None = None
|
||||
contact_url: str | None = None # PER04 per the CO Medicaid guide
|
||||
|
||||
|
||||
class Payee835(_Base):
|
||||
"""Loop 1000B — Payee Identification (NM1*PE)."""
|
||||
|
||||
name: str
|
||||
npi: str
|
||||
address: Address | None = None
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Claim / Service payment
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
class ClaimAdjustment(_Base):
|
||||
"""CAS — Claim Adjustment (one row per reason/amount pair)."""
|
||||
|
||||
group_code: str
|
||||
reason_code: str
|
||||
amount: Decimal
|
||||
quantity: Decimal | None = None
|
||||
|
||||
|
||||
class ServicePayment(_Base):
|
||||
"""Loop 2110 — Service Payment Information (SVC composite)."""
|
||||
|
||||
line_number: int
|
||||
procedure_qualifier: str
|
||||
procedure_code: str
|
||||
modifiers: list[str] = Field(default_factory=list)
|
||||
charge: Decimal
|
||||
payment: Decimal
|
||||
units: Decimal | None = None
|
||||
unit_type: str | None = None
|
||||
service_date: date | None = None
|
||||
adjustments: list[ClaimAdjustment] = Field(default_factory=list)
|
||||
ref_benefit_plan: str | None = None # REF*0K per the CO guide
|
||||
|
||||
|
||||
class ClaimPayment(_Base):
|
||||
"""Loop 2100 — Claim Payment Information (CLP composite)."""
|
||||
|
||||
payer_claim_control_number: str
|
||||
status_code: str
|
||||
status_label: str | None = None
|
||||
total_charge: Decimal
|
||||
total_paid: Decimal
|
||||
patient_responsibility: Decimal | None = None
|
||||
claim_filing_indicator: str | None = None # CLP06 (e.g. "MC" for Medicaid)
|
||||
original_claim_id: str | None = None # CLP07 — original 837P CLM01
|
||||
facility_type: str | None = None # CLP08 — original 837P CLM05-1
|
||||
frequency_code: str | None = None # CLP09 — original 837P CLM05-3
|
||||
per_diem_covered_days: Decimal | None = None # QTY*CA per the CO guide
|
||||
ref_benefit_plan: str | None = None # REF*CE per the CO guide
|
||||
service_payments: list[ServicePayment] = Field(default_factory=list)
|
||||
raw_segments: list[list[str]] = Field(default_factory=list)
|
||||
|
||||
@model_validator(mode="before")
|
||||
@classmethod
|
||||
def _decode_status_label(cls, data: Any) -> Any:
|
||||
"""Populate ``status_label`` from ``status_code`` when caller omitted it.
|
||||
|
||||
Callers that explicitly set ``status_label`` (e.g. round-tripping from
|
||||
JSON) keep their value.
|
||||
"""
|
||||
if isinstance(data, dict) and "status_label" not in data and "status_code" in data:
|
||||
data = {**data, "status_label": claim_status_label(str(data["status_code"]))}
|
||||
return data
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Top-level result
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
class ParseResult835(_Base):
|
||||
envelope: Envelope
|
||||
financial_info: FinancialInfo
|
||||
trace: ReassociationTrace
|
||||
payer: Payer835
|
||||
payee: Payee835
|
||||
claims: list[ClaimPayment] = Field(default_factory=list)
|
||||
summary: BatchSummary
|
||||
validation: ValidationReport | None = None # populated by the orchestrator
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Re-exports
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
__all__ = [
|
||||
"Address",
|
||||
"BatchSummary",
|
||||
"ClaimAdjustment",
|
||||
"ClaimPayment",
|
||||
"Envelope",
|
||||
"FinancialInfo",
|
||||
"ParseResult835",
|
||||
"Payer835",
|
||||
"Payee835",
|
||||
"ReassociationTrace",
|
||||
"ServicePayment",
|
||||
"ValidationIssue",
|
||||
"ValidationReport",
|
||||
"claim_status_label",
|
||||
]
|
||||
|
||||
|
||||
# Silence the "imported but unused" complaint on a few names that exist for
|
||||
# parity with the 837P module's public surface but are not yet referenced
|
||||
# in this file's implementation.
|
||||
_ = (Literal, BaseModel)
|
||||
@@ -0,0 +1,540 @@
|
||||
r"""Orchestrate parsing an X12 835 ERA file into a :class:`ParseResult835`.
|
||||
|
||||
Single-pass walker over the tokenized segment list:
|
||||
|
||||
- ISA / GS / ST / BPR / TRN — envelope + financial + trace
|
||||
- Loop 1000A (N1*PR + N3/N4/PER/REF) — payer
|
||||
- Loop 1000B (N1*PE + N3/N4) — payee
|
||||
- Loop 2100 (LX + CLP + CAS + NM1 + REF + DTM + QTY) — claim payment
|
||||
- Loop 2110 (SVC + CAS + REF + DTM + LQ) — service payment
|
||||
|
||||
Errors at the file level raise :class:`CycloneParseError`. Per-segment
|
||||
deficiencies are recorded in :class:`ValidationReport` and never abort
|
||||
the whole batch.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from datetime import date
|
||||
from decimal import Decimal
|
||||
from typing import Any
|
||||
|
||||
from cyclone.parsers.exceptions import CycloneParseError
|
||||
from cyclone.parsers.models import Address, BatchSummary, Envelope
|
||||
from cyclone.parsers.models_835 import (
|
||||
ClaimAdjustment,
|
||||
ClaimPayment,
|
||||
FinancialInfo,
|
||||
ParseResult835,
|
||||
Payer835,
|
||||
Payee835,
|
||||
ReassociationTrace,
|
||||
ServicePayment,
|
||||
claim_status_label,
|
||||
)
|
||||
from cyclone.parsers.payer import PayerConfig835
|
||||
from cyclone.parsers.segments import tokenize
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Date parsing
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def _parse_ccyymmdd(yymmdd: str) -> date:
|
||||
"""Parse a CCYYMMDD (8-char) or YY?? (6-char) date string."""
|
||||
if not yymmdd:
|
||||
raise ValueError("empty date")
|
||||
if len(yymmdd) == 8:
|
||||
return date(int(yymmdd[0:4]), int(yymmdd[4:6]), int(yymmdd[6:8]))
|
||||
if len(yymmdd) == 6:
|
||||
yy = int(yymmdd[0:2])
|
||||
mm = int(yymmdd[2:4])
|
||||
dd = int(yymmdd[4:6])
|
||||
# Pivot year 30: 00-29 = 2000s, 30-99 = 1900s.
|
||||
year = 2000 + yy if yy < 30 else 1900 + yy
|
||||
return date(year, mm, dd)
|
||||
raise ValueError(f"unexpected date length: {yymmdd!r}")
|
||||
|
||||
|
||||
def _safe_date(s: str) -> date | None:
|
||||
try:
|
||||
return _parse_ccyymmdd(s) if s else None
|
||||
except (ValueError, IndexError):
|
||||
return None
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Envelope / BPR / TRN
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def _build_envelope(segments: list[list[str]], input_file: str) -> tuple[Envelope, BatchSummary]:
|
||||
"""Build the 835 envelope from ISA/GS/ST. Raises on missing/malformed ISA."""
|
||||
summary = BatchSummary(input_file=input_file)
|
||||
envelope: Envelope | None = None
|
||||
for seg in segments:
|
||||
if seg[0] == "ISA":
|
||||
try:
|
||||
envelope = Envelope(
|
||||
sender_id=seg[6].strip(),
|
||||
receiver_id=seg[8].strip(),
|
||||
control_number=seg[13].strip(),
|
||||
transaction_date=date(2026, 1, 1), # overwritten by BPR
|
||||
implementation_guide=None,
|
||||
)
|
||||
except (IndexError, ValueError) as exc:
|
||||
raise CycloneParseError(f"Bad ISA: {exc}") from exc
|
||||
elif seg[0] == "ST" and envelope is not None:
|
||||
if len(seg) > 3:
|
||||
envelope = envelope.model_copy(update={"implementation_guide": seg[3]})
|
||||
if envelope is None:
|
||||
raise CycloneParseError("No ISA envelope found")
|
||||
return envelope, summary
|
||||
|
||||
|
||||
def _consume_bpr(segments: list[list[str]], idx: int) -> tuple[FinancialInfo, int]:
|
||||
seg = segments[idx]
|
||||
try:
|
||||
handling = seg[1] if len(seg) > 1 else ""
|
||||
paid = Decimal(seg[2]) if len(seg) > 2 and seg[2] else Decimal("0")
|
||||
credit_debit = seg[3] if len(seg) > 3 else ""
|
||||
method = seg[4] if len(seg) > 4 and seg[4] else None
|
||||
# BPR10 = Payer Identifier (Origin Company ID), per the CO Medicaid
|
||||
# guide: "81-1725341" (TXIX) or "84-0644739" (BHA).
|
||||
payer_tax = seg[10] if len(seg) > 10 and seg[10] else None
|
||||
# BPR16 = Effective Entry Date (CCYYMMDD). The segment is sparse —
|
||||
# positions 11..15 (BPR11..BPR15) are usually empty, so we look for
|
||||
# any 8-digit CCYYMMDD token from index 11 onward.
|
||||
payment_date = None
|
||||
for k in range(11, len(seg)):
|
||||
if seg[k] and len(seg[k]) == 8 and seg[k].isdigit():
|
||||
payment_date = _safe_date(seg[k])
|
||||
if payment_date is not None:
|
||||
break
|
||||
except (IndexError, ValueError) as exc:
|
||||
raise CycloneParseError(f"Bad BPR: {exc}", segment_index=idx) from exc
|
||||
return (
|
||||
FinancialInfo(
|
||||
handling_code=handling,
|
||||
paid_amount=paid,
|
||||
credit_debit_flag=credit_debit,
|
||||
payment_method=method,
|
||||
payer_tax_id=payer_tax,
|
||||
payment_date=payment_date,
|
||||
),
|
||||
idx + 1,
|
||||
)
|
||||
|
||||
|
||||
def _consume_trn(segments: list[list[str]], idx: int) -> tuple[ReassociationTrace, int]:
|
||||
seg = segments[idx]
|
||||
trace_type = seg[1] if len(seg) > 1 else ""
|
||||
trace_number = seg[2] if len(seg) > 2 else ""
|
||||
originating = seg[3] if len(seg) > 3 else ""
|
||||
payer_id = seg[4] if len(seg) > 4 and seg[4] else None
|
||||
return (
|
||||
ReassociationTrace(
|
||||
trace_type_code=trace_type,
|
||||
trace_number=trace_number,
|
||||
originating_company_id=originating,
|
||||
payer_id=payer_id,
|
||||
),
|
||||
idx + 1,
|
||||
)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Loop 1000A (payer) / 1000B (payee)
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def _consume_address(segments: list[list[str]], idx: int) -> tuple[Address | None, int]:
|
||||
"""Walk N3/N4 (if present) and return the assembled Address, plus new idx."""
|
||||
addr: Address | None = None
|
||||
while idx < len(segments) and segments[idx][0] in {"N3", "N4"}:
|
||||
seg = segments[idx]
|
||||
if seg[0] == "N3":
|
||||
line1 = seg[1] if len(seg) > 1 else ""
|
||||
line2 = seg[2] if len(seg) > 2 else None
|
||||
addr = (addr or Address(line1=line1, city="", state="", zip="")).model_copy(
|
||||
update={"line1": line1, "line2": line2}
|
||||
)
|
||||
elif seg[0] == "N4":
|
||||
city = seg[1] if len(seg) > 1 else ""
|
||||
state = seg[2] if len(seg) > 2 else ""
|
||||
zip_ = seg[3] if len(seg) > 3 else ""
|
||||
addr = (addr or Address(line1="", city=city, state=state, zip=zip_)).model_copy(
|
||||
update={"city": city, "state": state, "zip": zip_}
|
||||
)
|
||||
idx += 1
|
||||
return addr, idx
|
||||
|
||||
|
||||
def _consume_1000a(segments: list[list[str]], idx: int) -> tuple[Payer835, int]:
|
||||
"""Read N1*PR + N3 + N4 + PER* + REF* up to next N1 or CLP/LX/SE."""
|
||||
name = ""
|
||||
pid: str | None = None
|
||||
contact_url: str | None = None
|
||||
# First segment is N1*PR
|
||||
seg = segments[idx]
|
||||
if seg[0] == "N1" and len(seg) > 1 and seg[1] == "PR":
|
||||
# NM103 is the name (or organization name for non-person entities).
|
||||
name = seg[3] if len(seg) > 3 and seg[3] else (seg[2] if len(seg) > 2 else "")
|
||||
# NM108/NM109 carry the ID code qualifier and value. The CO guide
|
||||
# documents XV for the Health Plan ID, but other payers may use
|
||||
# PI / 24 / etc., so we accept any qualifier.
|
||||
for j, qual in enumerate(seg):
|
||||
if qual in {"XV", "PI", "24"} and j + 1 < len(seg):
|
||||
pid = seg[j + 1]
|
||||
break
|
||||
idx += 1
|
||||
addr: Address | None = None
|
||||
while idx < len(segments) and segments[idx][0] not in {"N1", "LX", "CLP", "SE"}:
|
||||
seg = segments[idx]
|
||||
if seg[0] in {"N3", "N4"}:
|
||||
addr, idx = _consume_address(segments, idx)
|
||||
continue
|
||||
if seg[0] == "PER":
|
||||
# PER03 is the communication number qualifier; PER04 is the
|
||||
# value. The CO guide documents PER04 as the URL but in real
|
||||
# 835s the URL is tagged with PER03=UR. We scan for the UR
|
||||
# qualifier to be safe.
|
||||
for j in range(1, len(seg) - 1, 2):
|
||||
if seg[j] == "UR" and j + 1 < len(seg):
|
||||
contact_url = seg[j + 1]
|
||||
break
|
||||
# Fall back to PER04 directly if it looks like a URL (matches
|
||||
# the literal CO guide wording).
|
||||
if contact_url is None and len(seg) > 4 and seg[4]:
|
||||
value = seg[4]
|
||||
if "@" in value or "http" in value or "." in value:
|
||||
contact_url = value
|
||||
idx += 1
|
||||
return Payer835(name=name, id=pid, address=addr, contact_url=contact_url), idx
|
||||
|
||||
|
||||
def _consume_1000b(segments: list[list[str]], idx: int) -> tuple[Payee835, int]:
|
||||
"""Read N1*PE + N3 + N4 up to next N1 or CLP/LX/SE."""
|
||||
name = ""
|
||||
npi = ""
|
||||
seg = segments[idx]
|
||||
if seg[0] == "N1" and len(seg) > 1 and seg[1] == "PE":
|
||||
name = seg[3] if len(seg) > 3 and seg[3] else (seg[2] if len(seg) > 2 else "")
|
||||
for j, qual in enumerate(seg):
|
||||
if qual == "XX" and j + 1 < len(seg):
|
||||
npi = seg[j + 1]
|
||||
break
|
||||
idx += 1
|
||||
addr: Address | None = None
|
||||
while idx < len(segments) and segments[idx][0] not in {"N1", "LX", "CLP", "SE"}:
|
||||
seg = segments[idx]
|
||||
if seg[0] in {"N3", "N4"}:
|
||||
addr, idx = _consume_address(segments, idx)
|
||||
continue
|
||||
idx += 1
|
||||
return Payee835(name=name, npi=npi, address=addr), idx
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Loop 2100 (claim payment) and 2110 (service payment)
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def _cas_segments(segments: list[list[str]], idx: int) -> tuple[list[ClaimAdjustment], int]:
|
||||
"""Read consecutive CAS segments starting at ``idx`` and return aggregated adjustments."""
|
||||
adjustments: list[ClaimAdjustment] = []
|
||||
while idx < len(segments) and segments[idx][0] == "CAS":
|
||||
seg = segments[idx]
|
||||
try:
|
||||
group = seg[1] if len(seg) > 1 else ""
|
||||
# Each CAS carries up to 6 reason/amount/quantity triplets
|
||||
# (elements 2..19). We expand each one into its own row.
|
||||
# Indexes 2,3,4 | 5,6,7 | 8,9,10 | 11,12,13 | 14,15,16 | 17,18,19
|
||||
triplet_starts = [2, 5, 8, 11, 14, 17]
|
||||
for start in triplet_starts:
|
||||
if len(seg) <= start:
|
||||
break
|
||||
reason = seg[start]
|
||||
if not reason:
|
||||
continue
|
||||
amount_s = seg[start + 1] if len(seg) > start + 1 else ""
|
||||
qty_s = seg[start + 2] if len(seg) > start + 2 else ""
|
||||
amount = Decimal(amount_s) if amount_s else Decimal("0")
|
||||
quantity = Decimal(qty_s) if qty_s else None
|
||||
adjustments.append(
|
||||
ClaimAdjustment(
|
||||
group_code=group,
|
||||
reason_code=reason,
|
||||
amount=amount,
|
||||
quantity=quantity,
|
||||
)
|
||||
)
|
||||
except (IndexError, ValueError) as exc:
|
||||
log.warning("Bad CAS at segment %d: %s", idx, exc)
|
||||
idx += 1
|
||||
return adjustments, idx
|
||||
|
||||
|
||||
def _consume_service_payment(segments: list[list[str]], idx: int, line_no: int) -> tuple[ServicePayment, int, list[list[str]]]:
|
||||
"""Read SVC + its child segments. Returns (ServicePayment, new_idx, raw_segments).
|
||||
|
||||
The third return value is the list of raw segments consumed (SVC itself
|
||||
plus any DTM/REF/CAS children) so the caller can fold them into the
|
||||
claim's raw_segments. :class:`ServicePayment` does not carry a
|
||||
``raw_segments`` field per the 835 spec, so we surface them at the
|
||||
claim level instead.
|
||||
"""
|
||||
seg = segments[idx]
|
||||
raw: list[list[str]] = [seg]
|
||||
qualifier = ""
|
||||
code = ""
|
||||
modifiers: list[str] = []
|
||||
if len(seg) > 1 and seg[1]:
|
||||
svc01 = seg[1].split(":")
|
||||
if len(svc01) > 0:
|
||||
qualifier = svc01[0]
|
||||
if len(svc01) > 1:
|
||||
code = svc01[1]
|
||||
modifiers = [m for m in svc01[2:] if m]
|
||||
charge = Decimal(seg[2]) if len(seg) > 2 and seg[2] else Decimal("0")
|
||||
payment = Decimal(seg[3]) if len(seg) > 3 and seg[3] else Decimal("0")
|
||||
units: Decimal | None = None
|
||||
unit_type: str | None = None
|
||||
if len(seg) > 4 and seg[4]:
|
||||
# In X12 835, SVC04 is "units paid" and SVC05 is the unit-of-measure.
|
||||
# SVC05 may carry a value (UN, MJ, etc.) or be empty.
|
||||
try:
|
||||
units = Decimal(seg[4])
|
||||
except Exception:
|
||||
units = None
|
||||
if len(seg) > 5 and seg[5]:
|
||||
unit_type = seg[5]
|
||||
elif len(seg) > 4 and seg[4]:
|
||||
# Some payers omit SVC05 but still mean "UN" when the units value
|
||||
# is present. Be conservative: only default to UN if no other value
|
||||
# was set above.
|
||||
if unit_type is None:
|
||||
unit_type = "UN"
|
||||
service_date: date | None = None
|
||||
ref_benefit_plan: str | None = None
|
||||
adjustments: list[ClaimAdjustment] = []
|
||||
|
||||
idx += 1
|
||||
while idx < len(segments) and segments[idx][0] not in {"SVC", "LX", "CLP", "SE"}:
|
||||
s = segments[idx]
|
||||
raw.append(s)
|
||||
if s[0] == "DTM" and len(s) > 2 and s[1] == "472":
|
||||
service_date = _safe_date(s[2]) if s[2] else None
|
||||
elif s[0] == "REF" and len(s) > 2 and s[1] == "0K":
|
||||
ref_benefit_plan = s[2]
|
||||
elif s[0] == "CAS":
|
||||
cas, idx = _cas_segments(segments, idx)
|
||||
raw.extend(c[0] for c in cas if False) # CAS rows are already in adjustments; nothing more to do
|
||||
adjustments.extend(cas)
|
||||
continue
|
||||
idx += 1
|
||||
|
||||
return (
|
||||
ServicePayment(
|
||||
line_number=line_no,
|
||||
procedure_qualifier=qualifier,
|
||||
procedure_code=code,
|
||||
modifiers=modifiers,
|
||||
charge=charge,
|
||||
payment=payment,
|
||||
units=units,
|
||||
unit_type=unit_type,
|
||||
service_date=service_date,
|
||||
adjustments=adjustments,
|
||||
ref_benefit_plan=ref_benefit_plan,
|
||||
),
|
||||
idx,
|
||||
raw,
|
||||
)
|
||||
|
||||
|
||||
def _consume_claim_payment(segments: list[list[str]], idx: int) -> tuple[ClaimPayment, int]:
|
||||
"""Read one CLP and all its child segments (CAS / NM1 / REF / DTM / QTY / SVC)."""
|
||||
clp = segments[idx]
|
||||
raw: list[list[str]] = [clp]
|
||||
try:
|
||||
pcn = clp[1] if len(clp) > 1 else ""
|
||||
status = clp[2] if len(clp) > 2 else ""
|
||||
total_charge = Decimal(clp[3]) if len(clp) > 3 and clp[3] else Decimal("0")
|
||||
total_paid = Decimal(clp[4]) if len(clp) > 4 and clp[4] else Decimal("0")
|
||||
patient_resp = Decimal(clp[5]) if len(clp) > 5 and clp[5] else None
|
||||
claim_filing = clp[6] if len(clp) > 6 and clp[6] else None
|
||||
original_claim_id = clp[7] if len(clp) > 7 and clp[7] else None
|
||||
facility_type = clp[8] if len(clp) > 8 and clp[8] else None
|
||||
# CLP09 is a single element (no composite qualifier on the 835 side).
|
||||
frequency = clp[9] if len(clp) > 9 and clp[9] else None
|
||||
except (IndexError, ValueError) as exc:
|
||||
raise CycloneParseError(f"Bad CLP: {exc}", segment_index=idx) from exc
|
||||
|
||||
service_payments: list[ServicePayment] = []
|
||||
ref_benefit_plan: str | None = None
|
||||
per_diem: Decimal | None = None
|
||||
status_label = claim_status_label(status)
|
||||
|
||||
idx += 1
|
||||
while idx < len(segments) and segments[idx][0] not in {"CLP", "SE"}:
|
||||
s = segments[idx]
|
||||
# Only consume this CLP's children; an LX starts a new 2100 loop.
|
||||
if s[0] == "LX":
|
||||
# Skip the LX and expect an SVC; if absent, stop.
|
||||
raw.append(s)
|
||||
idx += 1
|
||||
if idx < len(segments) and segments[idx][0] == "SVC":
|
||||
line_no = (
|
||||
int(s[1]) if len(s) > 1 and s[1].isdigit() else len(service_payments) + 1
|
||||
)
|
||||
sp, idx, svc_raw = _consume_service_payment(segments, idx, line_no)
|
||||
raw.extend(svc_raw)
|
||||
service_payments.append(sp)
|
||||
continue
|
||||
if s[0] == "SVC":
|
||||
sp, idx, svc_raw = _consume_service_payment(segments, idx, len(service_payments) + 1)
|
||||
raw.extend(svc_raw)
|
||||
service_payments.append(sp)
|
||||
continue
|
||||
raw.append(s)
|
||||
if s[0] == "CAS":
|
||||
cas, idx = _cas_segments(segments, idx)
|
||||
# Top-level claim adjustments are aggregated into a synthesized
|
||||
# ServicePayment with line_number=0 if there are no SVC rows.
|
||||
# Per the spec, ClaimPayment doesn't carry adjustments directly,
|
||||
# so we record them on the first service payment if any,
|
||||
# otherwise drop them on the floor (rare in practice).
|
||||
if service_payments:
|
||||
first = service_payments[0]
|
||||
service_payments[0] = first.model_copy(
|
||||
update={"adjustments": first.adjustments + cas}
|
||||
)
|
||||
continue
|
||||
if s[0] == "REF" and len(s) > 2 and s[1] == "CE":
|
||||
ref_benefit_plan = s[2]
|
||||
elif s[0] == "QTY" and len(s) > 2 and s[1] == "CA":
|
||||
try:
|
||||
per_diem = Decimal(s[2]) if s[2] else None
|
||||
except ValueError:
|
||||
per_diem = None
|
||||
elif s[0] == "NM1":
|
||||
# Patient (QC) / service-provider (1P) — captured in raw_segments.
|
||||
# The 835 spec doesn't require a structured patient model in v1.
|
||||
pass
|
||||
elif s[0] == "DTM":
|
||||
# Claim-level dates — captured in raw_segments.
|
||||
pass
|
||||
idx += 1
|
||||
|
||||
return (
|
||||
ClaimPayment(
|
||||
payer_claim_control_number=pcn,
|
||||
status_code=status,
|
||||
status_label=status_label,
|
||||
total_charge=total_charge,
|
||||
total_paid=total_paid,
|
||||
patient_responsibility=patient_resp,
|
||||
claim_filing_indicator=claim_filing,
|
||||
original_claim_id=original_claim_id,
|
||||
facility_type=facility_type,
|
||||
frequency_code=frequency,
|
||||
per_diem_covered_days=per_diem,
|
||||
ref_benefit_plan=ref_benefit_plan,
|
||||
service_payments=service_payments,
|
||||
raw_segments=raw,
|
||||
),
|
||||
idx,
|
||||
)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Top-level orchestrator
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def parse(text: str, payer_config: PayerConfig835, input_file: str = "") -> ParseResult835: # noqa: ARG001 — payer_config reserved for future use
|
||||
"""Parse a complete 835 ERA document and return a :class:`ParseResult835`.
|
||||
|
||||
``payer_config`` is accepted for symmetry with the 837P parser and
|
||||
for future payer-aware heuristics; in v1 the structural validation
|
||||
runs against the full result via :func:`cyclone.parsers.validator_835.validate`.
|
||||
"""
|
||||
segments = tokenize(text)
|
||||
envelope, summary = _build_envelope(segments, input_file=input_file)
|
||||
|
||||
financial: FinancialInfo | None = None
|
||||
trace: ReassociationTrace | None = None
|
||||
payer = Payer835(name="", id=None, address=None, contact_url=None)
|
||||
payee = Payee835(name="", npi="", address=None)
|
||||
claims: list[ClaimPayment] = []
|
||||
|
||||
i = 0
|
||||
# Pre-envelope segments (ISA/GS) are already consumed by _build_envelope.
|
||||
# Walk forward looking for the header (BPR/TRN), then loops.
|
||||
while i < len(segments):
|
||||
seg = segments[i]
|
||||
if seg[0] == "ST":
|
||||
# Skip the ST control segment.
|
||||
i += 1
|
||||
continue
|
||||
if seg[0] == "BPR":
|
||||
financial, i = _consume_bpr(segments, i)
|
||||
continue
|
||||
if seg[0] == "TRN":
|
||||
trace, i = _consume_trn(segments, i)
|
||||
continue
|
||||
if seg[0] == "N1" and len(seg) > 1 and seg[1] == "PR":
|
||||
payer, i = _consume_1000a(segments, i)
|
||||
continue
|
||||
if seg[0] == "N1" and len(seg) > 1 and seg[1] == "PE":
|
||||
payee, i = _consume_1000b(segments, i)
|
||||
continue
|
||||
if seg[0] == "CLP":
|
||||
try:
|
||||
claim, i = _consume_claim_payment(segments, i)
|
||||
claims.append(claim)
|
||||
except CycloneParseError:
|
||||
raise
|
||||
except Exception as exc: # pragma: no cover - safety net
|
||||
log.warning("Claim parse failed at segment %d: %s", i, exc)
|
||||
i += 1
|
||||
continue
|
||||
if seg[0] == "LX" and not claims and not payee.npi:
|
||||
# Defensive: a stray LX before any CLP (no 1000B seen) is ignored.
|
||||
i += 1
|
||||
continue
|
||||
i += 1
|
||||
|
||||
if financial is None:
|
||||
raise CycloneParseError("No BPR (Financial Information) segment found")
|
||||
if trace is None:
|
||||
raise CycloneParseError("No TRN (Reassociation Trace) segment found")
|
||||
|
||||
# Update the envelope's transaction date from the BPR payment date if available.
|
||||
if financial.payment_date is not None:
|
||||
envelope = envelope.model_copy(update={"transaction_date": financial.payment_date})
|
||||
|
||||
summary = summary.model_copy(update={
|
||||
"control_number": envelope.control_number,
|
||||
"transaction_date": envelope.transaction_date,
|
||||
"total_claims": len(claims),
|
||||
})
|
||||
|
||||
return ParseResult835(
|
||||
envelope=envelope,
|
||||
financial_info=financial,
|
||||
trace=trace,
|
||||
payer=payer,
|
||||
payee=payee,
|
||||
claims=claims,
|
||||
summary=summary,
|
||||
validation=None,
|
||||
)
|
||||
|
||||
|
||||
__all__ = ["parse"]
|
||||
@@ -1,13 +1,16 @@
|
||||
"""Payer-specific configuration for the 837P parser.
|
||||
"""Payer-specific configuration for the Cyclone parsers.
|
||||
|
||||
The same parser module can ingest 837P files from any payer, but the
|
||||
validation rules differ. ``PayerConfig`` carries the differences so
|
||||
``validator.validate(claim, config)`` can be payer-aware without branching
|
||||
in the parser.
|
||||
The same parser module can ingest 837P / 835 files from any payer, but
|
||||
the validation rules differ. ``PayerConfig`` (837P) and ``PayerConfig835``
|
||||
(835 ERA) carry the differences so the validator functions can be
|
||||
payer-aware without branching in the parser.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import Pattern
|
||||
|
||||
from pydantic import BaseModel, ConfigDict, Field
|
||||
|
||||
|
||||
@@ -26,6 +29,11 @@ CMS_PLACE_OF_SERVICE_CODES: set[str] = {
|
||||
}
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# 837P (Professional Claim)
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
class PayerConfig(BaseModel):
|
||||
model_config = ConfigDict(frozen=True)
|
||||
|
||||
@@ -82,3 +90,64 @@ class PayerConfig(BaseModel):
|
||||
allowed_facility_qualifiers={"B"},
|
||||
allowed_place_of_service_codes=set(CMS_PLACE_OF_SERVICE_CODES),
|
||||
)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# 835 ERA (Health Care Claim Payment/Advice)
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
class PayerConfig835(BaseModel):
|
||||
"""Payer-specific configuration for the 835 ERA parser.
|
||||
|
||||
Mirrors :class:`PayerConfig` but carries the 835-specific fields
|
||||
(handling codes, status codes, CO Medicaid tax IDs / health plan ID).
|
||||
Kept in a separate class — not a subclass — because the 835 and 837P
|
||||
validation surfaces don't share enough structure to make inheritance
|
||||
pay off.
|
||||
"""
|
||||
|
||||
model_config = ConfigDict(frozen=True, extra="ignore")
|
||||
|
||||
name: str
|
||||
allowed_handling_codes: set[str] = Field(
|
||||
default_factory=lambda: {"C", "D", "I", "X"}
|
||||
)
|
||||
allowed_status_codes: set[str] = Field(
|
||||
default_factory=lambda: {"1", "2", "3", "4", "19", "20", "21", "22", "23", "25"}
|
||||
)
|
||||
# CO Medicaid uses "81-1725341" (TXIX) and "84-0644739" (BHA).
|
||||
expected_payer_tax_ids: set[str] = Field(default_factory=set)
|
||||
# CO uses "7912900843" as the Health Plan ID on N104.
|
||||
expected_payer_health_plan_id: str = ""
|
||||
# Regex matched against Payer.name (N102). Example: r"^CO_(TXIX|BHA)$".
|
||||
payer_name_pattern: str = ""
|
||||
|
||||
def payer_name_regex(self) -> Pattern[str] | None:
|
||||
"""Compile ``payer_name_pattern`` (or return ``None`` if unset/blank)."""
|
||||
if not self.payer_name_pattern:
|
||||
return None
|
||||
return re.compile(self.payer_name_pattern)
|
||||
|
||||
@classmethod
|
||||
def co_medicaid_835(cls) -> "PayerConfig835":
|
||||
"""Defaults for Colorado Medical Assistance Program (835).
|
||||
|
||||
Source: ``docs/companionguides/835.md``.
|
||||
"""
|
||||
return cls(
|
||||
name="Colorado Medical Assistance Program (835)",
|
||||
expected_payer_tax_ids={"81-1725341", "84-0644739"},
|
||||
expected_payer_health_plan_id="7912900843",
|
||||
payer_name_pattern=r"^CO_(TXIX|BHA)$",
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def generic_835(cls) -> "PayerConfig835":
|
||||
"""Relaxed defaults for an unknown payer. Structural rules only."""
|
||||
return cls(
|
||||
name="Generic 835",
|
||||
expected_payer_tax_ids=set(),
|
||||
expected_payer_health_plan_id="",
|
||||
payer_name_pattern="",
|
||||
)
|
||||
|
||||
@@ -0,0 +1,218 @@
|
||||
"""Structural and payer-specific validation rules for 835 ERA payments.
|
||||
|
||||
Each rule is a pure function that takes a :class:`ParseResult835` and a
|
||||
:class:`PayerConfig835` and yields zero or more :class:`ValidationIssue`.
|
||||
The top-level :func:`validate` function runs every rule and aggregates
|
||||
results into a single :class:`ValidationReport` for the batch.
|
||||
|
||||
Unlike 837P (where each claim has its own report), the 835 ERA validates
|
||||
the whole batch as one unit because the BPR/CLP/SVC balancing rules are
|
||||
cross-cutting.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from collections.abc import Iterable
|
||||
from decimal import Decimal
|
||||
from typing import Callable
|
||||
|
||||
from cyclone.parsers.models import ValidationIssue, ValidationReport
|
||||
from cyclone.parsers.models_835 import ParseResult835
|
||||
from cyclone.parsers.payer import PayerConfig835
|
||||
|
||||
NPI_RE = re.compile(r"^\d{10}$")
|
||||
|
||||
Rule = Callable[[ParseResult835, PayerConfig835], Iterable[ValidationIssue]]
|
||||
|
||||
# Tolerance for monetary balancing (USD; the 835 standard's smallest
|
||||
# representable currency unit is 0.01).
|
||||
TOLERANCE = Decimal("0.01")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# BPR rules
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def _r001_bpr01_handling_code_allowed(result: ParseResult835, cfg: PayerConfig835) -> Iterable[ValidationIssue]:
|
||||
if result.financial_info.handling_code not in cfg.allowed_handling_codes:
|
||||
yield ValidationIssue(
|
||||
rule="R835_BPR01_handling_code_allowed",
|
||||
severity="error",
|
||||
message=f"BPR01 {result.financial_info.handling_code!r} not in {sorted(cfg.allowed_handling_codes)} for payer {cfg.name}",
|
||||
)
|
||||
|
||||
|
||||
def _r002_bpr02_positive(result: ParseResult835, _: PayerConfig835) -> Iterable[ValidationIssue]:
|
||||
if result.financial_info.paid_amount <= 0:
|
||||
yield ValidationIssue(
|
||||
rule="R835_BPR02_positive",
|
||||
severity="error",
|
||||
message=f"BPR02 (paid amount) must be > 0, got {result.financial_info.paid_amount}",
|
||||
)
|
||||
|
||||
|
||||
def _r003_bal_bpr_vs_clp04(result: ParseResult835, _: PayerConfig835) -> Iterable[ValidationIssue]:
|
||||
"""BPR02 must equal the sum of CLP04 across all claims in the batch."""
|
||||
if not result.claims:
|
||||
return
|
||||
clp_sum = sum((c.total_paid for c in result.claims), start=Decimal("0"))
|
||||
diff = (result.financial_info.paid_amount - clp_sum).copy_abs()
|
||||
if diff > TOLERANCE:
|
||||
yield ValidationIssue(
|
||||
rule="R835_BAL_BPR_vs_CLP04",
|
||||
severity="error",
|
||||
message=(
|
||||
f"BPR02 ({result.financial_info.paid_amount}) does not match "
|
||||
f"sum of CLP04 ({clp_sum}); diff={diff}"
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Per-claim CLP / SVC rules
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def _r004_bal_clp04_vs_svc03(result: ParseResult835, _: PayerConfig835) -> Iterable[ValidationIssue]:
|
||||
"""Each CLP04 must equal the sum of its SVC03 service payments."""
|
||||
for c in result.claims:
|
||||
if not c.service_payments:
|
||||
# No service lines — cannot balance, but R835_CLP03_present
|
||||
# already covers a zero-charge claim.
|
||||
continue
|
||||
svc_sum = sum((sp.payment for sp in c.service_payments), start=Decimal("0"))
|
||||
diff = (c.total_paid - svc_sum).copy_abs()
|
||||
if diff > TOLERANCE:
|
||||
yield ValidationIssue(
|
||||
rule="R835_BAL_CLP04_vs_SVC03",
|
||||
severity="error",
|
||||
message=(
|
||||
f"[claim {c.payer_claim_control_number}] CLP04 ({c.total_paid}) "
|
||||
f"does not match sum of SVC03 ({svc_sum}); diff={diff}"
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def _r005_clp01_present(result: ParseResult835, _: PayerConfig835) -> Iterable[ValidationIssue]:
|
||||
for c in result.claims:
|
||||
if not c.payer_claim_control_number or not c.payer_claim_control_number.strip():
|
||||
yield ValidationIssue(
|
||||
rule="R835_CLP01_present",
|
||||
severity="error",
|
||||
message="CLP01 (payer claim control number) is empty",
|
||||
)
|
||||
|
||||
|
||||
def _r006_clp02_status_allowed(result: ParseResult835, cfg: PayerConfig835) -> Iterable[ValidationIssue]:
|
||||
for c in result.claims:
|
||||
if c.status_code and c.status_code not in cfg.allowed_status_codes:
|
||||
yield ValidationIssue(
|
||||
rule="R835_CLP02_status_allowed",
|
||||
severity="error",
|
||||
message=(
|
||||
f"[claim {c.payer_claim_control_number}] CLP02 status "
|
||||
f"{c.status_code!r} not in {sorted(cfg.allowed_status_codes)} for payer {cfg.name}"
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def _r007_clp03_present(result: ParseResult835, _: PayerConfig835) -> Iterable[ValidationIssue]:
|
||||
for c in result.claims:
|
||||
if c.total_charge <= 0:
|
||||
yield ValidationIssue(
|
||||
rule="R835_CLP03_present",
|
||||
severity="error",
|
||||
message=(
|
||||
f"[claim {c.payer_claim_control_number}] CLP03 (total claim charge) "
|
||||
f"must be > 0, got {c.total_charge}"
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Payer / payee rules
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def _r008_payer_tax_id_matches(result: ParseResult835, cfg: PayerConfig835) -> Iterable[ValidationIssue]:
|
||||
if not cfg.expected_payer_tax_ids:
|
||||
return
|
||||
tax = result.financial_info.payer_tax_id
|
||||
if tax and tax not in cfg.expected_payer_tax_ids:
|
||||
yield ValidationIssue(
|
||||
rule="R835_PAYER_TAX_ID_matches",
|
||||
severity="warning",
|
||||
message=(
|
||||
f"BPR10 (payer tax id) {tax!r} not in "
|
||||
f"{sorted(cfg.expected_payer_tax_ids)} for payer {cfg.name}"
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def _r009_payer_health_plan_id_matches(result: ParseResult835, cfg: PayerConfig835) -> Iterable[ValidationIssue]:
|
||||
if not cfg.expected_payer_health_plan_id:
|
||||
return
|
||||
pid = result.payer.id
|
||||
if pid and pid != cfg.expected_payer_health_plan_id:
|
||||
yield ValidationIssue(
|
||||
rule="R835_PAYER_HEALTH_PLAN_ID_matches",
|
||||
severity="warning",
|
||||
message=(
|
||||
f"N104 (payer health plan id) {pid!r} != configured "
|
||||
f"{cfg.expected_payer_health_plan_id!r} for payer {cfg.name}"
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def _r010_payee_npi_format(result: ParseResult835, _: PayerConfig835) -> Iterable[ValidationIssue]:
|
||||
npi = result.payee.npi
|
||||
if npi and not NPI_RE.match(npi):
|
||||
yield ValidationIssue(
|
||||
rule="R835_PAYEE_NPI_format",
|
||||
severity="error",
|
||||
message=f"Payee NPI must be 10 digits, got {npi!r}",
|
||||
)
|
||||
|
||||
|
||||
def _r011_trn02_present(result: ParseResult835, _: PayerConfig835) -> Iterable[ValidationIssue]:
|
||||
if not result.trace.originating_company_id or not result.trace.originating_company_id.strip():
|
||||
yield ValidationIssue(
|
||||
rule="R835_REGN01_trn02_present",
|
||||
severity="error",
|
||||
message="TRN02 (originating company id / payee TPID) is empty",
|
||||
)
|
||||
|
||||
|
||||
_RULES: list[Rule] = [
|
||||
_r001_bpr01_handling_code_allowed,
|
||||
_r002_bpr02_positive,
|
||||
_r003_bal_bpr_vs_clp04,
|
||||
_r004_bal_clp04_vs_svc03,
|
||||
_r005_clp01_present,
|
||||
_r006_clp02_status_allowed,
|
||||
_r007_clp03_present,
|
||||
_r008_payer_tax_id_matches,
|
||||
_r009_payer_health_plan_id_matches,
|
||||
_r010_payee_npi_format,
|
||||
_r011_trn02_present,
|
||||
]
|
||||
|
||||
|
||||
def validate(result: ParseResult835, config: PayerConfig835) -> ValidationReport:
|
||||
"""Run every rule against the batch and return a single report.
|
||||
|
||||
Never raises — all issues are accumulated into the report. The
|
||||
orchestrator is responsible for promoting warnings to errors under
|
||||
``--strict`` and for setting ``BatchSummary.passed``.
|
||||
"""
|
||||
errors: list[ValidationIssue] = []
|
||||
warnings: list[ValidationIssue] = []
|
||||
for rule in _RULES:
|
||||
for issue in rule(result, config):
|
||||
(errors if issue.severity == "error" else warnings).append(issue)
|
||||
return ValidationReport(passed=not errors, errors=errors, warnings=warnings)
|
||||
|
||||
|
||||
__all__ = ["validate", "TOLERANCE"]
|
||||
@@ -0,0 +1,80 @@
|
||||
"""Write a :class:`ParseResult835` to disk as one JSON file per claim + summary.json.
|
||||
|
||||
Parallel to :mod:`cyclone.parsers.writer` for the 837P parser. Kept
|
||||
separate because the 835 writer is not just a duck-typed copy of the
|
||||
837P one — the ``payout`` payload includes the BPR/TRN/1000A/1000B
|
||||
headers alongside each claim file (so the UI can render the full ERA
|
||||
context per claim), and we want to be free to evolve the layout
|
||||
independently.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
from cyclone.parsers.models_835 import ParseResult835
|
||||
|
||||
_UNSAFE = re.compile(r"[^A-Za-z0-9._-]")
|
||||
|
||||
|
||||
def _sanitize(s: str) -> str:
|
||||
return _UNSAFE.sub("-", s).strip("-") or "claim"
|
||||
|
||||
|
||||
def _filename(control_number: str, transaction_date_iso: str, claim_id: str, used: set[str]) -> str:
|
||||
base = f"payout-{_sanitize(control_number)}-{_sanitize(transaction_date_iso.replace('-', ''))}-{_sanitize(claim_id)}.json"
|
||||
if base not in used:
|
||||
return base
|
||||
stem = base[:-5]
|
||||
n = 2
|
||||
while True:
|
||||
cand = f"{stem}-{n}.json"
|
||||
if cand not in used:
|
||||
return cand
|
||||
n += 1
|
||||
|
||||
|
||||
def write_outputs_835(result: ParseResult835, output_dir: Path) -> list[Path]:
|
||||
"""Write one JSON file per claim payment + a ``summary.json``.
|
||||
|
||||
Each claim file is enriched with the batch-level envelope/financial/
|
||||
trace/payer/payee so the UI can render the full ERA context without
|
||||
cross-referencing the summary. The ``summary.json`` is the canonical
|
||||
counts file.
|
||||
"""
|
||||
output_dir = Path(output_dir)
|
||||
output_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
txn_date = result.envelope.transaction_date.isoformat() if result.envelope else "unknown"
|
||||
control = result.envelope.control_number if result.envelope else "unknown"
|
||||
|
||||
used: set[str] = set()
|
||||
paths: list[Path] = []
|
||||
# Pre-serialize the batch-level fields so each claim file embeds them.
|
||||
envelope_obj = json.loads(result.envelope.model_dump_json())
|
||||
financial_obj = json.loads(result.financial_info.model_dump_json())
|
||||
trace_obj = json.loads(result.trace.model_dump_json())
|
||||
payer_obj = json.loads(result.payer.model_dump_json())
|
||||
payee_obj = json.loads(result.payee.model_dump_json())
|
||||
|
||||
for claim in result.claims:
|
||||
name = _filename(control, txn_date, claim.payer_claim_control_number, used)
|
||||
used.add(name)
|
||||
path = output_dir / name
|
||||
claim_obj = json.loads(claim.model_dump_json())
|
||||
payload = {
|
||||
"envelope": envelope_obj,
|
||||
"financial_info": financial_obj,
|
||||
"trace": trace_obj,
|
||||
"payer": payer_obj,
|
||||
"payee": payee_obj,
|
||||
"claim": claim_obj,
|
||||
}
|
||||
path.write_text(json.dumps(payload, indent=2))
|
||||
paths.append(path)
|
||||
|
||||
summary_path = output_dir / "summary.json"
|
||||
summary_path.write_text(result.summary.model_dump_json(indent=2))
|
||||
return paths
|
||||
Reference in New Issue
Block a user