feat(sp24): add cyclone.reissue subpackage with pure core functions

The script at scripts/reissue_claims.py was the operator's ad-hoc
workflow for rebuilding the 2026-07-08 dzinesco July-8 batches into
358 IG-correct single-claim X12 files. This commit lifts that
workflow into the cyclone namespace as cyclone.reissue — a pure
package with no Click imports inside.

Public surface (see cyclone/reissue/__init__.py):
  - parse_inputs(input_dir, payer_config) -> (claims, errors)
    AppleDouble-aware glob; per-file failure tolerance.
  - emit_outputs(claims, ..., output_dir, ...) -> list[Path]
    HCPF-spec filenames, per-claim 1ms offsets for uniqueness.
  - write_summary_sidecar(claims, written, output_dir) -> Path
    Operator-facing _serialize_summary.json sidecar.
  - zip_outputs(files, zip_path) -> None
    zipfile.ZIP_DEFLATED + testzip() integrity check.
  - ig_correctness_check(logger=None) -> bool
    Verifies serialize_837.PATIENT_LOOP_DEFAULT_INCLUDED is False.

The CLI wrapper (cyclone reissue-claims) lands in the next commit;
this commit is just the pure layer so the unit tests in
test_reissue_core.py can exercise the functions without a CliRunner.
This commit is contained in:
Nora
2026-07-08 22:28:59 -06:00
parent 9e422fdadb
commit df3e5e401d
2 changed files with 280 additions and 0 deletions
+43
View File
@@ -0,0 +1,43 @@
"""cyclone.reissue — offline 837P re-emission workflow.
Pure functions for parsing raw 837P files into ``ClaimOutput`` Pydantic
models and re-emitting one IG-correct single-claim X12 file per claim.
No Click imports inside :mod:`cyclone.reissue.core`; the CLI wrapper at
:mod:`cyclone.cli` is a thin shell over these functions.
Public surface:
- :func:`parse_inputs` — parse every ``*.x12`` / ``*.txt`` / ``*.edi``
under a directory; tolerates per-file failures and skips claims with
hard validation errors.
- :func:`emit_outputs` — write one X12 per claim to ``output_dir``
using HCPF-spec filenames; per-claim 1 ms timestamp offsets guarantee
unique filenames within a batch.
- :func:`write_summary_sidecar` — write a ``_serialize_summary.json``
sidecar with one row per emitted file (operator audit trail).
- :func:`zip_outputs` — zip the per-claim X12 files into a single flat
archive with a ``testzip()`` integrity check.
- :func:`ig_correctness_check` — verify the serializer's
``PATIENT_LOOP_DEFAULT_INCLUDED`` is ``False`` (the IG-correct
shape for SBR02 == "18" claims). See
``docs/superpowers/specs/2026-07-08-cyclone-reissue-claims-design.md``
for the SP24 rationale.
See `scripts/reissue_claims.py` for the deprecation shim; the
canonical entry point is ``cyclone reissue-claims`` (SP24).
"""
from cyclone.reissue.core import (
emit_outputs,
ig_correctness_check,
parse_inputs,
write_summary_sidecar,
zip_outputs,
)
__all__ = [
"parse_inputs",
"emit_outputs",
"write_summary_sidecar",
"zip_outputs",
"ig_correctness_check",
]
+237
View File
@@ -0,0 +1,237 @@
"""SP24 — pure functions for the offline 837P reissue workflow.
The functions here are deliberately Click-free so the CLI layer at
:mod:`cyclone.cli` stays a thin shell, and so unit tests can exercise
the behaviour without :class:`click.testing.CliRunner`.
Workflow contract:
>>> claims, errors = parse_inputs(input_dir, payer_config)
>>> written = emit_outputs(claims, payer_config=..., output_dir=..., ...)
>>> zip_outputs(written, zip_path)
``ig_correctness_check`` is the regression guard for the SP24 fix.
The serializer's ``PATIENT_LOOP_DEFAULT_INCLUDED`` constant MUST be
``False``; flipping it back to ``True`` would re-introduce the
Edifabric 999 rejection on every SBR02 == "18" claim.
"""
from __future__ import annotations
import json
import logging
import zipfile
from datetime import datetime, timedelta
from pathlib import Path
from zoneinfo import ZoneInfo
from cyclone.edi.filenames import build_outbound_filename
from cyclone.parsers.models import ClaimOutput
from cyclone.parsers.parse_837 import parse as parse_837_text
from cyclone.parsers.serialize_837 import (
PATIENT_LOOP_DEFAULT_INCLUDED,
serialize_837,
)
_log = logging.getLogger(__name__)
def ig_correctness_check(*, logger: logging.Logger | None = None) -> bool:
"""Return True iff the serializer's IG-correct patient-loop default is in place.
Per X12 005010X222A1, the 2000C Patient Hierarchical Level
(``HL*3 → PAT → NM1*QC``) MUST be absent when ``SBR02 == "18"``
(Self-pay, the CO-Medicaid IHSS workflow). The serializer encodes
this as a module-level constant ``PATIENT_LOOP_DEFAULT_INCLUDED``
in :mod:`cyclone.parsers.serialize_837`; this function checks
that constant.
Args:
logger: optional logger for the diagnostic WARNING when the
guard fires. Defaults to this module's logger.
Returns:
True if the constant is ``False`` (IG-correct). False
otherwise — callers should refuse to emit any X12 in that case.
"""
log = logger or _log
if PATIENT_LOOP_DEFAULT_INCLUDED is not False:
log.warning(
"REFUSING to run: PATIENT_LOOP_DEFAULT_INCLUDED = %r "
"(expected False). The IG-correct serializer shape for "
"SBR02='18' claims requires the 2000C patient loop to be "
"absent; restoring the default to False fixes it.",
PATIENT_LOOP_DEFAULT_INCLUDED,
)
return False
return True
def parse_inputs(
input_dir: Path,
payer_config,
) -> tuple[list[ClaimOutput], list[tuple[Path, str]]]:
"""Parse every ``*.x12`` / ``*.txt`` / ``*.edi`` under ``input_dir``.
AppleDouble metadata files (``._foo.x12``, the macOS resource
forks that Samba / Finder scatter) are skipped at the glob stage.
Per-file failures are tolerated: a file that raises during
``parse_837_text`` lands in the second tuple element with the
exception class + message. The first tuple element collects every
surviving :class:`ClaimOutput`, including those with **warnings**
(warnings don't abort). Claims with hard ``validation.errors``
are dropped with an error message in the second tuple element.
Args:
input_dir: directory to walk (one level).
payer_config: a :class:`PayerConfig` instance passed to the
parser — typically ``PayerConfig.co_medicaid()``.
Returns:
``(claims, errors)`` where ``claims`` is the surviving
:class:`ClaimOutput` list and ``errors`` is a list of
``(path, message)`` tuples for each per-file failure.
"""
raw_files: list[Path] = []
for ext in ("*.x12", "*.txt", "*.edi"):
for p in sorted(input_dir.glob(ext)):
# Skip macOS AppleDouble metadata (._<name>) — binary
# forks of the resource fork, not actual EDI.
if p.name.startswith("._"):
continue
raw_files.append(p)
claims: list[ClaimOutput] = []
errors: list[tuple[Path, str]] = []
for f in raw_files:
try:
text = f.read_text(encoding="utf-8")
result = parse_837_text(text, payer_config, input_file=str(f))
except Exception as exc: # noqa: BLE001 — log + skip
errors.append((f, f"{exc.__class__.__name__}: {exc}"))
continue
if not result.claims:
errors.append((f, "no claims parsed (empty CLM loop?)"))
continue
for c in result.claims:
if c.validation.errors:
msgs = [f"{i.rule}: {i.message}" for i in c.validation.errors]
errors.append((f, f"hard errors in {c.claim_id}: {msgs}"))
continue
claims.append(c)
return claims, errors
def emit_outputs(
claims: list[ClaimOutput],
*,
payer_config,
output_dir: Path,
sender_id: str,
receiver_id: str,
submitter_name: str,
submitter_contact_name: str,
submitter_contact_email: str,
receiver_name: str,
) -> list[Path]:
"""Write one X12 per claim under ``output_dir`` using HCPF-spec filenames.
Per-claim 1 ms timestamp offsets on the base datetime guarantee
unique filenames within a batch. The output directory is created
if it doesn't exist. Claims are sorted by ``claim_id`` before
emission so the per-claim filenames line up with the canonical
sort order.
Args:
claims: parsed :class:`ClaimOutput` instances (typically the
first tuple element from :func:`parse_inputs`).
payer_config: a :class:`PayerConfig` — threads
``sbr09_claim_filing`` into the SBR09 segment.
output_dir: directory to write the per-claim X12 files into.
sender_id, receiver_id, submitter_*, receiver_name: envelope
metadata threaded into the ISA/GS/NM1*41/NM1*40 segments.
Returns:
The list of written :class:`Path` instances, in the same
order as the sorted claims.
"""
output_dir.mkdir(parents=True, exist_ok=True)
# America/Denver is the operator's home tz and the canonical
# timezone for HCPF submission timestamps; matching the tz
# keeps the 999 round-trip deterministic for the operator's
# daily pull.
base_ts = datetime.now(tz=ZoneInfo("America/Denver"))
written: list[Path] = []
for i, claim in enumerate(sorted(claims, key=lambda c: c.claim_id), start=1):
body = serialize_837(
claim,
sender_id=sender_id,
receiver_id=receiver_id,
submitter_name=submitter_name,
submitter_contact_name=submitter_contact_name,
submitter_contact_email=submitter_contact_email,
receiver_name=receiver_name,
claim_filing_indicator_code=payer_config.sbr09_claim_filing,
interchange_control_number=f"{i:09d}",
group_control_number="1",
)
ts = base_ts + timedelta(milliseconds=i)
fname = build_outbound_filename(sender_id, "837P", now_mt=ts)
path = output_dir / fname
path.write_text(body, encoding="ascii")
written.append(path)
return written
def write_summary_sidecar(
claims: list[ClaimOutput],
written: list[Path],
output_dir: Path,
) -> Path:
"""Write ``_serialize_summary.json`` next to the per-claim X12 files.
Operator-facing sidecar for audit: one row per emitted file with
``claim_id``, ``output_file`` (absolute path), and ``byte_size``.
Both ``claims`` and ``written`` are zipped in the same order as
``emit_outputs`` produced them (i.e. sorted by ``claim_id``).
Returns the path of the written sidecar.
"""
summary = [
{
"claim_id": c.claim_id,
"output_file": str(p),
"byte_size": p.stat().st_size,
}
for c, p in zip(sorted(claims, key=lambda c: c.claim_id), written)
]
path = output_dir / "_serialize_summary.json"
path.write_text(json.dumps(summary, indent=2), encoding="utf-8")
return path
def zip_outputs(files: list[Path], zip_path: Path) -> None:
"""Zip the per-claim X12 files into a single flat archive.
Uses :class:`zipfile.ZIP_DEFLATED`. After writing, runs
:meth:`zipfile.ZipFile.testzip` to verify CRC integrity and
raises :class:`RuntimeError` if any entry is corrupt.
Args:
files: the per-claim X12 file paths (typically the return
value of :func:`emit_outputs`).
zip_path: destination archive path; parent directories are
created as needed.
Raises:
RuntimeError: if any entry fails the ``testzip()`` integrity
check.
"""
zip_path.parent.mkdir(parents=True, exist_ok=True)
with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
for p in files:
zf.write(p, arcname=p.name)
bad = zipfile.ZipFile(zip_path).testzip()
if bad is not None:
raise RuntimeError(f"zip integrity check failed: {bad} is corrupt")