"""FCC Carrier Compliance Checkup handler. Orchestrates the full compliance checkup for a telecom entity: 1. Runs compliance checks (CORES, RMD, STIR/SHAKEN, CPNI, 499-A/Q) 2. Generates the RMD certification letter (carrier-type-specific) 3. Generates Exhibit A if needed (partial STIR/SHAKEN) 4. Generates CPNI certification letter (if due/overdue) 5. Generates 499-A filing prep checklist 6. Fills the compliance status report template 7. Converts all documents to PDF 8. Computes ``recommended_slugs`` — the list of remediation services we should upsell the customer into based on what the checks flagged. These slugs feed the delivery-email deep-links and the bundle URL. """ from __future__ import annotations import logging import os from datetime import datetime from pathlib import Path from .base_handler import BaseServiceHandler logger = logging.getLogger(__name__) # Map checkup findings → remediation service slugs. The slugs match the # ERPNext Item codes / COMPLIANCE_SERVICES entries in # ``api/src/routes/compliance-orders.ts``. # # Order matters: this is also the natural display order in the upsell # email. Full-compliance bundle is handled separately downstream (the # recommendations endpoint swaps the individual slugs for the bundle # when 3+ remediations would fire). _CHECK_TO_SLUG = { "rmd_filing": "rmd-filing", "stir_shaken": "stir-shaken", "cpni": "cpni-certification", "form_499a": "fcc-499a", } class FCCComplianceCheckupHandler(BaseServiceHandler): SERVICE_SLUG = "fcc-compliance-checkup" SERVICE_NAME = "FCC Carrier Compliance Checkup" TEMPLATE_NAME = "fcc_compliance_report_template.docx" REQUIRES_LLM = False async def process(self, order_data: dict) -> list[str]: work_dir = self._make_work_dir() order_number = order_data["name"] entity = order_data.get("entity", {}) # Pull entity fields (passed in from job_server or looked up beforehand) entity_name = entity.get("legal_name", order_data.get("customer_name", "")) frn = entity.get("frn", "") filer_id = entity.get("filer_id_499", "") dba_name = entity.get("dba_name", "") # Contact contact_name = entity.get("contact_name", "") contact_title = entity.get("contact_title", "") contact_email = entity.get("contact_email", "") contact_phone = entity.get("contact_phone", "") ceo_name = entity.get("ceo_name", "") ceo_title = entity.get("ceo_title", "Chief Executive Officer") # Address address_street = entity.get("address_street", "") address_city = entity.get("address_city", "") address_state = entity.get("address_state", "") address_zip = entity.get("address_zip", "") # Classification carrier_category = entity.get("carrier_category", "interconnected_voip") infra_type = entity.get("infra_type", "facilities") is_wholesale = entity.get("is_wholesale", False) is_gateway_provider = entity.get("is_gateway_provider", False) is_international_only = entity.get("is_international_only", False) uses_ucaas_provider = entity.get("uses_ucaas_provider", False) carrier_metadata = entity.get("carrier_metadata", {}) stir_shaken_status = entity.get("stir_shaken_status", "complete_implementation") stir_shaken_cert_authority = entity.get("stir_shaken_cert_authority", "") upstream_provider_name = entity.get("upstream_provider_name", "") upstream_provider_frn = entity.get("upstream_provider_frn", "") # Revenue / classification is_deminimis = entity.get("is_deminimis", False) is_lire = entity.get("is_lire", False) service_categories = entity.get("service_categories", []) total_revenue_cents = entity.get("total_revenue_cents", 0) interstate_pct = entity.get("interstate_pct", 0) international_pct = entity.get("international_pct", 0) last_filing_year = entity.get("last_filing_year", 0) rmd_number = entity.get("rmd_number", "") # Compliance check results (passed in from the API or job_server) checks = order_data.get("compliance_checks", {}) generated_files: list[str] = [] date_str = datetime.now().strftime("%Y%m%d") # ── 1. RMD Certification Letter ────────────────────────── from scripts.document_gen.templates.rmd_letter_generator import ( generate_rmd_letter, ) rmd_docx = os.path.join( work_dir, f"rmd_certification_letter_{order_number}_{date_str}.docx" ) result = generate_rmd_letter( entity_name=entity_name, dba_name=dba_name, frn=frn, rmd_number=rmd_number, filer_id_499=filer_id, address_street=address_street, address_city=address_city, address_state=address_state, address_zip=address_zip, contact_name=contact_name, contact_title=contact_title, contact_email=contact_email, contact_phone=contact_phone, ceo_name=ceo_name, ceo_title=ceo_title, carrier_category=carrier_category, infra_type=infra_type, is_wholesale=is_wholesale, is_gateway_provider=is_gateway_provider, is_international_only=is_international_only, uses_ucaas_provider=uses_ucaas_provider, carrier_metadata=carrier_metadata, stir_shaken_status=stir_shaken_status, stir_shaken_cert_authority=stir_shaken_cert_authority, upstream_provider_name=upstream_provider_name, upstream_provider_frn=upstream_provider_frn, output_path=rmd_docx, ) if result: generated_files.append(result) try: generated_files.append(self._convert_to_pdf(result)) except Exception as exc: logger.warning("RMD letter PDF conversion failed: %s", exc) # ── 2. Exhibit A (if needed) ───────────────────────────── needs_exhibit_a = stir_shaken_status in ( "partial_implementation", "robocall_mitigation_only", "exempt_small_carrier", ) if needs_exhibit_a: from scripts.document_gen.templates.rmd_letter_generator import ( _determine_primary_role, ) from scripts.document_gen.templates.rmd_exhibit_a_generator import ( generate_exhibit_a, ) carrier_role = _determine_primary_role( is_gateway_provider=is_gateway_provider, uses_ucaas_provider=uses_ucaas_provider, is_wholesale=is_wholesale, is_international_only=is_international_only, infra_type=infra_type, ) exhibit_docx = os.path.join( work_dir, f"robocall_mitigation_program_{order_number}_{date_str}.docx", ) exhibit_result = generate_exhibit_a( entity_name=entity_name, frn=frn, carrier_role=carrier_role, carrier_metadata=carrier_metadata, upstream_provider_name=upstream_provider_name, llm_generate=self._call_llm if True else None, output_path=exhibit_docx, ) if exhibit_result: generated_files.append(exhibit_result) try: generated_files.append(self._convert_to_pdf(exhibit_result)) except Exception as exc: logger.warning("Exhibit A PDF conversion failed: %s", exc) # ── 3. CPNI Certification Letter ───────────────────────── from scripts.document_gen.templates.cpni_cert_letter_generator import ( generate_cpni_cert_letter, ) cpni_docx = os.path.join( work_dir, f"cpni_certification_{order_number}_{date_str}.docx" ) cpni_result = generate_cpni_cert_letter( entity_name=entity_name, frn=frn, filer_id_499=filer_id, address_street=address_street, address_city=address_city, address_state=address_state, address_zip=address_zip, officer_name=ceo_name or contact_name, officer_title=ceo_title, contact_email=contact_email, contact_phone=contact_phone, complaints_count=0, is_wholesale=is_wholesale, output_path=cpni_docx, ) if cpni_result: generated_files.append(cpni_result) try: generated_files.append(self._convert_to_pdf(cpni_result)) except Exception as exc: logger.warning("CPNI letter PDF conversion failed: %s", exc) # ── 4. 499-A Filing Prep Checklist ─────────────────────── from scripts.document_gen.templates.fcc_499a_checklist_generator import ( generate_499a_checklist, ) checklist_docx = os.path.join( work_dir, f"fcc_499a_checklist_{order_number}_{date_str}.docx" ) checklist_result = generate_499a_checklist( entity_name=entity_name, frn=frn, filer_id_499=filer_id, address_street=address_street, address_city=address_city, address_state=address_state, address_zip=address_zip, filer_type=carrier_category, infra_type=infra_type, service_categories=service_categories, is_deminimis=is_deminimis, is_lire=is_lire, total_revenue_cents=total_revenue_cents, interstate_pct=interstate_pct, international_pct=international_pct, last_filing_year=last_filing_year, output_path=checklist_docx, ) if checklist_result: generated_files.append(checklist_result) try: generated_files.append(self._convert_to_pdf(checklist_result)) except Exception as exc: logger.warning("499-A checklist PDF conversion failed: %s", exc) # ── 5. Compliance Status Report ────────────────────────── template_path = self._get_template_path() report_docx = os.path.join( work_dir, self._output_filename(order_number, "docx") ) # Build template variables from checks and entity data variables = self._build_report_variables( order_number=order_number, entity_name=entity_name, frn=frn, filer_id=filer_id, checks=checks, carrier_category=carrier_category, infra_type=infra_type, is_wholesale=is_wholesale, uses_ucaas_provider=uses_ucaas_provider, carrier_metadata=carrier_metadata, stir_shaken_status=stir_shaken_status, is_deminimis=is_deminimis, generated_files=generated_files, customer_name=order_data.get("customer_name", entity_name), ) self._fill_template(template_path, variables, report_docx) generated_files.append(report_docx) try: generated_files.append(self._convert_to_pdf(report_docx)) except Exception as exc: logger.warning("Compliance report PDF conversion failed: %s", exc) # ── 6. Compute recommended remediation slugs ───────────────────── # Any check that came back red or yellow triggers its corresponding # remediation slug. If 3+ trigger, we also include the full bundle # slug (consumed by the recommendations endpoint to offer the # bundled upsell at the 15% rate — see compliance-orders.ts:237). recommended: list[str] = [] for check_id, slug in _CHECK_TO_SLUG.items(): for c in checks.get("checks", []): if c.get("id") == check_id and c.get("status") in ("red", "yellow"): recommended.append(slug) break if len(recommended) >= 3: recommended.append("fcc-full-compliance") # Persist onto the PG compliance_orders row so the recommendations # API endpoint can serve it without re-running the checks. self._persist_recommendations(order_number, recommended) # ── 7. Re-render the PDF report with the bundle URL appended ──── # The checkup PDF is a takeaway artifact — customers who read it # offline should get the same one-click remediation link as the # delivery email. Append it in the appendix section we just wrote. if recommended: self._append_bundle_link_to_report( report_docx=report_docx, generated_files=generated_files, order_number=order_number, recommended=recommended, customer_email=order_data.get("customer_email", ""), customer_name=order_data.get("customer_name", ""), ) return generated_files # ------------------------------------------------------------------ # # PDF appendix — append the bundle URL # ------------------------------------------------------------------ # def _append_bundle_link_to_report( self, *, report_docx: str, generated_files: list[str], order_number: str, recommended: list[str], customer_email: str, customer_name: str, ) -> None: """Add a 'One-click remediation' paragraph + bundle URL to the DOCX. Only called when the PDF has already been converted; we update the DOCX and reconvert so both formats carry the link. Failures are non-fatal — the customer still gets the report (just without the deep-link). """ from urllib.parse import urlencode try: from docx import Document site_base = os.environ.get("SITE_URL", "https://performancewest.net") bundle_slugs = [s for s in recommended if s != "fcc-full-compliance"] if len(bundle_slugs) < 2: # Single-item — link straight to the /order/{slug} page. slug = bundle_slugs[0] if bundle_slugs else "" qs = urlencode({ "email": customer_email, "name": customer_name, "source": f"checkup:{order_number}", }) url = f"{site_base}/order/{slug}?{qs}" if slug else "" else: qs_pairs = [ ("email", customer_email), ("name", customer_name), ("source", f"checkup:{order_number}"), ] + [("service", s) for s in bundle_slugs] url = f"{site_base}/order/compliance-bundle?{urlencode(qs_pairs)}" if not url: return doc = Document(report_docx) doc.add_paragraph("") doc.add_heading("One-Click Remediation", level=2) doc.add_paragraph( "Fix the items flagged above in one checkout — the bundle " "link below applies the automatic 15% discount when two " "or more remediation services are selected." ) doc.add_paragraph(url) doc.save(report_docx) # Reconvert PDF so the delivery email attachment matches. try: new_pdf = self._convert_to_pdf(report_docx) # Replace the previous PDF path in generated_files if # present (paths are identical — same stem, same dir) but # the file has been rewritten on disk. if new_pdf not in generated_files: generated_files.append(new_pdf) except Exception as exc: logger.warning( "FCCComplianceCheckup: could not reconvert PDF after " "appending bundle link: %s", exc, ) except Exception as exc: logger.warning( "FCCComplianceCheckup: could not append bundle link to " "report DOCX: %s", exc, ) # ------------------------------------------------------------------ # # Recommendations persistence # ------------------------------------------------------------------ # def _persist_recommendations( self, order_number: str, recommended_slugs: list[str] ) -> None: """Store recommended_slugs on compliance_orders (migration 047).""" if not recommended_slugs: return try: import psycopg2 conn = psycopg2.connect(os.environ.get("DATABASE_URL", "")) with conn.cursor() as cur: cur.execute( "UPDATE compliance_orders SET recommended_slugs = %s " "WHERE order_number = %s", (recommended_slugs, order_number), ) conn.commit() conn.close() logger.info( "FCCComplianceCheckup: persisted %d recommendations for %s: %s", len(recommended_slugs), order_number, recommended_slugs, ) except Exception as exc: logger.warning( "FCCComplianceCheckup: could not persist recommendations for %s: %s", order_number, exc, ) def _build_report_variables( self, *, order_number: str, entity_name: str, frn: str, filer_id: str, checks: dict, carrier_category: str, infra_type: str, is_wholesale: bool, uses_ucaas_provider: bool, carrier_metadata: dict, stir_shaken_status: str, is_deminimis: bool, generated_files: list[str], customer_name: str, ) -> dict[str, str]: """Build the template variable dict for the compliance report.""" now = datetime.now() # Helper to extract check status def _check(check_id: str, field: str = "status") -> str: for c in checks.get("checks", []): if c.get("id") == check_id: return c.get(field, "unknown") return "unknown" def _check_detail(check_id: str) -> str: for c in checks.get("checks", []): if c.get("id") == check_id: return c.get("detail", "No data available.") return "No data available." # Status emoji mapping for report def _status_label(status: str) -> str: return { "green": "COMPLIANT", "yellow": "ATTENTION NEEDED", "red": "NON-COMPLIANT", "unknown": "UNABLE TO VERIFY", }.get(status, status.upper()) # Overall posture statuses = [_check(cid) for cid in [ "cores_registration", "rmd_filing", "stir_shaken", "cpni", "form_499a", ]] if "red" in statuses: overall = "RED — Immediate action required on one or more filings." elif "yellow" in statuses: overall = "YELLOW — Some items require attention within 30 days." elif all(s == "green" for s in statuses): overall = "GREEN — All checked items appear compliant." else: overall = "MIXED — Some items could not be verified programmatically." # Carrier classification display category_labels = { "interconnected_voip": "Interconnected VoIP", "non_interconnected_voip": "Non-Interconnected VoIP", "clec": "CLEC", "ixc": "Interexchange Carrier", "cmrs": "CMRS", "other": "Other", } ucaas_display = "N/A" if uses_ucaas_provider: ucaas_display = carrier_metadata.get("ucaas_provider", "Yes (provider not specified)") # Recommended actions actions = [] if _check("cores_registration") == "red": actions.append("[CRITICAL] Register with FCC CORES and obtain an FRN immediately.") if _check("rmd_filing") in ("red", "yellow"): actions.append("[HIGH] File or recertify in the Robocall Mitigation Database — use the attached RMD letter.") if _check("stir_shaken") in ("red", "yellow"): actions.append("[HIGH] Address STIR/SHAKEN implementation gaps.") if _check("cpni") in ("red", "yellow"): actions.append("[MEDIUM] File annual CPNI certification (due March 1) — use the attached CPNI letter.") if _check("form_499a") in ("red", "yellow"): actions.append("[MEDIUM] File Form 499-A (due April 1) — review the attached preparation checklist.") if not actions: actions.append("No immediate actions required. Continue to monitor filing deadlines.") # Appendix doc_names = [Path(f).name for f in generated_files if f.endswith(".pdf")] appendix = "\n".join(f"- {name}" for name in doc_names) if doc_names else "No documents attached." return { "order_number": order_number, "customer_name": customer_name, "entity_name": entity_name, "frn": frn or "Not on file", "date": now.strftime("%B %d, %Y"), "service_name": self.SERVICE_NAME, # Executive summary "executive_summary": ( f"This compliance checkup was conducted for {entity_name} " f"(FRN: {frn or 'pending'}) on {now.strftime('%B %d, %Y')}. " f"Overall compliance posture: {overall}" ), # CORES "cores_status": _status_label(_check("cores_registration")), "cores_red_light": _check_detail("cores_registration"), "cores_entity_name": checks.get("entity", {}).get("name", entity_name), "cores_address": checks.get("entity", {}).get("address", "Not available"), "cores_detail": _check_detail("cores_registration"), # RMD "rmd_status": _status_label(_check("rmd_filing")), "rmd_number": checks.get("rmd", {}).get("rmd_number", "Not on file"), "rmd_last_cert": checks.get("rmd", {}).get("certification_date", "Unknown"), "rmd_recert_due": _check("rmd_filing", "due_date") or "Unknown", "rmd_removal_status": "Not removed" if not checks.get("removed") else "REMOVED — see detail", "rmd_detail": _check_detail("rmd_filing"), # STIR/SHAKEN "stir_shaken_type": checks.get("rmd", {}).get("implementation_type", "Unknown"), "stir_shaken_cert": stir_shaken_status.replace("_", " ").title(), "stir_shaken_status": _status_label(_check("stir_shaken")), "stir_shaken_detail": _check_detail("stir_shaken"), # CPNI "cpni_status": _status_label(_check("cpni")), "cpni_due_date": "March 1", "cpni_detail": _check_detail("cpni"), # 499-A "f499a_status": _status_label(_check("form_499a")), "f499a_due_date": "April 1", "f499a_filer_id": filer_id or "Not on file", "f499a_detail": _check_detail("form_499a"), # 499-Q "f499q_detail": ( "De minimis provider — exempt from quarterly 499-Q filings." if is_deminimis else "Quarterly filings due: February 1, May 1, August 1, November 1. " "Status: " + _status_label(_check("form_499q")) ), # Classification "carrier_category": category_labels.get(carrier_category, carrier_category), "infra_type": infra_type.replace("_", " ").title() if infra_type else "Not classified", "is_wholesale": "Yes" if is_wholesale else "No", "ucaas_provider": ucaas_display, "classification_stir_shaken": stir_shaken_status.replace("_", " ").title(), "is_deminimis": "Yes" if is_deminimis else "No", # Actions "recommended_actions": "\n".join(actions), # Appendix "appendix_index": appendix, }