feat(backend): click CLI for parse-837

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()

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