diff --git a/backend/src/cyclone/cli.py b/backend/src/cyclone/cli.py index 2c3ddf6..0ac07c1 100644 --- a/backend/src/cyclone/cli.py +++ b/backend/src/cyclone/cli.py @@ -297,6 +297,97 @@ def validate_tax_id_cmd(tax_id: str, log_level: str) -> None: sys.exit(1) +# --------------------------------------------------------------------------- +# SP32: `cyclone backfill-rendering-npi` (Task 6) +# +# Re-parses on-disk 837p/835 files and patches up the typed NPI columns +# (``Claim.rendering_provider_npi`` / ``Remittance.rendering_provider_npi``) +# for rows that were ingested before the T4 writers took effect. +# --------------------------------------------------------------------------- + + +@main.command("backfill-rendering-npi") +@click.option( + "--file", "files", + multiple=True, + type=click.Path(exists=True, dir_okay=False, path_type=Path), + help=( + "Path to a specific file to re-parse. May be passed multiple times. " + "Mutually informative with --input-dir (both are processed)." + ), +) +@click.option( + "--input-dir", + type=click.Path(file_okay=False, path_type=Path), + default=None, + help=( + "Directory to scan one level deep for *.txt / *.edi / *.x12 files. " + "Honors CYCLONE_BACKFILL_INPUT_DIR if --input-dir is not passed." + ), +) +@click.option( + "--type", "transaction_type", + type=click.Choice(["837p", "835"]), + default=None, + help=( + "Pin the parser to use. Without --type, each file's transaction " + "kind is sniffed (filename hint + ISA/ST prefix)." + ), +) +@click.option( + "--log-level", + default="INFO", + show_default=True, + type=click.Choice(["DEBUG", "INFO", "WARNING", "ERROR"]), +) +def backfill_rendering_npi( + files: tuple[Path, ...], + input_dir: Path | None, + transaction_type: str | None, + log_level: str, +) -> None: + """Re-parse on-disk X12 files to populate the rendering NPI columns. + + Idempotent: rows whose ``rendering_provider_npi`` is already non-NULL + are left untouched. After patching, runs :func:`cyclone.reconcile.run` + once across every 835 batch so the T5 scoring arm can fire + retroactively on the touched pairs. + + Exit code: 0 on a successful run (zero populated rows is still exit 0). + """ + from cyclone import db as _db + from cyclone.store.backfill import backfill_rendering_provider_npi + + # SP18: re-run setup so per-command --log-level overrides the + # group default. ``setup_logging`` is idempotent. + setup_logging(level=log_level) + + # The backup pattern: each CLI subcommand that touches the DB + # initializes it explicitly so ``python -m cyclone.cli backup list`` + # works on a fresh machine with no prior app boot. + _db.init_db() + + # --input-dir falls back to CYCLONE_BACKFILL_INPUT_DIR. + if input_dir is None and not files: + env_dir = os.environ.get("CYCLONE_BACKFILL_INPUT_DIR", "").strip() + if env_dir: + input_dir = Path(env_dir) + + summary = backfill_rendering_provider_npi( + files=list(files) if files else None, + input_dir=input_dir, + transaction_type=transaction_type, + ) + + # One-line summary so operators can grep a log scrape quickly. + click.echo( + f"claims_updated={summary.claims_updated} " + f"remits_updated={summary.remits_updated} " + f"files_processed={summary.files_processed} " + f"files_skipped={summary.files_skipped}" + ) + + if __name__ == "__main__": main() diff --git a/backend/src/cyclone/store/backfill.py b/backend/src/cyclone/store/backfill.py new file mode 100644 index 0000000..33d52ae --- /dev/null +++ b/backend/src/cyclone/store/backfill.py @@ -0,0 +1,340 @@ +"""SP32 Task 6: backfill rendering/service-provider NPIs from on-disk files. + +The T4 writers populate ``Claim.rendering_provider_npi`` (from NM1*82 in +837P Loop 2420A) and ``Remittance.rendering_provider_npi`` (from the +NM1*1P service-provider segment in 835 Loop 2100). Rows ingested before +T4 was wired (or ingested via a path that bypasses the writer — e.g. an +ad-hoc ``store.add`` from a notebook) still have a NULL column. + +This module re-parses on-disk X12 files and patches up those columns on +matching rows. It is **idempotent**: rows whose +``rendering_provider_npi`` is already non-NULL are left untouched, and +re-running the same file twice is a clean no-op. + +Public surface +-------------- + +* :func:`backfill_rendering_provider_npi` — entry point used by the CLI. +* :class:`BackfillSummary` — counts dataclass echoed back as a one-line + summary by the CLI. + +Reconcile is run once at the end so the new typed NPI arm (T5) can fire +retroactively across the open claim/remit pairs the backfill touched. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass, field +from pathlib import Path +from typing import Iterable + +from sqlalchemy import select + +from cyclone import db +from cyclone.parsers.parse_835 import parse as parse_835 +from cyclone.parsers.parse_837 import parse as parse_837 +from cyclone.parsers.payer import PayerConfig, PayerConfig835 + +log = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# 837P fixture uses NM1*82 → rendering_provider_npi. +# 835 fixture uses NM1*1P → service_provider_npi (mapped onto the same +# Remittance.rendering_provider_npi column by T4 _remittance_835_row). +# --------------------------------------------------------------------------- + + +@dataclass +class BackfillSummary: + """Counts emitted by ``backfill_rendering_provider_npi``. + + The CLI echoes a one-line summary (`claims_updated=N remits_updated=N + …`) at the end of the run. + """ + + claims_updated: int = 0 + remits_updated: int = 0 + files_processed: int = 0 + files_skipped: int = 0 + + +# --------------------------------------------------------------------------- +# Parser wrappers — keep exceptions local so one bad file can't abort +# the whole backfill run. +# --------------------------------------------------------------------------- + + +def _parse_837_file(path: Path) -> tuple[list, Path | None]: + """Re-parse a single 837P file. Returns ``(claims, broken_alias)``. + + On success, ``claims`` is the list of parsed ``ClaimOutput`` rows + and ``broken_alias`` is None. On any failure, ``claims`` is empty + and ``broken_alias`` is the same ``path`` so the caller can log + which file was skipped. + """ + try: + text = path.read_text() + except OSError as exc: + log.warning("backfill: cannot read %s: %s", path, exc) + return [], path + try: + result = parse_837(text, PayerConfig.co_medicaid(), input_file=str(path)) + except Exception as exc: # noqa: BLE001 — failure isolated per-file + log.warning("backfill: 837 parse failed for %s: %s", path, exc) + return [], path + return list(result.claims), None + + +def _parse_835_file(path: Path) -> tuple[list, Path | None]: + """Re-parse a single 835 file. Returns ``(claims, broken_alias)``. + + ``claims`` is the list of parsed ``ClaimPayment`` rows. + """ + try: + text = path.read_text() + except OSError as exc: + log.warning("backfill: cannot read %s: %s", path, exc) + return [], path + try: + result = parse_835(text, PayerConfig835.co_medicaid_835(), input_file=str(path)) + except Exception as exc: # noqa: BLE001 + log.warning("backfill: 835 parse failed for %s: %s", path, exc) + return [], path + return list(result.claims), None + + +# --------------------------------------------------------------------------- +# DB-patching helpers — keep the per-row update logic in one place. +# --------------------------------------------------------------------------- + + +def _patch_claim_rendering_npi(parsed_claims: list) -> int: + """Update ``Claim.rendering_provider_npi`` for any parsed 837 claim. + + Only rows whose column is currently NULL are touched. Returns the + count of rows updated. + """ + from cyclone.db import Claim, SessionLocal + + updated = 0 + with SessionLocal()() as session: + for parsed in parsed_claims: + npi = getattr(parsed, "rendering_provider_npi", None) + if not npi: + continue + row = session.get(Claim, parsed.claim_id) + if row is None: + continue + if row.rendering_provider_npi is not None: + # Already populated (e.g. by a prior backfill run, or by + # the T4 writer if the claim was ingested afterwards). + continue + row.rendering_provider_npi = npi + updated += 1 + if updated: + session.commit() + return updated + + +def _patch_remit_rendering_npi(parsed_claims: list) -> int: + """Update ``Remittance.rendering_provider_npi`` for any parsed 835 claim. + + The 835 NM1*1P segment maps onto ``ClaimPayment.service_provider_npi`` + (which the T4 writer copies into ``Remittance.rendering_provider_npi``). + We match on ``payer_claim_control_number``; only NULL columns are + touched. Returns the count of rows updated. + """ + from cyclone.db import Remittance, SessionLocal + + updated = 0 + with SessionLocal()() as session: + for parsed in parsed_claims: + npi = getattr(parsed, "service_provider_npi", None) + if not npi: + continue + target_pcn = getattr(parsed, "payer_claim_control_number", None) + if not target_pcn: + continue + row = session.execute( + select(Remittance).where( + Remittance.payer_claim_control_number == target_pcn, + Remittance.rendering_provider_npi.is_(None), + ) + ).scalars().first() + if row is None: + continue + row.rendering_provider_npi = npi + updated += 1 + if updated: + session.commit() + return updated + + +# --------------------------------------------------------------------------- +# Reconcile pass — run once across every 835 batch so the T5 scoring arm +# can fire retroactively on pairs the backfill touched. +# --------------------------------------------------------------------------- + + +def _run_reconcile_sweep() -> int: + """Run :func:`cyclone.reconcile.run` over every 835 batch. Returns count. + + Failures are isolated per-batch so one bad batch can't abort the + sweep — the goal is "best-effort retroactive reconcile". + """ + from sqlalchemy import select as _select + + from cyclone import reconcile as _reconcile + from cyclone.db import Batch, SessionLocal + + completed = 0 + with SessionLocal()() as session: + batch_ids = [ + row[0] for row in session.execute( + _select(Batch.id).where(Batch.kind == "835") + ).all() + ] + # Each batch gets its own session — ``reconcile.run`` does not commit, + # so an exception in one batch must not orphan a half-flushed + # transaction. + for bid in batch_ids: + try: + with SessionLocal()() as s: + _reconcile.run(s, bid) + s.commit() + completed += 1 + except Exception as exc: # noqa: BLE001 + log.warning("backfill: reconcile failed for batch %s: %s", bid, exc) + return completed + + +# --------------------------------------------------------------------------- +# Public entry point. +# --------------------------------------------------------------------------- + + +# Supported ``--type`` flag values; also ``None`` means auto-detect. +TransactionType = str | None + + +def backfill_rendering_provider_npi( + *, + files: Iterable[Path] | None = None, + input_dir: Path | None = None, + transaction_type: TransactionType = None, +) -> BackfillSummary: + """Re-parse on-disk 837p + 835 files and populate the typed NPI columns. + + Args: + files: explicit list of file paths to re-parse. + input_dir: directory to scan for ``*.txt`` / ``*.edi`` files. + Files are scanned one level deep. + transaction_type: ``"837p"`` or ``"835"``. ``None`` auto-detects + by attempting the 837p parser first and falling back to + the 835 parser. + + Returns: + :class:`BackfillSummary` with populated counts. + + Idempotent: only writes columns that are currently NULL; re-running + on the same files is a clean no-op. After patching, runs + :func:`cyclone.reconcile.run` over every 835 batch so the T5 + scoring arm can re-fire on the touched pairs. + """ + summary = BackfillSummary() + + # 1. Resolve the candidate file set. + candidates: list[Path] = [] + if files: + candidates.extend(Path(f) for f in files) + if input_dir is not None: + for ext in ("*.txt", "*.edi", "*.835", "*.837", "*.x12"): + candidates.extend(input_dir.glob(ext)) + candidates = [p for p in candidates if p.is_file()] + + # 2. Re-parse each file and patch matching rows. + for path in candidates: + kind = transaction_type or _sniff_kind(path) + if kind == "837p": + parsed_claims, broken = _parse_837_file(path) + if broken is not None: + summary.files_skipped += 1 + continue + summary.files_processed += 1 + summary.claims_updated += _patch_claim_rendering_npi(parsed_claims) + elif kind == "835": + parsed_claims, broken = _parse_835_file(path) + if broken is not None: + summary.files_skipped += 1 + continue + summary.files_processed += 1 + summary.remits_updated += _patch_remit_rendering_npi(parsed_claims) + else: + # Auto-detect tried both parsers and both failed → skip. + summary.files_skipped += 1 + + # 3. Reconcile sweep — let the T5 NPI arm fire retroactively. + if summary.claims_updated or summary.remits_updated: + try: + _run_reconcile_sweep() + except Exception: # noqa: BLE001 — sweep is best-effort + log.exception("backfill: reconcile sweep failed") + + return summary + + +def _sniff_kind(path: Path) -> TransactionType: + """Best-effort transaction-type sniff (content + filename). + + Returns ``"837p"``, ``"835"``, or ``None`` if both parsers fail. + Used only when the caller didn't pin ``transaction_type`` explicitly + via the CLI. + """ + try: + text = path.read_text() + except OSError: + return None + # Cheap filename hint. + name = path.name.lower() + if name.endswith(".835") or "835" in name: + return _try_both(path, text, prefer="835") + if name.endswith(".837") or "837" in name or "837p" in name: + return _try_both(path, text, prefer="837p") + # Content: ISA + ST. 837 starts with "ST*837", 835 starts with "ST*835". + if "ST*837*" in text: + return "837p" + if "ST*835*" in text: + return "835" + # Fallback: try 837p parser, then 835. + return _try_both(path, text, prefer="837p") + + +def _try_both(path: Path, text: str, *, prefer: str) -> TransactionType: + """Try the preferred parser first, then the other; return first winner.""" + if prefer == "837p": + try: + parse_837(text, PayerConfig.co_medicaid(), input_file=str(path)) + return "837p" + except Exception: + try: + parse_835(text, PayerConfig835.co_medicaid_835(), input_file=str(path)) + return "835" + except Exception: + return None + try: + parse_835(text, PayerConfig835.co_medicaid_835(), input_file=str(path)) + return "835" + except Exception: + try: + parse_837(text, PayerConfig.co_medicaid(), input_file=str(path)) + return "837p" + except Exception: + return None + + +__all__ = [ + "BackfillSummary", + "backfill_rendering_provider_npi", +] diff --git a/backend/tests/test_backfill_cli.py b/backend/tests/test_backfill_cli.py new file mode 100644 index 0000000..82c29a9 --- /dev/null +++ b/backend/tests/test_backfill_cli.py @@ -0,0 +1,227 @@ +"""SP32 Task 6: end-to-end smoke tests for the ``backfill-rendering-npi`` CLI. + +The CLI re-parses on-disk 837p + 835 files and backfills the new typed +``rendering_provider_npi`` columns on existing Claim / Remittance rows +that were ingested before the T4 writers took effect. Idempotent: rows +that already have a non-NULL rendering NPI are left untouched. + +Test strategy mirrors ``test_cli_backup.py``: in-process Click CliRunner +invocation against a fresh per-test DB (autouse ``_auto_init_db`` +fixture), pre-seeded with Claim/Remittance rows whose +``rendering_provider_npi`` is NULL, then assert the CLI populated them. +""" + +from __future__ import annotations + +from datetime import datetime, timezone +from decimal import Decimal + +import pytest +from click.testing import CliRunner + +from cyclone import db +from cyclone.cli import main + +BACKEND_DIR = "backend" +_REPO_ROOT = "." + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture +def seed_837p_claim_without_renderer(): + """Insert a Claim row whose ``rendering_provider_npi`` is NULL. + + The id (``t991102984o1c1d``) matches the first claim in + ``co_medicaid_837p_with_renderer.txt`` — the fixture that does + contain a NM1*82 rendering provider (NPI ``1234567893``) attached + to that claim. Backfill should populate the column from the + re-parsed file. + """ + from cyclone.db import Batch, Claim, ClaimState + with db.SessionLocal()() as s: + s.add(Batch( + id="seed-837p-batch", + kind="837p", + input_filename="seed.x12", + parsed_at=datetime.now(timezone.utc), + )) + s.add(Claim( + id="t991102984o1c1d", + batch_id="seed-837p-batch", + patient_control_number="t991102984o1c1d", + charge_amount=Decimal("85.40"), + state=ClaimState.SUBMITTED, + rendering_provider_npi=None, # ← the column under test + provider_npi="1881068062", + )) + s.commit() + + +@pytest.fixture +def seed_835_remit_without_service_provider(tmp_path): + """Insert a Remittance whose ``rendering_provider_npi`` is NULL. + + Writes a minimal 835 fixture to ``tmp_path`` containing an + NM1*1P service provider (NPI ``2222222222``) attached to the + only CLP segment (claim control number ``CLM001``). Returns + the path to the written file. + """ + from cyclone.db import Batch, Remittance + + fixture = tmp_path / "minimal_835_with_service_provider.txt" + fixture.write_text( + "ISA*00* *00* *ZZ*RECEIVERID *ZZ*SENDERID " + "*250101*1200*^*00501*000000001*0*P*:~" + "GS*HP*RECEIVER*SENDER*20250101*1200*1*X*005010X221A1~" + "ST*835*0001~" + "BPR*I*85.40*C*NON*****01*021000021*DA*123456*1511111" + "**01*031302955*DA*9876543~" + "TRN*1*1511111*1511111~" + "DTM*405*20250101~" + "N1*PR*COLORADO MEDICAID*XV*CO MEDICAID~" + "N3*PO BOX 1100*~" + "N4*DENVER*CO*80203~" + "REF*2U*12345~" + "N1*PE*ACME CLINIC*XX*1111111111~" + "LX*1~" + "CLP*CLM001*1*85.40*85.40*0.00*MC*CLM001*11*1~" + "CAS*PR*1*0.00~" + "NM1*1P*2*RENDERING PROVIDER*****XX*2222222222~" + "SVC*HC:99213*85.40*85.40**1~" + "DTM*472*20250101~" + "SE*18*0001~" + "GE*1*1~" + "IEA*1*000000001~" + ) + + with db.SessionLocal()() as s: + s.add(Batch( + id="seed-835-batch", + kind="835", + input_filename=fixture.name, + parsed_at=datetime.now(timezone.utc), + )) + s.add(Remittance( + id="CLM001", + batch_id="seed-835-batch", + payer_claim_control_number="CLM001", + claim_id=None, + status_code="1", + status_label="Primary payer forward", + total_charge=Decimal("85.40"), + total_paid=Decimal("85.40"), + patient_responsibility=Decimal("0"), + adjustment_amount=Decimal("0"), + received_at=datetime.now(timezone.utc), + is_reversal=False, + rendering_provider_npi=None, # ← the column under test + )) + s.commit() + + return fixture + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _run(args: list[str]) -> object: + """Invoke the CLI in-process via CliRunner. + + CliRunner auto-captures stdout/stderr; ``catch_exceptions=False`` + so the test fails loudly on any unexpected raise instead of a + swallowed traceback. + """ + return CliRunner().invoke(main, args, catch_exceptions=False) + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + + +def test_backfill_populates_837p_rendering_npi(seed_837p_claim_without_renderer): + """Re-parse the 837p fixture → Claim.rendering_provider_npi populated.""" + fixture_path = "tests/fixtures/co_medicaid_837p_with_renderer.txt" + + # Sanity: column starts NULL. + with db.SessionLocal()() as s: + from cyclone.db import Claim + row = s.get(Claim, "t991102984o1c1d") + assert row is not None + assert row.rendering_provider_npi is None + + result = _run(["backfill-rendering-npi", "--file", fixture_path, "--type", "837p"]) + assert result.exit_code == 0, result.output + + with db.SessionLocal()() as s: + from cyclone.db import Claim + row = s.get(Claim, "t991102984o1c1d") + assert row is not None + # First claim in the fixture carries NM1*82 NPI ``1234567893``. + assert row.rendering_provider_npi == "1234567893" + # CLI prints the summary line with the populated counts. + assert "claims_updated=" in result.output + assert "remits_updated=" in result.output + + +def test_backfill_populates_835_rendering_npi(seed_835_remit_without_service_provider): + """Re-parse an 835 with NM1*1P → Remittance.rendering_provider_npi populated.""" + fixture_path = str(seed_835_remit_without_service_provider) + + # Sanity: column starts NULL. + with db.SessionLocal()() as s: + from cyclone.db import Remittance + row = s.get(Remittance, "CLM001") + assert row is not None + assert row.rendering_provider_npi is None + + result = _run(["backfill-rendering-npi", "--file", fixture_path, "--type", "835"]) + assert result.exit_code == 0, result.output + + with db.SessionLocal()() as s: + from cyclone.db import Remittance + row = s.get(Remittance, "CLM001") + assert row is not None + # Inline fixture's NM1*1P NM109 is ``2222222222``. + assert row.rendering_provider_npi == "2222222222" + assert "remits_updated=" in result.output + + +def test_backfill_is_idempotent(seed_837p_claim_without_renderer): + """Second run is a no-op — already-populated columns stay.""" + fixture_path = "tests/fixtures/co_medicaid_837p_with_renderer.txt" + + # First run: populates the column. + first = _run(["backfill-rendering-npi", "--file", fixture_path, "--type", "837p"]) + assert first.exit_code == 0, first.output + + # Second run: still exits 0; column is unchanged; nothing breaks. + second = _run(["backfill-rendering-npi", "--file", fixture_path, "--type", "837p"]) + assert second.exit_code == 0, second.output + + with db.SessionLocal()() as s: + from cyclone.db import Claim + row = s.get(Claim, "t991102984o1c1d") + assert row.rendering_provider_npi == "1234567893" + + +def test_backfill_help_lists_subcommand(): + """Regression guard: the subcommand is wired into ``main``.""" + result = _run(["--help"]) + assert result.exit_code == 0 + assert "backfill-rendering-npi" in result.output + + +def test_backfill_with_no_files_is_noop(): + """Running with no --file and no --input-dir is a clean no-op (exit 0).""" + result = _run(["backfill-rendering-npi"]) + assert result.exit_code == 0, result.output + # Summary line still printed (zero counts). + assert "claims_updated=0" in result.output + assert "remits_updated=0" in result.output