Add CMS-855 PDF filler + e-sign fulfillment for Medicare revalidation/enrollment

- cms855_pdf_filler.py: fills official CMS-855I/B/O/A AcroForms from intake
  (name, NPI, DOB, cert-page printed name) and records the signature anchor at
  the form's official /Sig box so the e-sign stamper lands on the cert line.
- npi_provider handlers (revalidation/reactivation/enrollment) now generate the
  paper CMS-855, upload it to MinIO, request_esign with anchors, and email the
  signing link. Human completes/verifies + USPS Priority Mails to the MAC.
- scripts/Dockerfile: copy the official CMS-855I/B/O/A forms into the image.
This commit is contained in:
justin 2026-06-05 02:27:11 -05:00
parent 31a53f89a6
commit e212f20a34
7 changed files with 359 additions and 2 deletions

View file

@ -110,6 +110,16 @@ _SLUG_META = {
},
}
# Slugs whose fulfilment includes a paper CMS-855 (auto-filled official form,
# e-signed, then printed + USPS Priority-mailed to the provider's MAC). The
# bundle's revalidation piece is handled by the dedicated revalidation order it
# spawns, so it is not listed here.
_PAPER_855_SLUGS = {
"npi-revalidation",
"npi-reactivation",
"medicare-enrollment",
}
class _BaseNPIHandler:
"""Shared review-staged behaviour for all NPI services."""
@ -133,6 +143,24 @@ class _BaseNPIHandler:
specialty = intake.get("specialty", "")
practice_state = intake.get("practice_state", "")
pecos_id = intake.get("pecos_enrollment_id", "")
customer_email = (
intake.get("email")
or order_data.get("customer_email")
or order_data.get("email")
or ""
)
# For PECOS enrollment/revalidation we generate the official CMS-855,
# send it for e-signature, then a human prints + USPS-mails it to the MAC.
paper_note = ""
if self.SERVICE_SLUG in _PAPER_855_SLUGS:
try:
paper_note = self._generate_855_for_signing(
order_number, intake, provider, customer_email
)
except Exception as exc: # never block the admin todo on PDF issues
LOG.error("[%s] CMS-855 generation failed: %s", order_number, exc)
paper_note = f"CMS-855 auto-generation FAILED ({exc}); prepare the form manually."
description = (
f"{meta['action']}\n\n"
@ -142,8 +170,11 @@ class _BaseNPIHandler:
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."
f"Access model: {meta['access']}\n"
+ (f"\n{paper_note}\n" if paper_note else "")
+ "\nReview-staged: complete/verify the form, get it signed, then "
"print and USPS Priority Mail it to the provider's MAC (or file in "
"PECOS if surrogate access was granted). Mark this order complete."
)
self._create_todo(
@ -155,6 +186,112 @@ class _BaseNPIHandler:
)
return []
def _generate_855_for_signing(self, order_number, intake, provider, customer_email) -> str:
"""Generate the official CMS-855, upload it, and request an e-signature.
Returns a human-readable note for the admin todo describing what was
generated and what still needs manual completion. The signed PDF is
printed and USPS Priority-mailed to the MAC by the fulfilment team.
"""
try:
from scripts.document_gen.templates.cms855_pdf_filler import (
determine_form_type, fill_cms855,
)
except ImportError:
from document_gen.templates.cms855_pdf_filler import ( # type: ignore
determine_form_type, fill_cms855,
)
form_type = determine_form_type(self.SERVICE_SLUG, intake)
pdf_bytes, anchors, missing = fill_cms855(form_type, intake, order_number)
# Upload the unsigned, partially-filled official form to MinIO.
document_key = f"compliance/{order_number}/cms{form_type}_unsigned.pdf"
try:
import tempfile
try:
from scripts.document_gen.minio_client import MinioStorage
except ImportError:
from document_gen.minio_client import MinioStorage # type: ignore
storage = MinioStorage()
with tempfile.NamedTemporaryFile(suffix=".pdf", delete=True) as tf:
tf.write(pdf_bytes)
tf.flush()
storage.upload(tf.name, document_key, content_type="application/pdf")
except Exception as exc:
LOG.error("[%s] CMS-855 upload failed: %s", order_number, exc)
return f"CMS-{form_type.upper()} generated but upload FAILED ({exc})."
# Create the esign record (via the shared helper, which also emails the
# signing link) and attach the official signature anchor so the stamper
# lands the signature on the certification line.
signed = self._create_855_esign_record(
order_number, intake, provider, customer_email,
form_type, document_key, anchors,
)
note_lines = [
f"PAPER CMS-{form_type.upper()} generated (official form, auto-filled where possible).",
f"Unsigned PDF: {document_key}",
]
if signed and customer_email:
note_lines.append(f"E-sign link emailed to {customer_email}. After signing, print + USPS Priority Mail to the MAC.")
else:
note_lines.append("No customer email or esign infra — send the form for wet signature manually.")
if missing:
note_lines.append("MANUAL COMPLETION NEEDED:")
note_lines.extend(f" - {m}" for m in missing)
return "\n".join(note_lines)
def _create_855_esign_record(self, order_number, intake, provider, customer_email,
form_type, document_key, anchors) -> bool:
"""Create the esign record + email the signing link via the shared helper,
then attach the official signature anchors. Returns True on success.
"""
if not customer_email:
return False
document_type = f"cms{form_type}"
document_title = f"Medicare Enrollment Form (CMS-{form_type.upper()})"
try:
try:
from scripts.workers.services.telecom.esign_helper import request_esign
except ImportError:
from .telecom.esign_helper import request_esign # type: ignore
import psycopg2
conn = psycopg2.connect(os.environ.get("DATABASE_URL", ""))
try:
esign_id = request_esign(
conn=conn,
order_number=order_number,
document_type=document_type,
document_title=document_title,
entity_name=provider,
customer_email=customer_email,
customer_name=provider,
document_minio_key=document_key,
requires_perjury=True,
metadata={"service_slug": self.SERVICE_SLUG, "npi": intake.get("npi", ""), "form_type": form_type},
expires_hours=21 * 24,
)
# request_esign does not persist signature anchors; attach them so
# the stamper places the signature on the certification line.
if esign_id and anchors:
with conn.cursor() as cur:
cur.execute(
"UPDATE esign_records SET signature_anchors = %s, updated_at = NOW() WHERE id = %s",
(json.dumps(anchors), esign_id),
)
conn.commit()
return esign_id is not None
finally:
conn.close()
except Exception as exc:
LOG.error("[%s] CMS-855 esign request failed: %s", order_number, exc)
return False
def _create_todo(self, order_number, intake, title, description, priority="normal"):
try:
import psycopg2