diff --git a/docs/CMS-855A Form.pdf b/docs/CMS-855A Form.pdf new file mode 100644 index 0000000..9c00299 Binary files /dev/null and b/docs/CMS-855A Form.pdf differ diff --git a/docs/CMS-855B Form.pdf b/docs/CMS-855B Form.pdf new file mode 100644 index 0000000..72de807 Binary files /dev/null and b/docs/CMS-855B Form.pdf differ diff --git a/docs/CMS-855I Form.pdf b/docs/CMS-855I Form.pdf new file mode 100644 index 0000000..dd461b4 Binary files /dev/null and b/docs/CMS-855I Form.pdf differ diff --git a/docs/CMS-855O Form.pdf b/docs/CMS-855O Form.pdf new file mode 100644 index 0000000..0877263 Binary files /dev/null and b/docs/CMS-855O Form.pdf differ diff --git a/scripts/Dockerfile b/scripts/Dockerfile index 17319ad..a6cf1a4 100644 --- a/scripts/Dockerfile +++ b/scripts/Dockerfile @@ -23,6 +23,10 @@ COPY docs/product-facts.md /app/docs/product-facts.md COPY ["docs/MCS-150 Form.pdf", "/app/docs/MCS-150 Form.pdf"] COPY ["docs/MCS-150B Form.pdf", "/app/docs/MCS-150B Form.pdf"] COPY ["docs/MCS-150C Form.pdf", "/app/docs/MCS-150C Form.pdf"] +COPY ["docs/CMS-855I Form.pdf", "/app/docs/CMS-855I Form.pdf"] +COPY ["docs/CMS-855B Form.pdf", "/app/docs/CMS-855B Form.pdf"] +COPY ["docs/CMS-855O Form.pdf", "/app/docs/CMS-855O Form.pdf"] +COPY ["docs/CMS-855A Form.pdf", "/app/docs/CMS-855A Form.pdf"] # Create data directories RUN mkdir -p /app/data/screenshots /app/data/documents /app/data/logs diff --git a/scripts/document_gen/templates/cms855_pdf_filler.py b/scripts/document_gen/templates/cms855_pdf_filler.py new file mode 100644 index 0000000..2cd8e9c --- /dev/null +++ b/scripts/document_gen/templates/cms855_pdf_filler.py @@ -0,0 +1,216 @@ +"""Fill an official CMS-855 enrollment PDF for Medicare revalidation/enrollment. + +We fill the *official* CMS form (downloaded into ``docs/CMS-855X Form.pdf``) +using its AcroForm fields, mirroring ``mcs150_pdf_filler.py``. The handful of +fields we can reliably populate from intake are mapped here; the rest are left +for a human to complete during review before the form is e-signed and mailed. + +After filling we record a signature anchor at the form's official signature box +(the ``/Sig`` annotation on the certification page) so the e-sign stamper lands +the provider's signature exactly on the certification line. The signed PDF is +then printed and mailed USPS Priority Mail to the provider's MAC. + +Supported forms (by slug -> form type): + npi-revalidation / medicare-enrollment (individual) -> 855I + medicare-enrollment (group/supplier) -> 855B + medicare-enrollment (ordering/referring only) -> 855O + +Usage: + from scripts.document_gen.templates.cms855_pdf_filler import fill_cms855 + pdf_bytes, anchors, missing = fill_cms855("855i", intake, order_number) +""" +from __future__ import annotations + +import io +import logging +from pathlib import Path + +LOG = logging.getLogger("cms855_pdf_filler") + +# docs/ lives two levels up from scripts/document_gen/templates/ +DOCS_DIR = Path(__file__).resolve().parents[3] / "docs" + +FORM_PATHS = { + "855i": DOCS_DIR / "CMS-855I Form.pdf", + "855b": DOCS_DIR / "CMS-855B Form.pdf", + "855o": DOCS_DIR / "CMS-855O Form.pdf", + "855a": DOCS_DIR / "CMS-855A Form.pdf", +} + +# Map intake keys -> AcroForm field names, per form. Field names are taken from +# the official PDF's AcroForm (verified against each form's /TU tooltips). Only +# the reliably-derivable identity fields are mapped; a human completes the rest. +FIELD_MAPS = { + # 855I — individual practitioner (most common for revalidation) + "855i": { + # Section 2A: individual personal identifying info + "first_name": "5-1", + "middle_init": "5-2", + "last_name": "5-3", + "npi": "5-13", + "dob": "5-11", + # Section 15: certification — printed signer name + "sign_first": "24-1", + "sign_middle": "24-2", + "sign_last": "24-3", + }, + # 855B — clinic/group practice + "855b": { + # Org NPI appears in section 4 area; org name fields vary, so we map the + # ones we can verify and leave the rest for human completion. + "org_npi": "11-4", + }, + # 855O — ordering/referring (no billing). Minimal identity fields. + "855o": {}, + "855a": {}, +} + +# Official signature annotation(s) on the certification page, by form. Recorded +# from the PDF's /Sig field rects (page index is 0-based). These let the e-sign +# stamper place the provider's signature on the exact certification line. +# rect = [llx, lly, urx, ury] in points; page_w/page_h = mediabox dims. +SIGNATURE_FIELDS = { + "855i": [ + {"field": "signer", "page": 24, "rect": [44.9, 522.8, 370.7, 541.2], "page_w": 612.0, "page_h": 792.0}, + ], + # 855b/855o/855a signature rects are populated when those flows go live; the + # filler still works without them (human places signature during review). + "855b": [], + "855o": [], + "855a": [], +} + + +def _split_name(full: str) -> tuple[str, str, str]: + """Split 'First Middle Last' into (first, middle_initial, last).""" + parts = (full or "").split() + if not parts: + return "", "", "" + if len(parts) == 1: + return parts[0], "", "" + if len(parts) == 2: + return parts[0], "", parts[1] + return parts[0], parts[1][0], " ".join(parts[2:]) + + +def determine_form_type(slug: str, intake: dict) -> str: + """Pick which 855 form applies for the order.""" + enum_type = (intake.get("enumeration_type") or "").upper() + if slug == "npi-revalidation": + # Org NPIs (Type 2) revalidate on 855B; individuals on 855I. + return "855b" if enum_type in ("NPI-2", "2", "ORGANIZATION") else "855i" + if slug == "medicare-enrollment": + return "855b" if enum_type in ("NPI-2", "2", "ORGANIZATION") else "855i" + return "855i" + + +def fill_cms855(form_type: str, intake: dict, order_number: str = "") -> tuple[bytes, list[dict], list[str]]: + """Fill the official CMS-855 form. + + Returns (pdf_bytes, signature_anchors, unmapped_fields_note). + + ``signature_anchors`` is the list the e-sign stamper consumes. + ``unmapped_fields_note`` lists data we could NOT auto-place, for the human + reviewer to complete before mailing. + """ + try: + from pypdf import PdfReader, PdfWriter + except ImportError as exc: # pragma: no cover + raise RuntimeError("pypdf is required to fill CMS-855 forms") from exc + + form_type = (form_type or "855i").lower() + form_path = FORM_PATHS.get(form_type) + if not form_path or not form_path.exists(): + raise FileNotFoundError(f"Official form not found for {form_type}: {form_path}") + + reader = PdfReader(str(form_path)) + writer = PdfWriter() + writer.append(reader) + + fmap = FIELD_MAPS.get(form_type, {}) + + # Derive values from intake. + provider = intake.get("provider_name", "") + first, mid, last = _split_name(provider) + npi = intake.get("npi", "") + dob = intake.get("dob", "") # MMDDYYYY if provided + + values = { + "first_name": intake.get("first_name", first), + "middle_init": intake.get("middle_initial", mid), + "last_name": intake.get("last_name", last), + "npi": npi, + "dob": dob, + "sign_first": intake.get("first_name", first), + "sign_middle": intake.get("middle_initial", mid), + "sign_last": intake.get("last_name", last), + "org_npi": npi, + } + + field_updates = {} + for intake_key, pdf_field in fmap.items(): + v = values.get(intake_key, "") + if v: + field_updates[pdf_field] = str(v) + + # Apply to every page that contains these fields. + for page in writer.pages: + try: + writer.update_page_form_field_values(page, field_updates) + except Exception: + # update only the fields that exist on this page; ignore the rest + pass + + # Make filled values render in all viewers. + try: + writer.set_need_appearances_writer(True) + except Exception: + pass + + buf = io.BytesIO() + writer.write(buf) + pdf_bytes = buf.getvalue() + + # Build signature anchors from the official /Sig rects. + anchors = [] + for sig in SIGNATURE_FIELDS.get(form_type, []): + llx, lly, urx, ury = sig["rect"] + anchors.append({ + "field": sig["field"], + "page": sig["page"], + "x": float(llx) + 4, + "y": float(lly) + 1, + "w": float(urx - llx) - 8, + "h": max(float(ury - lly), 18.0), + "page_w": sig["page_w"], + "page_h": sig["page_h"], + }) + + # What we could not auto-fill (so the human reviewer knows to complete it). + missing = [] + if not anchors: + missing.append(f"No signature anchor mapped for {form_type} — place signature manually before mailing.") + if form_type in ("855b", "855a") or (form_type == "855o"): + missing.append(f"{form_type.upper()} is partially mapped — verify all sections before signing/mailing.") + if not intake.get("dob"): + missing.append("Date of birth not collected — required on the 855; confirm with provider.") + if not intake.get("practice_state"): + missing.append("Practice/MAC routing state not collected — needed to address the USPS envelope to the correct MAC.") + + return pdf_bytes, anchors, missing + + +if __name__ == "__main__": # quick local render for visual verification + sample = { + "provider_name": "Jane Q Smith", + "npi": "1234567893", + "dob": "01011980", + "practice_state": "CA", + "enumeration_type": "NPI-1", + } + pdf, anchors, missing = fill_cms855("855i", sample, "CO-TEST1234") + with open("/tmp/cms855i_filled.pdf", "wb") as f: + f.write(pdf) + print("wrote /tmp/cms855i_filled.pdf") + print("anchors:", anchors) + print("missing:", missing) diff --git a/scripts/workers/services/npi_provider.py b/scripts/workers/services/npi_provider.py index cdc01a6..c1fcbb7 100644 --- a/scripts/workers/services/npi_provider.py +++ b/scripts/workers/services/npi_provider.py @@ -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