tyler/cyclone
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(sp32): add backfill-rendering-npi CLI subcommand

Browse Source
This commit is contained in:
Nora
2026-07-02 17:16:59 -06:00
parent de77f19d9d
commit 0f3b264e41
3 changed files with 658 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/cyclone/cli.py
+91
View File
@@ -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()
backend/src/cyclone/store/backfill.py
+340
View File
@@ -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",
]