From 28835e2f1d324dea649a43055aa5572e69c066d6 Mon Sep 17 00:00:00 2001 From: Nora Date: Tue, 7 Jul 2026 14:33:38 -0600 Subject: [PATCH] feat(sp25): cyclone recover-ingest CLI for stranded source files --- backend/src/cyclone/cli.py | 83 ++++++++ backend/src/cyclone/submission/recover.py | 247 ++++++++++++++++++++++ backend/tests/test_cli_recover_ingest.py | 161 ++++++++++++++ 3 files changed, 491 insertions(+) create mode 100644 backend/src/cyclone/submission/recover.py create mode 100644 backend/tests/test_cli_recover_ingest.py diff --git a/backend/src/cyclone/cli.py b/backend/src/cyclone/cli.py index 6b8a4f3..c962847 100644 --- a/backend/src/cyclone/cli.py +++ b/backend/src/cyclone/cli.py @@ -1376,5 +1376,88 @@ def ack_orphans_reconcile_cmd(dry_run: bool) -> None: click.echo("(dry-run — no rows inserted)") +# --------------------------------------------------------------------------- +# SP25: ``cyclone recover-ingest`` — offline parse + DB-write helper for +# stranded source files in ingest/. Mirrors the data-writing half of +# ``submit-batch`` but does NOT SFTP-upload (the files were already +# uploaded by whatever path got them into ingest/) and does NOT write +# to audit_log (the submit event is already in history). +# +# Supports 837P and 835. 999/TA1/277CA ingestion lives on +# ``cyclone pull-inbound``; this command is for the larger source files +# that need a typed BatchRecord + the canonical CycloneStore.add path so +# live-tail pages light up. +# +# Exit codes (mirrors submit-batch): +# 0 — at least one file ingested OK (or all were duplicates). +# 2 — a file failed to parse or DB-write. +# 1 — unexpected exception (DB unreachable, etc.). +# --------------------------------------------------------------------------- + + +@main.command("recover-ingest") +@click.option( + "--file", "files", multiple=True, required=True, + type=click.Path(exists=True, dir_okay=False, path_type=Path), + help="One or more X12 files to recover. Repeatable.", +) +@click.option( + "--sftp-block-name", "sftp_block_name", + default="manual-recover", show_default=True, + help="Dedup key in processed_inbound_files. Defaults to " + "'manual-recover' so it doesn't collide with the real " + "dzinesco block.", +) +@click.option( + "--actor", default="recover-ingest", show_default=True, + help="Audit actor tag (reserved — recovery doesn't audit by default).", +) +def recover_ingest_cmd( + files: tuple[Path, ...], + sftp_block_name: str, + actor: str, # noqa: ARG001 — reserved +) -> None: + """Parse X12 files from a local path and write them to the DB. + + Detects the kind (837P / 835) from the first ST segment; supports + only those two. Dedupes by ``(sftp_block_name, filename)`` — a + second invocation against the same file is a no-op. + + Use this when an 837P / 835 file landed in ``ingest/`` but the + cyclic ``parse-837`` / ``parse-835`` CLI only emits JSON (it does + NOT write to the DB) and ``submit-batch`` was never called. + """ + from cyclone.submission.recover import recover_file + + click.echo(f"Recovering {len(files)} file(s) under sftp_block={sftp_block_name!r}") + any_ok = False + failed = 0 + for f in files: + result = recover_file(f, sftp_block_name=sftp_block_name) + if result["status"] == "ok": + click.echo( + f" ok {f} kind={result['kind']} " + f"batch_id={result['batch_id']}" + ) + any_ok = True + elif result["status"] == "duplicate": + click.echo(f" skipped {f} (already processed)") + else: # failed + failed += 1 + click.echo( + f" FAILED {f} error={result['error']}", err=True, + ) + + if failed and not any_ok: + click.echo(f"\n{failed} file(s) failed; nothing ingested.", err=True) + sys.exit(2) + if failed: + # Mixed bag — some ok some failed. + click.echo(f"\n{failed} file(s) failed; the rest ingested.", err=True) + sys.exit(2) + if any_ok: + click.echo("\nDone.") + + if __name__ == "__main__": main() diff --git a/backend/src/cyclone/submission/recover.py b/backend/src/cyclone/submission/recover.py new file mode 100644 index 0000000..48640e5 --- /dev/null +++ b/backend/src/cyclone/submission/recover.py @@ -0,0 +1,247 @@ +"""SP25 Task 5: ``cyclone recover-ingest`` parse + DB-write helper. + +What this module is for +------------------------ +On 2026-07-07 the dashboard was empty even though four 837P files +and one 835 file were sitting in ``ingest/``. The 837Ps had been +SFTP-shipped to Gainwell (the operator's SFTP client, manually), +but cyclone never recorded them in the DB — ``parse-837`` only +emits JSON, and ``submit-batch`` was never called for that run. +The 835 came back from the payer and was dropped into ``ingest/``, +but ``pull-inbound`` was never run for it. + +This helper is the recovery path: parse the local file, build a +typed :class:`BatchRecord`, call :meth:`CycloneStore.add` (which +publishes events to the live-tail bus), record the file in +``processed_inbound_files`` for dedup, and stop — no SFTP, no audit +log because the file was already uploaded (or already came back). + +Public surface +-------------- +- :func:`recover_file` — one file. Returns a structured result dict. +- :func:`detect_kind` — ST01 helper, exposed for tests. + +Why this is its own module (not a new branch in submit_file) +------------------------------------------------------------ +``submit_file`` is the SFTP-write path; its invariants are +"DB write → SFTP upload → audit event" and removing the SFTP half +would mutate that contract for every caller. Recovery is a +deliberately offline path and a separate surface keeps the two +flows readable. ``recover_file`` shares the same parse + store +write code path (``CycloneStore.add``) so the live-tail pages +light up exactly the way they would for a fresh submission. +""" +from __future__ import annotations + +import logging +import uuid +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +from cyclone import db as db_mod +from cyclone.store import store as cycl_store +from cyclone.db import ProcessedInboundFile +from cyclone.parsers.parse_837 import parse as parse_837_text +from cyclone.parsers.parse_835 import parse as parse_835_text +from cyclone.parsers.payer import PayerConfig, PayerConfig835 +from cyclone.parsers.segments import tokenize as _tokenize_segments +from cyclone.store.records import BatchRecord835, BatchRecord837 + +log = logging.getLogger(__name__) + +# Default dedup key for the sftp_block_name column. Distinct from +# ``dzinesco`` so a real remote-pull can't accidentally mask a +# manual recovery (or vice versa). +DEFAULT_SFTP_BLOCK_NAME = "manual-recover" + + +def detect_kind(text: str) -> str: + """Return the ST01 of the *first* ST segment, or ``"unknown"``. + + X12 envelopes can contain multiple transactions; the first ST is + the canonical kind for the whole file under our parsing rules. + + Mirrors the SP35 envelope check in ``api_routers/parse.py`` + (auto-detect layer A; this is layer B for offline recovery). + """ + try: + segments = _tokenize_segments(text) + except Exception: # noqa: BLE001 — recover-context, log only + log.warning("recover-ingest: tokenize failed", exc_info=True) + return "unknown" + for seg in segments: + if seg and seg[0] == "ST" and len(seg) > 1: + return seg[1] + return "unknown" + + +def _normalize_kind(st01: str) -> str: + """Map ``"837"`` → ``"837p"`` (parser-level kind); pass others through. + + The parser distinguishes ``837p`` (professional) from the + 837I/D variants; only 837P is supported in the recovery path + today. Anything else is rejected with a clear error message. + """ + if st01 == "837": + return "837p" + if st01 in {"835", "837p"}: + return st01 + return st01 + + +def _already_processed(name: str, sftp_block_name: str) -> bool: + """Return True if ``(sftp_block_name, name)`` was already recovered. + + Mirrors the dedup index ``ux_processed_inbound_files_block_name`` + (UNIQUE on sftp_block_name, name). + """ + s = db_mod.SessionLocal()() + try: + return s.query(ProcessedInboundFile).filter_by( + sftp_block_name=sftp_block_name, name=name + ).first() is not None + finally: + s.close() + + +def _mark_processed( + *, + sftp_block_name: str, + path: Path, + file_type: str, + parser_used: str, + claim_count: int, +) -> None: + """Insert a ``processed_inbound_files`` row for the recovered file. + + Status hardcoded to ``"ok"`` because by the time we reach this + function the parse + store add have both succeeded. + """ + stat = path.stat() + now = datetime.now(tz=timezone.utc) + s = db_mod.SessionLocal()() + try: + s.add(ProcessedInboundFile( + sftp_block_name=sftp_block_name, + name=path.name, + size=stat.st_size, + modified_at=datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc), + file_type=file_type, + processed_at=now, + parser_used=parser_used, + claim_count=claim_count, + status="ok", + error_message=None, + )) + s.commit() + finally: + s.close() + + +def recover_file( + path: str | Path, + *, + sftp_block_name: str = DEFAULT_SFTP_BLOCK_NAME, + actor: str = "recover-ingest", +) -> dict: + """Recover one X12 file from a local path: parse → DB write (no SFTP). + + Args: + path: local file. 837P / 835 only. + sftp_block_name: dedup key in ``processed_inbound_files``. + Defaults to ``"manual-recover"`` so the recovery doesn't + collide with the real ``dzinesco`` SFTP block. + actor: reserved for audit. Recovery doesn't audit-log per + design (the file was already uploaded by whatever path + got it into ingest/), but the parameter is kept so a + future audit-enabled variant doesn't break callers. + + Returns: + dict with keys: + - ``file``: path as passed in (str) + - ``kind``: ``"837p"`` / ``"835"`` / ``None`` + - ``status``: ``"ok"`` / ``"duplicate"`` / ``"failed"`` + - ``batch_id``: hex uuid4 when status=="ok", else None + - ``error``: str when status=="failed", else None + """ + p = Path(path) + if not p.exists(): + return {"file": str(p), "kind": None, "status": "failed", + "batch_id": None, "error": f"file does not exist: {p}"} + if not p.is_file(): + return {"file": str(p), "kind": None, "status": "failed", + "batch_id": None, "error": f"not a regular file: {p}"} + + # 1. Dedup — short-circuit if this exact name was already recovered. + if _already_processed(p.name, sftp_block_name): + return {"file": str(p), "kind": None, "status": "duplicate", + "batch_id": None, "error": None} + + # 2. Read + detect kind. + try: + text = p.read_text(encoding="utf-8") + except UnicodeDecodeError as exc: + return {"file": str(p), "kind": None, "status": "failed", + "batch_id": None, "error": f"encoding: {exc!r}"} + + raw_kind = detect_kind(text) + kind = _normalize_kind(raw_kind) + if kind not in {"837p", "835"}: + return {"file": str(p), "kind": raw_kind, "status": "failed", + "batch_id": None, + "error": f"unsupported kind {raw_kind!r} (recovery supports 837p + 835)"} + + # 3. Parse through the canonical parser for the detected kind. + parsed: Any = None + if kind == "837p": + try: + parsed = parse_837_text(text, PayerConfig.co_medicaid(), input_file=p.name) + except Exception as exc: # noqa: BLE001 + return {"file": str(p), "kind": kind, "status": "failed", + "batch_id": None, "error": f"parse_837: {exc!r}"} + else: # "835" + try: + parsed = parse_835_text(text, PayerConfig835.co_medicaid_835(), input_file=p.name) + except Exception as exc: # noqa: BLE001 + return {"file": str(p), "kind": kind, "status": "failed", + "batch_id": None, "error": f"parse_835: {exc!r}"} + + # 4. Build a typed BatchRecord (mirrors api.py parse-837 happy path). + batch_id = uuid.uuid4().hex + now = datetime.now(tz=timezone.utc) + if kind == "837p": + record = BatchRecord837( + id=batch_id, input_filename=p.name, parsed_at=now, result=parsed, + ) + else: + record = BatchRecord835( + id=batch_id, input_filename=p.name, parsed_at=now, result=parsed, + ) + + # 5. DB write — canonical path. CycloneStore.add publishes events on + # the bundled bus so live-tail subscribers see new claims/remits. + try: + cycl_store.add(record) + except Exception as exc: # noqa: BLE001 + return {"file": str(p), "kind": kind, "status": "failed", + "batch_id": None, "error": f"db_write: {exc!r}"} + + # 6. Record success for dedup. + claim_count = len(getattr(parsed, "claims", []) or []) + try: + _mark_processed( + sftp_block_name=sftp_block_name, + path=p, + file_type=kind, + parser_used=f"parse_{kind}", + claim_count=claim_count, + ) + except Exception as exc: # noqa: BLE001 + # The DB write succeeded but dedup-record failed — log and + # still report ok (re-running will hit dedup once it's + # recorded; until then a re-run is a real duplicate insert). + log.warning("recover-ingest: dedup-write failed for %s: %r", p, exc) + + return {"file": str(p), "kind": kind, "status": "ok", + "batch_id": batch_id, "error": None} diff --git a/backend/tests/test_cli_recover_ingest.py b/backend/tests/test_cli_recover_ingest.py new file mode 100644 index 0000000..a6a833f --- /dev/null +++ b/backend/tests/test_cli_recover_ingest.py @@ -0,0 +1,161 @@ +"""SP25: integration tests for ``cyclone recover-ingest`` (parse + DB-write recovery). + +Fixtures used (all real production EDI from the test fixture corpus): + - ``co_medicaid_837p.txt`` — 1-claim 837P fixture (small, fast). + - ``co_medicaid_835.txt`` — minimal 835 fixture (real CO Medicaid). + +These tests do NOT cover the operator's full ingest/ batch — that's +Task 6 in the plan and runs against the live DB after these unit +tests are green. +""" +from __future__ import annotations + +from pathlib import Path + +import pytest +from click.testing import CliRunner + +from cyclone import db, store +from cyclone.cli import main as cli_main +from cyclone.db import Batch, Claim, ProcessedInboundFile, Remittance + +FIX_837P = Path(__file__).parent / "fixtures" / "co_medicaid_837p.txt" +FIX_835 = Path(__file__).parent / "fixtures" / "co_medicaid_835.txt" + + +@pytest.fixture +def runner(): + return CliRunner() + + +# --- helpers -------------------------------------------------------------- + +def _count_claims() -> int: + s = db.SessionLocal()() + try: + return s.query(Claim).count() + finally: + s.close() + + +def _count_remittances() -> int: + s = db.SessionLocal()() + try: + return s.query(Remittance).count() + finally: + s.close() + + +def _count_batches() -> int: + s = db.SessionLocal()() + try: + return s.query(Batch).count() + finally: + s.close() + + +# --- tests ---------------------------------------------------------------- + +def test_recover_ingest_837p_persists_claims(runner): + """One 837P ingest via the CLI lands a Batch + ≥1 Claim row + a dedup row.""" + pre_claims = _count_claims() + pre_batches = _count_batches() + result = runner.invoke( + cli_main, + ["recover-ingest", "--file", str(FIX_837P), + "--sftp-block-name", "test-recover-837p"], + catch_exceptions=False, + ) + assert result.exit_code == 0, result.stderr or result.stdout + out = (result.stdout or "") + (result.stderr or "") + assert "ok" in out + assert FIX_837P.name in out + + assert _count_claims() > pre_claims, "expected new Claim rows" + assert _count_batches() > pre_batches, "expected new Batch row" + + # Dedup row created. + s = db.SessionLocal()() + try: + row = s.query(ProcessedInboundFile).filter_by( + sftp_block_name="test-recover-837p", name=FIX_837P.name, + ).first() + finally: + s.close() + assert row is not None, "expected processed_inbound_files row" + assert row.status == "ok" + assert row.file_type == "837p" + + +def test_recover_ingest_835_persists_remittance(runner): + """One 835 ingest via the CLI lands a Batch + Remittance row + dedup row.""" + pre_batches = _count_batches() + pre_remits = _count_remittances() + result = runner.invoke( + cli_main, + ["recover-ingest", "--file", str(FIX_835), + "--sftp-block-name", "test-recover-835"], + catch_exceptions=False, + ) + assert result.exit_code == 0, result.stderr or result.stdout + assert "ok" in (result.stdout or "") + (result.stderr or "") + + assert _count_batches() > pre_batches + assert _count_remittances() > pre_remits + + s = db.SessionLocal()() + try: + row = s.query(ProcessedInboundFile).filter_by( + sftp_block_name="test-recover-835", name=FIX_835.name, + ).first() + finally: + s.close() + assert row is not None + assert row.status == "ok" + assert row.file_type == "835" + assert row.claim_count >= 1 + + +def test_recover_ingest_is_idempotent(runner): + """Second invocation with the same (block, file) is a no-op (duplicate).""" + # First call — must ingest. + r1 = runner.invoke( + cli_main, + ["recover-ingest", "--file", str(FIX_837P), + "--sftp-block-name", "test-recover-idem"], + catch_exceptions=False, + ) + assert r1.exit_code == 0, r1.stderr or r1.stdout + + pre_claims = _count_claims() + pre_batches = _count_batches() + # Second call — must skip. + r2 = runner.invoke( + cli_main, + ["recover-ingest", "--file", str(FIX_837P), + "--sftp-block-name", "test-recover-idem"], + catch_exceptions=False, + ) + assert r2.exit_code == 0, r2.stderr or r2.stdout + assert "skipped" in (r2.stdout or "") + (r2.stderr or "") + + # No new rows on second call. + assert _count_claims() == pre_claims + assert _count_batches() == pre_batches + + +def test_recover_ingest_missing_file_returns_failed(runner, tmp_path): + """Missing file → Click rejects with usage error (the CLI uses exists=True).""" + missing = tmp_path / "does-not-exist.837" + pre_claims = _count_claims() + result = runner.invoke( + cli_main, + ["recover-ingest", "--file", str(missing), + "--sftp-block-name", "test-recover-missing"], + catch_exceptions=False, + ) + # Click rejects up-front; usage error exits with code 2. + assert result.exit_code == 2 + out = (result.stdout or "") + (result.stderr or "") + assert "does not exist" in out or "Invalid value" in out + assert _count_claims() == pre_claims