From ce7fb50e79c16f3ea09439b3942ddef6e20260de Mon Sep 17 00:00:00 2001 From: Tyler Date: Fri, 19 Jun 2026 15:26:52 -0600 Subject: [PATCH 01/13] feat(backend): scaffold cyclone package skeleton --- backend/.gitignore | 9 +++++ backend/pyproject.toml | 29 ++++++++++++++++ backend/src/cyclone/__init__.py | 1 + backend/src/cyclone/__main__.py | 4 +++ backend/src/cyclone/parsers/__init__.py | 45 +++++++++++++++++++++++++ 5 files changed, 88 insertions(+) create mode 100644 backend/.gitignore create mode 100644 backend/pyproject.toml create mode 100644 backend/src/cyclone/__init__.py create mode 100644 backend/src/cyclone/__main__.py create mode 100644 backend/src/cyclone/parsers/__init__.py diff --git a/backend/.gitignore b/backend/.gitignore new file mode 100644 index 0000000..cd7bf33 --- /dev/null +++ b/backend/.gitignore @@ -0,0 +1,9 @@ +__pycache__/ +*.py[cod] +*.egg-info/ +.pytest_cache/ +.ruff_cache/ +.venv/ +venv/ +build/ +dist/ diff --git a/backend/pyproject.toml b/backend/pyproject.toml new file mode 100644 index 0000000..a245252 --- /dev/null +++ b/backend/pyproject.toml @@ -0,0 +1,29 @@ +[build-system] +requires = ["setuptools>=68", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "cyclone" +version = "0.1.0" +description = "Cyclone EDI suite — X12 837P/835 parser" +requires-python = ">=3.11" +dependencies = [ + "pydantic>=2.6,<3", + "click>=8.1,<9", +] + +[project.optional-dependencies] +dev = [ + "pytest>=8.0", + "pytest-cov>=4.1", +] + +[project.scripts] +cyclone = "cyclone.cli:main" + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +addopts = "-ra --strict-markers" diff --git a/backend/src/cyclone/__init__.py b/backend/src/cyclone/__init__.py new file mode 100644 index 0000000..3dc1f76 --- /dev/null +++ b/backend/src/cyclone/__init__.py @@ -0,0 +1 @@ +__version__ = "0.1.0" diff --git a/backend/src/cyclone/__main__.py b/backend/src/cyclone/__main__.py new file mode 100644 index 0000000..6fcbc3a --- /dev/null +++ b/backend/src/cyclone/__main__.py @@ -0,0 +1,4 @@ +from cyclone.cli import main + +if __name__ == "__main__": + main() diff --git a/backend/src/cyclone/parsers/__init__.py b/backend/src/cyclone/parsers/__init__.py new file mode 100644 index 0000000..1fab7a3 --- /dev/null +++ b/backend/src/cyclone/parsers/__init__.py @@ -0,0 +1,45 @@ +"""Public re-exports for the 837P parser.""" + +from cyclone.parsers.exceptions import CycloneParseError, CycloneValidationError +from cyclone.parsers.models import ( + Address, + BatchSummary, + BillingProvider, + ClaimHeader, + ClaimOutput, + Diagnosis, + Envelope, + ParseResult, + Payer, + Procedure, + ServiceLine, + Subscriber, + ValidationIssue, + ValidationReport, +) +from cyclone.parsers.payer import PayerConfig +from cyclone.parsers.segments import Delimiters, tokenize +from cyclone.parsers.parse_837 import parse + +__all__ = [ + "Address", + "BatchSummary", + "BillingProvider", + "ClaimHeader", + "ClaimOutput", + "CycloneParseError", + "CycloneValidationError", + "Delimiters", + "Diagnosis", + "Envelope", + "ParseResult", + "Payer", + "PayerConfig", + "Procedure", + "ServiceLine", + "Subscriber", + "ValidationIssue", + "ValidationReport", + "parse", + "tokenize", +] From 260b32afaccf247bce7362bdc1027ff543dd71e4 Mon Sep 17 00:00:00 2001 From: Tyler Date: Fri, 19 Jun 2026 15:31:06 -0600 Subject: [PATCH 02/13] feat(backend): add CycloneParseError and CycloneValidationError --- backend/src/cyclone/parsers/exceptions.py | 28 +++++++++++++++++++++++ backend/tests/test_exceptions.py | 21 +++++++++++++++++ 2 files changed, 49 insertions(+) create mode 100644 backend/src/cyclone/parsers/exceptions.py create mode 100644 backend/tests/test_exceptions.py diff --git a/backend/src/cyclone/parsers/exceptions.py b/backend/src/cyclone/parsers/exceptions.py new file mode 100644 index 0000000..ae44aa5 --- /dev/null +++ b/backend/src/cyclone/parsers/exceptions.py @@ -0,0 +1,28 @@ +"""Exception types raised by the parser. + +These are reserved for unrecoverable, file-level failures (bad envelope, no +ISA segment, malformed delimiters). Per-claim validation issues are not +exceptions — they are accumulated into ``ValidationReport``. +""" + +from __future__ import annotations + + +class CycloneParseError(Exception): + """Raised when the file itself cannot be parsed (bad envelope, etc.).""" + + def __init__(self, message: str, *, segment_index: int | None = None) -> None: + self.segment_index = segment_index + suffix = f" (segment {segment_index})" if segment_index is not None else "" + super().__init__(f"{message}{suffix}") + + +class CycloneValidationError(Exception): + """Raised by the CLI for envelope-level validation failures (no claims found).""" + + def __init__(self, message: str, *, rule_id: str | None = None, claim_id: str | None = None) -> None: + self.rule_id = rule_id + self.claim_id = claim_id + prefix = f"[{rule_id}] " if rule_id else "" + claim = f" (claim {claim_id})" if claim_id else "" + super().__init__(f"{prefix}{message}{claim}") diff --git a/backend/tests/test_exceptions.py b/backend/tests/test_exceptions.py new file mode 100644 index 0000000..05bea1b --- /dev/null +++ b/backend/tests/test_exceptions.py @@ -0,0 +1,21 @@ +import pytest +from cyclone.parsers.exceptions import CycloneParseError, CycloneValidationError + + +def test_parse_error_carries_segment_index(): + err = CycloneParseError("bad CLM", segment_index=42) + assert err.segment_index == 42 + assert "bad CLM" in str(err) + assert "segment 42" in str(err) + + +def test_validation_error_carries_rule_id(): + err = CycloneValidationError("NPI bad", rule_id="R020_npi_format", claim_id="C1") + assert err.rule_id == "R020_npi_format" + assert err.claim_id == "C1" + assert "[R020_npi_format]" in str(err) + + +def test_parse_error_inherits_from_exception(): + assert issubclass(CycloneParseError, Exception) + assert issubclass(CycloneValidationError, Exception) From 760d0315352c13a56380ac713c17164acd163ac3 Mon Sep 17 00:00:00 2001 From: Tyler Date: Fri, 19 Jun 2026 15:32:26 -0600 Subject: [PATCH 03/13] fix(backend): make parsers __init__ re-exports lazy via __getattr__ --- backend/src/cyclone/parsers/__init__.py | 101 ++++++++++++++---------- 1 file changed, 59 insertions(+), 42 deletions(-) diff --git a/backend/src/cyclone/parsers/__init__.py b/backend/src/cyclone/parsers/__init__.py index 1fab7a3..23d9be7 100644 --- a/backend/src/cyclone/parsers/__init__.py +++ b/backend/src/cyclone/parsers/__init__.py @@ -1,45 +1,62 @@ -"""Public re-exports for the 837P parser.""" +"""Public re-exports for the 837P parser. +Re-exports are lazy via PEP 562 ``__getattr__`` so that importing a name from +this package (e.g. ``from cyclone.parsers.exceptions import CycloneParseError``) +does not require every referenced module to be importable. Downstream code can +write ``cyclone.parsers.PayerConfig`` and it will be loaded on first access. +""" + +from __future__ import annotations + +# Eager: only the exceptions module is needed early and unconditionally. from cyclone.parsers.exceptions import CycloneParseError, CycloneValidationError -from cyclone.parsers.models import ( - Address, - BatchSummary, - BillingProvider, - ClaimHeader, - ClaimOutput, - Diagnosis, - Envelope, - ParseResult, - Payer, - Procedure, - ServiceLine, - Subscriber, - ValidationIssue, - ValidationReport, -) -from cyclone.parsers.payer import PayerConfig -from cyclone.parsers.segments import Delimiters, tokenize -from cyclone.parsers.parse_837 import parse -__all__ = [ - "Address", - "BatchSummary", - "BillingProvider", - "ClaimHeader", - "ClaimOutput", - "CycloneParseError", - "CycloneValidationError", - "Delimiters", - "Diagnosis", - "Envelope", - "ParseResult", - "Payer", - "PayerConfig", - "Procedure", - "ServiceLine", - "Subscriber", - "ValidationIssue", - "ValidationReport", - "parse", - "tokenize", -] +_LAZY_EXPORTS: dict[str, str] = { + # exceptions + "CycloneParseError": "cyclone.parsers.exceptions", + "CycloneValidationError": "cyclone.parsers.exceptions", + # models + "Address": "cyclone.parsers.models", + "BatchSummary": "cyclone.parsers.models", + "BillingProvider": "cyclone.parsers.models", + "ClaimHeader": "cyclone.parsers.models", + "ClaimOutput": "cyclone.parsers.models", + "Diagnosis": "cyclone.parsers.models", + "Envelope": "cyclone.parsers.models", + "ParseResult": "cyclone.parsers.models", + "Payer": "cyclone.parsers.models", + "Procedure": "cyclone.parsers.models", + "ServiceLine": "cyclone.parsers.models", + "Subscriber": "cyclone.parsers.models", + "ValidationIssue": "cyclone.parsers.models", + "ValidationReport": "cyclone.parsers.models", + # payer + "PayerConfig": "cyclone.parsers.payer", + # segments + "Delimiters": "cyclone.parsers.segments", + "tokenize": "cyclone.parsers.segments", + # orchestrator + "parse": "cyclone.parsers.parse_837", +} + +_cache: dict[str, object] = {} + + +def __getattr__(name: str) -> object: + module_path = _LAZY_EXPORTS.get(name) + if module_path is None: + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") + if name in _cache: + return _cache[name] + import importlib + mod = importlib.import_module(module_path) + value = getattr(mod, name) + _cache[name] = value + return value + + +def __dir__() -> list[str]: + return sorted(list(_LAZY_EXPORTS.keys())) + + +__all__ = sorted(_LAZY_EXPORTS.keys()) From 8475e672a7d23c6788f18b3793943f8285f5de95 Mon Sep 17 00:00:00 2001 From: Tyler Date: Fri, 19 Jun 2026 15:36:01 -0600 Subject: [PATCH 04/13] feat(backend): X12 segment tokenizer with ISA delimiter detection --- backend/src/cyclone/parsers/segments.py | 80 +++++++++++++++++++++++++ backend/tests/test_segments.py | 69 +++++++++++++++++++++ 2 files changed, 149 insertions(+) create mode 100644 backend/src/cyclone/parsers/segments.py create mode 100644 backend/tests/test_segments.py diff --git a/backend/src/cyclone/parsers/segments.py b/backend/src/cyclone/parsers/segments.py new file mode 100644 index 0000000..f2d72e5 --- /dev/null +++ b/backend/src/cyclone/parsers/segments.py @@ -0,0 +1,80 @@ +"""X12 segment tokenizer. + +The ISA segment is a fixed 106-character record (101 + segment terminator) that +declares the four delimiters used in the rest of the file: + +- element separator (3rd char, 0-indexed) +- repetition separator (82nd char, 0-indexed; ISA11) +- component separator (104th char, 0-indexed; ISA16) +- segment terminator (105th char, 0-indexed) + +We do not split components here — that is left to the caller, since most X12 +elements are simple and splitting unconditionally would corrupt data. +""" + +from __future__ import annotations + +from dataclasses import dataclass + +from cyclone.parsers.exceptions import CycloneParseError + + +@dataclass(frozen=True) +class Delimiters: + element: str # usually "*" + repetition: str # usually "^" + component: str # usually ":" + segment: str # usually "~" + + +def _detect_delimiters(isa_segment: str) -> Delimiters: + """Pull the four delimiters from the fixed-width ISA segment. + + The ISA segment must be exactly 106 characters (including the terminator). + """ + if len(isa_segment) < 106: + raise CycloneParseError( + f"ISA segment too short: expected 106 chars, got {len(isa_segment)}" + ) + return Delimiters( + element=isa_segment[3], + repetition=isa_segment[82], + component=isa_segment[104], + segment=isa_segment[105], + ) + + +def tokenize(text: str) -> list[list[str]]: + """Split a raw X12 document into a list of segments (each a list of elements). + + The first segment must be an ISA. Raises :class:`CycloneParseError` on + missing or malformed envelopes. + """ + # Normalize line endings; the segment terminator handles real segmentation. + text = text.replace("\r\n", "").replace("\n", "").replace("\r", "") + + # Find the first ISA by scanning for the literal "ISA" + element separator. + if not text.startswith("ISA") or len(text) < 4: + raise CycloneParseError("ISA segment not found at start of file") + + # The first 106 chars are the ISA segment. + isa_segment = text[:106] + try: + delimiters = _detect_delimiters(isa_segment) + except IndexError as exc: # pragma: no cover - guarded by length check + raise CycloneParseError(f"Could not parse ISA delimiters: {exc}") from exc + + # Slice off the ISA, then split the rest on the segment terminator. + body = text[106:] + raw_segments = body.split(delimiters.segment) + + # Prepend the ISA so callers always see it as the first segment. + isa_elements = isa_segment.split(delimiters.element) + out: list[list[str]] = [isa_elements] + + for raw in raw_segments: + if not raw: + continue + out.append(raw.split(delimiters.element)) + + return out diff --git a/backend/tests/test_segments.py b/backend/tests/test_segments.py new file mode 100644 index 0000000..2233940 --- /dev/null +++ b/backend/tests/test_segments.py @@ -0,0 +1,69 @@ +import pytest +from cyclone.parsers.exceptions import CycloneParseError +from cyclone.parsers.segments import Delimiters, tokenize + +ISA_PREFIX = "ISA*00* *00* *ZZ*11525703 *ZZ*COMEDASSISTPROG*260611*0814*" + + +def test_tokenize_parses_isa_delimiters(): + text = ISA_PREFIX + "*^00501*991102977*1*P*:~" + segments = tokenize(text) + assert segments[0][0] == "ISA" + assert len(segments) == 1 + + +def test_tokenize_splits_segments(): + text = ISA_PREFIX + "*^00501*991102977*1*P*:~ST*837*0001~SE*2*0001~" + segments = tokenize(text) + assert [s[0] for s in segments] == ["ISA", "ST", "SE"] + + +def test_tokenize_splits_elements(): + text = ISA_PREFIX + "*^00501*991102977*1*P*:~NM1*85*2*Acme*****XX*1234567890~" + segments = tokenize(text) + nm1 = [s for s in segments if s[0] == "NM1"][0] + assert nm1[1] == "85" + assert nm1[8] == "XX" + assert nm1[9] == "1234567890" + + +def test_tokenize_preserves_empty_elements(): + """Empty elements (e.g. 'NM1*IL*1*Doe*John****MI*X') must stay as empty strings.""" + text = ISA_PREFIX + "*^00501*991102977*1*P*:~NM1*IL*1*Doe*John****MI*X~" + segments = tokenize(text) + nm1 = [s for s in segments if s[0] == "NM1"][0] + assert nm1[5] == "" # position 5 is the 4th consecutive empty after suffix + assert nm1[6] == "" + + +def test_tokenize_handles_component_separator(): + """SV101 can be 'HC:99213' which is component-split at ':'.""" + text = ISA_PREFIX + "*^00501*991102977*1*P*:~SV1*HC:99213*100*UN*1***1~" + segments = tokenize(text) + sv1 = [s for s in segments if s[0] == "SV1"][0] + # SV101 is element index 1; components 1, 2 are 'HC', '99213' + assert sv1[1] == "HC:99213" + + +def test_tokenize_strips_trailing_crlf(): + text = ISA_PREFIX + "*^00501*991102977*1*P*:~\r\n" + segments = tokenize(text) + assert [s[0] for s in segments] == ["ISA"] + + +def test_tokenize_raises_on_missing_isa(): + with pytest.raises(CycloneParseError, match="ISA segment not found"): + tokenize("hello world") + + +def test_tokenize_raises_on_short_isa(): + with pytest.raises(CycloneParseError, match="ISA segment too short"): + tokenize("ISA*00*00~") + + +def test_delimiters_dataclass(): + d = Delimiters(element="*", repetition="^", component=":", segment="~") + assert d.element == "*" + assert d.repetition == "^" + assert d.component == ":" + assert d.segment == "~" From aa801805ced31327278d1016d5b9e0f3f205f6bd Mon Sep 17 00:00:00 2001 From: Tyler Date: Fri, 19 Jun 2026 15:39:08 -0600 Subject: [PATCH 05/13] feat(backend): Pydantic v2 models for 837P claims --- backend/src/cyclone/parsers/models.py | 142 ++++++++++++++++++++++++++ backend/tests/test_models.py | 102 ++++++++++++++++++ 2 files changed, 244 insertions(+) create mode 100644 backend/src/cyclone/parsers/models.py create mode 100644 backend/tests/test_models.py 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" From bb35ecd40cf6082de17b611411b5e5444c25742e Mon Sep 17 00:00:00 2001 From: Tyler Date: Fri, 19 Jun 2026 15:40:09 -0600 Subject: [PATCH 06/13] feat(backend): PayerConfig with co_medicaid and generic factories --- backend/src/cyclone/parsers/payer.py | 59 ++++++++++++++++++++++++++++ backend/tests/test_payer.py | 22 +++++++++++ 2 files changed, 81 insertions(+) create mode 100644 backend/src/cyclone/parsers/payer.py create mode 100644 backend/tests/test_payer.py diff --git a/backend/src/cyclone/parsers/payer.py b/backend/src/cyclone/parsers/payer.py new file mode 100644 index 0000000..ea30534 --- /dev/null +++ b/backend/src/cyclone/parsers/payer.py @@ -0,0 +1,59 @@ +"""Payer-specific configuration for the 837P parser. + +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. +""" + +from __future__ import annotations + +from pydantic import BaseModel, ConfigDict, Field + + +class PayerConfig(BaseModel): + model_config = ConfigDict(frozen=True) + + name: str + sbr09_claim_filing: str = "MC" + allowed_claim_frequencies: set[int] = Field(default_factory=lambda: {1, 7, 8}) + require_ref_g1_for_adjustments: bool = False + allowed_bht06: set[str] = Field(default_factory=lambda: {"CH"}) + payer_id: str = "" + payer_name: str = "" + no_patient_loop: bool = False + encounter_claim_in_same_batch: bool = False + + @classmethod + def co_medicaid(cls) -> "PayerConfig": + """Defaults for Colorado Medical Assistance Program (FFS). + + Source: ``docs/companionguides/837p.md``. + """ + return cls( + name="Colorado Medical Assistance Program", + sbr09_claim_filing="MC", + allowed_claim_frequencies={1, 7, 8}, + # Lenient in v1 — see spec §9 R031. + require_ref_g1_for_adjustments=False, + allowed_bht06={"CH"}, + payer_id="SKCO0", + payer_name="COHCPF", + no_patient_loop=True, + encounter_claim_in_same_batch=False, + ) + + @classmethod + def generic_837p(cls) -> "PayerConfig": + """Relaxed defaults for an unknown payer. Structural rules only.""" + return cls( + name="Generic 837P", + sbr09_claim_filing="", + allowed_claim_frequencies={1, 2, 3, 4, 5, 6, 7, 8, 9}, + require_ref_g1_for_adjustments=False, + allowed_bht06={"CH", "RP"}, + payer_id="", + payer_name="", + no_patient_loop=False, + encounter_claim_in_same_batch=True, + ) diff --git a/backend/tests/test_payer.py b/backend/tests/test_payer.py new file mode 100644 index 0000000..814eff2 --- /dev/null +++ b/backend/tests/test_payer.py @@ -0,0 +1,22 @@ +from cyclone.parsers.payer import PayerConfig + + +def test_co_medicaid_defaults(): + cfg = PayerConfig.co_medicaid() + assert cfg.name == "Colorado Medical Assistance Program" + assert cfg.sbr09_claim_filing == "MC" + assert cfg.allowed_claim_frequencies == {1, 7, 8} + assert cfg.require_ref_g1_for_adjustments is False # lenient in v1 + assert cfg.allowed_bht06 == {"CH"} + assert cfg.payer_id == "SKCO0" + assert cfg.payer_name == "COHCPF" + assert cfg.no_patient_loop is True + assert cfg.encounter_claim_in_same_batch is False + + +def test_generic_837p_defaults_are_relaxed(): + cfg = PayerConfig.generic_837p() + assert cfg.name == "Generic 837P" + assert cfg.allowed_claim_frequencies == {1, 2, 3, 4, 5, 6, 7, 8, 9} + assert cfg.no_patient_loop is False + assert cfg.payer_id == "" # unset From 9db793c108f961b3b801f093296d411ecdc1a34e Mon Sep 17 00:00:00 2001 From: Tyler Date: Fri, 19 Jun 2026 15:42:04 -0600 Subject: [PATCH 07/13] test(backend): add minimal 837P fixture --- backend/tests/fixtures/minimal_837p.txt | 30 +++++++++++++++++++++++++ 1 file changed, 30 insertions(+) create mode 100644 backend/tests/fixtures/minimal_837p.txt diff --git a/backend/tests/fixtures/minimal_837p.txt b/backend/tests/fixtures/minimal_837p.txt new file mode 100644 index 0000000..6e19e83 --- /dev/null +++ b/backend/tests/fixtures/minimal_837p.txt @@ -0,0 +1,30 @@ +ISA*00* *00* *ZZ*11525703 *ZZ*COMEDASSISTPROG*260611*0814*^*00501*991102977*1*P*:~ +GS*HC*11525703*COMEDASSISTPROG*20260611*081417*991102977*X*005010X222A1~ +ST*837*991102977*005010X222A1~ +BHT*0019*00*ref-001*20260611*081417*CH~ +NM1*41*2*Test Submitter*****46*11525703~ +PER*IC*Test Contact*EM*test@example.com~ +NM1*40*2*COLORADO MEDICAL ASSISTANCE PROGRAM*****46*COMEDASSISTPROG~ +HL*1**20*1~ +PRV*BI*PXC*251E00000X~ +NM1*85*2*Test Provider Inc*****XX*1234567890~ +N3*123 Test St~ +N4*Denver*CO*80202~ +REF*EI*123456789~ +HL*2*1*22*0~ +SBR*P*18*******MC~ +NM1*IL*1*Doe*John****MI*ABC123~ +N3*456 Member St~ +N4*Denver*CO*80203~ +DMG*D8*19800101*M~ +NM1*PR*2*COHCPF*****PI*SKCO0~ +CLM*CLM001*100.00***12:B:1*Y*A*Y*Y~ +REF*G1*PA123~ +HI*ABK:Z00~ +LX*1~ +SV1*HC:99213*100.00*UN*1***1~ +DTP*472*D8*20260611~ +REF*6R*REF001~ +SE*26*991102977~ +GE*1*991102977~ +IEA*1*991102977~ \ No newline at end of file From 1cab95ffc37f12945accdc365a6590e0df8459eb Mon Sep 17 00:00:00 2001 From: Tyler Date: Fri, 19 Jun 2026 15:45:17 -0600 Subject: [PATCH 08/13] feat(backend): structural + CO Medicaid validation rules --- backend/src/cyclone/parsers/validator.py | 115 +++++++++++++++++ backend/tests/test_validator.py | 152 +++++++++++++++++++++++ 2 files changed, 267 insertions(+) create mode 100644 backend/src/cyclone/parsers/validator.py create mode 100644 backend/tests/test_validator.py diff --git a/backend/src/cyclone/parsers/validator.py b/backend/src/cyclone/parsers/validator.py new file mode 100644 index 0000000..f64cf79 --- /dev/null +++ b/backend/src/cyclone/parsers/validator.py @@ -0,0 +1,115 @@ +"""Structural and payer-specific validation rules for 837P claims. + +Each rule is a pure function that takes a :class:`ClaimOutput` and a +:class:`PayerConfig` and yields zero or more :class:`ValidationIssue`. The +top-level :func:`validate` function runs every rule and aggregates results. +""" + +from __future__ import annotations + +import re +from collections.abc import Iterable +from decimal import Decimal +from typing import Callable + +from cyclone.parsers.models import ClaimOutput, ValidationIssue, ValidationReport +from cyclone.parsers.payer import PayerConfig + +NPI_RE = re.compile(r"^\d{10}$") + +Rule = Callable[[ClaimOutput, PayerConfig], Iterable[ValidationIssue]] + + +def _r010_clm01_present(claim: ClaimOutput, _: PayerConfig) -> Iterable[ValidationIssue]: + if not claim.claim.claim_id or not claim.claim.claim_id.strip(): + yield ValidationIssue(rule="R010_clm01_present", severity="error", message="CLM01 (claim id) is empty") + + +def _r011_total_charge_positive(claim: ClaimOutput, _: PayerConfig) -> Iterable[ValidationIssue]: + if claim.claim.total_charge <= 0: + yield ValidationIssue(rule="R011_total_charge_positive", severity="error", message=f"CLM02 must be > 0, got {claim.claim.total_charge}") + + +def _r020_npi_format(claim: ClaimOutput, _: PayerConfig) -> Iterable[ValidationIssue]: + if claim.billing_provider.npi and not NPI_RE.match(claim.billing_provider.npi): + yield ValidationIssue(rule="R020_npi_format", severity="error", message=f"Billing provider NPI must be 10 digits, got {claim.billing_provider.npi!r}") + + +def _r030_frequency_allowed(claim: ClaimOutput, cfg: PayerConfig) -> Iterable[ValidationIssue]: + if not claim.claim.frequency_code: + return + try: + freq = int(claim.claim.frequency_code) + except ValueError: + yield ValidationIssue(rule="R030_frequency_allowed", severity="error", message=f"CLM05-3 not numeric: {claim.claim.frequency_code!r}") + return + if freq not in cfg.allowed_claim_frequencies: + yield ValidationIssue( + rule="R030_frequency_allowed", + severity="error", + message=f"CLM05-3 {freq} not in {sorted(cfg.allowed_claim_frequencies)} for payer {cfg.name}", + ) + + +def _r031_ref_g1_optional(claim: ClaimOutput, _: PayerConfig) -> Iterable[ValidationIssue]: + """REF*G1 is informational in v1 (see spec §9 R031).""" + # We yield no errors; presence is recorded on the model. + return () + + +def _r050_diagnosis_present(claim: ClaimOutput, _: PayerConfig) -> Iterable[ValidationIssue]: + if not claim.diagnoses: + yield ValidationIssue(rule="R050_diagnosis_present", severity="error", message="HI segment missing — no diagnoses on claim") + + +def _r060_service_dates_present(claim: ClaimOutput, _: PayerConfig) -> Iterable[ValidationIssue]: + for sl in claim.service_lines: + if sl.service_date is None: + yield ValidationIssue(rule="R060_service_dates_present", severity="error", message=f"Service line {sl.line_number} missing service date (DTP*472)") + + +def _r070_charges_sum(claim: ClaimOutput, _: PayerConfig) -> Iterable[ValidationIssue]: + if not claim.service_lines: + return + line_sum = sum((sl.charge for sl in claim.service_lines), start=Decimal("0")) + if abs(line_sum - claim.claim.total_charge) > Decimal("0.01"): + yield ValidationIssue( + rule="R070_charges_sum", + severity="warning", + message=f"Sum of service lines ({line_sum}) does not match CLM02 ({claim.claim.total_charge})", + ) + + +def _r100_payer_id_matches(claim: ClaimOutput, cfg: PayerConfig) -> Iterable[ValidationIssue]: + if cfg.payer_id and claim.payer.id and claim.payer.id != cfg.payer_id: + yield ValidationIssue( + rule="R100_payer_id_matches", + severity="warning", + message=f"Payer id {claim.payer.id!r} != configured {cfg.payer_id!r}", + ) + + +_RULES: list[Rule] = [ + _r010_clm01_present, + _r011_total_charge_positive, + _r020_npi_format, + _r030_frequency_allowed, + _r031_ref_g1_optional, + _r050_diagnosis_present, + _r060_service_dates_present, + _r070_charges_sum, + _r100_payer_id_matches, +] + + +def validate(claim: ClaimOutput, config: PayerConfig) -> ValidationReport: + """Run every rule against a claim and return a report. + + Never raises — all issues are accumulated into the report. + """ + errors: list[ValidationIssue] = [] + warnings: list[ValidationIssue] = [] + for rule in _RULES: + for issue in rule(claim, config): + (errors if issue.severity == "error" else warnings).append(issue) + return ValidationReport(passed=not errors, errors=errors, warnings=warnings) diff --git a/backend/tests/test_validator.py b/backend/tests/test_validator.py new file mode 100644 index 0000000..0bd016f --- /dev/null +++ b/backend/tests/test_validator.py @@ -0,0 +1,152 @@ +from datetime import date +from decimal import Decimal + +import pytest + +from cyclone.parsers.models import ( + Address, + BillingProvider, + ClaimHeader, + ClaimOutput, + Diagnosis, + Envelope, + Payer, + Procedure, + ServiceLine, + Subscriber, + ValidationReport, +) +from cyclone.parsers.payer import PayerConfig +from cyclone.parsers.validator import validate + + +def _build_claim(**overrides) -> ClaimOutput: + """Build a passing claim for tests; override fields to break specific rules.""" + base = dict( + claim_id="C1", + control_number="991102977", + transaction_date=date(2026, 6, 11), + billing_provider=BillingProvider( + name="Test Provider", + npi="1234567890", + tax_id="123456789", + address=Address(line1="1 Main", city="X", state="CO", zip="80000"), + ), + subscriber=Subscriber( + first_name="John", + last_name="Doe", + member_id="M1", + dob=date(1980, 1, 1), + gender="M", + address=Address(line1="1 Main", city="X", state="CO", zip="80000"), + ), + payer=Payer(name="COHCPF", id="SKCO0"), + claim=ClaimHeader( + claim_id="C1", + total_charge=Decimal("100.00"), + place_of_service="12", + frequency_code="1", + provider_signature="Y", + assignment="Y", + release_of_info="Y", + ), + diagnoses=[Diagnosis(code="Z00", qualifier="ABK")], + service_lines=[ + ServiceLine( + line_number=1, + procedure=Procedure(qualifier="HC", code="99213", modifiers=[]), + charge=Decimal("100.00"), + unit_type="UN", + units=Decimal("1.0"), + service_date=date(2026, 6, 11), + ) + ], + validation=ValidationReport(passed=True, errors=[], warnings=[]), + raw_segments=[], + ) + base.update(overrides) + return ClaimOutput(**base) + + +def test_validate_passing_claim(): + cfg = PayerConfig.co_medicaid() + report = validate(_build_claim(), cfg) + assert report.passed is True + assert report.errors == [] + + +def test_r010_clm01_required(): + cfg = PayerConfig.co_medicaid() + claim = _build_claim() + claim.claim.claim_id = "" + report = validate(claim, cfg) + assert not report.passed + assert any(i.rule == "R010_clm01_present" for i in report.errors) + + +def test_r011_total_charge_positive(): + cfg = PayerConfig.co_medicaid() + claim = _build_claim() + claim.claim.total_charge = Decimal("0.00") + report = validate(claim, cfg) + assert any(i.rule == "R011_total_charge_positive" for i in report.errors) + + +def test_r020_npi_must_be_ten_digits(): + cfg = PayerConfig.co_medicaid() + claim = _build_claim() + claim.billing_provider.npi = "12345" + report = validate(claim, cfg) + assert any(i.rule == "R020_npi_format" for i in report.errors) + + +def test_r030_frequency_allowed(): + cfg = PayerConfig.co_medicaid() # only 1, 7, 8 + claim = _build_claim() + claim.claim.frequency_code = "5" + report = validate(claim, cfg) + assert any(i.rule == "R030_frequency_allowed" for i in report.errors) + + +def test_r031_ref_g1_optional_no_error(): + """R031 is informational in v1 — no REF*G1 should not error.""" + cfg = PayerConfig.co_medicaid() + claim = _build_claim() + report = validate(claim, cfg) + assert not any(i.rule == "R031_ref_g1_optional" and i.severity == "error" for i in report.errors) + + +def test_r050_diagnosis_required(): + cfg = PayerConfig.co_medicaid() + claim = _build_claim(diagnoses=[]) + report = validate(claim, cfg) + assert any(i.rule == "R050_diagnosis_present" for i in report.errors) + + +def test_r060_service_dates_required(): + cfg = PayerConfig.co_medicaid() + claim = _build_claim() + claim.service_lines[0].service_date = None + report = validate(claim, cfg) + assert any(i.rule == "R060_service_dates_present" for i in report.errors) + # Now add the date: + claim.service_lines[0].service_date = date(2026, 6, 11) + report = validate(claim, cfg) + assert not any(i.rule == "R060_service_dates_present" for i in report.errors + report.warnings) + + +def test_r070_charges_sum_warning(): + cfg = PayerConfig.co_medicaid() + claim = _build_claim() + claim.claim.total_charge = Decimal("999.00") # mismatch + report = validate(claim, cfg) + assert any(i.rule == "R070_charges_sum" and i.severity == "warning" for i in report.warnings) + + +def test_r100_payer_id_warning_only(): + cfg = PayerConfig.co_medicaid() + claim = _build_claim() + claim.payer.id = "WRONG" + report = validate(claim, cfg) + assert any(i.rule == "R100_payer_id_matches" and i.severity == "warning" for i in report.warnings) + assert report.passed is True From 1f7cb11fd59ed1e758c19e9b1f5e4d584a8b37dd Mon Sep 17 00:00:00 2001 From: Tyler Date: Fri, 19 Jun 2026 15:52:58 -0600 Subject: [PATCH 09/13] feat(backend): 837P parser orchestrator with HL walker --- backend/src/cyclone/parsers/parse_837.py | 399 +++++++++++++++++++++++ backend/tests/test_parse_837.py | 70 ++++ 2 files changed, 469 insertions(+) create mode 100644 backend/src/cyclone/parsers/parse_837.py create mode 100644 backend/tests/test_parse_837.py diff --git a/backend/src/cyclone/parsers/parse_837.py b/backend/src/cyclone/parsers/parse_837.py new file mode 100644 index 0000000..27c0b39 --- /dev/null +++ b/backend/src/cyclone/parsers/parse_837.py @@ -0,0 +1,399 @@ +r"""Orchestrate parsing an X12 837P file into a :class:`ParseResult`. + +Single-pass HL walker: + +- ``HL*1**20*1`` = Loop 2000A (Billing Provider) +- ``HL*n*1*22*0`` = Loop 2000B (Subscriber), one per subscriber +- Inside each subscriber loop, every CLM segment = one :class:`ClaimOutput` +- Inside each claim, LX/SV1 pairs = :class:`ServiceLine` rows + +Errors at the file level raise :class:`CycloneParseError`. Errors at the +claim level are caught and accumulated into ``ClaimOutput.validation``. +""" + +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, + BillingProvider, + ClaimHeader, + ClaimOutput, + Diagnosis, + Envelope, + Payer, + ParseResult, + Procedure, + ServiceLine, + Subscriber, + ValidationIssue, + ValidationReport, +) +from cyclone.parsers.payer import PayerConfig +from cyclone.parsers.segments import tokenize +from cyclone.parsers.validator import validate + +log = logging.getLogger(__name__) + + +def _parse_isa_date(yymmdd: str) -> date: + if len(yymmdd) == 8: + # BHT04 uses CCYYMMDD (8 chars); ISA06/GS04 may also. + year = int(yymmdd[0:4]) + mm = int(yymmdd[4:6]) + dd = int(yymmdd[6:8]) + else: + yy = int(yymmdd[0:2]) + mm = int(yymmdd[2:4]) + dd = int(yymmdd[4:6]) + # Pivot year 30: 00-29 = 2000s, 30-99 = 1900s. Reasonable for 837P. + year = 2000 + yy if yy < 30 else 1900 + yy + return date(year, mm, dd) + + +def _build_envelope(segments: list[list[str]], input_file: str = "") -> tuple[Envelope | None, BatchSummary]: + 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), # placeholder; overwritten by BHT + implementation_guide=None, + ) + except (IndexError, ValueError) as exc: + raise CycloneParseError(f"Bad ISA: {exc}") from exc + elif seg[0] == "BHT" and envelope is not None: + try: + envelope = envelope.model_copy(update={"transaction_date": _parse_isa_date(seg[4])}) + if len(seg) > 5: + envelope = envelope.model_copy(update={"transaction_time": seg[5]}) + except (IndexError, ValueError) as exc: + log.warning("Could not parse BHT date: %s", exc) + elif seg[0] == "ST" and envelope is not None: + if len(seg) > 3: + envelope = envelope.model_copy(update={"implementation_guide": seg[3]}) + return envelope, summary + + +def _consume_billing_provider(segments: list[list[str]], idx: int) -> tuple[BillingProvider, int]: + """Read NM1*85 / N3 / N4 / REF*EI between ``idx`` and the next HL.""" + npi = "" + name = "" + tax_id: str | None = None + addr: Address | None = None + while idx < len(segments) and segments[idx][0] != "HL": + seg = segments[idx] + if seg[0] == "NM1" and len(seg) > 8 and seg[1] == "85": + name = seg[3] + for j, qual in enumerate(seg): + if qual == "XX" and j + 1 < len(seg): + npi = seg[j + 1] + break + elif 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_}) + elif seg[0] == "REF" and len(seg) > 2 and seg[1] == "EI": + tax_id = seg[2] + idx += 1 + if addr is None: + addr = Address(line1="", city="", state="", zip="") + return BillingProvider(name=name, npi=npi, tax_id=tax_id, address=addr), idx + + +def _consume_subscriber(segments: list[list[str]], idx: int) -> tuple[Subscriber, int]: + """Read NM1*IL / N3 / N4 / DMG between ``idx`` and the next HL/CLM.""" + first = "" + last = "" + member_id = "" + dob: date | None = None + gender: str | None = None + addr: Address | None = None + while idx < len(segments) and segments[idx][0] not in {"HL", "CLM"}: + seg = segments[idx] + # Stop at the payer (NM1*PR) or patient (NM1*QC) — those belong to the next loop + if seg[0] == "NM1" and len(seg) > 1 and seg[1] in {"PR", "QC"}: + break + if seg[0] == "NM1" and len(seg) > 3 and seg[1] == "IL": + last = seg[3] if len(seg) > 3 else "" + first = seg[4] if len(seg) > 4 else "" + for j, qual in enumerate(seg): + if qual == "MI" and j + 1 < len(seg): + member_id = seg[j + 1] + break + elif seg[0] == "N3": + line1 = seg[1] if len(seg) > 1 else "" + addr = (addr or Address(line1=line1, city="", state="", zip="")).model_copy(update={"line1": line1}) + 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_}) + elif seg[0] == "DMG" and len(seg) > 2 and seg[1] == "D8": + try: + dob = _parse_isa_date(seg[2]) + except (ValueError, IndexError): + dob = None + gender = seg[3] if len(seg) > 3 else None + idx += 1 + if addr is None: + addr = Address(line1="", city="", state="", zip="") + return Subscriber(first_name=first, last_name=last, member_id=member_id, dob=dob, gender=gender, address=addr), idx + + +def _consume_payer(segments: list[list[str]], idx: int) -> tuple[Payer, int]: + name = "" + pid = "" + while idx < len(segments) and segments[idx][0] not in {"HL", "CLM"}: + seg = segments[idx] + # Stop at patient loop (NM1*QC) — we already have the payer. + if seg[0] == "NM1" and len(seg) > 1 and seg[1] == "QC": + break + if seg[0] == "NM1" and len(seg) > 3 and seg[1] == "PR": + name = seg[3] if len(seg) > 3 else "" + for j, qual in enumerate(seg): + if qual == "PI" and j + 1 < len(seg): + pid = seg[j + 1] + break + idx += 1 + return Payer(name=name, id=pid), idx + + +def _consume_claim(segments: list[list[str]], idx: int) -> tuple[ClaimOutput, int]: + """Read one CLM and all its child segments up to the next CLM/HL/SE.""" + seg = segments[idx] + clm = seg + claim_id = clm[1] if len(clm) > 1 else "" + total_charge = Decimal(clm[2]) if len(clm) > 2 and clm[2] else Decimal("0") + pos = "" + freq = "" + if len(clm) > 5: + clm05 = clm[5].split(":") + if len(clm05) > 0: + pos = clm05[0] + if len(clm05) > 2: + freq = clm05[2] + claim_header = ClaimHeader( + claim_id=claim_id, + total_charge=total_charge, + place_of_service=pos or None, + frequency_code=freq or None, + provider_signature=clm[6] if len(clm) > 6 else None, + assignment=clm[7] if len(clm) > 7 else None, + release_of_info=clm[9] if len(clm) > 9 else None, + ) + + diagnoses: list[Diagnosis] = [] + service_lines: list[ServiceLine] = [] + raw: list[list[str]] = [seg] + prior_auth: str | None = None + + idx += 1 + while idx < len(segments) and segments[idx][0] not in {"HL", "CLM", "SE"}: + s = segments[idx] + raw.append(s) + if s[0] == "REF" and len(s) > 2 and s[1] == "G1": + prior_auth = s[2] + elif s[0] == "HI": + for h in s[1:]: + if not h: + continue + parts = h.split(":") + qualifier = parts[0] if parts else None + for code in parts[1:]: + if code: + diagnoses.append(Diagnosis(code=code, qualifier=qualifier)) + elif s[0] == "LX": + line_no = int(s[1]) if len(s) > 1 and s[1].isdigit() else len(service_lines) + 1 + # LX is just a separator — the actual service line data is in the next SV1. + # Capture SV1 in the claim's raw_segments, then hand off to the consumer. + if idx + 1 < len(segments) and segments[idx + 1][0] == "SV1": + raw.append(segments[idx + 1]) + sl, idx = _consume_service_line(segments, idx + 1, line_no) + else: + # No SV1 after LX — skip the LX and create a stub line. + idx += 1 + sl = ServiceLine( + line_number=line_no, + procedure=Procedure(qualifier="", code=""), + charge=Decimal("0"), + ) + service_lines.append(sl) + continue # _consume_service_line already advanced idx + idx += 1 + + claim_header.prior_auth = prior_auth + return ( + ClaimOutput( + claim_id=claim_id, + control_number="", # filled in by caller + transaction_date=date(2026, 1, 1), # filled in by caller + billing_provider=BillingProvider(name="", npi=""), # filled in by caller + subscriber=Subscriber(first_name="", last_name="", member_id="", address=Address(line1="", city="", state="", zip="")), + payer=Payer(name="", id=""), + claim=claim_header, + diagnoses=diagnoses, + service_lines=service_lines, + validation=ValidationReport(passed=True, errors=[], warnings=[]), + raw_segments=raw, + ), + idx, + ) + + +def _consume_service_line(segments: list[list[str]], idx: int, line_no: int) -> tuple[ServiceLine, int]: + seg = segments[idx] + raw: list[list[str]] = [seg] + qualifier = "" + code = "" + modifiers: list[str] = [] + if len(seg) > 1: + sv101 = seg[1].split(":") + if len(sv101) > 0: + qualifier = sv101[0] + if len(sv101) > 1: + code = sv101[1] + modifiers = [m for m in sv101[2:] if m] + charge = Decimal(seg[2]) if len(seg) > 2 and seg[2] else Decimal("0") + unit_type = seg[3] if len(seg) > 3 else None + units: Decimal | None = None + if len(seg) > 4 and seg[4]: + try: + units = Decimal(seg[4]) + except Exception: + units = None + place_of_service = seg[5] if len(seg) > 5 else None + service_date: date | None = None + provider_ref: str | None = None + + idx += 1 + while idx < len(segments) and segments[idx][0] not in {"LX", "HL", "CLM", "SE"}: + s = segments[idx] + raw.append(s) + if s[0] == "DTP" and len(s) > 2 and s[1] == "472": + try: + service_date = _parse_isa_date(s[3]) if len(s) > 3 and s[2] == "D8" and len(s) > 3 else None + except (ValueError, IndexError): + service_date = None + elif s[0] == "REF" and len(s) > 2 and s[1] == "6R": + provider_ref = s[2] + idx += 1 + + return ( + ServiceLine( + line_number=line_no, + procedure=Procedure(qualifier=qualifier, code=code, modifiers=modifiers), + charge=charge, + unit_type=unit_type, + units=units, + place_of_service=place_of_service, + service_date=service_date, + provider_reference=provider_ref, + ), + idx, + ) + + +def parse(text: str, payer_config: PayerConfig, input_file: str = "") -> ParseResult: + """Parse a complete 837P document and return a :class:`ParseResult`.""" + segments = tokenize(text) + envelope, summary = _build_envelope(segments, input_file=input_file) + if envelope is None: + raise CycloneParseError("No ISA / BHT envelope found") + + claims: list[ClaimOutput] = [] + issues_by_rule: dict[str, int] = {} + + i = 0 + while i < len(segments): + seg = segments[i] + if seg[0] == "HL" and len(seg) > 3 and seg[3] == "20": + # Billing provider HL + provider, i = _consume_billing_provider(segments, i + 1) + continue + if seg[0] == "HL" and len(seg) > 3 and seg[3] == "22": + # Subscriber HL + try: + subscriber, i = _consume_subscriber(segments, i + 1) + except Exception as exc: # pragma: no cover - safety net + log.warning("Subscriber parse failed at segment %d: %s", i, exc) + i += 1 + continue + try: + payer, i = _consume_payer(segments, i) + except Exception as exc: # pragma: no cover + log.warning("Payer parse failed at segment %d: %s", i, exc) + payer = Payer(name="", id="") + # Consume all CLMs in this subscriber loop + while i < len(segments) and segments[i][0] != "HL": + if segments[i][0] == "CLM": + try: + claim, i = _consume_claim(segments, i) + except Exception as exc: + log.warning("Claim parse failed at segment %d: %s", i, exc) + i += 1 + continue + # Fill in envelope-bound fields + claim = claim.model_copy(update={ + "control_number": envelope.control_number, + "transaction_date": envelope.transaction_date, + "billing_provider": provider, + "subscriber": subscriber, + "payer": payer, + }) + # Run validation + report = validate(claim, payer_config) + # Structural check not covered by the validator rules: subscriber + # member id (NM109/MI) must be present. The validator's + # `R040_member_id_present` style rule lives here as a guard + # so a truncated MI segment does not silently produce a + # "passed" claim. + if not claim.subscriber.member_id: + extra = ValidationIssue( + rule="R040_member_id_present", + severity="error", + message="Subscriber member id (NM109/MI) is empty", + ) + report = report.model_copy(update={ + "errors": report.errors + [extra], + "passed": False, + }) + claim = claim.model_copy(update={"validation": report}) + for issue in report.errors + report.warnings: + issues_by_rule[issue.rule] = issues_by_rule.get(issue.rule, 0) + 1 + claims.append(claim) + else: + i += 1 + continue + i += 1 + + passed = sum(1 for c in claims if c.validation.passed) + failed = len(claims) - passed + failed_ids = [c.claim_id for c in claims if not c.validation.passed] + + summary = summary.model_copy(update={ + "control_number": envelope.control_number, + "transaction_date": envelope.transaction_date, + "total_claims": len(claims), + "passed": passed, + "failed": failed, + "failed_claim_ids": failed_ids, + "issues_by_rule": issues_by_rule, + }) + + return ParseResult(envelope=envelope, claims=claims, summary=summary) diff --git a/backend/tests/test_parse_837.py b/backend/tests/test_parse_837.py new file mode 100644 index 0000000..b5510ec --- /dev/null +++ b/backend/tests/test_parse_837.py @@ -0,0 +1,70 @@ +from datetime import date +from pathlib import Path + +from cyclone.parsers.parse_837 import parse +from cyclone.parsers.payer import PayerConfig + + +FIXTURE = Path(__file__).parent / "fixtures" / "minimal_837p.txt" + + +def test_parse_minimal_fixture_returns_one_claim(): + text = FIXTURE.read_text() + result = parse(text, PayerConfig.co_medicaid()) + assert len(result.claims) == 1 + claim = result.claims[0] + assert claim.claim_id == "CLM001" + assert claim.billing_provider.npi == "1234567890" + assert claim.subscriber.last_name == "Doe" + assert claim.subscriber.first_name == "John" + assert claim.subscriber.member_id == "ABC123" + assert claim.payer.id == "SKCO0" + assert claim.claim.frequency_code == "1" + assert claim.claim.place_of_service == "12" + assert claim.claim.prior_auth == "PA123" + assert claim.diagnoses[0].code == "Z00" + assert claim.diagnoses[0].qualifier == "ABK" + assert len(claim.service_lines) == 1 + sl = claim.service_lines[0] + assert sl.procedure.code == "99213" + assert sl.procedure.qualifier == "HC" + assert sl.charge == 100.00 + + +def test_parse_envelope_captured(): + result = parse(FIXTURE.read_text(), PayerConfig.co_medicaid()) + assert result.envelope is not None + assert result.envelope.control_number == "991102977" + assert result.envelope.transaction_date == date(2026, 6, 11) + assert result.envelope.sender_id == "11525703" + + +def test_parse_summary_counts(): + result = parse(FIXTURE.read_text(), PayerConfig.co_medicaid()) + assert result.summary.total_claims == 1 + assert result.summary.passed == 1 + assert result.summary.failed == 0 + + +def test_parse_continues_on_malformed_subscriber(): + """A bad subscriber record must not abort the whole batch.""" + text = FIXTURE.read_text().replace("MI*ABC123", "MI*") # truncate member id segment + result = parse(text, PayerConfig.co_medicaid()) + # Should still find the claim, but with errors + assert len(result.claims) == 1 + assert result.claims[0].validation.passed is False + assert result.summary.failed == 1 + + +def test_parse_preserves_raw_segments(): + result = parse(FIXTURE.read_text(), PayerConfig.co_medicaid()) + claim = result.claims[0] + assert any(s[0] == "CLM" for s in claim.raw_segments) + assert any(s[0] == "SV1" for s in claim.raw_segments) + + +def test_parse_uses_generic_config_when_requested(): + result = parse(FIXTURE.read_text(), PayerConfig.generic_837p()) + assert len(result.claims) == 1 + # No patient-loop rule on generic config + assert result.claims[0].validation.passed is True From 2d509d92f4babe441c061b5070c5bf573ae6050f Mon Sep 17 00:00:00 2001 From: Tyler Date: Fri, 19 Jun 2026 15:54:46 -0600 Subject: [PATCH 10/13] test(backend): multi-claim CO Medicaid fixture + integration test --- backend/tests/fixtures/co_medicaid_837p.txt | 60 +++++++++++++++++++++ backend/tests/test_co_medicaid_fixture.py | 40 ++++++++++++++ 2 files changed, 100 insertions(+) create mode 100644 backend/tests/fixtures/co_medicaid_837p.txt create mode 100644 backend/tests/test_co_medicaid_fixture.py diff --git a/backend/tests/fixtures/co_medicaid_837p.txt b/backend/tests/fixtures/co_medicaid_837p.txt new file mode 100644 index 0000000..54def3f --- /dev/null +++ b/backend/tests/fixtures/co_medicaid_837p.txt @@ -0,0 +1,60 @@ +ISA*00* *00* *ZZ*11525703 *ZZ*COMEDASSISTPROG*260617*1937*^*00501*991102984*1*P*:~ +GS*HC*11525703*COMEDASSISTPROG*20260617*193715*991102984*X*005010X222A1~ +ST*837*991102984*005010X222A1~ +BHT*0019*00*co-fixture-001*20260617*193715*CH~ +NM1*41*2*Dzinesco*****46*11525703~ +PER*IC*Tester*EM*tester@example.com~ +NM1*40*2*COLORADO MEDICAL ASSISTANCE PROGRAM*****46*COMEDASSISTPROG~ +HL*1**20*1~ +PRV*BI*PXC*251E00000X~ +NM1*85*2*TOC, Inc.*****XX*1881068062~ +N3*1100 East Main St*Suite A~ +N4*Montrose*CO*814014063~ +REF*EI*721587149~ +HL*2*1*22*0~ +SBR*P*18*******MC~ +NM1*IL*1*Balliache*Marianela****MI*P060946~ +N3*1811 PAVILION DR APT 303~ +N4*Montrose*CO*814016072~ +DMG*D8*19590223*F~ +NM1*PR*2*COHCPF*****PI*SKCO0~ +CLM*t991102984o1c1d*85.40***12:B:1*Y*A*Y*Y~ +REF*G1*3173~ +HI*ABK:R69~ +LX*1~ +SV1*HC:T1019:U2*42.70*UN*6.00***1~ +DTP*472*D8*20260602~ +REF*6R*t991102984v769804d~ +LX*2~ +SV1*HC:T1019:U2*42.70*UN*6.00***1~ +DTP*472*D8*20260603~ +REF*6R*t991102984v770058d~ +HL*3*1*22*0~ +SBR*P*18*******MC~ +NM1*IL*1*Barella*Victoria****MI*H582447~ +N3*1900 Kellie DR~ +N4*Montrose*CO*814019524~ +DMG*D8*19570727*F~ +NM1*PR*2*COHCPF*****PI*SKCO0~ +CLM*t991102984o1c2d*155.76***12:B:1*Y*A*Y*Y~ +REF*G1*3173~ +HI*ABK:R69~ +LX*1~ +SV1*HC:S5130:U2*38.94*UN*6.00***1~ +DTP*472*D8*20260602~ +REF*6R*t991102984v769805d~ +LX*2~ +SV1*HC:S5130:U2*38.94*UN*6.00***1~ +DTP*472*D8*20260603~ +REF*6R*t991102984v770059d~ +LX*3~ +SV1*HC:S5130:U2*38.94*UN*6.00***1~ +DTP*472*D8*20260604~ +REF*6R*t991102984v770308d~ +LX*4~ +SV1*HC:S5130:U2*38.94*UN*6.00***1~ +DTP*472*D8*20260605~ +REF*6R*t991102984v770668d~ +SE*44*991102984~ +GE*1*991102984~ +IEA*1*991102984~ \ No newline at end of file diff --git a/backend/tests/test_co_medicaid_fixture.py b/backend/tests/test_co_medicaid_fixture.py new file mode 100644 index 0000000..e15e49d --- /dev/null +++ b/backend/tests/test_co_medicaid_fixture.py @@ -0,0 +1,40 @@ +from pathlib import Path + +from cyclone.parsers.parse_837 import parse +from cyclone.parsers.payer import PayerConfig + + +FIXTURE = Path(__file__).parent / "fixtures" / "co_medicaid_837p.txt" + + +def test_co_fixture_two_claims_pass(): + result = parse(FIXTURE.read_text(), PayerConfig.co_medicaid()) + assert result.summary.total_claims == 2 + assert result.summary.passed == 2 + assert result.summary.failed == 0 + + +def test_first_claim_has_two_service_lines(): + result = parse(FIXTURE.read_text(), PayerConfig.co_medicaid()) + c1 = result.claims[0] + assert c1.claim_id == "t991102984o1c1d" + assert len(c1.service_lines) == 2 + assert c1.service_lines[0].procedure.code == "T1019" + assert c1.service_lines[0].procedure.modifiers == ["U2"] + assert c1.service_lines[1].service_date is not None + + +def test_second_claim_has_four_service_lines(): + result = parse(FIXTURE.read_text(), PayerConfig.co_medicaid()) + c2 = result.claims[1] + assert c2.claim_id == "t991102984o1c2d" + assert len(c2.service_lines) == 4 + + +def test_subscriber_payer_reused_across_claims(): + """Both claims share the same billing provider; payers are recorded on each claim.""" + result = parse(FIXTURE.read_text(), PayerConfig.co_medicaid()) + assert result.claims[0].billing_provider.npi == "1881068062" + assert result.claims[1].billing_provider.npi == "1881068062" + assert result.claims[0].payer.id == "SKCO0" + assert result.claims[1].payer.id == "SKCO0" From f1e86979f816f39ddfd97dfe3cc8a0499ca89e32 Mon Sep 17 00:00:00 2001 From: Tyler Date: Fri, 19 Jun 2026 15:56:09 -0600 Subject: [PATCH 11/13] feat(backend): per-claim JSON writer + summary.json --- backend/src/cyclone/parsers/__init__.py | 1 + backend/src/cyclone/parsers/writer.py | 54 +++++++++++++++ backend/tests/test_writer.py | 90 +++++++++++++++++++++++++ 3 files changed, 145 insertions(+) create mode 100644 backend/src/cyclone/parsers/writer.py create mode 100644 backend/tests/test_writer.py diff --git a/backend/src/cyclone/parsers/__init__.py b/backend/src/cyclone/parsers/__init__.py index 23d9be7..dcd37e4 100644 --- a/backend/src/cyclone/parsers/__init__.py +++ b/backend/src/cyclone/parsers/__init__.py @@ -37,6 +37,7 @@ _LAZY_EXPORTS: dict[str, str] = { "tokenize": "cyclone.parsers.segments", # orchestrator "parse": "cyclone.parsers.parse_837", + "write_outputs": "cyclone.parsers.writer", } _cache: dict[str, object] = {} diff --git a/backend/src/cyclone/parsers/writer.py b/backend/src/cyclone/parsers/writer.py new file mode 100644 index 0000000..1b89be5 --- /dev/null +++ b/backend/src/cyclone/parsers/writer.py @@ -0,0 +1,54 @@ +"""Write a :class:`ParseResult` to disk as one JSON file per claim + summary.json.""" + +from __future__ import annotations + +import json +import re +from pathlib import Path + +from cyclone.parsers.models import ParseResult + +_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"claim-{_sanitize(control_number)}-{_sanitize(transaction_date_iso.replace('-', ''))}-{_sanitize(claim_id)}.json" + if base not in used: + return base + # Disambiguate: append -2, -3, ... + stem = base[:-5] + n = 2 + while True: + cand = f"{stem}-{n}.json" + if cand not in used: + return cand + n += 1 + + +def write_outputs(result: ParseResult, output_dir: Path) -> list[Path]: + """Write one JSON file per claim and a ``summary.json``. + + Returns the list of written claim-file paths, in input order. + """ + 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] = [] + for claim in result.claims: + name = _filename(control, txn_date, claim.claim_id, used) + used.add(name) + path = output_dir / name + path.write_text(claim.model_dump_json(indent=2)) + paths.append(path) + + summary_path = output_dir / "summary.json" + summary_path.write_text(result.summary.model_dump_json(indent=2)) + return paths diff --git a/backend/tests/test_writer.py b/backend/tests/test_writer.py new file mode 100644 index 0000000..3f81f48 --- /dev/null +++ b/backend/tests/test_writer.py @@ -0,0 +1,90 @@ +import json +from datetime import date +from decimal import Decimal +from pathlib import Path + +import pytest + +from cyclone.parsers.models import ( + Address, + BatchSummary, + BillingProvider, + ClaimHeader, + ClaimOutput, + Envelope, + Payer, + Subscriber, + ValidationReport, +) +from cyclone.parsers.parse_837 import ParseResult +from cyclone.parsers.writer import write_outputs + + +def _claim(cid: str) -> ClaimOutput: + return ClaimOutput( + claim_id=cid, + control_number="991102977", + transaction_date=date(2026, 6, 11), + billing_provider=BillingProvider(name="X", npi="1234567890", 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=cid, total_charge=Decimal("100"), place_of_service="12", frequency_code="1", provider_signature="Y", assignment="Y", release_of_info="Y"), + diagnoses=[], + service_lines=[], + validation=ValidationReport(passed=True, errors=[], warnings=[]), + raw_segments=[], + ) + + +def _result(*cids: str) -> ParseResult: + return ParseResult( + envelope=Envelope(sender_id="X", receiver_id="Y", control_number="991102977", transaction_date=date(2026, 6, 11)), + claims=[_claim(c) for c in cids], + summary=BatchSummary(input_file="f", control_number="991102977", transaction_date=date(2026, 6, 11), total_claims=len(cids), passed=len(cids), failed=0), + ) + + +def test_write_outputs_creates_one_file_per_claim(tmp_path: Path): + result = _result("C1", "C2") + paths = write_outputs(result, tmp_path) + assert len(paths) == 2 + for p in paths: + assert p.exists() + assert p.suffix == ".json" + data = json.loads(p.read_text()) + assert "validation" in data + + +def test_filename_pattern(tmp_path: Path): + result = _result("C1") + paths = write_outputs(result, tmp_path) + assert paths[0].name == "claim-991102977-20260611-C1.json" + + +def test_filename_sanitizes_unsafe_chars(tmp_path: Path): + bad = _claim("C/1:*?") + result = ParseResult( + envelope=result.envelope if False else None, # type: ignore + claims=[bad], + summary=BatchSummary(input_file="f", total_claims=1, passed=1, failed=0), + ) # use explicit envelope + # Simpler: rebuild + from cyclone.parsers.models import Envelope as E + result = ParseResult( + envelope=E(sender_id="X", receiver_id="Y", control_number="991102977", transaction_date=date(2026, 6, 11)), + claims=[bad], + summary=BatchSummary(input_file="f", total_claims=1, passed=1, failed=0), + ) + paths = write_outputs(result, tmp_path) + # All non-alnum chars except - and _ replaced with - + assert all(c.isalnum() or c in "-_" for c in paths[0].stem) + + +def test_summary_json_written(tmp_path: Path): + result = _result("C1") + write_outputs(result, tmp_path) + summary_path = tmp_path / "summary.json" + assert summary_path.exists() + data = json.loads(summary_path.read_text()) + assert data["total_claims"] == 1 + assert data["passed"] == 1 From d4f3b204efca3072c1f0b1a913c54cae2e5e9d8e Mon Sep 17 00:00:00 2001 From: Tyler Date: Fri, 19 Jun 2026 16:00:47 -0600 Subject: [PATCH 12/13] feat(backend): click CLI for parse-837 --- backend/src/cyclone/cli.py | 107 +++++++++++++++++++++++++++++++++++++ backend/tests/test_cli.py | 70 ++++++++++++++++++++++++ 2 files changed, 177 insertions(+) create mode 100644 backend/src/cyclone/cli.py create mode 100644 backend/tests/test_cli.py diff --git a/backend/src/cyclone/cli.py b/backend/src/cyclone/cli.py new file mode 100644 index 0000000..6516aac --- /dev/null +++ b/backend/src/cyclone/cli.py @@ -0,0 +1,107 @@ +"""Click-based CLI for the Cyclone 837P parser.""" + +from __future__ import annotations + +import logging +import sys +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.writer import write_outputs + + +_PAYER_FACTORIES = { + "co_medicaid": PayerConfig.co_medicaid, + "generic_837p": PayerConfig.generic_837p, +} + + +def _payer(name: str) -> PayerConfig: + if name not in _PAYER_FACTORIES: + raise click.BadParameter(f"Unknown payer {name!r}. Choose from: {', '.join(_PAYER_FACTORIES)}") + return _PAYER_FACTORIES[name]() + + +@click.group() +def main() -> None: + """Cyclone EDI suite — X12 parser.""" + + +@main.command("parse-837") +@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", show_default=True, help="Payer configuration preset") +@click.option("--strict", is_flag=True, help="Promote warnings to errors") +@click.option("--max-retries", default=0, show_default=True, type=int, help="Re-validate failed claims (no fix in v1)") +@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_837( + input_file: Path, + output_dir: Path, + payer: str, + strict: bool, + max_retries: int, + include_raw_segments: bool, + log_level: str, +) -> None: + """Parse an X12 837P file into one JSON per claim.""" + logging.basicConfig(level=getattr(logging, log_level)) + + text = input_file.read_text() + config = _payer(payer) + + try: + result = parse(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 claims found in file") from None + + # Strict mode: rewrite reports + if strict: + for claim in result.claims: + promoted = [ + issue.model_copy(update={"severity": "error"}) + for issue in claim.validation.warnings + ] + new_errors = claim.validation.errors + promoted + claim.validation = claim.validation.model_copy( + update={"errors": new_errors, "passed": not new_errors} + ) + # Recompute summary + passed = sum(1 for c in result.claims if c.validation.passed) + result.summary = result.summary.model_copy(update={ + "passed": passed, + "failed": len(result.claims) - passed, + "failed_claim_ids": [c.claim_id for c in result.claims if not c.validation.passed], + }) + + # Drop raw_segments if disabled + if not include_raw_segments: + for claim in result.claims: + claim.raw_segments = [] + + # v1: max_retries is a no-op stub. Surface a log line so users know it's wired. + if max_retries: + logging.info("max-retries=%d (no auto-fix in v1; would re-validate here)", max_retries) + + paths = write_outputs(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}" + ) + + +if __name__ == "__main__": + main() diff --git a/backend/tests/test_cli.py b/backend/tests/test_cli.py new file mode 100644 index 0000000..4e018ec --- /dev/null +++ b/backend/tests/test_cli.py @@ -0,0 +1,70 @@ +import json +from pathlib import Path + +import pytest +from click.testing import CliRunner + +from cyclone.cli import main +from cyclone.parsers.payer import PayerConfig + + +FIXTURE = Path(__file__).parent / "fixtures" / "minimal_837p.txt" + + +def test_cli_parse_837_writes_outputs(tmp_path: Path): + runner = CliRunner() + result = runner.invoke(main, ["parse-837", str(FIXTURE), "--output-dir", str(tmp_path)]) + assert result.exit_code == 0, result.output + written = list(tmp_path.glob("claim-*.json")) + assert len(written) == 1 + summary = json.loads((tmp_path / "summary.json").read_text()) + assert summary["total_claims"] == 1 + assert summary["passed"] == 1 + + +def test_cli_parse_837_with_strict(tmp_path: Path): + runner = CliRunner() + result = runner.invoke(main, ["parse-837", str(FIXTURE), "--output-dir", str(tmp_path), "--strict"]) + assert result.exit_code == 0 + # Strict should not change anything on a passing claim + summary = json.loads((tmp_path / "summary.json").read_text()) + assert summary["passed"] == 1 + + +def test_cli_parse_837_with_no_raw_segments(tmp_path: Path): + runner = CliRunner() + result = runner.invoke(main, ["parse-837", str(FIXTURE), "--output-dir", str(tmp_path), "--no-raw-segments"]) + assert result.exit_code == 0 + data = json.loads(next(tmp_path.glob("claim-*.json")).read_text()) + assert data["raw_segments"] == [] + + +def test_cli_unknown_payer_uses_generic(tmp_path: Path): + runner = CliRunner() + result = runner.invoke(main, ["parse-837", str(FIXTURE), "--output-dir", str(tmp_path), "--payer", "generic_837p"]) + assert result.exit_code == 0 + + +def test_cli_missing_file_exits_2(tmp_path: Path): + runner = CliRunner() + result = runner.invoke(main, ["parse-837", "/no/such/file.txt", "--output-dir", str(tmp_path)]) + assert result.exit_code == 2 + + +@pytest.mark.skipif( + not (Path(__file__).parent.parent.parent / "docs" / "prodfiles" / "837p-from-axiscare").exists(), + reason="prodfile corpus not present", +) +def test_cli_runs_against_prodfiles(tmp_path: Path): + """Smoke test: run the CLI against a real production file if available.""" + from glob import glob + runner = CliRunner() + corpus = Path(__file__).parent.parent.parent / "docs" / "prodfiles" / "837p-from-axiscare" + files = sorted(glob(str(corpus / "*.txt"))) + if not files: + pytest.skip("no prodfile .txt files in corpus") + sample = files[0] + result = runner.invoke(main, ["parse-837", sample, "--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"] >= 1 From 464a31be09a738a823dfd56db73815b65ce9115c Mon Sep 17 00:00:00 2001 From: Tyler Date: Fri, 19 Jun 2026 16:01:25 -0600 Subject: [PATCH 13/13] docs(backend): README with install, CLI, and programmatic usage --- backend/README.md | 68 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 68 insertions(+) create mode 100644 backend/README.md diff --git a/backend/README.md b/backend/README.md new file mode 100644 index 0000000..6aaf6e7 --- /dev/null +++ b/backend/README.md @@ -0,0 +1,68 @@ +# Cyclone Backend — X12 837P 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 (later via a FastAPI service). + +## Install + +```bash +cd backend +python -m venv .venv +. .venv/bin/activate +pip install -e ".[dev]" +``` + +## CLI + +```bash +python -m cyclone.cli parse-837 path/to/837p.txt --output-dir ./claims +``` + +Options: +- `--payer {co_medicaid,generic_837p}` — default `co_medicaid` +- `--strict` — promote warnings to errors +- `--max-retries N` — stub in v1 (no auto-fix yet) +- `--include-raw-segments / --no-raw-segments` — default: include +- `--log-level {DEBUG,INFO,WARNING,ERROR}` + +Exit codes: +- `0` — every claim parsed +- `2` — file-level failure (no envelope, no claims) +- `1` — unexpected exception + +## Output + +``` +claims/ +├── claim-991102977-20260611-CLM001.json # one per claim +├── claim-991102977-20260611-CLM002.json +└── summary.json # batch summary +``` + +## Programmatic use + +```python +from pathlib import Path +from cyclone.parsers.parse_837 import parse +from cyclone.parsers.payer import PayerConfig +from cyclone.parsers.writer import write_outputs + +result = parse(Path("input.txt").read_text(), PayerConfig.co_medicaid()) +write_outputs(result, Path("./claims")) +print(f"passed={result.summary.passed} failed={result.summary.failed}") +``` + +## Tests + +```bash +pytest -v +``` + +The CLI smoke test against `docs/prodfiles/837p-from-axiscare/*.txt` is +auto-skipped if no production files are present. + +## Spec & plan + +- Design: `docs/superpowers/specs/2026-06-19-cyclone-837p-parser-design.md` +- Plan: `docs/superpowers/plans/2026-06-19-cyclone-837p-parser.md`