From d4f3b204efca3072c1f0b1a913c54cae2e5e9d8e Mon Sep 17 00:00:00 2001 From: Tyler Date: Fri, 19 Jun 2026 16:00:47 -0600 Subject: [PATCH] feat(backend): click CLI for parse-837 --- backend/src/cyclone/cli.py | 107 +++++++++++++++++++++++++++++++++++++ backend/tests/test_cli.py | 70 ++++++++++++++++++++++++ 2 files changed, 177 insertions(+) create mode 100644 backend/src/cyclone/cli.py create mode 100644 backend/tests/test_cli.py diff --git a/backend/src/cyclone/cli.py b/backend/src/cyclone/cli.py new file mode 100644 index 0000000..6516aac --- /dev/null +++ b/backend/src/cyclone/cli.py @@ -0,0 +1,107 @@ +"""Click-based CLI for the Cyclone 837P parser.""" + +from __future__ import annotations + +import logging +import sys +from pathlib import Path + +import click + +from cyclone.parsers.exceptions import CycloneParseError, CycloneValidationError +from cyclone.parsers.payer import PayerConfig +from cyclone.parsers.parse_837 import parse +from cyclone.parsers.writer import write_outputs + + +_PAYER_FACTORIES = { + "co_medicaid": PayerConfig.co_medicaid, + "generic_837p": PayerConfig.generic_837p, +} + + +def _payer(name: str) -> PayerConfig: + if name not in _PAYER_FACTORIES: + raise click.BadParameter(f"Unknown payer {name!r}. Choose from: {', '.join(_PAYER_FACTORIES)}") + return _PAYER_FACTORIES[name]() + + +@click.group() +def main() -> None: + """Cyclone EDI suite — X12 parser.""" + + +@main.command("parse-837") +@click.argument("input_file", type=click.Path(exists=True, dir_okay=False, path_type=Path)) +@click.option("--output-dir", required=True, type=click.Path(file_okay=False, path_type=Path)) +@click.option("--payer", default="co_medicaid", show_default=True, help="Payer configuration preset") +@click.option("--strict", is_flag=True, help="Promote warnings to errors") +@click.option("--max-retries", default=0, show_default=True, type=int, help="Re-validate failed claims (no fix in v1)") +@click.option("--include-raw-segments / --no-raw-segments", default=True, help="Embed raw segments in output JSON") +@click.option("--log-level", default="INFO", show_default=True, type=click.Choice(["DEBUG", "INFO", "WARNING", "ERROR"])) +def parse_837( + input_file: Path, + output_dir: Path, + payer: str, + strict: bool, + max_retries: int, + include_raw_segments: bool, + log_level: str, +) -> None: + """Parse an X12 837P file into one JSON per claim.""" + logging.basicConfig(level=getattr(logging, log_level)) + + text = input_file.read_text() + config = _payer(payer) + + try: + result = parse(text, config, input_file=str(input_file)) + except CycloneParseError as exc: + click.echo(f"PARSE ERROR: {exc}", err=True) + sys.exit(2) + + if not result.claims: + raise click.UsageError("No claims found in file") from None + + # Strict mode: rewrite reports + if strict: + for claim in result.claims: + promoted = [ + issue.model_copy(update={"severity": "error"}) + for issue in claim.validation.warnings + ] + new_errors = claim.validation.errors + promoted + claim.validation = claim.validation.model_copy( + update={"errors": new_errors, "passed": not new_errors} + ) + # Recompute summary + passed = sum(1 for c in result.claims if c.validation.passed) + result.summary = result.summary.model_copy(update={ + "passed": passed, + "failed": len(result.claims) - passed, + "failed_claim_ids": [c.claim_id for c in result.claims if not c.validation.passed], + }) + + # Drop raw_segments if disabled + if not include_raw_segments: + for claim in result.claims: + claim.raw_segments = [] + + # v1: max_retries is a no-op stub. Surface a log line so users know it's wired. + if max_retries: + logging.info("max-retries=%d (no auto-fix in v1; would re-validate here)", max_retries) + + paths = write_outputs(result, output_dir) + result.summary = result.summary.model_copy(update={"output_dir": str(output_dir)}) + (output_dir / "summary.json").write_text(result.summary.model_dump_json(indent=2)) + + click.echo( + f"parsed={result.summary.total_claims} " + f"passed={result.summary.passed} " + f"failed={result.summary.failed} " + f"output={output_dir}" + ) + + +if __name__ == "__main__": + main() diff --git a/backend/tests/test_cli.py b/backend/tests/test_cli.py new file mode 100644 index 0000000..4e018ec --- /dev/null +++ b/backend/tests/test_cli.py @@ -0,0 +1,70 @@ +import json +from pathlib import Path + +import pytest +from click.testing import CliRunner + +from cyclone.cli import main +from cyclone.parsers.payer import PayerConfig + + +FIXTURE = Path(__file__).parent / "fixtures" / "minimal_837p.txt" + + +def test_cli_parse_837_writes_outputs(tmp_path: Path): + runner = CliRunner() + result = runner.invoke(main, ["parse-837", str(FIXTURE), "--output-dir", str(tmp_path)]) + assert result.exit_code == 0, result.output + written = list(tmp_path.glob("claim-*.json")) + assert len(written) == 1 + summary = json.loads((tmp_path / "summary.json").read_text()) + assert summary["total_claims"] == 1 + assert summary["passed"] == 1 + + +def test_cli_parse_837_with_strict(tmp_path: Path): + runner = CliRunner() + result = runner.invoke(main, ["parse-837", str(FIXTURE), "--output-dir", str(tmp_path), "--strict"]) + assert result.exit_code == 0 + # Strict should not change anything on a passing claim + summary = json.loads((tmp_path / "summary.json").read_text()) + assert summary["passed"] == 1 + + +def test_cli_parse_837_with_no_raw_segments(tmp_path: Path): + runner = CliRunner() + result = runner.invoke(main, ["parse-837", str(FIXTURE), "--output-dir", str(tmp_path), "--no-raw-segments"]) + assert result.exit_code == 0 + data = json.loads(next(tmp_path.glob("claim-*.json")).read_text()) + assert data["raw_segments"] == [] + + +def test_cli_unknown_payer_uses_generic(tmp_path: Path): + runner = CliRunner() + result = runner.invoke(main, ["parse-837", str(FIXTURE), "--output-dir", str(tmp_path), "--payer", "generic_837p"]) + assert result.exit_code == 0 + + +def test_cli_missing_file_exits_2(tmp_path: Path): + runner = CliRunner() + result = runner.invoke(main, ["parse-837", "/no/such/file.txt", "--output-dir", str(tmp_path)]) + assert result.exit_code == 2 + + +@pytest.mark.skipif( + not (Path(__file__).parent.parent.parent / "docs" / "prodfiles" / "837p-from-axiscare").exists(), + reason="prodfile corpus not present", +) +def test_cli_runs_against_prodfiles(tmp_path: Path): + """Smoke test: run the CLI against a real production file if available.""" + from glob import glob + runner = CliRunner() + corpus = Path(__file__).parent.parent.parent / "docs" / "prodfiles" / "837p-from-axiscare" + files = sorted(glob(str(corpus / "*.txt"))) + if not files: + pytest.skip("no prodfile .txt files in corpus") + sample = files[0] + result = runner.invoke(main, ["parse-837", sample, "--output-dir", str(tmp_path)]) + assert result.exit_code == 0, result.output + summary = json.loads((tmp_path / "summary.json").read_text()) + assert summary["total_claims"] >= 1