justin/new-site
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
new-site/scripts/workers/services/fcc_compliance_checkup.py
justin f8cd37ac8c Initial commit — Performance West telecom compliance platform 
Includes: API (Express/TypeScript), Astro site, Python workers,
document generators, FCC compliance tools, Canada CRTC formation,
Ansible infrastructure, and deployment scripts.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-27 06:54:22 -05:00

586 lines
24 KiB
Python
Raw Blame History

"""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,
}
Reference in a new issue View git blame Copy permalink