feat(backend): X12 segment tokenizer with ISA delimiter detection
This commit is contained in:
@@ -0,0 +1,80 @@
|
||||
"""X12 segment tokenizer.
|
||||
|
||||
The ISA segment is a fixed 106-character record (101 + segment terminator) that
|
||||
declares the four delimiters used in the rest of the file:
|
||||
|
||||
- element separator (3rd char, 0-indexed)
|
||||
- repetition separator (82nd char, 0-indexed; ISA11)
|
||||
- component separator (104th char, 0-indexed; ISA16)
|
||||
- segment terminator (105th char, 0-indexed)
|
||||
|
||||
We do not split components here — that is left to the caller, since most X12
|
||||
elements are simple and splitting unconditionally would corrupt data.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
|
||||
from cyclone.parsers.exceptions import CycloneParseError
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Delimiters:
|
||||
element: str # usually "*"
|
||||
repetition: str # usually "^"
|
||||
component: str # usually ":"
|
||||
segment: str # usually "~"
|
||||
|
||||
|
||||
def _detect_delimiters(isa_segment: str) -> Delimiters:
|
||||
"""Pull the four delimiters from the fixed-width ISA segment.
|
||||
|
||||
The ISA segment must be exactly 106 characters (including the terminator).
|
||||
"""
|
||||
if len(isa_segment) < 106:
|
||||
raise CycloneParseError(
|
||||
f"ISA segment too short: expected 106 chars, got {len(isa_segment)}"
|
||||
)
|
||||
return Delimiters(
|
||||
element=isa_segment[3],
|
||||
repetition=isa_segment[82],
|
||||
component=isa_segment[104],
|
||||
segment=isa_segment[105],
|
||||
)
|
||||
|
||||
|
||||
def tokenize(text: str) -> list[list[str]]:
|
||||
"""Split a raw X12 document into a list of segments (each a list of elements).
|
||||
|
||||
The first segment must be an ISA. Raises :class:`CycloneParseError` on
|
||||
missing or malformed envelopes.
|
||||
"""
|
||||
# Normalize line endings; the segment terminator handles real segmentation.
|
||||
text = text.replace("\r\n", "").replace("\n", "").replace("\r", "")
|
||||
|
||||
# Find the first ISA by scanning for the literal "ISA" + element separator.
|
||||
if not text.startswith("ISA") or len(text) < 4:
|
||||
raise CycloneParseError("ISA segment not found at start of file")
|
||||
|
||||
# The first 106 chars are the ISA segment.
|
||||
isa_segment = text[:106]
|
||||
try:
|
||||
delimiters = _detect_delimiters(isa_segment)
|
||||
except IndexError as exc: # pragma: no cover - guarded by length check
|
||||
raise CycloneParseError(f"Could not parse ISA delimiters: {exc}") from exc
|
||||
|
||||
# Slice off the ISA, then split the rest on the segment terminator.
|
||||
body = text[106:]
|
||||
raw_segments = body.split(delimiters.segment)
|
||||
|
||||
# Prepend the ISA so callers always see it as the first segment.
|
||||
isa_elements = isa_segment.split(delimiters.element)
|
||||
out: list[list[str]] = [isa_elements]
|
||||
|
||||
for raw in raw_segments:
|
||||
if not raw:
|
||||
continue
|
||||
out.append(raw.split(delimiters.element))
|
||||
|
||||
return out
|
||||
@@ -0,0 +1,69 @@
|
||||
import pytest
|
||||
from cyclone.parsers.exceptions import CycloneParseError
|
||||
from cyclone.parsers.segments import Delimiters, tokenize
|
||||
|
||||
ISA_PREFIX = "ISA*00* *00* *ZZ*11525703 *ZZ*COMEDASSISTPROG*260611*0814*"
|
||||
|
||||
|
||||
def test_tokenize_parses_isa_delimiters():
|
||||
text = ISA_PREFIX + "*^00501*991102977*1*P*:~"
|
||||
segments = tokenize(text)
|
||||
assert segments[0][0] == "ISA"
|
||||
assert len(segments) == 1
|
||||
|
||||
|
||||
def test_tokenize_splits_segments():
|
||||
text = ISA_PREFIX + "*^00501*991102977*1*P*:~ST*837*0001~SE*2*0001~"
|
||||
segments = tokenize(text)
|
||||
assert [s[0] for s in segments] == ["ISA", "ST", "SE"]
|
||||
|
||||
|
||||
def test_tokenize_splits_elements():
|
||||
text = ISA_PREFIX + "*^00501*991102977*1*P*:~NM1*85*2*Acme*****XX*1234567890~"
|
||||
segments = tokenize(text)
|
||||
nm1 = [s for s in segments if s[0] == "NM1"][0]
|
||||
assert nm1[1] == "85"
|
||||
assert nm1[8] == "XX"
|
||||
assert nm1[9] == "1234567890"
|
||||
|
||||
|
||||
def test_tokenize_preserves_empty_elements():
|
||||
"""Empty elements (e.g. 'NM1*IL*1*Doe*John****MI*X') must stay as empty strings."""
|
||||
text = ISA_PREFIX + "*^00501*991102977*1*P*:~NM1*IL*1*Doe*John****MI*X~"
|
||||
segments = tokenize(text)
|
||||
nm1 = [s for s in segments if s[0] == "NM1"][0]
|
||||
assert nm1[5] == "" # position 5 is the 4th consecutive empty after suffix
|
||||
assert nm1[6] == ""
|
||||
|
||||
|
||||
def test_tokenize_handles_component_separator():
|
||||
"""SV101 can be 'HC:99213' which is component-split at ':'."""
|
||||
text = ISA_PREFIX + "*^00501*991102977*1*P*:~SV1*HC:99213*100*UN*1***1~"
|
||||
segments = tokenize(text)
|
||||
sv1 = [s for s in segments if s[0] == "SV1"][0]
|
||||
# SV101 is element index 1; components 1, 2 are 'HC', '99213'
|
||||
assert sv1[1] == "HC:99213"
|
||||
|
||||
|
||||
def test_tokenize_strips_trailing_crlf():
|
||||
text = ISA_PREFIX + "*^00501*991102977*1*P*:~\r\n"
|
||||
segments = tokenize(text)
|
||||
assert [s[0] for s in segments] == ["ISA"]
|
||||
|
||||
|
||||
def test_tokenize_raises_on_missing_isa():
|
||||
with pytest.raises(CycloneParseError, match="ISA segment not found"):
|
||||
tokenize("hello world")
|
||||
|
||||
|
||||
def test_tokenize_raises_on_short_isa():
|
||||
with pytest.raises(CycloneParseError, match="ISA segment too short"):
|
||||
tokenize("ISA*00*00~")
|
||||
|
||||
|
||||
def test_delimiters_dataclass():
|
||||
d = Delimiters(element="*", repetition="^", component=":", segment="~")
|
||||
assert d.element == "*"
|
||||
assert d.repetition == "^"
|
||||
assert d.component == ":"
|
||||
assert d.segment == "~"
|
||||
Reference in New Issue
Block a user