diff --git a/backend/src/cyclone/parsers/models.py b/backend/src/cyclone/parsers/models.py new file mode 100644 index 0000000..23860bc --- /dev/null +++ b/backend/src/cyclone/parsers/models.py @@ -0,0 +1,142 @@ +"""Pydantic v2 models for parsed 837P claims. + +Models mirror the X12 837P structure (Loop 2000A → 2000B → 2300 → 2400) but +flattened to one ``ClaimOutput`` per CLM segment, suitable for JSON serialization +and consumption by the Cyclone frontend. +""" + +from __future__ import annotations + +from datetime import date +from decimal import Decimal +from typing import Literal + +from pydantic import BaseModel, ConfigDict, Field, model_serializer + + +class _Base(BaseModel): + 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 + + +class Address(_Base): + line1: str + line2: str | None = None + city: str + state: str + zip: str + + +class BillingProvider(_Base): + name: str + npi: str + tax_id: str | None = None + address: Address | None = None + + +class Subscriber(_Base): + first_name: str + last_name: str + member_id: str + dob: date | None = None + gender: Literal["M", "F", "U"] | None = None + address: Address | None = None + + +class Payer(_Base): + name: str + id: str + + +class ClaimHeader(_Base): + claim_id: str + total_charge: Decimal + place_of_service: str | None = None + frequency_code: str | None = None + provider_signature: str | None = None + assignment: str | None = None + release_of_info: str | None = None + prior_auth: str | None = None + + +class Diagnosis(_Base): + code: str + qualifier: str | None = None + + +class Procedure(_Base): + qualifier: str + code: str + modifiers: list[str] = Field(default_factory=list) + + +class ServiceLine(_Base): + line_number: int + procedure: Procedure + charge: Decimal + unit_type: str | None = None + units: Decimal | None = None + place_of_service: str | None = None + service_date: date | None = None + provider_reference: str | None = None + + +class ValidationIssue(_Base): + rule: str + severity: Literal["error", "warning"] + message: str + segment_index: int | None = None + + +class ValidationReport(_Base): + passed: bool + errors: list[ValidationIssue] = Field(default_factory=list) + warnings: list[ValidationIssue] = Field(default_factory=list) + + +class Envelope(_Base): + sender_id: str + receiver_id: str + control_number: str + transaction_date: date + transaction_time: str | None = None + implementation_guide: str | None = None + + +class BatchSummary(_Base): + input_file: str + control_number: str | None = None + transaction_date: date | None = None + total_claims: int = 0 + passed: int = 0 + failed: int = 0 + failed_claim_ids: list[str] = Field(default_factory=list) + issues_by_rule: dict[str, int] = Field(default_factory=dict) + output_dir: str | None = None + + +class ClaimOutput(_Base): + claim_id: str + control_number: str + transaction_date: date + billing_provider: BillingProvider + subscriber: Subscriber + payer: Payer + claim: ClaimHeader + diagnoses: list[Diagnosis] = Field(default_factory=list) + service_lines: list[ServiceLine] = Field(default_factory=list) + validation: ValidationReport + raw_segments: list[list[str]] = Field(default_factory=list) + + +class ParseResult(_Base): + envelope: Envelope | None = None + claims: list[ClaimOutput] = Field(default_factory=list) + summary: BatchSummary diff --git a/backend/tests/test_models.py b/backend/tests/test_models.py new file mode 100644 index 0000000..1fb381a --- /dev/null +++ b/backend/tests/test_models.py @@ -0,0 +1,102 @@ +import json +from datetime import date +from decimal import Decimal + +import pytest +from pydantic import ValidationError as PydanticValidationError + +from cyclone.parsers.models import ( + Address, + BillingProvider, + ClaimHeader, + ClaimOutput, + Diagnosis, + Envelope, + Payer, + Procedure, + ServiceLine, + Subscriber, + ValidationIssue, + ValidationReport, +) + + +def test_address_round_trip(): + a = Address(line1="1100 East Main St", line2="Suite A", city="Montrose", state="CO", zip="81401") + d = a.model_dump() + assert d["line1"] == "1100 East Main St" + assert d["line2"] == "Suite A" + assert Address.model_validate(d) == a + + +def test_address_line2_optional(): + a = Address(line1="1 Main", city="X", state="CO", zip="80000") + assert a.line2 is None + + +def test_subscriber_splits_name(): + s = Subscriber(first_name="John", last_name="Doe", member_id="ABC", dob=date(1980, 1, 1), gender="M", address=Address(line1="x", city="x", state="CO", zip="80000")) + assert s.first_name == "John" + assert s.gender == "M" + + +def test_service_line_units_are_decimal(): + sl = ServiceLine( + line_number=1, + procedure=Procedure(qualifier="HC", code="99213", modifiers=[]), + charge=Decimal("100.00"), + unit_type="UN", + units=Decimal("1.0"), + ) + assert sl.units == Decimal("1.0") + + +def test_claim_header_total_charge_decimal(): + h = ClaimHeader(claim_id="C1", total_charge=Decimal("151.72"), place_of_service="12", frequency_code="1", provider_signature="Y", assignment="Y", release_of_info="Y", prior_auth="3173") + assert h.total_charge == Decimal("151.72") + + +def test_validation_issue_strict_severity(): + issue = ValidationIssue(rule="R001", severity="error", message="bad") + assert issue.severity == "error" + with pytest.raises(PydanticValidationError): + ValidationIssue(rule="R001", severity="WTF", message="bad") + + +def test_validation_report_passed_true_when_no_errors(): + r = ValidationReport(passed=True, errors=[], warnings=[]) + assert r.passed is True + + +def test_envelope_serializes_to_iso_date(): + e = Envelope( + sender_id="11525703", + receiver_id="COMEDASSISTPROG", + control_number="991102977", + transaction_date=date(2026, 6, 11), + transaction_time="081417", + implementation_guide="005010X222A1", + ) + d = e.model_dump() + assert d["transaction_date"] == "2026-06-11" + + +def test_claim_output_full_round_trip(): + claim = ClaimOutput( + claim_id="C1", + control_number="991102977", + transaction_date=date(2026, 6, 11), + billing_provider=BillingProvider(name="X", npi="1234567890", tax_id="123456789", address=Address(line1="1", city="x", state="CO", zip="80000")), + subscriber=Subscriber(first_name="J", last_name="D", member_id="M", dob=date(1980, 1, 1), gender="M", address=Address(line1="1", city="x", state="CO", zip="80000")), + payer=Payer(name="P", id="SKCO0"), + claim=ClaimHeader(claim_id="C1", total_charge=Decimal("100"), place_of_service="12", frequency_code="1", provider_signature="Y", assignment="Y", release_of_info="Y"), + diagnoses=[Diagnosis(code="R69", qualifier="ABK")], + service_lines=[], + validation=ValidationReport(passed=True, errors=[], warnings=[]), + raw_segments=[["CLM", "C1", "100"]], + ) + blob = json.loads(claim.model_dump_json()) + assert blob["claim_id"] == "C1" + assert blob["billing_provider"]["npi"] == "1234567890" + assert blob["raw_segments"] == [["CLM", "C1", "100"]] + assert ClaimOutput.model_validate(blob).claim_id == "C1"