"""CPNI Annual Certification filing handler. Submits the Annual 47 C.F.R. § 64.2009(e) CPNI Certification to the FCC's Electronic Comment Filing System (ECFS) under WC Docket No. 06-36. The filing window opens each year in January and closes March 1. Flow: 1. Generate the CPNI certification letter via the existing generator. 2. Convert to PDF. 3. Launch undetected browser, navigate to ECFS express comment form, upload the signed letter, and submit under docket 06-36. 4. Capture confirmation page (ECFS assigns a document number like ``2026XXXXXXXX``) and persist it. Idempotency: if the carrier already has a CPNI cert on record for the current calendar year, skip the submission and deliver docs only. """ from __future__ import annotations import logging import os from datetime import datetime from .base_handler import BaseServiceHandler from .telecom import filing_state from .telecom.auto_filing import check_auto_filing, request_admin_review from .telecom.undetected_browser import undetected_browser, human_delay, type_slowly logger = logging.getLogger(__name__) ECFS_URL = os.environ.get( "FCC_ECFS_URL", "https://www.fcc.gov/ecfs/search/proceedings?name=06-36", ) ECFS_UPLOAD_URL = os.environ.get( "FCC_ECFS_UPLOAD_URL", "https://www.fcc.gov/ecfs/upload/express", ) CPNI_DOCKET = "06-36" class CPNIFilingHandler(BaseServiceHandler): SERVICE_SLUG = "cpni-certification" SERVICE_NAME = "CPNI Annual Certification" 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", {}) entity_id = entity.get("id") date_str = datetime.now().strftime("%Y%m%d") # ── Guard: require entity data before generating documents ─────── legal_name = entity.get("legal_name", "").strip() if not legal_name: logger.warning( "CPNIFilingHandler: no entity data for %s — pausing for intake", order_number, ) self._request_entity_intake(order_data) return [] # CPNI questionnaire answers from intake_data (filled by CPNIStep) intake = order_data.get("intake_data", {}) or {} cpni_answers = intake.get("cpni", {}) generated: list[str] = [] # ── 1. Generate packet ─────────────────────────────────────────── cpni_docx = os.path.join( work_dir, f"cpni_certification_{order_number}_{date_str}.docx" ) # Pick the Line 105 primary-category-specific variant if available. variant_fn = self._pick_template_generator(entity) if variant_fn is not None: logger.info( "CPNIFilingHandler: using variant generator %s for entity %s", getattr(variant_fn, "__name__", ""), entity.get("id"), ) result = variant_fn( output_path=cpni_docx, entity_name=entity.get("legal_name", ""), frn=entity.get("frn", ""), filer_id_499=entity.get("filer_id_499", ""), officer_name=entity.get("ceo_name") or entity.get("contact_name", ""), officer_title=entity.get("ceo_title", "Chief Executive Officer"), complaints_count=entity.get("cpni_complaints_count", 0), has_data_broker_inquiries=bool( entity.get("has_data_broker_inquiries", False) ), reporting_year=entity.get("reporting_year", 0), address_street=entity.get("address_street", ""), address_city=entity.get("address_city", ""), address_state=entity.get("address_state", ""), address_zip=entity.get("address_zip", ""), contact_email=entity.get("contact_email", ""), contact_phone=entity.get("contact_phone", ""), ) else: from scripts.document_gen.templates.cpni_cert_letter_generator import ( generate_cpni_cert_letter, ) result = generate_cpni_cert_letter( entity_name=entity.get("legal_name", ""), frn=entity.get("frn", ""), filer_id_499=entity.get("filer_id_499", ""), address_street=entity.get("address_street", ""), address_city=entity.get("address_city", ""), address_state=entity.get("address_state", ""), address_zip=entity.get("address_zip", ""), officer_name=entity.get("ceo_name") or entity.get("contact_name", ""), officer_title=entity.get("ceo_title", "Chief Executive Officer"), contact_email=entity.get("contact_email", ""), contact_phone=entity.get("contact_phone", ""), complaints_count=int(cpni_answers.get("complaints_count", 0) or 0), complaints_description=cpni_answers.get("complaints_description", ""), # Breaches — build the list[dict] format the generator expects breaches=( [{ "date": "during the reporting period", "description": cpni_answers.get("breaches_description", ""), "records_affected": 0, "notified_fcc": cpni_answers.get("breaches_notified", "yes") in ("yes", "partial"), "notified_customers": cpni_answers.get("breaches_notified", "yes") == "yes", }] if cpni_answers.get("breaches", "no") != "no" else [] ), disciplinary_actions_taken=cpni_answers.get("disciplinary", "no") != "no", disciplinary_actions_description=cpni_answers.get("disciplinary_description", ""), data_broker_actions=cpni_answers.get("data_brokers_description", ""), uses_cpni_for_marketing=cpni_answers.get("marketing_usage", "no") != "no", cpni_approval_method=cpni_answers.get("marketing_usage", "no") if cpni_answers.get("marketing_usage", "no") != "no" else "opt_in", reporting_year=int(cpni_answers.get("reporting_year", 0) or 0), is_wholesale=entity.get("is_wholesale", False), output_path=cpni_docx, ) packet_pdf: str | None = None if result: generated.append(result) try: packet_pdf = self._convert_to_pdf(result) generated.append(packet_pdf) except Exception as exc: logger.warning("CPNI letter PDF conversion failed: %s", exc) # ── 1b. CPNI Procedure Statement (47 CFR § 64.2008 annual notice) ─ # Separate deliverable from the ECFS cert letter: the internal # policy document every example filing includes alongside the # cert. Customers post this on their website and send it to # customers annually. try: from scripts.document_gen.templates.cpni_procedure_statement_generator import ( generate_cpni_procedure_statement, ) policy_docx = os.path.join( work_dir, f"cpni_procedure_statement_{order_number}_{date_str}.docx" ) policy_path = generate_cpni_procedure_statement( entity_name=entity.get("legal_name", ""), support_email=entity.get("contact_email", ""), website=entity.get("website", ""), signatory_name=entity.get("ceo_name") or entity.get("contact_name", ""), signatory_title=entity.get("ceo_title", "Chief Executive Officer"), is_wholesale=entity.get("is_wholesale", False), output_path=policy_docx, ) if policy_path: generated.append(policy_path) try: generated.append(self._convert_to_pdf(policy_path)) except Exception as exc: logger.warning( "CPNI procedure statement PDF conversion failed: %s", exc, ) except Exception as exc: logger.warning("CPNI procedure statement generation failed: %s", exc) # ── 2. Idempotency ────────────────────────────────────────────── if entity_id and filing_state.already_filed(entity_id, "cpni"): logger.info( "CPNIFilingHandler: already on file for entity %s this cycle", entity_id ) return generated # ── 2a. Auto-filing toggle ────────────────────────────────────── decision = check_auto_filing(order_data) if not decision.may_submit: logger.info( "CPNIFilingHandler: %s — staging for admin review (order=%s)", decision.reason, order_number, ) request_admin_review( order_number=order_number, service_slug=self.SERVICE_SLUG, service_name=self.SERVICE_NAME, entity_name=entity.get("legal_name", ""), frn=entity.get("frn", ""), packet_minio_paths=[f"compliance/{order_number}/{os.path.basename(p)}" for p in generated], admin_email=decision.admin_email, summary=( f"CPNI annual certification packet ready for ECFS docket " f"{CPNI_DOCKET}. 2026 deadline is March 2." ), ) return generated # ── 3. Submit to ECFS ─────────────────────────────────────────── confirmation_path, confirmation_number = await self._submit_to_ecfs( order_number=order_number, entity=entity, packet_pdf=packet_pdf or cpni_docx, work_dir=work_dir, ) if confirmation_path: generated.append(confirmation_path) # ── 4. Persist success + email confirmation to client ────────── if entity_id and confirmation_number: filing_state.record_cpni_filing(entity_id, confirmation_number) if confirmation_number: try: from scripts.workers.job_server import _send_filing_confirmation from scripts.document_gen import MinioStorage customer_email = order_data.get("customer_email") or entity.get("contact_email") customer_name = order_data.get("customer_name") or entity.get("contact_name", "") if customer_email: conf_paths = [p for p in generated if "confirmation" in p.lower()] _send_filing_confirmation( customer_email=customer_email, customer_name=customer_name, order_number=order_number, service_name=self.SERVICE_NAME, confirmation_number=confirmation_number, authority="FCC Electronic Comment Filing System (ECFS)", minio_paths=[f"compliance/{order_number}/{os.path.basename(p)}" for p in conf_paths], storage=MinioStorage(), ) except Exception as exc: logger.warning("CPNI filing confirmation email failed: %s", exc) return generated # ------------------------------------------------------------------ # # Variant selection # ------------------------------------------------------------------ # def _pick_template_generator(self, entity: dict): """ Resolve the right CPNI generator based on the entity's Line 105 primary category and that category's infra_type. Returns a callable ``generate_cpni_(...)`` on success, or ``None`` to indicate the caller should fall back to the generic ``cpni_cert_letter_generator.generate_cpni_cert_letter``. """ primary = entity.get("line_105_primary") or entity.get( "filer_type", "voip_interconnected" ) cats = entity.get("line_105_categories") or [] entry = next((c for c in cats if c.get("id") == primary), {}) infra = entry.get("infra_type", "facilities") mapping = { ("clec", "facilities"): "cpni_clec_generator", ("clec", "reseller"): "cpni_clec_reseller_generator", ("ixc", "facilities"): "cpni_ixc_generator", ("ixc", "reseller"): "cpni_ixc_reseller_generator", ("wireless", "facilities"): "cpni_wireless_generator", ("wireless", "mvno"): "cpni_wireless_mvno_generator", ("satellite", "facilities"): "cpni_satellite_generator", ("audio_bridging", "facilities"): "cpni_audio_bridge_generator", ("private_line", "facilities"): "cpni_private_line_generator", } module_name = mapping.get((primary, infra)) if not module_name: return None try: import importlib mod = importlib.import_module( f"scripts.document_gen.templates.{module_name}" ) fn_name = "generate_" + module_name.replace("_generator", "") return getattr(mod, fn_name) except Exception as exc: logger.warning( "CPNIFilingHandler: failed to load CPNI variant %s: %s", module_name, exc, ) return None # ------------------------------------------------------------------ # # ECFS Express Upload flow # ------------------------------------------------------------------ # async def _submit_to_ecfs( self, *, order_number: str, entity: dict, packet_pdf: str, work_dir: str, ) -> tuple[str | None, str]: confirmation_path = os.path.join( work_dir, f"cpni_confirmation_{order_number}.pdf" ) confirmation_number = "" try: async with undetected_browser(headless=True) as (ctx, page): await page.goto(ECFS_UPLOAD_URL, wait_until="domcontentloaded") await human_delay(1.5, 3.0) # Express comment form — fields are: Proceedings, Name of # filer, Filer's address, upload, type of filing. await type_slowly(page, 'input[name="proceedings"]', CPNI_DOCKET) # Dropdown suggestion click await page.wait_for_selector( f'li:has-text("{CPNI_DOCKET}")', timeout=10000 ) await page.click(f'li:has-text("{CPNI_DOCKET}")') await human_delay() await type_slowly( page, 'input[name="name_of_filer"]', entity.get("legal_name", "") ) await type_slowly( page, 'input[name="filer_email"]', entity.get("contact_email", "") ) # Address await page.fill( 'input[name="address_line_1"]', entity.get("address_street", "") ) await page.fill('input[name="city"]', entity.get("address_city", "")) await page.fill('input[name="state"]', entity.get("address_state", "")) await page.fill('input[name="zip_code"]', entity.get("address_zip", "")) # Type of filing — "Certification" await page.select_option( 'select[name="type_of_filing"]', label="Certification" ) # Upload the signed PDF. await page.set_input_files('input[type="file"]', packet_pdf) await human_delay(1.0, 2.0) await page.click('button:has-text("Continue")') await page.wait_for_selector("text=Review", timeout=30000) await human_delay(2.0, 4.0) await page.click('button:has-text("Submit")') await page.wait_for_selector("text=Confirmation", timeout=60000) body = await page.locator("body").inner_text() for line in body.splitlines(): if "Filing ID" in line or "Confirmation" in line: parts = line.split(":", 1) if len(parts) == 2 and parts[1].strip(): confirmation_number = parts[1].strip() break await page.pdf(path=confirmation_path, format="Letter") logger.info( "CPNIFilingHandler: submitted under docket %s, filing ID %s", CPNI_DOCKET, confirmation_number, ) return confirmation_path, confirmation_number except Exception as exc: logger.exception("CPNIFilingHandler: ECFS submission failed: %s", exc) self._create_admin_todo( order_number, f"ECFS CPNI certification submission failed: {exc}. " f"Packet is available in MinIO under compliance/{order_number}/. " "File manually at https://www.fcc.gov/ecfs/ under docket 06-36.", ) return None, "" def _create_admin_todo(self, order_number: str, description: str) -> None: try: from scripts.workers.erpnext_client import ERPNextClient ERPNextClient().create_resource( "ToDo", { "description": ( f"[{self.SERVICE_SLUG}] {order_number}\n\n{description}" ), "priority": "High", "role": "Accounting Advisor", }, ) except Exception as exc: logger.error("Could not create admin ToDo: %s", exc)