Tyler
200 lines
7.5 KiB
Python
200 lines
7.5 KiB
Python
"""Click-based CLI for the Cyclone 837P and 835 ERA parsers."""
|
|
|
|
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, PayerConfig835
|
|
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.validator_835 import validate as validate_835
|
|
from cyclone.parsers.writer import write_outputs
|
|
from cyclone.parsers.writer_835 import write_outputs_835
|
|
|
|
|
|
_PAYER_FACTORIES = {
|
|
"co_medicaid": PayerConfig.co_medicaid,
|
|
"generic_837p": PayerConfig.generic_837p,
|
|
}
|
|
|
|
_PAYER_FACTORIES_835 = {
|
|
"co_medicaid_835": PayerConfig835.co_medicaid_835,
|
|
"generic_835": PayerConfig835.generic_835,
|
|
}
|
|
|
|
|
|
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]()
|
|
|
|
|
|
def _payer_835(name: str) -> PayerConfig835:
|
|
if name not in _PAYER_FACTORIES_835:
|
|
raise click.BadParameter(f"Unknown payer {name!r}. Choose from: {', '.join(_PAYER_FACTORIES_835)}")
|
|
return _PAYER_FACTORIES_835[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_837_text(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}"
|
|
)
|
|
|
|
|
|
@main.command("parse-835")
|
|
@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_835", show_default=True, help="Payer configuration preset")
|
|
@click.option("--strict", is_flag=True, help="Promote warnings to errors")
|
|
@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_835(
|
|
input_file: Path,
|
|
output_dir: Path,
|
|
payer: str,
|
|
strict: bool,
|
|
include_raw_segments: bool,
|
|
log_level: str,
|
|
) -> None:
|
|
"""Parse an X12 835 ERA file into one JSON per claim payment."""
|
|
logging.basicConfig(level=getattr(logging, log_level))
|
|
|
|
text = input_file.read_text()
|
|
config = _payer_835(payer)
|
|
|
|
try:
|
|
result = parse_835_text(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 claim payments found in file") from None
|
|
|
|
# Run validation; populate summary.passed/failed from the report.
|
|
# 835 validation is batch-level (BPR/CLP/SVC balancing is cross-cutting),
|
|
# so pass/fail applies uniformly to every claim payment in the batch:
|
|
# when the batch passes, all N claims "passed" (passed=N, failed=0);
|
|
# when the batch fails, all N claims "failed" (passed=0, failed=N).
|
|
report = validate_835(result, config)
|
|
if strict:
|
|
promoted = [i.model_copy(update={"severity": "error"}) for i in report.warnings]
|
|
new_errors = report.errors + promoted
|
|
report = report.model_copy(update={"errors": new_errors, "passed": not new_errors})
|
|
n = len(result.claims)
|
|
claim_ids = [c.payer_claim_control_number for c in result.claims]
|
|
if report.passed:
|
|
passed, failed, failed_claim_ids = n, 0, []
|
|
else:
|
|
passed, failed, failed_claim_ids = 0, n, claim_ids
|
|
result = result.model_copy(update={"validation": report})
|
|
result.summary = result.summary.model_copy(update={
|
|
"passed": passed,
|
|
"failed": failed,
|
|
"failed_claim_ids": failed_claim_ids,
|
|
"issues_by_rule": _count_issues(report),
|
|
})
|
|
|
|
# Drop raw_segments if disabled
|
|
if not include_raw_segments:
|
|
result = result.model_copy(update={"claims": [c.model_copy(update={"raw_segments": []}) for c in result.claims]})
|
|
|
|
write_outputs_835(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}"
|
|
)
|
|
|
|
|
|
def _count_issues(report) -> dict[str, int]:
|
|
"""Build an issues_by_rule dict from a ValidationReport (for the summary)."""
|
|
counts: dict[str, int] = {}
|
|
for issue in report.errors + report.warnings:
|
|
counts[issue.rule] = counts.get(issue.rule, 0) + 1
|
|
return counts
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|