feat(backend): 837P parser orchestrator with HL walker
This commit is contained in:
Tyler
@@ -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)
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user