feat(sp20): NPI Luhn checksum + Tax ID (EIN) format validation
Adds pure local validators for the 10-digit NPI Luhn checksum (CMS- published algorithm with the '80840' NPPES prefix) and 9-digit EIN format (rejects reserved prefixes 00/07/80-89). No NPPES round-trip, no IRS e-file lookup — catches the 99% typo case at parse time. Surface: - cyclone.npi.is_valid_npi / is_valid_tax_id / normalize_tax_id - CLI: 'cyclone validate-npi <npi>' and 'cyclone validate-tax-id <ein>' - API: GET /api/admin/validate-provider?npi=&tax_id= - Parser validator: new R021_npi_checksum rule (warning, not error, to keep test fixtures with placeholder NPIs ingestible) - minimal_837p.txt fixture NPI updated from '1234567890' to the Luhn-valid '1993999998' so strict-mode CLI parses still pass Tests: - test_npi.py — 27 cases (Luhn math, valid/invalid NPIs, EIN cases, normalize_tax_id edge cases) - test_api_validate_provider.py — 4 cases (both valid, both invalid, omitted NPI, omitted tax_id) - test_cli_validate.py — 8 cases (valid/invalid for both subcommands, exit codes, malformed inputs) - test_validator.py — 4 new R021 cases (valid Luhn silent, bad Luhn warning, skipped when format bad, skipped when NPI missing) Total: 923 tests pass.
This commit is contained in:
Tyler
@@ -248,11 +248,12 @@ app.add_middleware(SecurityHeadersMiddleware)
|
||||
# and are wired in here. (Kept as a top-level package rather than nested
|
||||
# under `cyclone.api` so the existing ``cyclone.api`` module path keeps
|
||||
# working — Python prefers packages over same-named modules.)
|
||||
from cyclone.api_routers import acks, health, ta1_acks # noqa: E402
|
||||
from cyclone.api_routers import acks, admin, health, ta1_acks # noqa: E402
|
||||
|
||||
app.include_router(health.router)
|
||||
app.include_router(acks.router)
|
||||
app.include_router(ta1_acks.router)
|
||||
app.include_router(admin.router)
|
||||
|
||||
|
||||
def _resolve_payer(name: str) -> PayerConfig:
|
||||
|
||||
@@ -0,0 +1,60 @@
|
||||
"""``/api/admin/validate-provider`` — NPI + Tax ID liveness probe (SP20).
|
||||
|
||||
Pure read-only endpoint that runs the local NPI Luhn + EIN format checks
|
||||
without touching the DB. Useful for:
|
||||
- operators vetting a new provider before adding them to the registry,
|
||||
- the dashboard's "validate" button on a Provider row,
|
||||
- smoke-testing the SP20 checks after a deploy.
|
||||
|
||||
Both query params are optional; omitting one just skips that check.
|
||||
Returns the per-check result dict so the caller can distinguish "bad
|
||||
format" from "bad checksum".
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from fastapi import APIRouter, Query
|
||||
|
||||
from cyclone.npi import is_valid_npi, is_valid_tax_id, normalize_tax_id
|
||||
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
@router.get("/api/admin/validate-provider")
|
||||
def validate_provider(
|
||||
npi: str | None = Query(None, description="10-digit NPI to validate (Luhn checksum)"),
|
||||
tax_id: str | None = Query(None, description="9-digit EIN to validate (format + reserved-prefix check)"),
|
||||
) -> dict:
|
||||
"""Return per-field validation results for ``npi`` and ``tax_id``.
|
||||
|
||||
Each field's payload is the same shape:
|
||||
|
||||
* ``valid`` — bool, the operator's "yes/no" answer
|
||||
* ``normalized`` — for ``tax_id``: the 9-digit plain form, or null
|
||||
if the input is unparseable
|
||||
|
||||
An empty/unset query param returns ``{"valid": None, "skipped": true}``
|
||||
so the caller can render "no check performed" rather than treating
|
||||
``None`` as a hard fail.
|
||||
"""
|
||||
result: dict = {}
|
||||
|
||||
if npi is None or npi == "":
|
||||
result["npi"] = {"valid": None, "skipped": True}
|
||||
else:
|
||||
result["npi"] = {
|
||||
"valid": is_valid_npi(npi),
|
||||
"skipped": False,
|
||||
}
|
||||
|
||||
if tax_id is None or tax_id == "":
|
||||
result["tax_id"] = {"valid": None, "skipped": True, "normalized": None}
|
||||
else:
|
||||
normalized = normalize_tax_id(tax_id)
|
||||
result["tax_id"] = {
|
||||
"valid": is_valid_tax_id(tax_id),
|
||||
"skipped": False,
|
||||
"normalized": normalized,
|
||||
}
|
||||
|
||||
return result
|
||||
@@ -228,6 +228,63 @@ def _count_issues(report) -> dict[str, int]:
|
||||
return counts
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# SP20: `cyclone validate-npi` + `cyclone validate-tax-id`
|
||||
#
|
||||
# Pure local validators. No DB, no Keychain, no network — operators can
|
||||
# run them on a developer laptop without standing up the full Cyclone
|
||||
# stack. Exit code is 0 (valid) or 1 (invalid) so they compose with
|
||||
# shell scripting / CI gates.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@main.command("validate-npi")
|
||||
@click.argument("npi")
|
||||
@click.option("--log-level", default="WARNING", show_default=True, type=click.Choice(["DEBUG", "INFO", "WARNING", "ERROR"]))
|
||||
def validate_npi_cmd(npi: str, log_level: str) -> None:
|
||||
"""Validate a 10-digit NPI's Luhn checksum locally (SP20).
|
||||
|
||||
Exit 0 if valid, 1 if not. No logging of the value itself — NPIs
|
||||
are PHI under HIPAA, so the operator's CLI history is the only
|
||||
audit trail.
|
||||
"""
|
||||
# SP18: re-run so --log-level overrides the group default.
|
||||
setup_logging(level=log_level)
|
||||
from cyclone.npi import is_valid_npi
|
||||
if is_valid_npi(npi):
|
||||
click.echo(f"OK: {len(npi)}-digit NPI passes Luhn checksum")
|
||||
return
|
||||
click.echo(f"INVALID: {npi!r} fails NPI Luhn checksum", err=True)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
@main.command("validate-tax-id")
|
||||
@click.argument("tax_id")
|
||||
@click.option("--log-level", default="WARNING", show_default=True, type=click.Choice(["DEBUG", "INFO", "WARNING", "ERROR"]))
|
||||
def validate_tax_id_cmd(tax_id: str, log_level: str) -> None:
|
||||
"""Validate a 9-digit EIN's format + prefix locally (SP20).
|
||||
|
||||
Accepts both ``XX-XXXXXXX`` and ``XXXXXXXXX``. Exit 0 if valid,
|
||||
1 if not. EIN is sensitive (PII), so we don't echo the value back
|
||||
on failure — only the validation verdict.
|
||||
"""
|
||||
# SP18: re-run so --log-level overrides the group default.
|
||||
setup_logging(level=log_level)
|
||||
from cyclone.npi import is_valid_tax_id, normalize_tax_id
|
||||
plain = normalize_tax_id(tax_id)
|
||||
if plain is None:
|
||||
click.echo("INVALID: input is not a 9-digit EIN (XX-XXXXXXX or XXXXXXXXX)", err=True)
|
||||
sys.exit(1)
|
||||
if is_valid_tax_id(tax_id):
|
||||
click.echo(f"OK: 9-digit EIN (normalized={plain})")
|
||||
return
|
||||
click.echo(
|
||||
f"INVALID: 9-digit EIN has reserved prefix ({plain[:2]}); EIN is not assignable by IRS",
|
||||
err=True,
|
||||
)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
|
||||
@@ -0,0 +1,159 @@
|
||||
"""SP20 — NPI checksum + Tax ID format validation.
|
||||
|
||||
The National Provider Identifier (NPI) is a 10-digit number where
|
||||
the last digit is a **Luhn checksum** over the 9 preceding digits
|
||||
prefixed with the constant ``80840`` (the NPPES "healthcare
|
||||
provider identifier" prefix). See CMS / HHS NPI Standard:
|
||||
|
||||
https://www.cms.gov/medicare/health-care-provider-identifier
|
||||
|
||||
The Tax ID (EIN) is a 9-digit number, optionally formatted with a
|
||||
hyphen after the second digit (``XX-XXXXXXX``). We don't validate
|
||||
against the IRS (that needs their e-file schema), but we *do* catch
|
||||
the 99% typo case at parse time.
|
||||
|
||||
Everything in this module is local — no NPPES, no network. Operators
|
||||
who want real NPPES verification can wire it in later; this module
|
||||
catches typos (an off-by-one in a 10-digit NPI, a letter in an EIN,
|
||||
an extra digit, the all-zeros EIN prefix ``00`` / ``07``).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
|
||||
# NPPES prefix per the NPI Luhn algorithm. Prepended to the 9-digit
|
||||
# NPI body before running the Luhn check.
|
||||
_NPPES_PREFIX = "80840"
|
||||
|
||||
|
||||
def npi_checksum(npi_body: str) -> int:
|
||||
"""Compute the Luhn check digit for a 9-digit NPI body.
|
||||
|
||||
``npi_body`` must be exactly 9 digits; the caller is responsible
|
||||
for length + character validation. Returns the check digit (0–9).
|
||||
"""
|
||||
if not npi_body.isdigit() or len(npi_body) != 9:
|
||||
raise ValueError(f"npi_body must be 9 digits, got {npi_body!r}")
|
||||
digits = _NPPES_PREFIX + npi_body
|
||||
return _luhn_check_digit(digits)
|
||||
|
||||
|
||||
def is_valid_npi(npi: str | None) -> bool:
|
||||
"""True if ``npi`` is a well-formed 10-digit NPI with valid checksum.
|
||||
|
||||
Returns False for ``None`` / empty string / non-strings / wrong
|
||||
length / non-digit characters / wrong Luhn check digit. Doesn't
|
||||
call NPPES — see module docstring for why.
|
||||
|
||||
>>> is_valid_npi("1234567893") # CMS-published example NPI
|
||||
True
|
||||
>>> is_valid_npi("1234567894") # last digit off by one
|
||||
False
|
||||
>>> is_valid_npi("1234567890") # passes digit but fails Luhn
|
||||
False
|
||||
>>> is_valid_npi("")
|
||||
False
|
||||
"""
|
||||
if not isinstance(npi, str):
|
||||
return False
|
||||
if len(npi) != 10 or not npi.isdigit():
|
||||
return False
|
||||
return npi[-1] == str(npi_checksum(npi[:-1]))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tax ID (EIN)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
# EIN prefix table (subset). The IRS publishes a full table; the
|
||||
# common "this is obviously a typo" prefixes we reject are:
|
||||
# 00 — reserved / never assigned
|
||||
# 07 — campus prefixes reserved for future use
|
||||
# 8X — formerly used by the IRS Pension Plan Branch
|
||||
# Other 00-prefixed EINs (e.g., 000000000) are technically not
|
||||
# assigned but we don't reject them here — the operator might have
|
||||
# a deliberate placeholder.
|
||||
_EIN_FORBIDDEN_PREFIXES = {"00", "07"}
|
||||
_EIN_RESERVED_PREFIX_8X = re.compile(r"^8\d$")
|
||||
|
||||
# 9 digits, optionally formatted as XX-XXXXXXX.
|
||||
_EIN_FORMATTED = re.compile(r"^\d{2}-\d{7}$")
|
||||
_EIN_PLAIN = re.compile(r"^\d{9}$")
|
||||
|
||||
|
||||
def normalize_tax_id(tax_id: str | None) -> str | None:
|
||||
"""Return ``tax_id`` in 9-digit plain form, or None if it's malformed.
|
||||
|
||||
>>> normalize_tax_id("72-1587149")
|
||||
'721587149'
|
||||
>>> normalize_tax_id("721587149")
|
||||
'721587149'
|
||||
>>> normalize_tax_id("not-an-ein")
|
||||
None
|
||||
"""
|
||||
if not isinstance(tax_id, str):
|
||||
return None
|
||||
s = tax_id.strip()
|
||||
if _EIN_FORMATTED.match(s):
|
||||
return s.replace("-", "")
|
||||
if _EIN_PLAIN.match(s):
|
||||
return s
|
||||
return None
|
||||
|
||||
|
||||
def is_valid_tax_id(tax_id: str | None) -> bool:
|
||||
"""True if ``tax_id`` is a 9-digit EIN (formatted or plain) with
|
||||
a non-reserved prefix.
|
||||
|
||||
>>> is_valid_tax_id("72-1587149") # Touch of Care
|
||||
True
|
||||
>>> is_valid_tax_id("00-1234567") # reserved prefix
|
||||
False
|
||||
>>> is_valid_tax_id("07-1234567") # reserved prefix
|
||||
False
|
||||
>>> is_valid_tax_id("not-an-ein")
|
||||
False
|
||||
"""
|
||||
plain = normalize_tax_id(tax_id)
|
||||
if plain is None:
|
||||
return False
|
||||
prefix = plain[:2]
|
||||
if prefix in _EIN_FORBIDDEN_PREFIXES:
|
||||
return False
|
||||
if _EIN_RESERVED_PREFIX_8X.match(prefix):
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Luhn internals
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _luhn_check_digit(digits: str) -> int:
|
||||
"""Return the Luhn check digit for ``digits``.
|
||||
|
||||
The Luhn algorithm doubles every second digit starting from the
|
||||
RIGHTMOST position (i.e., the first digit doubled is the rightmost
|
||||
character of ``digits``). If the doubled value exceeds 9, subtract
|
||||
9. Sum all digits; the check digit is ``(10 - sum % 10) % 10``.
|
||||
|
||||
``digits`` here is the body WITHOUT the check digit — for the NPI
|
||||
case it's the 14-character ``80840`` + 9-digit NPI body. The
|
||||
CMS-published example ``123456789`` (body) yields check digit
|
||||
``3`` → full NPI ``1234567893`` (verified against
|
||||
https://www.cms.gov/.../NPIcheckdigit.pdf).
|
||||
"""
|
||||
total = 0
|
||||
# The rightmost digit of ``digits`` is the FIRST one doubled (i=0
|
||||
# in the reversed iteration). Per CMS, doubling starts at the
|
||||
# rightmost and alternates leftward.
|
||||
for i, ch in enumerate(reversed(digits)):
|
||||
d = int(ch)
|
||||
if i % 2 == 0: # rightmost, third-from-right, fifth-from-right, ...
|
||||
d *= 2
|
||||
if d > 9:
|
||||
d -= 9
|
||||
total += d
|
||||
return (10 - total % 10) % 10
|
||||
@@ -35,6 +35,36 @@ def _r020_npi_format(claim: ClaimOutput, _: PayerConfig) -> Iterable[ValidationI
|
||||
yield ValidationIssue(rule="R020_npi_format", severity="error", message=f"Billing provider NPI must be 10 digits, got {claim.billing_provider.npi!r}")
|
||||
|
||||
|
||||
def _r021_npi_checksum(claim: ClaimOutput, _: PayerConfig) -> Iterable[ValidationIssue]:
|
||||
"""SP20: validate the billing-provider NPI's Luhn check digit.
|
||||
|
||||
A 10-digit NPI whose body passes R020's format check can still have
|
||||
a bad Luhn check digit (a typo at the end). Yielded as a WARNING —
|
||||
not an error — because operators sometimes ingest test fixtures with
|
||||
placeholder NPIs (e.g. all-same-digit) and we don't want to block
|
||||
that path. Local-only check, no NPPES round-trip.
|
||||
"""
|
||||
npi = claim.billing_provider.npi
|
||||
if not npi:
|
||||
return
|
||||
# Skip silently if R020 already flagged the format — we don't want to
|
||||
# duplicate the operator's screen with a second issue about the same NPI.
|
||||
if not NPI_RE.match(npi):
|
||||
return
|
||||
# Lazy import keeps the validator module importable even if
|
||||
# ``cyclone.npi`` is unavailable (e.g. in some legacy test setups).
|
||||
try:
|
||||
from cyclone.npi import is_valid_npi
|
||||
except ImportError: # pragma: no cover — defensive
|
||||
return
|
||||
if not is_valid_npi(npi):
|
||||
yield ValidationIssue(
|
||||
rule="R021_npi_checksum",
|
||||
severity="warning",
|
||||
message=f"Billing provider NPI {npi!r} fails Luhn checksum (likely typo)",
|
||||
)
|
||||
|
||||
|
||||
def _r030_frequency_allowed(claim: ClaimOutput, cfg: PayerConfig) -> Iterable[ValidationIssue]:
|
||||
if not claim.claim.frequency_code:
|
||||
return
|
||||
@@ -392,6 +422,7 @@ _RULES: list[Rule] = [
|
||||
_r010_clm01_present,
|
||||
_r011_total_charge_positive,
|
||||
_r020_npi_format,
|
||||
_r021_npi_checksum,
|
||||
_r030_frequency_allowed,
|
||||
_r031_ref_g1_optional,
|
||||
_r034_ref_g1_required,
|
||||
|
||||
Reference in New Issue
Block a user