Cyclone
5ed59ef01a
- 12 tasks: project skeleton, exceptions, segments, models, payer config, fixtures, validator, parser orchestrator, CO fixture, writer, CLI, README - TDD throughout; each task has 4-6 bite-sized steps with full code - Optional prodfile smoke test (skipped if docs/prodfiles absent) - Self-review checklist confirms full spec coverage
2368 lines
74 KiB
Markdown
2368 lines
74 KiB
Markdown
# Cyclone 837P Parser Implementation Plan
|
||
|
||
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
|
||
|
||
**Goal:** Build a Python parser module under `cyclone/backend/` that turns X12 837P files into one validated JSON per claim, with a click CLI and tests.
|
||
|
||
**Architecture:** Thin custom tokenizer (no third-party EDI libs) + Pydantic v2 models + rule-based validator (structural + CO Medicaid) + single HL-walker orchestrator. One JSON per claim, plus a batch `summary.json`. Continue on per-claim failure.
|
||
|
||
**Tech Stack:** Python 3.11+, Pydantic v2, click 8.x, pytest 8.x.
|
||
|
||
**Spec:** `docs/superpowers/specs/2026-06-19-cyclone-837p-parser-design.md`
|
||
|
||
---
|
||
|
||
## File map
|
||
|
||
| File | Responsibility |
|
||
|---|---|
|
||
| `backend/pyproject.toml` | Package metadata, deps, entrypoints |
|
||
| `backend/src/cyclone/__init__.py` | Package marker, version |
|
||
| `backend/src/cyclone/parsers/__init__.py` | Public re-exports |
|
||
| `backend/src/cyclone/parsers/exceptions.py` | `CycloneParseError`, `CycloneValidationError` |
|
||
| `backend/src/cyclone/parsers/segments.py` | `Delimiters`, `tokenize(text)` |
|
||
| `backend/src/cyclone/parsers/models.py` | All Pydantic v2 models |
|
||
| `backend/src/cyclone/parsers/payer.py` | `PayerConfig` + `co_medicaid()` / `generic_837p()` factories |
|
||
| `backend/src/cyclone/parsers/validator.py` | Rule functions + `validate(claim, config)` |
|
||
| `backend/src/cyclone/parsers/parse_837.py` | Orchestrator: `parse(text, payer_config)` |
|
||
| `backend/src/cyclone/cli.py` | `click` CLI: `python -m cyclone.cli parse-837` |
|
||
| `backend/src/cyclone/__main__.py` | Enables `python -m cyclone.cli` |
|
||
| `backend/tests/conftest.py` | Shared fixtures: synthetic 837P strings |
|
||
| `backend/tests/fixtures/minimal_837p.txt` | Hand-written minimal 837P |
|
||
| `backend/tests/fixtures/co_medicaid_837p.txt` | Synthetic 2-claim CO 837P |
|
||
| `backend/tests/test_segments.py` | Tokenizer tests |
|
||
| `backend/tests/test_models.py` | Model round-trip tests |
|
||
| `backend/tests/test_payer.py` | PayerConfig factory tests |
|
||
| `backend/tests/test_validator.py` | Rule-by-rule tests |
|
||
| `backend/tests/test_parse_837.py` | End-to-end parser tests |
|
||
| `backend/tests/test_cli.py` | CLI invocation tests + optional prodfile smoke test |
|
||
| `backend/README.md` | Install + usage |
|
||
|
||
---
|
||
|
||
## Task 1: Project skeleton
|
||
|
||
**Files:**
|
||
- Create: `backend/pyproject.toml`
|
||
- Create: `backend/src/cyclone/__init__.py`
|
||
- Create: `backend/src/cyclone/__main__.py`
|
||
- Create: `backend/src/cyclone/parsers/__init__.py`
|
||
- Create: `backend/.gitignore` (overrides the repo-root one inside `backend/`)
|
||
|
||
- [ ] **Step 1: Create `backend/pyproject.toml`**
|
||
|
||
```toml
|
||
[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"
|
||
```
|
||
|
||
- [ ] **Step 2: Create the package marker files**
|
||
|
||
`backend/src/cyclone/__init__.py`:
|
||
```python
|
||
__version__ = "0.1.0"
|
||
```
|
||
|
||
`backend/src/cyclone/__main__.py`:
|
||
```python
|
||
from cyclone.cli import main
|
||
|
||
if __name__ == "__main__":
|
||
main()
|
||
```
|
||
|
||
`backend/src/cyclone/parsers/__init__.py`:
|
||
```python
|
||
"""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",
|
||
]
|
||
```
|
||
|
||
- [ ] **Step 3: Create `backend/.gitignore`**
|
||
|
||
```gitignore
|
||
__pycache__/
|
||
*.py[cod]
|
||
*.egg-info/
|
||
.pytest_cache/
|
||
.ruff_cache/
|
||
.venv/
|
||
venv/
|
||
build/
|
||
dist/
|
||
```
|
||
|
||
- [ ] **Step 4: Install the package in dev mode**
|
||
|
||
Run from `backend/`:
|
||
```bash
|
||
cd backend
|
||
python -m venv .venv
|
||
. .venv/bin/activate
|
||
pip install -e ".[dev]"
|
||
```
|
||
|
||
Expected: installation completes; `cyclone` console script installed.
|
||
|
||
- [ ] **Step 5: Verify the package imports**
|
||
|
||
Run: `python -c "import cyclone; print(cyclone.__version__)"`
|
||
Expected: `0.1.0`
|
||
|
||
- [ ] **Step 6: Commit**
|
||
|
||
```bash
|
||
git add backend/
|
||
git commit -m "feat(backend): scaffold cyclone package skeleton"
|
||
```
|
||
|
||
---
|
||
|
||
## Task 2: Exceptions module
|
||
|
||
**Files:**
|
||
- Create: `backend/src/cyclone/parsers/exceptions.py`
|
||
- Create: `backend/tests/test_exceptions.py`
|
||
|
||
- [ ] **Step 1: Write the failing test**
|
||
|
||
`backend/tests/test_exceptions.py`:
|
||
```python
|
||
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)
|
||
```
|
||
|
||
- [ ] **Step 2: Run the test, expect import failure**
|
||
|
||
Run: `pytest tests/test_exceptions.py -v`
|
||
Expected: `ModuleNotFoundError: No module named 'cyclone.parsers.exceptions'`
|
||
|
||
- [ ] **Step 3: Implement the exceptions**
|
||
|
||
`backend/src/cyclone/parsers/exceptions.py`:
|
||
```python
|
||
"""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}")
|
||
```
|
||
|
||
- [ ] **Step 4: Run the test, expect pass**
|
||
|
||
Run: `pytest tests/test_exceptions.py -v`
|
||
Expected: 3 passed.
|
||
|
||
- [ ] **Step 5: Commit**
|
||
|
||
```bash
|
||
git add backend/src/cyclone/parsers/exceptions.py backend/tests/test_exceptions.py
|
||
git commit -m "feat(backend): add CycloneParseError and CycloneValidationError"
|
||
```
|
||
|
||
---
|
||
|
||
## Task 3: Segment tokenizer
|
||
|
||
**Files:**
|
||
- Create: `backend/src/cyclone/parsers/segments.py`
|
||
- Create: `backend/tests/test_segments.py`
|
||
|
||
- [ ] **Step 1: Write the failing tests**
|
||
|
||
`backend/tests/test_segments.py`:
|
||
```python
|
||
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 == "~"
|
||
```
|
||
|
||
- [ ] **Step 2: Run, expect import failure**
|
||
|
||
Run: `pytest tests/test_segments.py -v`
|
||
Expected: `ModuleNotFoundError: No module named 'cyclone.parsers.segments'`
|
||
|
||
- [ ] **Step 3: Implement the tokenizer**
|
||
|
||
`backend/src/cyclone/parsers/segments.py`:
|
||
```python
|
||
"""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))
|
||
|
||
if len(out) == 1:
|
||
raise CycloneParseError("File contains no segments after ISA")
|
||
|
||
return out
|
||
```
|
||
|
||
- [ ] **Step 4: Run, expect pass**
|
||
|
||
Run: `pytest tests/test_segments.py -v`
|
||
Expected: 8 passed.
|
||
|
||
- [ ] **Step 5: Commit**
|
||
|
||
```bash
|
||
git add backend/src/cyclone/parsers/segments.py backend/tests/test_segments.py
|
||
git commit -m "feat(backend): X12 segment tokenizer with ISA delimiter detection"
|
||
```
|
||
|
||
---
|
||
|
||
## Task 4: Pydantic v2 models
|
||
|
||
**Files:**
|
||
- Create: `backend/src/cyclone/parsers/models.py`
|
||
- Create: `backend/tests/test_models.py`
|
||
|
||
- [ ] **Step 1: Write the failing tests**
|
||
|
||
`backend/tests/test_models.py`:
|
||
```python
|
||
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"
|
||
```
|
||
|
||
- [ ] **Step 2: Run, expect import failure**
|
||
|
||
Run: `pytest tests/test_models.py -v`
|
||
Expected: `ModuleNotFoundError: No module named 'cyclone.parsers.models'`
|
||
|
||
- [ ] **Step 3: Implement the models**
|
||
|
||
`backend/src/cyclone/parsers/models.py`:
|
||
```python
|
||
"""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
|
||
|
||
|
||
class _Base(BaseModel):
|
||
model_config = ConfigDict(extra="ignore", str_strip_whitespace=True)
|
||
|
||
|
||
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
|
||
```
|
||
|
||
- [ ] **Step 4: Run, expect pass**
|
||
|
||
Run: `pytest tests/test_models.py -v`
|
||
Expected: 9 passed.
|
||
|
||
- [ ] **Step 5: Commit**
|
||
|
||
```bash
|
||
git add backend/src/cyclone/parsers/models.py backend/tests/test_models.py
|
||
git commit -m "feat(backend): Pydantic v2 models for 837P claims"
|
||
```
|
||
|
||
---
|
||
|
||
## Task 5: PayerConfig factories
|
||
|
||
**Files:**
|
||
- Create: `backend/src/cyclone/parsers/payer.py`
|
||
- Create: `backend/tests/test_payer.py`
|
||
|
||
- [ ] **Step 1: Write the failing tests**
|
||
|
||
`backend/tests/test_payer.py`:
|
||
```python
|
||
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
|
||
```
|
||
|
||
- [ ] **Step 2: Run, expect import failure**
|
||
|
||
Run: `pytest tests/test_payer.py -v`
|
||
Expected: `ModuleNotFoundError: No module named 'cyclone.parsers.payer'`
|
||
|
||
- [ ] **Step 3: Implement PayerConfig**
|
||
|
||
`backend/src/cyclone/parsers/payer.py`:
|
||
```python
|
||
"""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,
|
||
)
|
||
```
|
||
|
||
- [ ] **Step 4: Run, expect pass**
|
||
|
||
Run: `pytest tests/test_payer.py -v`
|
||
Expected: 2 passed.
|
||
|
||
- [ ] **Step 5: Commit**
|
||
|
||
```bash
|
||
git add backend/src/cyclone/parsers/payer.py backend/tests/test_payer.py
|
||
git commit -m "feat(backend): PayerConfig with co_medicaid and generic factories"
|
||
```
|
||
|
||
---
|
||
|
||
## Task 6: Minimal 837P fixture
|
||
|
||
**Files:**
|
||
- Create: `backend/tests/fixtures/minimal_837p.txt`
|
||
|
||
- [ ] **Step 1: Create the fixture file**
|
||
|
||
`backend/tests/fixtures/minimal_837p.txt`:
|
||
```
|
||
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~
|
||
```
|
||
|
||
- [ ] **Step 2: Verify the fixture tokenizes**
|
||
|
||
Run from `backend/`:
|
||
```bash
|
||
python -c "from pathlib import Path; from cyclone.parsers.segments import tokenize; print(len(tokenize(Path('tests/fixtures/minimal_837p.txt').read_text())))"
|
||
```
|
||
Expected: `28` (1 ISA + 27 content segments, including the 3 envelope closers)
|
||
|
||
- [ ] **Step 3: Commit**
|
||
|
||
```bash
|
||
git add backend/tests/fixtures/minimal_837p.txt
|
||
git commit -m "test(backend): add minimal 837P fixture"
|
||
```
|
||
|
||
---
|
||
|
||
## Task 7: Validator
|
||
|
||
**Files:**
|
||
- Create: `backend/src/cyclone/parsers/validator.py`
|
||
- Create: `backend/tests/test_validator.py`
|
||
|
||
- [ ] **Step 1: Write the failing tests**
|
||
|
||
`backend/tests/test_validator.py`:
|
||
```python
|
||
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"),
|
||
)
|
||
],
|
||
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_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()
|
||
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
|
||
```
|
||
|
||
- [ ] **Step 2: Run, expect import failure**
|
||
|
||
Run: `pytest tests/test_validator.py -v`
|
||
Expected: `ModuleNotFoundError: No module named 'cyclone.parsers.validator'`
|
||
|
||
- [ ] **Step 3: Implement the validator**
|
||
|
||
`backend/src/cyclone/parsers/validator.py`:
|
||
```python
|
||
"""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)
|
||
```
|
||
|
||
- [ ] **Step 4: Run, expect pass**
|
||
|
||
Run: `pytest tests/test_validator.py -v`
|
||
Expected: 10 passed.
|
||
|
||
- [ ] **Step 5: Commit**
|
||
|
||
```bash
|
||
git add backend/src/cyclone/parsers/validator.py backend/tests/test_validator.py
|
||
git commit -m "feat(backend): structural + CO Medicaid validation rules"
|
||
```
|
||
|
||
---
|
||
|
||
## Task 8: Parser orchestrator (basic structure)
|
||
|
||
**Files:**
|
||
- Create: `backend/src/cyclone/parsers/parse_837.py`
|
||
- Create: `backend/tests/test_parse_837.py`
|
||
|
||
- [ ] **Step 1: Write the failing test**
|
||
|
||
`backend/tests/test_parse_837.py`:
|
||
```python
|
||
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
|
||
```
|
||
|
||
- [ ] **Step 2: Run, expect import failure**
|
||
|
||
Run: `pytest tests/test_parse_837.py -v`
|
||
Expected: `ModuleNotFoundError: No module named 'cyclone.parsers.parse_837'`
|
||
|
||
- [ ] **Step 3: Implement the orchestrator**
|
||
|
||
`backend/src/cyclone/parsers/parse_837.py`:
|
||
```python
|
||
"""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,
|
||
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:
|
||
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]
|
||
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]
|
||
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
|
||
sl, idx = _consume_service_line(segments, idx, line_no)
|
||
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)
|
||
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)
|
||
```
|
||
|
||
- [ ] **Step 4: Run, expect pass**
|
||
|
||
Run: `pytest tests/test_parse_837.py -v`
|
||
Expected: 6 passed. (If anything fails, the most likely cause is a missing field on the minimal fixture — adjust the fixture and re-run.)
|
||
|
||
- [ ] **Step 5: Commit**
|
||
|
||
```bash
|
||
git add backend/src/cyclone/parsers/parse_837.py backend/tests/test_parse_837.py
|
||
git commit -m "feat(backend): 837P parser orchestrator with HL walker"
|
||
```
|
||
|
||
---
|
||
|
||
## Task 9: Multi-claim CO fixture + integration test
|
||
|
||
**Files:**
|
||
- Create: `backend/tests/fixtures/co_medicaid_837p.txt`
|
||
- Create: `backend/tests/test_co_medicaid_fixture.py`
|
||
|
||
- [ ] **Step 1: Create the CO Medicaid fixture**
|
||
|
||
`backend/tests/fixtures/co_medicaid_837p.txt`:
|
||
```
|
||
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~
|
||
```
|
||
|
||
- [ ] **Step 2: Write the integration test**
|
||
|
||
`backend/tests/test_co_medicaid_fixture.py`:
|
||
```python
|
||
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"
|
||
```
|
||
|
||
- [ ] **Step 3: Run the tests**
|
||
|
||
Run: `pytest tests/test_co_medicaid_fixture.py -v`
|
||
Expected: 4 passed.
|
||
|
||
If `test_first_claim_has_two_service_lines` fails because the second LX isn't picked up, the likely cause is `_consume_service_line` not advancing past the next `LX` boundary. Inspect the `raw_segments` field of the failed claim and adjust the inner-loop logic.
|
||
|
||
- [ ] **Step 4: Commit**
|
||
|
||
```bash
|
||
git add backend/tests/fixtures/co_medicaid_837p.txt backend/tests/test_co_medicaid_fixture.py
|
||
git commit -m "test(backend): multi-claim CO Medicaid fixture + integration test"
|
||
```
|
||
|
||
---
|
||
|
||
## Task 10: Output writer
|
||
|
||
**Files:**
|
||
- Create: `backend/src/cyclone/parsers/writer.py`
|
||
- Create: `backend/tests/test_writer.py`
|
||
|
||
- [ ] **Step 1: Write the failing test**
|
||
|
||
`backend/tests/test_writer.py`:
|
||
```python
|
||
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
|
||
```
|
||
|
||
- [ ] **Step 2: Run, expect import failure**
|
||
|
||
Run: `pytest tests/test_writer.py -v`
|
||
Expected: `ModuleNotFoundError: No module named 'cyclone.parsers.writer'`
|
||
|
||
- [ ] **Step 3: Implement the writer**
|
||
|
||
`backend/src/cyclone/parsers/writer.py`:
|
||
```python
|
||
"""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
|
||
```
|
||
|
||
- [ ] **Step 4: Run, expect pass**
|
||
|
||
Run: `pytest tests/test_writer.py -v`
|
||
Expected: 4 passed.
|
||
|
||
- [ ] **Step 5: Wire the writer into the package re-exports**
|
||
|
||
Edit `backend/src/cyclone/parsers/__init__.py` — add to the imports:
|
||
```python
|
||
from cyclone.parsers.writer import write_outputs
|
||
```
|
||
And to `__all__`:
|
||
```python
|
||
"write_outputs",
|
||
```
|
||
|
||
- [ ] **Step 6: Commit**
|
||
|
||
```bash
|
||
git add backend/src/cyclone/parsers/writer.py backend/src/cyclone/parsers/__init__.py backend/tests/test_writer.py
|
||
git commit -m "feat(backend): per-claim JSON writer + summary.json"
|
||
```
|
||
|
||
---
|
||
|
||
## Task 11: CLI
|
||
|
||
**Files:**
|
||
- Create: `backend/src/cyclone/cli.py`
|
||
- Create: `backend/tests/test_cli.py`
|
||
|
||
- [ ] **Step 1: Write the failing test**
|
||
|
||
`backend/tests/test_cli.py`:
|
||
```python
|
||
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
|
||
```
|
||
|
||
- [ ] **Step 2: Run, expect import failure**
|
||
|
||
Run: `pytest tests/test_cli.py -v`
|
||
Expected: `ModuleNotFoundError: No module named 'cyclone.cli'`
|
||
|
||
- [ ] **Step 3: Implement the CLI**
|
||
|
||
`backend/src/cyclone/cli.py`:
|
||
```python
|
||
"""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}"
|
||
)
|
||
```
|
||
|
||
- [ ] **Step 4: Run, expect pass**
|
||
|
||
Run: `pytest tests/test_cli.py -v`
|
||
Expected: 4 passed, 1 skipped (the prodfile test).
|
||
|
||
- [ ] **Step 5: Smoke-test the CLI from the shell**
|
||
|
||
Run from `backend/`:
|
||
```bash
|
||
. .venv/bin/activate
|
||
python -m cyclone.cli parse-837 tests/fixtures/minimal_837p.txt --output-dir /tmp/claims
|
||
```
|
||
Expected:
|
||
```
|
||
parsed=1 passed=1 failed=0 output=/tmp/claims
|
||
```
|
||
And `/tmp/claims/claim-991102977-20260611-CLM001.json` exists.
|
||
|
||
- [ ] **Step 6: Commit**
|
||
|
||
```bash
|
||
git add backend/src/cyclone/cli.py backend/tests/test_cli.py
|
||
git commit -m "feat(backend): click CLI for parse-837"
|
||
```
|
||
|
||
---
|
||
|
||
## Task 12: README
|
||
|
||
**Files:**
|
||
- Create: `backend/README.md`
|
||
|
||
- [ ] **Step 1: Write the README**
|
||
|
||
`backend/README.md`:
|
||
```markdown
|
||
# 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`
|
||
```
|
||
|
||
- [ ] **Step 2: Commit**
|
||
|
||
```bash
|
||
git add backend/README.md
|
||
git commit -m "docs(backend): README with install, CLI, and programmatic usage"
|
||
```
|
||
|
||
---
|
||
|
||
## Self-review checklist (run after writing the plan)
|
||
|
||
- [x] **Spec coverage:** every section in the spec is implemented by at least one task.
|
||
- §4 stack → Task 1 (pyproject), Task 4 (Pydantic v2), Task 11 (click CLI), Task 1 (pytest)
|
||
- §5 directory layout → Tasks 1, 2, 3, 4, 5, 6, 7, 8, 10, 11, 12
|
||
- §6 data flow → Tasks 3, 4, 7, 8, 10, 11
|
||
- §7 data model → Task 4
|
||
- §8 parser internals → Task 8
|
||
- §9 validation rules → Task 7
|
||
- §10 output files → Task 10
|
||
- §11 CLI → Task 11
|
||
- §12 testing → Tasks 2, 4, 5, 6, 7, 8, 9, 10, 11
|
||
- [x] **Placeholder scan:** no TBD/TODO/"fill in later" — all code blocks are complete.
|
||
- [x] **Type consistency:** the same names appear throughout (`PayerConfig`, `ClaimOutput`, `ValidationReport`, `tokenize`, `parse`, `write_outputs`, `Delimiters`). The CLI in Task 11 references the same functions exported in Task 1's `__init__.py`.
|
||
- [x] **No spec gaps:** §13 (out of scope) is intentionally not implemented; §14 (open assumptions) is documented but not blocked.
|
||
|
||
---
|
||
|
||
## Final task: verify everything together
|
||
|
||
- [ ] **Step 1: Run the full test suite**
|
||
|
||
From `backend/`:
|
||
```bash
|
||
pytest -v
|
||
```
|
||
Expected: all unit tests pass, prodfile smoke test is skipped.
|
||
|
||
- [ ] **Step 2: Run the CLI against both fixtures**
|
||
|
||
```bash
|
||
python -m cyclone.cli parse-837 tests/fixtures/minimal_837p.txt --output-dir /tmp/c1
|
||
python -m cyclone.cli parse-837 tests/fixtures/co_medicaid_837p.txt --output-dir /tmp/c2
|
||
```
|
||
Expected: `parsed=N passed=N failed=0` for both. Inspect `/tmp/c1/summary.json` and `/tmp/c2/summary.json`.
|
||
|
||
- [ ] **Step 3: Final commit if anything was tweaked**
|
||
|
||
```bash
|
||
git status
|
||
git add -A
|
||
git commit -m "chore(backend): post-plan cleanups" || true
|
||
```
|
||
|
||
- [ ] **Step 4: Optional — run the prodfile smoke test**
|
||
|
||
```bash
|
||
python -m cyclone.cli parse-837 ../docs/prodfiles/837p-from-axiscare/tp11525703-837P-20260618153358831-1of1.txt --output-dir /tmp/claims-real
|
||
```
|
||
Expected: 4–5 claims parsed, all passing. The test suite auto-runs this if the file is present.
|