31a53f89a6
- order-confirmation email presents both filing methods: paper CMS-855 (no account needed, client e-signs one page, we print+mail to their MAC) and I&A surrogacy (faster, needs CMS account). NPPES-only services note that surrogacy is required (web-only). - npi_provider handlers record the access model per service in admin todos. - marketing copy leads with the lowest-friction paper option.
203 lines
7.5 KiB
Python
203 lines
7.5 KiB
Python
"""Healthcare / NPI provider compliance handlers.
|
|
|
|
Review-staged: each NPI service generates an admin todo with the provider's
|
|
NPI + intake details for a human to file in CMS PECOS / NPPES. This mirrors the
|
|
FCC auto-filing-off safety default — no automated submission to government
|
|
portals until the Playwright flows are proven.
|
|
|
|
Covers slugs:
|
|
npi-revalidation Medicare PECOS revalidation (5-yr cycle)
|
|
npi-reactivation reactivate a deactivated NPI
|
|
nppes-update NPPES data update / attestation
|
|
medicare-enrollment new Medicare enrollment via PECOS
|
|
oig-sam-screening OIG LEIE + SAM exclusion screening (annual)
|
|
provider-compliance-bundle revalidation watch + screening + NPPES upkeep
|
|
|
|
Intake data needed (collected by the npi-intake wizard step):
|
|
- npi provider's 10-digit NPI
|
|
- provider_name legal/provider name
|
|
- email contact email
|
|
- pecos_enrollment_id (optional) PECOS enrollment ID
|
|
- specialty (optional) taxonomy/specialty
|
|
- practice_state (optional) practice location state
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
|
|
LOG = logging.getLogger("workers.services.npi_provider")
|
|
|
|
# Per-slug admin todo metadata: human-readable action + the portal a human uses.
|
|
_SLUG_META = {
|
|
"npi-revalidation": {
|
|
"name": "Medicare PECOS Revalidation Filing",
|
|
"portal": "https://pecos.cms.hhs.gov/",
|
|
"action": (
|
|
"File the Medicare revalidation in PECOS for this provider. Confirm "
|
|
"the revalidation due date on the CMS revalidation list, update the "
|
|
"enrollment record, and submit. Capture the PECOS tracking ID."
|
|
),
|
|
"access": (
|
|
"PECOS via CMS I&A surrogacy (preferred). Fallback: paper CMS-855I/B/R, provider wet-signs cert page, mail to provider's MAC."
|
|
),
|
|
"priority": "high",
|
|
},
|
|
"npi-reactivation": {
|
|
"name": "NPI Reactivation",
|
|
"portal": "https://nppes.cms.hhs.gov/",
|
|
"action": (
|
|
"Reactivate the deactivated NPI in NPPES. Verify the deactivation "
|
|
"reason, correct any stale data, and re-certify the record."
|
|
),
|
|
"access": (
|
|
"NPPES via CMS I&A surrogacy. No paper option (NPPES is web-only)."
|
|
),
|
|
"priority": "high",
|
|
},
|
|
"nppes-update": {
|
|
"name": "NPPES Data Update / Attestation",
|
|
"portal": "https://nppes.cms.hhs.gov/",
|
|
"action": (
|
|
"Update + re-attest the provider's NPPES record (CMS requires "
|
|
"updates within 30 days of any change). Apply the requested field "
|
|
"changes and certify."
|
|
),
|
|
"access": (
|
|
"NPPES via CMS I&A surrogacy. No paper option (NPPES is web-only)."
|
|
),
|
|
"priority": "normal",
|
|
},
|
|
"medicare-enrollment": {
|
|
"name": "Medicare Enrollment (PECOS)",
|
|
"portal": "https://pecos.cms.hhs.gov/",
|
|
"action": (
|
|
"Complete the provider's Medicare enrollment in PECOS (CMS-855). "
|
|
"Confirm taxonomy, practice location, and authorized official."
|
|
),
|
|
"access": (
|
|
"PECOS via CMS I&A surrogacy (preferred). Fallback: paper CMS-855, provider wet-signs, mail to MAC."
|
|
),
|
|
"priority": "high",
|
|
},
|
|
"oig-sam-screening": {
|
|
"name": "OIG/SAM Exclusion Screening (Annual)",
|
|
"portal": "https://oig.hhs.gov/exclusions/ + https://sam.gov/",
|
|
"action": (
|
|
"Run the provider (and any listed staff) against the OIG LEIE and "
|
|
"SAM exclusion lists. Produce the screening certificate and flag any "
|
|
"matches for escalation."
|
|
),
|
|
"access": (
|
|
"No client access needed - OIG LEIE + SAM.gov are public. Screen by NPI/name, issue certificate."
|
|
),
|
|
"priority": "normal",
|
|
},
|
|
"provider-compliance-bundle": {
|
|
"name": "Provider Compliance Bundle (Annual)",
|
|
"portal": "https://pecos.cms.hhs.gov/ + https://nppes.cms.hhs.gov/",
|
|
"action": (
|
|
"Onboard the provider into the annual compliance bundle: enroll in "
|
|
"revalidation watch, run OIG/SAM screening, and refresh the NPPES "
|
|
"record. Set the next revalidation reminder."
|
|
),
|
|
"access": (
|
|
"PECOS/NPPES via CMS I&A surrogacy; screening is public. Paper CMS-855 fallback for the enrollment/revalidation piece."
|
|
),
|
|
"priority": "high",
|
|
},
|
|
}
|
|
|
|
|
|
class _BaseNPIHandler:
|
|
"""Shared review-staged behaviour for all NPI services."""
|
|
|
|
SERVICE_SLUG = ""
|
|
|
|
async def process(self, order_data: dict) -> list[str]:
|
|
order_number = order_data.get("order_number", order_data.get("name", ""))
|
|
return await self.handle(order_data, order_number)
|
|
|
|
async def handle(self, order_data: dict, order_number: str) -> list[str]:
|
|
meta = _SLUG_META[self.SERVICE_SLUG]
|
|
LOG.info("[%s] Processing %s", order_number, self.SERVICE_SLUG)
|
|
|
|
intake = order_data.get("intake_data") or {}
|
|
if isinstance(intake, str):
|
|
intake = json.loads(intake)
|
|
|
|
npi = intake.get("npi", "N/A")
|
|
provider = intake.get("provider_name", order_data.get("customer_name", "Unknown"))
|
|
specialty = intake.get("specialty", "")
|
|
practice_state = intake.get("practice_state", "")
|
|
pecos_id = intake.get("pecos_enrollment_id", "")
|
|
|
|
description = (
|
|
f"{meta['action']}\n\n"
|
|
f"Provider: {provider}\n"
|
|
f"NPI: {npi}\n"
|
|
f"PECOS Enrollment ID: {pecos_id or 'not provided'}\n"
|
|
f"Specialty: {specialty or 'not provided'}\n"
|
|
f"Practice state: {practice_state or 'not provided'}\n"
|
|
f"Portal: {meta['portal']}\n"
|
|
f"Access model: {meta['access']}\n\n"
|
|
f"Review-staged: file manually, then mark this order complete."
|
|
)
|
|
|
|
self._create_todo(
|
|
order_number,
|
|
intake,
|
|
title=f"{meta['name']} — {provider} (NPI {npi})",
|
|
description=description,
|
|
priority=meta["priority"],
|
|
)
|
|
return []
|
|
|
|
def _create_todo(self, order_number, intake, title, description, priority="normal"):
|
|
try:
|
|
import psycopg2
|
|
|
|
conn = psycopg2.connect(os.environ.get("DATABASE_URL", ""))
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO admin_todos (
|
|
title, category, priority, order_number, service_slug,
|
|
description, data, status
|
|
) VALUES (%s, %s, %s, %s, %s, %s, %s, 'pending')
|
|
""",
|
|
(
|
|
title, "filing", priority, order_number,
|
|
self.SERVICE_SLUG, description, json.dumps(intake),
|
|
),
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
except Exception as exc:
|
|
LOG.error("[%s] Failed to create NPI todo: %s", order_number, exc)
|
|
|
|
|
|
class NPIRevalidationHandler(_BaseNPIHandler):
|
|
SERVICE_SLUG = "npi-revalidation"
|
|
|
|
|
|
class NPIReactivationHandler(_BaseNPIHandler):
|
|
SERVICE_SLUG = "npi-reactivation"
|
|
|
|
|
|
class NPPESUpdateHandler(_BaseNPIHandler):
|
|
SERVICE_SLUG = "nppes-update"
|
|
|
|
|
|
class MedicareEnrollmentHandler(_BaseNPIHandler):
|
|
SERVICE_SLUG = "medicare-enrollment"
|
|
|
|
|
|
class OIGSAMScreeningHandler(_BaseNPIHandler):
|
|
SERVICE_SLUG = "oig-sam-screening"
|
|
|
|
|
|
class ProviderComplianceBundleHandler(_BaseNPIHandler):
|
|
SERVICE_SLUG = "provider-compliance-bundle"
|