Merge feat/835-era-parser: 835 ERA parser with BPR/CLP/SVC balancing, CO Medicaid config, CLI, and FastAPI route
This commit is contained in:
Tyler
+94
-5
@@ -1,8 +1,9 @@
|
||||
# Cyclone Backend — X12 837P Parser
|
||||
# Cyclone Backend — X12 837P / 835 ERA Parser
|
||||
|
||||
Python module + CLI for parsing X12 837P professional claim files into one
|
||||
validated JSON per claim. Output is consumed by the Cyclone Vite/React
|
||||
frontend via a FastAPI service (see **REST API** below).
|
||||
Python module + CLI for parsing X12 837P professional claim files and X12
|
||||
835 ERA (Health Care Claim Payment/Advice) files into one validated JSON
|
||||
per claim / payout. Output is consumed by the Cyclone Vite/React frontend
|
||||
via a FastAPI service (see **REST API** below).
|
||||
|
||||
## Install
|
||||
|
||||
@@ -80,7 +81,8 @@ python -m cyclone serve
|
||||
| Method | Path | Purpose |
|
||||
| ------ | ---------------- | -------------------------------------------------- |
|
||||
| GET | `/api/health` | Liveness probe: `{"status": "ok", "version": ...}` |
|
||||
| POST | `/api/parse-837` | Upload an X12 file, get parsed claims back |
|
||||
| POST | `/api/parse-837` | Upload an X12 837P file, get parsed claims back |
|
||||
| POST | `/api/parse-835` | Upload an X12 835 ERA file, get parsed payouts back |
|
||||
|
||||
`POST /api/parse-837` accepts `multipart/form-data` with a single `file`
|
||||
field. Optional query parameters:
|
||||
@@ -152,3 +154,90 @@ while (true) {
|
||||
|
||||
- Design: `docs/superpowers/specs/2026-06-19-cyclone-837p-parser-design.md`
|
||||
- Plan: `docs/superpowers/plans/2026-06-19-cyclone-837p-parser.md`
|
||||
|
||||
## 835 ERA parser
|
||||
|
||||
The 835 ERA parser mirrors the 837P architecture but is a separate
|
||||
module (`cyclone.parsers.parse_835`) with its own models
|
||||
(`cyclone.parsers.models_835`), writer (`cyclone.parsers.writer_835`),
|
||||
validator (`cyclone.parsers.validator_835`), and `PayerConfig835` preset.
|
||||
|
||||
It parses one 835 batch into:
|
||||
|
||||
- `envelope` — ISA / GS / ST control segments
|
||||
- `financial_info` — BPR (total paid, handling code, payer tax id, date)
|
||||
- `trace` — TRN (reassociation trace number + originating company id)
|
||||
- `payer` — Loop 1000A (NM1*PR + N3/N4/PER contact URL)
|
||||
- `payee` — Loop 1000B (NM1*PE + N3/N4)
|
||||
- `claims` — one `ClaimPayment` per Loop 2100 CLP, with embedded
|
||||
`ServicePayment` rows for each Loop 2110 SVC
|
||||
- `summary` — same `BatchSummary` used by 837P
|
||||
|
||||
### 835 CLI
|
||||
|
||||
```bash
|
||||
python -m cyclone.cli parse-835 path/to/835.txt --output-dir ./payouts
|
||||
```
|
||||
|
||||
Options:
|
||||
- `--payer {co_medicaid_835,generic_835}` — default `co_medicaid_835`
|
||||
- `--strict` — promote warnings to errors
|
||||
- `--include-raw-segments / --no-raw-segments` — default: include
|
||||
- `--log-level {DEBUG,INFO,WARNING,ERROR}`
|
||||
|
||||
Output:
|
||||
|
||||
```
|
||||
payouts/
|
||||
├── payout-991102984-20260617-CLM001.json # one per claim, with envelope/financial/trace/payer/payee embedded
|
||||
├── payout-991102984-20260617-CLM002.json
|
||||
└── summary.json
|
||||
```
|
||||
|
||||
### 835 programmatic use
|
||||
|
||||
```python
|
||||
from pathlib import Path
|
||||
from cyclone.parsers.parse_835 import parse
|
||||
from cyclone.parsers.payer import PayerConfig835
|
||||
from cyclone.parsers.validator_835 import validate
|
||||
from cyclone.parsers.writer_835 import write_outputs_835
|
||||
|
||||
result = parse(Path("input.txt").read_text(), PayerConfig835.co_medicaid_835())
|
||||
report = validate(result, PayerConfig835.co_medicaid_835())
|
||||
print(f"passed={report.passed} errors={len(report.errors)}")
|
||||
write_outputs_835(result, Path("./payouts"))
|
||||
```
|
||||
|
||||
### 835 payout balancing
|
||||
|
||||
The 835 validator enforces two monetary-balancing rules:
|
||||
|
||||
- `R835_BAL_BPR_vs_CLP04` — the BPR02 total paid must equal the sum of
|
||||
`CLP04` (total claim paid) across all claims in the batch. Tolerance
|
||||
is **$0.01**.
|
||||
- `R835_BAL_CLP04_vs_SVC03` — each `CLP04` must equal the sum of its
|
||||
embedded `SVC03` service payment amounts. Tolerance is **$0.01**.
|
||||
|
||||
These rules catch the common "BPR doesn't match the sum of the claims"
|
||||
mismatch (often caused by a missing or duplicated claim in the batch).
|
||||
|
||||
### 835 REST API
|
||||
|
||||
`POST /api/parse-835` mirrors `POST /api/parse-837` exactly:
|
||||
|
||||
- `multipart/form-data` with a single `file` field
|
||||
- Query params: `payer` (default `co_medicaid_835`), `include_raw_segments`
|
||||
(default `true`), `strict` (default `false`)
|
||||
- Content negotiation: `Accept: application/json` → single `ParseResult835`
|
||||
object; default → NDJSON stream with `envelope → financial_info → trace
|
||||
→ payer → payee → claim_payment (×N) → summary` lines
|
||||
- `400` on `CycloneParseError` or unknown `payer`
|
||||
- `422` when `validate()` reports errors (body still includes the full
|
||||
`ParseResult835` so the UI can show the issues)
|
||||
|
||||
```bash
|
||||
curl -X POST http://localhost:8000/api/parse-835 \
|
||||
-F "file=@./samples/co_835.txt" \
|
||||
-H "Accept: application/json"
|
||||
```
|
||||
|
||||
+152
-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,126 @@ 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.
|
||||
# 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()),
|
||||
)
|
||||
|
||||
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,83 @@ 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.
|
||||
# 835 validation is batch-level (BPR/CLP/SVC balancing is cross-cutting),
|
||||
# so pass/fail applies uniformly to every claim payment in the batch:
|
||||
# when the batch passes, all N claims "passed" (passed=N, failed=0);
|
||||
# when the batch fails, all N claims "failed" (passed=0, failed=N).
|
||||
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})
|
||||
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})
|
||||
result.summary = result.summary.model_copy(update={
|
||||
"passed": passed,
|
||||
"failed": failed,
|
||||
"failed_claim_ids": failed_claim_ids,
|
||||
"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
|
||||
+42
@@ -0,0 +1,42 @@
|
||||
ISA*00* *00* *ZZ*COMEDASSISTPROG*ZZ*TPID001 *260617*1937*^*00501*991102986*1*P*:~ST*835*991102986*005010X221A1~
|
||||
BPR*C*241.16*C*ACH*CCP*01*021000021*DA*123456789*81-1725341******20260617~
|
||||
TRN*1*TPID001*81-1725341~
|
||||
DTM*405*20260617~
|
||||
N1*PR*2*CO_TXIX*****XV*7912900843~
|
||||
N3*PO BOX 11000~
|
||||
N4*DENVER*CO*802110000~
|
||||
PER*BL*Customer Service*TE*8001234567*UR*hcpf.colorado.gov~
|
||||
N1*PE*2*TOC, Inc.*****XX*1881068062~
|
||||
N3*1100 East Main St*Suite A~
|
||||
N4*Montrose*CO*814014063~
|
||||
REF*TJ*123456789~
|
||||
LX*1~
|
||||
CLP*t991102986o1c1d*1*85.40*85.40*0.00*MC*t991102986o1c1d*12*1~
|
||||
CAS*CO*45*0.00~
|
||||
REF*CE*PLAN_A~
|
||||
NM1*QC*1*Balliache*Marianela~
|
||||
SVC*HC:T1019:U2*42.70*42.70*UN*6.00*20260602~
|
||||
DTM*472*20260602~
|
||||
SVC*HC:T1019:U2*42.70*42.70*UN*6.00*20260603~
|
||||
DTM*472*20260603~
|
||||
LX*2~
|
||||
CLP*t991102986o1c2d*1*155.76*155.76*0.00*MC*t991102986o1c2d*12*1~
|
||||
CAS*CO*45*0.00~
|
||||
REF*CE*PLAN_A~
|
||||
QTY*CA*4~
|
||||
NM1*QC*1*Barella*Victoria~
|
||||
SVC*HC:S5130:U2*38.94*38.94*UN*6.00*20260602~
|
||||
DTM*472*20260602~
|
||||
REF*0K*PLAN_A~
|
||||
SVC*HC:S5130:U2*38.94*38.94*UN*6.00*20260603~
|
||||
DTM*472*20260603~
|
||||
REF*0K*PLAN_A~
|
||||
SVC*HC:S5130:U2*38.94*38.94*UN*6.00*20260604~
|
||||
DTM*472*20260604~
|
||||
REF*0K*PLAN_A~
|
||||
SVC*HC:S5130:U2*38.94*38.94*UN*6.00*20260605~
|
||||
DTM*472*20260605~
|
||||
REF*0K*PLAN_A~
|
||||
SE*45*991102986~
|
||||
GE*1*991102986~
|
||||
IEA*1*991102986~
|
||||
+21
@@ -0,0 +1,21 @@
|
||||
ISA*00* *00* *ZZ*COMEDASSISTPROG*ZZ*TPID001 *260617*1937*^*00501*991102984*1*P*:~ST*835*991102984*005010X221A1~
|
||||
BPR*C*85.00*C*ACH*CCP*01*021000021*DA*123456789*81-1725341******20260617~
|
||||
TRN*1*TPID001*81-1725341~
|
||||
DTM*405*20260617~
|
||||
N1*PR*2*CO_TXIX*****XV*7912900843~
|
||||
N3*PO BOX 11000~
|
||||
N4*DENVER*CO*802110000~
|
||||
PER*BL*Customer Service*TE*8001234567*UR*hcpf.colorado.gov~
|
||||
N1*PE*2*TOC, Inc.*****XX*1881068062~
|
||||
N3*1100 East Main St*Suite A~
|
||||
N4*Montrose*CO*814014063~
|
||||
LX*1~
|
||||
CLP*CLM001*1*100.00*85.00*15.00*MC*CLM001*12*1~
|
||||
CAS*CO*45*15.00~
|
||||
SVC*HC:S5130:U2*60.00*50.00*7.73*UN*20260602~
|
||||
DTM*472*20260602~
|
||||
SVC*HC:S5130:U2*40.00*35.00*7.73*UN*20260603~
|
||||
DTM*472*20260603~
|
||||
SE*22*991102984~
|
||||
GE*1*991102984~
|
||||
IEA*1*991102984~
|
||||
+21
@@ -0,0 +1,21 @@
|
||||
ISA*00* *00* *ZZ*COMEDASSISTPROG*ZZ*TPID001 *260617*1937*^*00501*991102985*1*P*:~ST*835*991102985*005010X221A1~
|
||||
BPR*C*85.50*C*ACH*CCP*01*021000021*DA*123456789*81-1725341******20260617~
|
||||
TRN*1*TPID001*81-1725341~
|
||||
DTM*405*20260617~
|
||||
N1*PR*2*CO_TXIX*****XV*7912900843~
|
||||
N3*PO BOX 11000~
|
||||
N4*DENVER*CO*802110000~
|
||||
PER*BL*Customer Service*TE*8001234567*UR*hcpf.colorado.gov~
|
||||
N1*PE*2*TOC, Inc.*****XX*1881068062~
|
||||
N3*1100 East Main St*Suite A~
|
||||
N4*Montrose*CO*814014063~
|
||||
LX*1~
|
||||
CLP*CLM001*1*100.00*85.00*15.00*MC*CLM001*12*1~
|
||||
CAS*CO*45*15.00~
|
||||
SVC*HC:S5130:U2*60.00*50.00*7.73*UN*20260602~
|
||||
DTM*472*20260602~
|
||||
SVC*HC:S5130:U2*40.00*35.00*7.73*UN*20260603~
|
||||
DTM*472*20260603~
|
||||
SE*22*991102985~
|
||||
GE*1*991102985~
|
||||
IEA*1*991102985~
|
||||
@@ -0,0 +1,171 @@
|
||||
"""Tests for the FastAPI surface in ``cyclone.api`` for the 835 ERA endpoint."""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from cyclone.api import app
|
||||
|
||||
|
||||
FIXTURE = Path(__file__).parent / "fixtures" / "co_medicaid_835.txt"
|
||||
UNBALANCED = Path(__file__).parent / "fixtures" / "unbalanced_835.txt"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client() -> TestClient:
|
||||
return TestClient(app)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# JSON response path
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def test_parse_835_endpoint_returns_json(client: TestClient):
|
||||
text = FIXTURE.read_text()
|
||||
resp = client.post(
|
||||
"/api/parse-835",
|
||||
files={"file": ("co_medicaid_835.txt", text, "text/plain")},
|
||||
headers={"Accept": "application/json"},
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
body = resp.json()
|
||||
assert "envelope" in body
|
||||
assert "financial_info" in body
|
||||
assert "trace" in body
|
||||
assert "payer" in body
|
||||
assert "payee" in body
|
||||
assert "claims" in body and len(body["claims"]) == 2
|
||||
assert "summary" in body
|
||||
assert body["summary"]["total_claims"] == 2
|
||||
assert body["summary"]["passed"] == 2
|
||||
assert body["summary"]["failed"] == 0
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# NDJSON streaming path
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def test_parse_835_endpoint_streams_ndjson(client: TestClient):
|
||||
text = FIXTURE.read_text()
|
||||
resp = client.post(
|
||||
"/api/parse-835",
|
||||
files={"file": ("co_medicaid_835.txt", text, "text/plain")},
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
assert resp.headers["content-type"].startswith("application/x-ndjson")
|
||||
|
||||
lines = list(resp.iter_lines())
|
||||
# 1 envelope + 1 financial_info + 1 trace + 1 payer + 1 payee + 2 claims + 1 summary = 8
|
||||
assert len(lines) == 8
|
||||
|
||||
parsed = [json.loads(line) for line in lines]
|
||||
assert parsed[0]["type"] == "envelope"
|
||||
assert parsed[1]["type"] == "financial_info"
|
||||
assert parsed[2]["type"] == "trace"
|
||||
assert parsed[3]["type"] == "payer"
|
||||
assert parsed[4]["type"] == "payee"
|
||||
assert parsed[5]["type"] == "claim_payment"
|
||||
assert parsed[6]["type"] == "claim_payment"
|
||||
assert parsed[7]["type"] == "summary"
|
||||
|
||||
# When include_raw_segments defaults to True, each claim carries raw segments.
|
||||
for obj in parsed[5:7]:
|
||||
assert "raw_segments" in obj["data"]
|
||||
assert isinstance(obj["data"]["raw_segments"], list)
|
||||
|
||||
# Summary numbers match the JSON path.
|
||||
assert parsed[7]["data"]["total_claims"] == 2
|
||||
assert parsed[7]["data"]["passed"] == 2
|
||||
|
||||
|
||||
def test_parse_835_endpoint_streams_ndjson_without_raw_segments(client: TestClient):
|
||||
text = FIXTURE.read_text()
|
||||
resp = client.post(
|
||||
"/api/parse-835?include_raw_segments=false",
|
||||
files={"file": ("co_medicaid_835.txt", text, "text/plain")},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.headers["content-type"].startswith("application/x-ndjson")
|
||||
|
||||
claims = [
|
||||
json.loads(line)
|
||||
for line in resp.iter_lines()
|
||||
if json.loads(line)["type"] == "claim_payment"
|
||||
]
|
||||
assert len(claims) == 2
|
||||
for c in claims:
|
||||
assert c["data"]["raw_segments"] == []
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Validation / error paths
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def test_parse_835_endpoint_rejects_missing_file(client: TestClient):
|
||||
# FastAPI's `File(...)` (no default) → 422 Unprocessable Entity.
|
||||
resp = client.post("/api/parse-835")
|
||||
assert resp.status_code == 422
|
||||
|
||||
|
||||
def test_parse_835_endpoint_handles_payer_query_param(client: TestClient):
|
||||
text = FIXTURE.read_text()
|
||||
for payer in ("co_medicaid_835", "generic_835"):
|
||||
resp = client.post(
|
||||
f"/api/parse-835?payer={payer}",
|
||||
files={"file": ("co_medicaid_835.txt", text, "text/plain")},
|
||||
headers={"Accept": "application/json"},
|
||||
)
|
||||
assert resp.status_code == 200, (payer, resp.text)
|
||||
body = resp.json()
|
||||
assert body["summary"]["total_claims"] == 2
|
||||
assert body["summary"]["passed"] == 2
|
||||
|
||||
|
||||
def test_parse_835_endpoint_rejects_unknown_payer(client: TestClient):
|
||||
text = FIXTURE.read_text()
|
||||
resp = client.post(
|
||||
"/api/parse-835?payer=does_not_exist",
|
||||
files={"file": ("co_medicaid_835.txt", text, "text/plain")},
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
|
||||
|
||||
def test_parse_835_endpoint_unbalanced_returns_422(client: TestClient):
|
||||
text = UNBALANCED.read_text()
|
||||
resp = client.post(
|
||||
"/api/parse-835",
|
||||
files={"file": ("unbalanced_835.txt", text, "text/plain")},
|
||||
headers={"Accept": "application/json"},
|
||||
)
|
||||
assert resp.status_code == 422, resp.text
|
||||
body = resp.json()
|
||||
assert body["summary"]["failed"] == 1
|
||||
# The R835_BAL_BPR_vs_CLP04 rule should have fired.
|
||||
assert any(
|
||||
issue["rule"] == "R835_BAL_BPR_vs_CLP04"
|
||||
for issue in body["validation"]["errors"]
|
||||
)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# CORS
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def test_cors_headers_present_for_835(client: TestClient):
|
||||
resp = client.options(
|
||||
"/api/parse-835",
|
||||
headers={
|
||||
"Origin": "http://localhost:5173",
|
||||
"Access-Control-Request-Method": "POST",
|
||||
"Access-Control-Request-Headers": "content-type",
|
||||
},
|
||||
)
|
||||
assert resp.headers.get("access-control-allow-origin") == "http://localhost:5173"
|
||||
assert "POST" in resp.headers.get("access-control-allow-methods", "").upper()
|
||||
@@ -0,0 +1,92 @@
|
||||
"""Tests for the `parse-835` Click subcommand."""
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from click.testing import CliRunner
|
||||
|
||||
from cyclone.cli import main
|
||||
|
||||
FIXTURE = Path(__file__).parent / "fixtures" / "minimal_835.txt"
|
||||
UNBALANCED = Path(__file__).parent / "fixtures" / "unbalanced_835.txt"
|
||||
CO_FIXTURE = Path(__file__).parent / "fixtures" / "co_medicaid_835.txt"
|
||||
|
||||
|
||||
def test_cli_parse_835_writes_outputs(tmp_path: Path):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(main, ["parse-835", str(FIXTURE), "--output-dir", str(tmp_path)])
|
||||
assert result.exit_code == 0, result.output
|
||||
written = list(tmp_path.glob("payout-*.json"))
|
||||
assert len(written) == 1
|
||||
summary = json.loads((tmp_path / "summary.json").read_text())
|
||||
assert summary["total_claims"] == 1
|
||||
assert summary["passed"] == 1
|
||||
assert summary["failed"] == 0
|
||||
|
||||
|
||||
def test_cli_parse_835_strict_promotes_warnings(tmp_path: Path):
|
||||
runner = CliRunner()
|
||||
# Strict mode: even warnings cause failed=1. Minimal fixture is fully
|
||||
# passing so this is a smoke test that the flag is wired.
|
||||
result = runner.invoke(main, ["parse-835", str(FIXTURE), "--output-dir", str(tmp_path), "--strict"])
|
||||
assert result.exit_code == 0
|
||||
summary = json.loads((tmp_path / "summary.json").read_text())
|
||||
assert summary["passed"] == 1
|
||||
|
||||
|
||||
def test_cli_parse_835_no_raw_segments(tmp_path: Path):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(main, ["parse-835", str(FIXTURE), "--output-dir", str(tmp_path), "--no-raw-segments"])
|
||||
assert result.exit_code == 0
|
||||
data = json.loads(next(tmp_path.glob("payout-*.json")).read_text())
|
||||
# The claim lives under the "claim" key in each payout file.
|
||||
assert data["claim"]["raw_segments"] == []
|
||||
|
||||
|
||||
def test_cli_parse_835_unbalanced_strict_fails(tmp_path: Path):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(main, ["parse-835", str(UNBALANCED), "--output-dir", str(tmp_path), "--strict"])
|
||||
assert result.exit_code == 0, result.output
|
||||
summary = json.loads((tmp_path / "summary.json").read_text())
|
||||
assert summary["passed"] == 0
|
||||
assert summary["failed"] == 1
|
||||
# R835_BAL_BPR_vs_CLP04 is the rule that fired.
|
||||
assert "R835_BAL_BPR_vs_CLP04" in summary["issues_by_rule"]
|
||||
|
||||
|
||||
def test_cli_parse_835_unknown_payer_errors(tmp_path: Path):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(main, ["parse-835", str(FIXTURE), "--output-dir", str(tmp_path), "--payer", "no_such_payer"])
|
||||
assert result.exit_code != 0
|
||||
|
||||
|
||||
def test_cli_parse_835_missing_file_exits_2(tmp_path: Path):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(main, ["parse-835", "/no/such/file.txt", "--output-dir", str(tmp_path)])
|
||||
assert result.exit_code == 2
|
||||
|
||||
|
||||
def test_cli_parse_835_generic_payer(tmp_path: Path):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(main, ["parse-835", str(FIXTURE), "--output-dir", str(tmp_path), "--payer", "generic_835"])
|
||||
assert result.exit_code == 0, result.output
|
||||
summary = json.loads((tmp_path / "summary.json").read_text())
|
||||
assert summary["passed"] == 1
|
||||
|
||||
|
||||
def test_cli_parse_835_summary_counts_consistent(tmp_path: Path):
|
||||
"""Regression: passed + failed must equal total_claims for any batch size.
|
||||
|
||||
The CO Medicaid fixture has 2 claim payments; if the batch validates
|
||||
cleanly, both must show as passed (passed=2, failed=0). A pre-fix
|
||||
implementation set passed=1 regardless of claim count.
|
||||
"""
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(main, ["parse-835", str(CO_FIXTURE), "--output-dir", str(tmp_path)])
|
||||
assert result.exit_code == 0, result.output
|
||||
summary = json.loads((tmp_path / "summary.json").read_text())
|
||||
assert summary["total_claims"] == 2
|
||||
assert summary["passed"] + summary["failed"] == summary["total_claims"]
|
||||
assert summary["passed"] == 2
|
||||
assert summary["failed"] == 0
|
||||
assert summary["failed_claim_ids"] == []
|
||||
@@ -0,0 +1,60 @@
|
||||
"""Integration test: load the realistic CO Medicaid 835 fixture and validate it end-to-end."""
|
||||
from decimal import Decimal
|
||||
from pathlib import Path
|
||||
|
||||
from cyclone.parsers.parse_835 import parse
|
||||
from cyclone.parsers.payer import PayerConfig835
|
||||
from cyclone.parsers.validator_835 import validate
|
||||
|
||||
FIXTURE = Path(__file__).parent / "fixtures" / "co_medicaid_835.txt"
|
||||
|
||||
|
||||
def test_co_fixture_two_claims_pass():
|
||||
result = parse(FIXTURE.read_text(), PayerConfig835.co_medicaid_835())
|
||||
cfg = PayerConfig835.co_medicaid_835()
|
||||
report = validate(result, cfg)
|
||||
assert result.summary.total_claims == 2
|
||||
assert report.passed is True
|
||||
assert report.errors == []
|
||||
|
||||
|
||||
def test_first_claim_has_two_service_lines():
|
||||
result = parse(FIXTURE.read_text(), PayerConfig835.co_medicaid_835())
|
||||
c1 = result.claims[0]
|
||||
assert c1.payer_claim_control_number == "t991102986o1c1d"
|
||||
assert len(c1.service_payments) == 2
|
||||
assert c1.service_payments[0].procedure_code == "T1019"
|
||||
assert c1.service_payments[0].modifiers == ["U2"]
|
||||
assert c1.service_payments[1].service_date is not None
|
||||
|
||||
|
||||
def test_second_claim_has_four_service_lines():
|
||||
result = parse(FIXTURE.read_text(), PayerConfig835.co_medicaid_835())
|
||||
c2 = result.claims[1]
|
||||
assert c2.payer_claim_control_number == "t991102986o1c2d"
|
||||
assert len(c2.service_payments) == 4
|
||||
assert c2.per_diem_covered_days == Decimal("4")
|
||||
|
||||
|
||||
def test_payer_and_payee_match_co_config():
|
||||
result = parse(FIXTURE.read_text(), PayerConfig835.co_medicaid_835())
|
||||
assert result.payer.name == "CO_TXIX"
|
||||
assert result.payer.id == "7912900843"
|
||||
assert result.financial_info.payer_tax_id in {"81-1725341", "84-0644739"}
|
||||
assert result.payee.npi == "1881068062"
|
||||
assert result.payee.name == "TOC, Inc."
|
||||
|
||||
|
||||
def test_bpr_vs_clp_balance_holds():
|
||||
"""BPR02 (241.16) = CLP04 sum (85.40 + 155.76)."""
|
||||
result = parse(FIXTURE.read_text(), PayerConfig835.co_medicaid_835())
|
||||
clp_sum = sum((c.total_paid for c in result.claims), start=Decimal("0"))
|
||||
assert result.financial_info.paid_amount == clp_sum
|
||||
assert result.financial_info.paid_amount == Decimal("241.16")
|
||||
|
||||
|
||||
def test_per_claim_clp_vs_svc_balance_holds():
|
||||
result = parse(FIXTURE.read_text(), PayerConfig835.co_medicaid_835())
|
||||
for c in result.claims:
|
||||
svc_sum = sum((sp.payment for sp in c.service_payments), start=Decimal("0"))
|
||||
assert c.total_paid == svc_sum, f"claim {c.payer_claim_control_number}: CLP04={c.total_paid} != sum(SVC03)={svc_sum}"
|
||||
@@ -0,0 +1,171 @@
|
||||
"""Tests for the 835 ERA Pydantic models."""
|
||||
import json
|
||||
from datetime import date
|
||||
from decimal import Decimal
|
||||
|
||||
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,
|
||||
)
|
||||
|
||||
|
||||
def _build_result() -> ParseResult835:
|
||||
return ParseResult835(
|
||||
envelope=Envelope(
|
||||
sender_id="COMEDASSISTPROG",
|
||||
receiver_id="TPID001",
|
||||
control_number="991102984",
|
||||
transaction_date=date(2026, 6, 17),
|
||||
implementation_guide="005010X221A1",
|
||||
),
|
||||
financial_info=FinancialInfo(
|
||||
handling_code="C",
|
||||
paid_amount=Decimal("85.00"),
|
||||
credit_debit_flag="C",
|
||||
payment_method="ACH",
|
||||
payer_tax_id="81-1725341",
|
||||
payment_date=date(2026, 6, 17),
|
||||
),
|
||||
trace=ReassociationTrace(
|
||||
trace_type_code="1",
|
||||
trace_number="991102984",
|
||||
originating_company_id="TPID001",
|
||||
payer_id="81-1725341",
|
||||
),
|
||||
payer=Payer835(
|
||||
name="CO_TXIX",
|
||||
id="7912900843",
|
||||
contact_url="hcpf.colorado.gov",
|
||||
address=Address(line1="PO BOX 11000", city="DENVER", state="CO", zip="802110000"),
|
||||
),
|
||||
payee=Payee835(
|
||||
name="TOC, Inc.",
|
||||
npi="1881068062",
|
||||
address=Address(line1="1100 East Main St", line2="Suite A", city="Montrose", state="CO", zip="814014063"),
|
||||
),
|
||||
claims=[
|
||||
ClaimPayment(
|
||||
payer_claim_control_number="CLM001",
|
||||
status_code="1",
|
||||
status_label="Processed as Primary",
|
||||
total_charge=Decimal("100.00"),
|
||||
total_paid=Decimal("85.00"),
|
||||
patient_responsibility=Decimal("15.00"),
|
||||
claim_filing_indicator="MC",
|
||||
original_claim_id="CLM001",
|
||||
facility_type="12",
|
||||
frequency_code="1",
|
||||
per_diem_covered_days=None,
|
||||
ref_benefit_plan=None,
|
||||
service_payments=[
|
||||
ServicePayment(
|
||||
line_number=1,
|
||||
procedure_qualifier="HC",
|
||||
procedure_code="S5130",
|
||||
modifiers=["U2"],
|
||||
charge=Decimal("60.00"),
|
||||
payment=Decimal("50.00"),
|
||||
units=Decimal("7.73"),
|
||||
unit_type="UN",
|
||||
service_date=date(2026, 6, 2),
|
||||
adjustments=[
|
||||
ClaimAdjustment(group_code="CO", reason_code="45", amount=Decimal("10.00"), quantity=None),
|
||||
],
|
||||
ref_benefit_plan=None,
|
||||
),
|
||||
],
|
||||
raw_segments=[["CLP", "CLM001", "1", "100.00", "85.00"]],
|
||||
)
|
||||
],
|
||||
summary=BatchSummary(
|
||||
input_file="f",
|
||||
control_number="991102984",
|
||||
transaction_date=date(2026, 6, 17),
|
||||
total_claims=1,
|
||||
passed=1,
|
||||
failed=0,
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def test_financial_info_decimal_paid_amount():
|
||||
f = FinancialInfo(handling_code="C", paid_amount=Decimal("85.00"), credit_debit_flag="C")
|
||||
assert f.paid_amount == Decimal("85.00")
|
||||
assert f.payer_tax_id is None
|
||||
assert f.payment_date is None
|
||||
|
||||
|
||||
def test_reassociation_trace_required():
|
||||
t = ReassociationTrace(trace_type_code="1", trace_number="X", originating_company_id="Y")
|
||||
assert t.payer_id is None
|
||||
|
||||
|
||||
def test_payer835_optional_fields():
|
||||
p = Payer835(name="X")
|
||||
assert p.id is None
|
||||
assert p.address is None
|
||||
assert p.contact_url is None
|
||||
|
||||
|
||||
def test_payee835_required_npi():
|
||||
p = Payee835(name="TOC", npi="1881068062")
|
||||
assert p.npi == "1881068062"
|
||||
|
||||
|
||||
def test_claim_adjustment_quantity_optional():
|
||||
a = ClaimAdjustment(group_code="CO", reason_code="45", amount=Decimal("10.00"))
|
||||
assert a.quantity is None
|
||||
|
||||
|
||||
def test_service_payment_modifiers_default_empty():
|
||||
sp = ServicePayment(
|
||||
line_number=1,
|
||||
procedure_qualifier="HC",
|
||||
procedure_code="99213",
|
||||
charge=Decimal("100.00"),
|
||||
payment=Decimal("85.00"),
|
||||
)
|
||||
assert sp.modifiers == []
|
||||
assert sp.adjustments == []
|
||||
|
||||
|
||||
def test_claim_payment_status_label_decoded():
|
||||
c = ClaimPayment(
|
||||
payer_claim_control_number="X",
|
||||
status_code="1",
|
||||
total_charge=Decimal("100"),
|
||||
total_paid=Decimal("85"),
|
||||
)
|
||||
assert c.status_label == "Processed as Primary"
|
||||
assert claim_status_label("4") == "Denied"
|
||||
assert claim_status_label("25") == "Predetermination Pricing Only"
|
||||
assert claim_status_label("99") is None # unknown → None
|
||||
|
||||
|
||||
def test_parse_result835_date_serializes_to_iso():
|
||||
r = _build_result()
|
||||
blob = r.model_dump()
|
||||
assert blob["envelope"]["transaction_date"] == "2026-06-17"
|
||||
assert blob["financial_info"]["payment_date"] == "2026-06-17"
|
||||
assert blob["claims"][0]["service_payments"][0]["service_date"] == "2026-06-02"
|
||||
|
||||
|
||||
def test_parse_result835_full_round_trip():
|
||||
r = _build_result()
|
||||
blob = json.loads(r.model_dump_json())
|
||||
assert blob["envelope"]["control_number"] == "991102984"
|
||||
assert blob["payer"]["name"] == "CO_TXIX"
|
||||
assert blob["payee"]["npi"] == "1881068062"
|
||||
assert blob["claims"][0]["payer_claim_control_number"] == "CLM001"
|
||||
# Round-trip back to a model
|
||||
r2 = ParseResult835.model_validate(blob)
|
||||
assert r2.claims[0].total_paid == Decimal("85.00")
|
||||
assert r2.payer.contact_url == "hcpf.colorado.gov"
|
||||
@@ -0,0 +1,126 @@
|
||||
"""Tests for the 835 ERA parser orchestrator."""
|
||||
from datetime import date
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from cyclone.parsers.exceptions import CycloneParseError
|
||||
from cyclone.parsers.parse_835 import parse
|
||||
from cyclone.parsers.payer import PayerConfig835
|
||||
|
||||
FIXTURE = Path(__file__).parent / "fixtures" / "minimal_835.txt"
|
||||
|
||||
|
||||
def test_parse_minimal_fixture_returns_one_claim():
|
||||
text = FIXTURE.read_text()
|
||||
result = parse(text, PayerConfig835.co_medicaid_835())
|
||||
assert len(result.claims) == 1
|
||||
c = result.claims[0]
|
||||
assert c.payer_claim_control_number == "CLM001"
|
||||
assert c.status_code == "1"
|
||||
assert c.status_label == "Processed as Primary"
|
||||
assert c.total_charge == 100.00
|
||||
assert c.total_paid == 85.00
|
||||
assert c.patient_responsibility == 15.00
|
||||
assert c.claim_filing_indicator == "MC"
|
||||
assert c.facility_type == "12"
|
||||
assert c.frequency_code == "1"
|
||||
|
||||
|
||||
def test_parse_envelope_captured():
|
||||
result = parse(FIXTURE.read_text(), PayerConfig835.co_medicaid_835())
|
||||
assert result.envelope is not None
|
||||
assert result.envelope.control_number == "991102984"
|
||||
assert result.envelope.transaction_date == date(2026, 6, 17)
|
||||
assert result.envelope.sender_id == "COMEDASSISTPROG"
|
||||
assert result.envelope.implementation_guide == "005010X221A1"
|
||||
|
||||
|
||||
def test_parse_financial_info():
|
||||
result = parse(FIXTURE.read_text(), PayerConfig835.co_medicaid_835())
|
||||
f = result.financial_info
|
||||
assert f.handling_code == "C"
|
||||
assert f.paid_amount == 85.00
|
||||
assert f.credit_debit_flag == "C"
|
||||
assert f.payment_method == "ACH"
|
||||
assert f.payer_tax_id == "81-1725341"
|
||||
assert f.payment_date == date(2026, 6, 17)
|
||||
|
||||
|
||||
def test_parse_trace():
|
||||
result = parse(FIXTURE.read_text(), PayerConfig835.co_medicaid_835())
|
||||
t = result.trace
|
||||
assert t.trace_type_code == "1"
|
||||
# X12 standard: TRN02 = Reference Identification (trace number),
|
||||
# TRN03 = Originating Company Identifier. The CO guide labels
|
||||
# TRN03 as the Payer Tax ID, so for CO Medicaid 835s the
|
||||
# originating_company_id is "81-1725341" and the trace number is
|
||||
# the provider's TPID.
|
||||
assert t.trace_number == "TPID001"
|
||||
assert t.originating_company_id == "81-1725341"
|
||||
assert t.payer_id is None # TRN04 is not present in this fixture
|
||||
|
||||
|
||||
def test_parse_payer_and_payee():
|
||||
result = parse(FIXTURE.read_text(), PayerConfig835.co_medicaid_835())
|
||||
p = result.payer
|
||||
assert p.name == "CO_TXIX"
|
||||
assert p.id == "7912900843"
|
||||
assert p.contact_url == "hcpf.colorado.gov"
|
||||
assert p.address is not None
|
||||
assert p.address.city == "DENVER"
|
||||
pe = result.payee
|
||||
assert pe.name == "TOC, Inc."
|
||||
assert pe.npi == "1881068062"
|
||||
assert pe.address is not None
|
||||
assert pe.address.zip == "814014063"
|
||||
|
||||
|
||||
def test_parse_claim_has_two_service_lines():
|
||||
result = parse(FIXTURE.read_text(), PayerConfig835.co_medicaid_835())
|
||||
c = result.claims[0]
|
||||
assert len(c.service_payments) == 2
|
||||
s1, s2 = c.service_payments
|
||||
assert s1.line_number == 1
|
||||
assert s1.procedure_code == "S5130"
|
||||
assert s1.procedure_qualifier == "HC"
|
||||
assert s1.modifiers == ["U2"]
|
||||
assert s1.payment == 50.00
|
||||
assert s1.service_date == date(2026, 6, 2)
|
||||
assert s2.line_number == 2
|
||||
assert s2.payment == 35.00
|
||||
assert s2.service_date == date(2026, 6, 3)
|
||||
|
||||
|
||||
def test_parse_summary_counts():
|
||||
result = parse(FIXTURE.read_text(), PayerConfig835.co_medicaid_835())
|
||||
assert result.summary.total_claims == 1
|
||||
assert result.summary.input_file == ""
|
||||
# The orchestrator sets passed/failed from the validation report.
|
||||
# Without running validation it remains 0 — the CLI / API does that.
|
||||
assert result.summary.passed == 0
|
||||
|
||||
|
||||
def test_parse_preserves_raw_segments():
|
||||
result = parse(FIXTURE.read_text(), PayerConfig835.co_medicaid_835())
|
||||
c = result.claims[0]
|
||||
assert any(s[0] == "CLP" for s in c.raw_segments)
|
||||
assert any(s[0] == "SVC" for s in c.raw_segments)
|
||||
|
||||
|
||||
def test_parse_uses_generic_config():
|
||||
"""Generic config shouldn't suppress parser errors but accepts any payer."""
|
||||
result = parse(FIXTURE.read_text(), PayerConfig835.generic_835())
|
||||
assert len(result.claims) == 1
|
||||
|
||||
|
||||
def test_parse_no_bpr_raises():
|
||||
text = (
|
||||
"ISA*00* *00* *ZZ*COMEDASSISTPROG*ZZ*TPID001 *260617*1937*^*00501*991102990*1*P*:~"
|
||||
"ST*835*991102990*005010X221A1~"
|
||||
"SE*1*991102990~"
|
||||
"GE*1*991102990~"
|
||||
"IEA*1*991102990~"
|
||||
)
|
||||
with pytest.raises(CycloneParseError, match="No BPR"):
|
||||
parse(text, PayerConfig835.co_medicaid_835())
|
||||
@@ -0,0 +1,260 @@
|
||||
"""Tests for the 835 ERA validator rules."""
|
||||
from datetime import date
|
||||
from decimal import Decimal
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from cyclone.parsers.models import Address, BatchSummary, Envelope
|
||||
from cyclone.parsers.models_835 import (
|
||||
ClaimAdjustment,
|
||||
ClaimPayment,
|
||||
FinancialInfo,
|
||||
ParseResult835,
|
||||
Payer835,
|
||||
Payee835,
|
||||
ReassociationTrace,
|
||||
ServicePayment,
|
||||
)
|
||||
from cyclone.parsers.payer import PayerConfig835
|
||||
from cyclone.parsers.validator_835 import validate
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Builders
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def _build_result(**overrides) -> ParseResult835:
|
||||
"""Build a passing CO Medicaid 835 result; override fields to break rules."""
|
||||
financial = FinancialInfo(
|
||||
handling_code="C",
|
||||
paid_amount=Decimal("85.00"),
|
||||
credit_debit_flag="C",
|
||||
payment_method="ACH",
|
||||
payer_tax_id="81-1725341",
|
||||
payment_date=date(2026, 6, 17),
|
||||
)
|
||||
trace = ReassociationTrace(
|
||||
trace_type_code="1",
|
||||
trace_number="TPID001",
|
||||
originating_company_id="TPID001",
|
||||
payer_id="81-1725341",
|
||||
)
|
||||
payer = Payer835(
|
||||
name="CO_TXIX",
|
||||
id="7912900843",
|
||||
address=Address(line1="PO BOX 11000", city="DENVER", state="CO", zip="802110000"),
|
||||
)
|
||||
payee = Payee835(
|
||||
name="TOC, Inc.",
|
||||
npi="1881068062",
|
||||
address=Address(line1="1100 East Main St", city="Montrose", state="CO", zip="814014063"),
|
||||
)
|
||||
claim = ClaimPayment(
|
||||
payer_claim_control_number="CLM001",
|
||||
status_code="1",
|
||||
status_label="Processed as Primary",
|
||||
total_charge=Decimal("100.00"),
|
||||
total_paid=Decimal("85.00"),
|
||||
patient_responsibility=Decimal("15.00"),
|
||||
claim_filing_indicator="MC",
|
||||
original_claim_id="CLM001",
|
||||
facility_type="12",
|
||||
frequency_code="1",
|
||||
per_diem_covered_days=None,
|
||||
ref_benefit_plan=None,
|
||||
service_payments=[
|
||||
ServicePayment(
|
||||
line_number=1,
|
||||
procedure_qualifier="HC",
|
||||
procedure_code="S5130",
|
||||
modifiers=["U2"],
|
||||
charge=Decimal("60.00"),
|
||||
payment=Decimal("50.00"),
|
||||
units=Decimal("7.73"),
|
||||
unit_type="UN",
|
||||
service_date=date(2026, 6, 2),
|
||||
adjustments=[],
|
||||
ref_benefit_plan=None,
|
||||
),
|
||||
ServicePayment(
|
||||
line_number=2,
|
||||
procedure_qualifier="HC",
|
||||
procedure_code="S5130",
|
||||
modifiers=["U2"],
|
||||
charge=Decimal("40.00"),
|
||||
payment=Decimal("35.00"),
|
||||
units=Decimal("7.73"),
|
||||
unit_type="UN",
|
||||
service_date=date(2026, 6, 3),
|
||||
adjustments=[],
|
||||
ref_benefit_plan=None,
|
||||
),
|
||||
],
|
||||
raw_segments=[],
|
||||
)
|
||||
summary = BatchSummary(input_file="f", total_claims=1, passed=1, failed=0)
|
||||
result = ParseResult835(
|
||||
envelope=Envelope(
|
||||
sender_id="COMEDASSISTPROG",
|
||||
receiver_id="TPID001",
|
||||
control_number="991102984",
|
||||
transaction_date=date(2026, 6, 17),
|
||||
),
|
||||
financial_info=financial,
|
||||
trace=trace,
|
||||
payer=payer,
|
||||
payee=payee,
|
||||
claims=[claim],
|
||||
summary=summary,
|
||||
)
|
||||
if overrides:
|
||||
result = result.model_copy(update=overrides)
|
||||
return result
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Sanity: balanced claim passes
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def test_balanced_claim_passes_all_rules():
|
||||
cfg = PayerConfig835.co_medicaid_835()
|
||||
report = validate(_build_result(), cfg)
|
||||
assert report.passed is True
|
||||
assert report.errors == []
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# BPR rules
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def test_r835_bpr01_handling_code_disallowed():
|
||||
cfg = PayerConfig835.co_medicaid_835()
|
||||
result = _build_result(financial_info=_build_result().financial_info.model_copy(update={"handling_code": "Z"}))
|
||||
report = validate(result, cfg)
|
||||
assert any(i.rule == "R835_BPR01_handling_code_allowed" for i in report.errors)
|
||||
assert report.passed is False
|
||||
|
||||
|
||||
def test_r835_bpr02_positive():
|
||||
cfg = PayerConfig835.co_medicaid_835()
|
||||
result = _build_result(financial_info=_build_result().financial_info.model_copy(update={"paid_amount": Decimal("0.00")}))
|
||||
report = validate(result, cfg)
|
||||
assert any(i.rule == "R835_BPR02_positive" for i in report.errors)
|
||||
|
||||
|
||||
def test_r835_bal_bpr_vs_clp04_unbalanced():
|
||||
"""BPR02 (85.00) must equal sum of CLP04 (85.00). Off by 0.50 → error."""
|
||||
cfg = PayerConfig835.co_medicaid_835()
|
||||
result = _build_result(financial_info=_build_result().financial_info.model_copy(update={"paid_amount": Decimal("85.50")}))
|
||||
report = validate(result, cfg)
|
||||
assert any(i.rule == "R835_BAL_BPR_vs_CLP04" for i in report.errors)
|
||||
assert report.passed is False
|
||||
|
||||
|
||||
def test_r835_bal_bpr_vs_clp04_within_tolerance_passes():
|
||||
cfg = PayerConfig835.co_medicaid_835()
|
||||
result = _build_result(financial_info=_build_result().financial_info.model_copy(update={"paid_amount": Decimal("85.005")}))
|
||||
report = validate(result, cfg)
|
||||
# 0.005 difference is below tolerance (0.01) — no error.
|
||||
assert not any(i.rule == "R835_BAL_BPR_vs_CLP04" for i in report.errors)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# CLP rules
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def test_r835_bal_clp04_vs_svc03_unbalanced():
|
||||
cfg = PayerConfig835.co_medicaid_835()
|
||||
# The default fixture has CLP04=85.00 and SVC03 sum = 50+35 = 85.00.
|
||||
# Change one SVC payment to 40 → sum = 40+35 = 75, off by 10.
|
||||
base = _build_result()
|
||||
svc = base.claims[0].service_payments[0].model_copy(update={"payment": Decimal("40.00")})
|
||||
result = base.model_copy(update={"claims": [base.claims[0].model_copy(update={"service_payments": [svc, base.claims[0].service_payments[1]]})]})
|
||||
report = validate(result, cfg)
|
||||
assert any(i.rule == "R835_BAL_CLP04_vs_SVC03" for i in report.errors)
|
||||
assert "CLM001" in [i.message for i in report.errors if i.rule == "R835_BAL_CLP04_vs_SVC03"][0]
|
||||
|
||||
|
||||
def test_r835_clp01_present():
|
||||
cfg = PayerConfig835.co_medicaid_835()
|
||||
result = _build_result(claims=[_build_result().claims[0].model_copy(update={"payer_claim_control_number": ""})])
|
||||
report = validate(result, cfg)
|
||||
assert any(i.rule == "R835_CLP01_present" for i in report.errors)
|
||||
|
||||
|
||||
def test_r835_clp02_status_disallowed():
|
||||
cfg = PayerConfig835.co_medicaid_835()
|
||||
result = _build_result(claims=[_build_result().claims[0].model_copy(update={"status_code": "9", "status_label": None})])
|
||||
report = validate(result, cfg)
|
||||
assert any(i.rule == "R835_CLP02_status_allowed" for i in report.errors)
|
||||
|
||||
|
||||
def test_r835_clp03_present():
|
||||
cfg = PayerConfig835.co_medicaid_835()
|
||||
result = _build_result(claims=[_build_result().claims[0].model_copy(update={"total_charge": Decimal("0.00")})])
|
||||
report = validate(result, cfg)
|
||||
assert any(i.rule == "R835_CLP03_present" for i in report.errors)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Payer / payee rules
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def test_r835_payer_tax_id_mismatch_warning():
|
||||
cfg = PayerConfig835.co_medicaid_835()
|
||||
result = _build_result(financial_info=_build_result().financial_info.model_copy(update={"payer_tax_id": "99-9999999"}))
|
||||
report = validate(result, cfg)
|
||||
assert any(i.rule == "R835_PAYER_TAX_ID_matches" and i.severity == "warning" for i in report.warnings)
|
||||
# Warning only — report still passes.
|
||||
assert report.passed is True
|
||||
|
||||
|
||||
def test_r835_payer_tax_id_skipped_when_set_empty():
|
||||
cfg = PayerConfig835.co_medicaid_835()
|
||||
result = _build_result(financial_info=_build_result().financial_info.model_copy(update={"payer_tax_id": None}))
|
||||
report = validate(result, cfg)
|
||||
# When BPR10 is empty, the rule is skipped silently (per spec wording).
|
||||
assert not any(i.rule == "R835_PAYER_TAX_ID_matches" for i in report.errors + report.warnings)
|
||||
|
||||
|
||||
def test_r835_payer_health_plan_id_mismatch_warning():
|
||||
cfg = PayerConfig835.co_medicaid_835()
|
||||
result = _build_result(payer=_build_result().payer.model_copy(update={"id": "12345"}))
|
||||
report = validate(result, cfg)
|
||||
assert any(i.rule == "R835_PAYER_HEALTH_PLAN_ID_matches" and i.severity == "warning" for i in report.warnings)
|
||||
assert report.passed is True
|
||||
|
||||
|
||||
def test_r835_payee_npi_format():
|
||||
cfg = PayerConfig835.co_medicaid_835()
|
||||
result = _build_result(payee=_build_result().payee.model_copy(update={"npi": "12345"}))
|
||||
report = validate(result, cfg)
|
||||
assert any(i.rule == "R835_PAYEE_NPI_format" for i in report.errors)
|
||||
assert report.passed is False
|
||||
|
||||
|
||||
def test_r835_trn02_originating_company_id_present():
|
||||
cfg = PayerConfig835.co_medicaid_835()
|
||||
result = _build_result(trace=_build_result().trace.model_copy(update={"originating_company_id": ""}))
|
||||
report = validate(result, cfg)
|
||||
assert any(i.rule == "R835_REGN01_trn02_present" for i in report.errors)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Optional: warnings don't fail the report
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def test_warnings_do_not_set_passed_false():
|
||||
cfg = PayerConfig835.co_medicaid_835()
|
||||
# A wrong payer tax id is a warning; report should still pass.
|
||||
result = _build_result(financial_info=_build_result().financial_info.model_copy(update={"payer_tax_id": "BAD"}))
|
||||
report = validate(result, cfg)
|
||||
assert report.passed is True
|
||||
assert any(i.severity == "warning" for i in report.warnings)
|
||||
Reference in New Issue
Block a user