Tyler
2893676c0b
X12 835 SVC segment: SVC01 = composite procedure SVC02 = charge SVC03 = payment SVC04 = Unit or Basis for Measurement Code (UN, MJ, DA, ...) SVC05 = Service Unit Count The parser previously read SVC04 as the units count and SVC05 as the unit type — backwards. On real 835s (and the canonical minimal fixture), SVC04 carries the code 'UN' which fails Decimal parsing, so the units always came out as None and the code string was assigned to unit_type. SP7's line-level matcher couldn't compare units on the SVC side against the claim side because of this. - _consume_service_payment: SVC04 → unit_type, SVC05 → units count - Default unit_type to 'UN' when only the count is present - minimal_835.txt + unbalanced_835.txt: swap positions to match spec - Add 2 regression tests (units-and-unit-type, default-unit-type-to-UN)
554 lines
22 KiB
Python
554 lines
22 KiB
Python
r"""Orchestrate parsing an X12 835 ERA file into a :class:`ParseResult835`.
|
|
|
|
Single-pass walker over the tokenized segment list:
|
|
|
|
- ISA / GS / ST / BPR / TRN — envelope + financial + trace
|
|
- Loop 1000A (N1*PR + N3/N4/PER/REF) — payer
|
|
- Loop 1000B (N1*PE + N3/N4) — payee
|
|
- Loop 2100 (LX + CLP + CAS + NM1 + REF + DTM + QTY) — claim payment
|
|
- Loop 2110 (SVC + CAS + REF + DTM + LQ) — service payment
|
|
|
|
Errors at the file level raise :class:`CycloneParseError`. Per-segment
|
|
deficiencies are recorded in :class:`ValidationReport` and never abort
|
|
the whole batch.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from datetime import date
|
|
from decimal import Decimal
|
|
from typing import Any
|
|
|
|
from cyclone.parsers.exceptions import CycloneParseError
|
|
from cyclone.parsers.models import Address, BatchSummary, Envelope
|
|
from cyclone.parsers.models_835 import (
|
|
ClaimAdjustment,
|
|
ClaimPayment,
|
|
FinancialInfo,
|
|
ParseResult835,
|
|
Payer835,
|
|
Payee835,
|
|
ReassociationTrace,
|
|
ServicePayment,
|
|
claim_status_label,
|
|
)
|
|
from cyclone.parsers.payer import PayerConfig835
|
|
from cyclone.parsers.segments import tokenize
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Date parsing
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def _parse_ccyymmdd(yymmdd: str) -> date:
|
|
"""Parse a CCYYMMDD (8-char) or YY?? (6-char) date string."""
|
|
if not yymmdd:
|
|
raise ValueError("empty date")
|
|
if len(yymmdd) == 8:
|
|
return date(int(yymmdd[0:4]), int(yymmdd[4:6]), int(yymmdd[6:8]))
|
|
if len(yymmdd) == 6:
|
|
yy = int(yymmdd[0:2])
|
|
mm = int(yymmdd[2:4])
|
|
dd = int(yymmdd[4:6])
|
|
# Pivot year 30: 00-29 = 2000s, 30-99 = 1900s.
|
|
year = 2000 + yy if yy < 30 else 1900 + yy
|
|
return date(year, mm, dd)
|
|
raise ValueError(f"unexpected date length: {yymmdd!r}")
|
|
|
|
|
|
def _safe_date(s: str) -> date | None:
|
|
try:
|
|
return _parse_ccyymmdd(s) if s else None
|
|
except (ValueError, IndexError):
|
|
return None
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Envelope / BPR / TRN
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def _build_envelope(segments: list[list[str]], input_file: str) -> tuple[Envelope, BatchSummary]:
|
|
"""Build the 835 envelope from ISA/GS/ST. Raises on missing/malformed ISA."""
|
|
summary = BatchSummary(input_file=input_file)
|
|
envelope: Envelope | None = None
|
|
for seg in segments:
|
|
if seg[0] == "ISA":
|
|
try:
|
|
envelope = Envelope(
|
|
sender_id=seg[6].strip(),
|
|
receiver_id=seg[8].strip(),
|
|
control_number=seg[13].strip(),
|
|
transaction_date=date(2026, 1, 1), # overwritten by BPR
|
|
implementation_guide=None,
|
|
)
|
|
except (IndexError, ValueError) as exc:
|
|
raise CycloneParseError(f"Bad ISA: {exc}") from exc
|
|
elif seg[0] == "ST" and envelope is not None:
|
|
if len(seg) > 3:
|
|
envelope = envelope.model_copy(update={"implementation_guide": seg[3]})
|
|
if envelope is None:
|
|
raise CycloneParseError("No ISA envelope found")
|
|
return envelope, summary
|
|
|
|
|
|
def _consume_bpr(segments: list[list[str]], idx: int) -> tuple[FinancialInfo, int]:
|
|
seg = segments[idx]
|
|
try:
|
|
handling = seg[1] if len(seg) > 1 else ""
|
|
paid = Decimal(seg[2]) if len(seg) > 2 and seg[2] else Decimal("0")
|
|
credit_debit = seg[3] if len(seg) > 3 else ""
|
|
method = seg[4] if len(seg) > 4 and seg[4] else None
|
|
# BPR10 = Payer Identifier (Origin Company ID), per the CO Medicaid
|
|
# guide: "81-1725341" (TXIX) or "84-0644739" (BHA).
|
|
payer_tax = seg[10] if len(seg) > 10 and seg[10] else None
|
|
# BPR16 = Effective Entry Date (CCYYMMDD). The segment is sparse —
|
|
# positions 11..15 (BPR11..BPR15) are usually empty, so we look for
|
|
# any 8-digit CCYYMMDD token from index 11 onward.
|
|
payment_date = None
|
|
for k in range(11, len(seg)):
|
|
if seg[k] and len(seg[k]) == 8 and seg[k].isdigit():
|
|
payment_date = _safe_date(seg[k])
|
|
if payment_date is not None:
|
|
break
|
|
except (IndexError, ValueError) as exc:
|
|
raise CycloneParseError(f"Bad BPR: {exc}", segment_index=idx) from exc
|
|
return (
|
|
FinancialInfo(
|
|
handling_code=handling,
|
|
paid_amount=paid,
|
|
credit_debit_flag=credit_debit,
|
|
payment_method=method,
|
|
payer_tax_id=payer_tax,
|
|
payment_date=payment_date,
|
|
),
|
|
idx + 1,
|
|
)
|
|
|
|
|
|
def _consume_trn(segments: list[list[str]], idx: int) -> tuple[ReassociationTrace, int]:
|
|
seg = segments[idx]
|
|
trace_type = seg[1] if len(seg) > 1 else ""
|
|
trace_number = seg[2] if len(seg) > 2 else ""
|
|
originating = seg[3] if len(seg) > 3 else ""
|
|
payer_id = seg[4] if len(seg) > 4 and seg[4] else None
|
|
return (
|
|
ReassociationTrace(
|
|
trace_type_code=trace_type,
|
|
trace_number=trace_number,
|
|
originating_company_id=originating,
|
|
payer_id=payer_id,
|
|
),
|
|
idx + 1,
|
|
)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Loop 1000A (payer) / 1000B (payee)
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def _consume_address(segments: list[list[str]], idx: int) -> tuple[Address | None, int]:
|
|
"""Walk N3/N4 (if present) and return the assembled Address, plus new idx."""
|
|
addr: Address | None = None
|
|
while idx < len(segments) and segments[idx][0] in {"N3", "N4"}:
|
|
seg = segments[idx]
|
|
if seg[0] == "N3":
|
|
line1 = seg[1] if len(seg) > 1 else ""
|
|
line2 = seg[2] if len(seg) > 2 else None
|
|
addr = (addr or Address(line1=line1, city="", state="", zip="")).model_copy(
|
|
update={"line1": line1, "line2": line2}
|
|
)
|
|
elif seg[0] == "N4":
|
|
city = seg[1] if len(seg) > 1 else ""
|
|
state = seg[2] if len(seg) > 2 else ""
|
|
zip_ = seg[3] if len(seg) > 3 else ""
|
|
addr = (addr or Address(line1="", city=city, state=state, zip=zip_)).model_copy(
|
|
update={"city": city, "state": state, "zip": zip_}
|
|
)
|
|
idx += 1
|
|
return addr, idx
|
|
|
|
|
|
def _consume_1000a(segments: list[list[str]], idx: int) -> tuple[Payer835, int]:
|
|
"""Read N1*PR + N3 + N4 + PER* + REF* up to next N1 or CLP/LX/SE."""
|
|
name = ""
|
|
pid: str | None = None
|
|
contact_url: str | None = None
|
|
# First segment is N1*PR
|
|
seg = segments[idx]
|
|
if seg[0] == "N1" and len(seg) > 1 and seg[1] == "PR":
|
|
# NM103 is the name (or organization name for non-person entities).
|
|
name = seg[3] if len(seg) > 3 and seg[3] else (seg[2] if len(seg) > 2 else "")
|
|
# NM108/NM109 carry the ID code qualifier and value. The CO guide
|
|
# documents XV for the Health Plan ID, but other payers may use
|
|
# PI / 24 / etc., so we accept any qualifier.
|
|
for j, qual in enumerate(seg):
|
|
if qual in {"XV", "PI", "24"} and j + 1 < len(seg):
|
|
pid = seg[j + 1]
|
|
break
|
|
idx += 1
|
|
addr: Address | None = None
|
|
while idx < len(segments) and segments[idx][0] not in {"N1", "LX", "CLP", "SE"}:
|
|
seg = segments[idx]
|
|
if seg[0] in {"N3", "N4"}:
|
|
addr, idx = _consume_address(segments, idx)
|
|
continue
|
|
if seg[0] == "PER":
|
|
# PER03 is the communication number qualifier; PER04 is the
|
|
# value. The CO guide documents PER04 as the URL but in real
|
|
# 835s the URL is tagged with PER03=UR. We scan for the UR
|
|
# qualifier to be safe.
|
|
for j in range(1, len(seg) - 1, 2):
|
|
if seg[j] == "UR" and j + 1 < len(seg):
|
|
contact_url = seg[j + 1]
|
|
break
|
|
# Fall back to PER04 directly if it looks like a URL (matches
|
|
# the literal CO guide wording).
|
|
if contact_url is None and len(seg) > 4 and seg[4]:
|
|
value = seg[4]
|
|
if "@" in value or "http" in value or "." in value:
|
|
contact_url = value
|
|
idx += 1
|
|
return Payer835(name=name, id=pid, address=addr, contact_url=contact_url), idx
|
|
|
|
|
|
def _consume_1000b(segments: list[list[str]], idx: int) -> tuple[Payee835, int]:
|
|
"""Read N1*PE + N3 + N4 up to next N1 or CLP/LX/SE."""
|
|
name = ""
|
|
npi = ""
|
|
seg = segments[idx]
|
|
if seg[0] == "N1" and len(seg) > 1 and seg[1] == "PE":
|
|
name = seg[3] if len(seg) > 3 and seg[3] else (seg[2] if len(seg) > 2 else "")
|
|
for j, qual in enumerate(seg):
|
|
if qual == "XX" and j + 1 < len(seg):
|
|
npi = seg[j + 1]
|
|
break
|
|
idx += 1
|
|
addr: Address | None = None
|
|
while idx < len(segments) and segments[idx][0] not in {"N1", "LX", "CLP", "SE"}:
|
|
seg = segments[idx]
|
|
if seg[0] in {"N3", "N4"}:
|
|
addr, idx = _consume_address(segments, idx)
|
|
continue
|
|
idx += 1
|
|
return Payee835(name=name, npi=npi, address=addr), idx
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Loop 2100 (claim payment) and 2110 (service payment)
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def _cas_segments(segments: list[list[str]], idx: int) -> tuple[list[ClaimAdjustment], int]:
|
|
"""Read consecutive CAS segments starting at ``idx`` and return aggregated adjustments."""
|
|
adjustments: list[ClaimAdjustment] = []
|
|
while idx < len(segments) and segments[idx][0] == "CAS":
|
|
seg = segments[idx]
|
|
try:
|
|
group = seg[1] if len(seg) > 1 else ""
|
|
# Each CAS carries up to 6 reason/amount/quantity triplets
|
|
# (elements 2..19). We expand each one into its own row.
|
|
# Indexes 2,3,4 | 5,6,7 | 8,9,10 | 11,12,13 | 14,15,16 | 17,18,19
|
|
triplet_starts = [2, 5, 8, 11, 14, 17]
|
|
for start in triplet_starts:
|
|
if len(seg) <= start:
|
|
break
|
|
reason = seg[start]
|
|
if not reason:
|
|
continue
|
|
amount_s = seg[start + 1] if len(seg) > start + 1 else ""
|
|
qty_s = seg[start + 2] if len(seg) > start + 2 else ""
|
|
amount = Decimal(amount_s) if amount_s else Decimal("0")
|
|
quantity = Decimal(qty_s) if qty_s else None
|
|
adjustments.append(
|
|
ClaimAdjustment(
|
|
group_code=group,
|
|
reason_code=reason,
|
|
amount=amount,
|
|
quantity=quantity,
|
|
)
|
|
)
|
|
except (IndexError, ValueError) as exc:
|
|
log.warning("Bad CAS at segment %d: %s", idx, exc)
|
|
idx += 1
|
|
return adjustments, idx
|
|
|
|
|
|
def _consume_service_payment(segments: list[list[str]], idx: int, line_no: int) -> tuple[ServicePayment, int, list[list[str]]]:
|
|
"""Read SVC + its child segments. Returns (ServicePayment, new_idx, raw_segments).
|
|
|
|
The third return value is the list of raw segments consumed (SVC itself
|
|
plus any DTM/REF/CAS children) so the caller can fold them into the
|
|
claim's raw_segments. :class:`ServicePayment` does not carry a
|
|
``raw_segments`` field per the 835 spec, so we surface them at the
|
|
claim level instead.
|
|
"""
|
|
seg = segments[idx]
|
|
raw: list[list[str]] = [seg]
|
|
qualifier = ""
|
|
code = ""
|
|
modifiers: list[str] = []
|
|
if len(seg) > 1 and seg[1]:
|
|
svc01 = seg[1].split(":")
|
|
if len(svc01) > 0:
|
|
qualifier = svc01[0]
|
|
if len(svc01) > 1:
|
|
code = svc01[1]
|
|
modifiers = [m for m in svc01[2:] if m]
|
|
charge = Decimal(seg[2]) if len(seg) > 2 and seg[2] else Decimal("0")
|
|
payment = Decimal(seg[3]) if len(seg) > 3 and seg[3] else Decimal("0")
|
|
units: Decimal | None = None
|
|
unit_type: str | None = None
|
|
# X12 005010X221A1: SVC04 = Unit or Basis for Measurement Code
|
|
# (UN, MJ, DA, ...); SVC05 = Service Unit Count.
|
|
if len(seg) > 4 and seg[4]:
|
|
unit_type = seg[4]
|
|
if len(seg) > 5 and seg[5]:
|
|
try:
|
|
units = Decimal(seg[5])
|
|
except Exception:
|
|
units = None
|
|
# Some payers omit SVC04 but still mean "UN" when SVC05 carries a
|
|
# count. Be conservative: only default to UN if no other value was
|
|
# set above.
|
|
if unit_type is None and units is not None:
|
|
unit_type = "UN"
|
|
service_date: date | None = None
|
|
ref_benefit_plan: str | None = None
|
|
adjustments: list[ClaimAdjustment] = []
|
|
|
|
idx += 1
|
|
while idx < len(segments) and segments[idx][0] not in {"SVC", "LX", "CLP", "SE"}:
|
|
s = segments[idx]
|
|
raw.append(s)
|
|
if s[0] == "DTM" and len(s) > 2 and s[1] == "472":
|
|
service_date = _safe_date(s[2]) if s[2] else None
|
|
elif s[0] == "REF" and len(s) > 2 and s[1] == "0K":
|
|
ref_benefit_plan = s[2]
|
|
elif s[0] == "CAS":
|
|
cas, idx = _cas_segments(segments, idx)
|
|
raw.extend(c[0] for c in cas if False) # CAS rows are already in adjustments; nothing more to do
|
|
adjustments.extend(cas)
|
|
continue
|
|
idx += 1
|
|
|
|
return (
|
|
ServicePayment(
|
|
line_number=line_no,
|
|
procedure_qualifier=qualifier,
|
|
procedure_code=code,
|
|
modifiers=modifiers,
|
|
charge=charge,
|
|
payment=payment,
|
|
units=units,
|
|
unit_type=unit_type,
|
|
service_date=service_date,
|
|
adjustments=adjustments,
|
|
ref_benefit_plan=ref_benefit_plan,
|
|
),
|
|
idx,
|
|
raw,
|
|
)
|
|
|
|
|
|
def _consume_claim_payment(segments: list[list[str]], idx: int) -> tuple[ClaimPayment, int]:
|
|
"""Read one CLP and all its child segments (CAS / NM1 / REF / DTM / QTY / SVC)."""
|
|
clp = segments[idx]
|
|
raw: list[list[str]] = [clp]
|
|
try:
|
|
pcn = clp[1] if len(clp) > 1 else ""
|
|
status = clp[2] if len(clp) > 2 else ""
|
|
total_charge = Decimal(clp[3]) if len(clp) > 3 and clp[3] else Decimal("0")
|
|
total_paid = Decimal(clp[4]) if len(clp) > 4 and clp[4] else Decimal("0")
|
|
patient_resp = Decimal(clp[5]) if len(clp) > 5 and clp[5] else None
|
|
claim_filing = clp[6] if len(clp) > 6 and clp[6] else None
|
|
original_claim_id = clp[7] if len(clp) > 7 and clp[7] else None
|
|
facility_type = clp[8] if len(clp) > 8 and clp[8] else None
|
|
# CLP09 is a single element (no composite qualifier on the 835 side).
|
|
frequency = clp[9] if len(clp) > 9 and clp[9] else None
|
|
except (IndexError, ValueError) as exc:
|
|
raise CycloneParseError(f"Bad CLP: {exc}", segment_index=idx) from exc
|
|
|
|
service_payments: list[ServicePayment] = []
|
|
ref_benefit_plan: str | None = None
|
|
per_diem: Decimal | None = None
|
|
status_label = claim_status_label(status)
|
|
|
|
idx += 1
|
|
while idx < len(segments) and segments[idx][0] not in {"CLP", "SE"}:
|
|
s = segments[idx]
|
|
# Only consume this CLP's children; an LX starts a new 2100 loop.
|
|
if s[0] == "LX":
|
|
# Skip the LX and expect an SVC; if absent, stop.
|
|
raw.append(s)
|
|
idx += 1
|
|
if idx < len(segments) and segments[idx][0] == "SVC":
|
|
line_no = (
|
|
int(s[1]) if len(s) > 1 and s[1].isdigit() else len(service_payments) + 1
|
|
)
|
|
sp, idx, svc_raw = _consume_service_payment(segments, idx, line_no)
|
|
raw.extend(svc_raw)
|
|
service_payments.append(sp)
|
|
continue
|
|
if s[0] == "SVC":
|
|
sp, idx, svc_raw = _consume_service_payment(segments, idx, len(service_payments) + 1)
|
|
raw.extend(svc_raw)
|
|
service_payments.append(sp)
|
|
continue
|
|
raw.append(s)
|
|
if s[0] == "CAS":
|
|
cas, idx = _cas_segments(segments, idx)
|
|
# Top-level claim adjustments are aggregated into a synthesized
|
|
# ServicePayment with line_number=0 if there are no SVC rows.
|
|
# Per the spec, ClaimPayment doesn't carry adjustments directly,
|
|
# so we record them on the first service payment if any,
|
|
# otherwise drop them on the floor (rare in practice).
|
|
if service_payments:
|
|
first = service_payments[0]
|
|
service_payments[0] = first.model_copy(
|
|
update={"adjustments": first.adjustments + cas}
|
|
)
|
|
continue
|
|
if s[0] == "REF" and len(s) > 2 and s[1] == "CE":
|
|
ref_benefit_plan = s[2]
|
|
elif s[0] == "QTY" and len(s) > 2 and s[1] == "CA":
|
|
try:
|
|
per_diem = Decimal(s[2]) if s[2] else None
|
|
except ValueError:
|
|
per_diem = None
|
|
elif s[0] == "NM1":
|
|
# Patient (QC) / service-provider (1P) — captured in raw_segments.
|
|
# The 835 spec doesn't require a structured patient model in v1.
|
|
pass
|
|
elif s[0] == "DTM":
|
|
# Claim-level dates — captured in raw_segments.
|
|
pass
|
|
idx += 1
|
|
|
|
return (
|
|
ClaimPayment(
|
|
payer_claim_control_number=pcn,
|
|
status_code=status,
|
|
status_label=status_label,
|
|
total_charge=total_charge,
|
|
total_paid=total_paid,
|
|
patient_responsibility=patient_resp,
|
|
claim_filing_indicator=claim_filing,
|
|
original_claim_id=original_claim_id,
|
|
facility_type=facility_type,
|
|
frequency_code=frequency,
|
|
per_diem_covered_days=per_diem,
|
|
ref_benefit_plan=ref_benefit_plan,
|
|
service_payments=service_payments,
|
|
raw_segments=raw,
|
|
),
|
|
idx,
|
|
)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Top-level orchestrator
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def parse(text: str, payer_config: PayerConfig835, input_file: str = "") -> ParseResult835: # noqa: ARG001 — payer_config reserved for future use
|
|
"""Parse a complete 835 ERA document and return a :class:`ParseResult835`.
|
|
|
|
``payer_config`` is accepted for symmetry with the 837P parser and
|
|
for future payer-aware heuristics; in v1 the structural validation
|
|
runs against the full result via :func:`cyclone.parsers.validator_835.validate`.
|
|
"""
|
|
segments = tokenize(text)
|
|
envelope, summary = _build_envelope(segments, input_file=input_file)
|
|
|
|
financial: FinancialInfo | None = None
|
|
trace: ReassociationTrace | None = None
|
|
payer = Payer835(name="", id=None, address=None, contact_url=None)
|
|
payee = Payee835(name="", npi="", address=None)
|
|
claims: list[ClaimPayment] = []
|
|
|
|
i = 0
|
|
n_bprs = 0 # CO Medicaid sends multiple BPRs in one 835; we sum them.
|
|
# Pre-envelope segments (ISA/GS) are already consumed by _build_envelope.
|
|
# Walk forward looking for the header (BPR/TRN), then loops.
|
|
while i < len(segments):
|
|
seg = segments[i]
|
|
if seg[0] == "ST":
|
|
# Skip the ST control segment.
|
|
i += 1
|
|
continue
|
|
if seg[0] == "BPR":
|
|
new_fin, i = _consume_bpr(segments, i)
|
|
if financial is None:
|
|
financial = new_fin
|
|
else:
|
|
# CO Medicaid "split payment" pattern: multiple BPRs whose
|
|
# BPR02 amounts sum to the total paid. X12 835 only allows
|
|
# one BPR per transaction set; we accept the data and warn.
|
|
# Keep the first BPR's handling_code, tax_id, payment_date;
|
|
# only the paid_amount is additive.
|
|
financial = financial.model_copy(update={
|
|
"paid_amount": financial.paid_amount + new_fin.paid_amount,
|
|
})
|
|
n_bprs += 1
|
|
continue
|
|
if seg[0] == "TRN":
|
|
trace, i = _consume_trn(segments, i)
|
|
continue
|
|
if seg[0] == "N1" and len(seg) > 1 and seg[1] == "PR":
|
|
payer, i = _consume_1000a(segments, i)
|
|
continue
|
|
if seg[0] == "N1" and len(seg) > 1 and seg[1] == "PE":
|
|
payee, i = _consume_1000b(segments, i)
|
|
continue
|
|
if seg[0] == "CLP":
|
|
try:
|
|
claim, i = _consume_claim_payment(segments, i)
|
|
claims.append(claim)
|
|
except CycloneParseError:
|
|
raise
|
|
except Exception as exc: # pragma: no cover - safety net
|
|
log.warning("Claim parse failed at segment %d: %s", i, exc)
|
|
i += 1
|
|
continue
|
|
if seg[0] == "LX" and not claims and not payee.npi:
|
|
# Defensive: a stray LX before any CLP (no 1000B seen) is ignored.
|
|
i += 1
|
|
continue
|
|
i += 1
|
|
|
|
if financial is None:
|
|
raise CycloneParseError("No BPR (Financial Information) segment found")
|
|
if trace is None:
|
|
raise CycloneParseError("No TRN (Reassociation Trace) segment found")
|
|
|
|
# Update the envelope's transaction date from the BPR payment date if available.
|
|
if financial.payment_date is not None:
|
|
envelope = envelope.model_copy(update={"transaction_date": financial.payment_date})
|
|
|
|
summary = summary.model_copy(update={
|
|
"control_number": envelope.control_number,
|
|
"transaction_date": envelope.transaction_date,
|
|
"total_claims": len(claims),
|
|
})
|
|
|
|
return ParseResult835(
|
|
envelope=envelope,
|
|
financial_info=financial,
|
|
trace=trace,
|
|
payer=payer,
|
|
payee=payee,
|
|
claims=claims,
|
|
multi_bpr=(n_bprs > 1),
|
|
summary=summary,
|
|
validation=None,
|
|
)
|
|
|
|
|
|
__all__ = ["parse"]
|