diff --git a/scripts/workers/services/fmcsa_web_submitter.py b/scripts/workers/services/fmcsa_web_submitter.py new file mode 100644 index 0000000..9e000ee --- /dev/null +++ b/scripts/workers/services/fmcsa_web_submitter.py @@ -0,0 +1,273 @@ +"""FMCSA Web Submitter — submit MCS-150 updates via ask.fmcsa.dot.gov. + +Automates the FMCSA "Ask FMCSA" ticket form to submit MCS-150 updates +electronically instead of (or in addition to) faxing. Attaches the filled +PDF + photo ID, takes a screenshot of the confirmation for attestation. + +The submission goes to: https://ask.fmcsa.dot.gov/app/ticket +FMCSA replies to the email we provide from: + FMCSA Customer Service + +We use fmcsa-filings@performancewest.net (monitored via IMAP) so we can +track their responses and confirmations. +""" + +from __future__ import annotations + +import asyncio +import logging +import os +import tempfile +from datetime import datetime, timezone +from pathlib import Path + +LOG = logging.getLogger("workers.services.fmcsa_web_submitter") + +# SOCKS proxy relay (local unauthenticated relay to authenticated upstream) +PROXY_SERVER = os.getenv("FMCSA_PROXY", "socks5://127.0.0.1:11081") + +# Email for FMCSA to reply to — must be monitored via IMAP +FILING_EMAIL = os.getenv("FMCSA_FILING_EMAIL", "fmcsa-filings@performancewest.net") + +FMCSA_TICKET_URL = "https://ask.fmcsa.dot.gov/app/ticket" + + +async def submit_mcs150( + pdf_path: str, + photo_id_path: str | None = None, + dot_number: str = "", + mc_number: str = "", + entity_name: str = "", + contact_first: str = "Compliance", + contact_last: str = "Department", + contact_email: str = FILING_EMAIL, + screenshot_dir: str = "", +) -> dict: + """Submit an MCS-150 update via the FMCSA Ask ticket form. + + Args: + pdf_path: Path to the filled MCS-150 PDF. + photo_id_path: Path to the signer's photo ID (optional). + dot_number: USDOT number. + mc_number: MC/FF/MX number (optional). + entity_name: Legal entity name (for the question text). + contact_first: First name for the submission. + contact_last: Last name for the submission. + contact_email: Email for FMCSA to reply to. + screenshot_dir: Where to save confirmation screenshots. + + Returns: + dict with keys: success, screenshot_path, submitted_at, confirmation_text, error + """ + if not screenshot_dir: + screenshot_dir = tempfile.mkdtemp(prefix="pw_fmcsa_") + + submitted_at = datetime.now(timezone.utc) + + try: + from patchright.async_api import async_playwright + except ImportError: + from playwright.async_api import async_playwright + + try: + async with async_playwright() as p: + browser = await p.chromium.launch( + headless=True, + proxy={"server": PROXY_SERVER} if PROXY_SERVER else None, + ) + ctx = await browser.new_context( + user_agent=( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " + "AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/131.0.0.0 Safari/537.36" + ), + viewport={"width": 1920, "height": 1080}, + locale="en-US", + timezone_id="America/Chicago", + ) + + page = await ctx.new_page() + + # Navigate to the ticket form + LOG.info("[fmcsa] Navigating to %s", FMCSA_TICKET_URL) + resp = await page.goto(FMCSA_TICKET_URL, wait_until="networkidle", timeout=30000) + await page.wait_for_timeout(2000) + + if resp and resp.status != 200: + await browser.close() + return { + "success": False, + "screenshot_path": None, + "submitted_at": submitted_at.isoformat(), + "confirmation_text": None, + "error": f"FMCSA returned HTTP {resp.status}", + } + + # ── Fill the form fields using exact Oracle Service Cloud IDs ── + LOG.info("[fmcsa] Filling form for DOT %s (%s)", dot_number, entity_name) + + # Email + await page.fill("#rn_TextInput_5_Contact\\.Emails\\.PRIMARY\\.Address", + contact_email, timeout=5000) + + # First Name + await page.fill("#rn_TextInput_7_Contact\\.Name\\.First", + contact_first, timeout=5000) + + # Last Name + await page.fill("#rn_TextInput_9_Contact\\.Name\\.Last", + contact_last, timeout=5000) + + # USDOT Number + if dot_number: + await page.fill("#rn_TextInput_11_Incident\\.CustomFields\\.c\\.dot_number", + dot_number, timeout=3000) + + # MC Number + if mc_number: + mc_clean = mc_number.replace("MC-", "").replace("FF-", "").replace("MX-", "") + await page.fill("#rn_TextInput_13_Incident\\.CustomFields\\.c\\.mc_number", + mc_clean, timeout=3000) + + # Inquiry Type — select "USDOT Number" (value=175) + await page.select_option( + "#rn_ProductCategoryInputSimpleFiltered_20_Incident\\.Product_lvl1", + value="175", timeout=5000, + ) + await page.wait_for_timeout(1000) + + # Question text + question_text = ( + f"MCS-150 Biennial Update for {entity_name} (USDOT# {dot_number}).\n\n" + f"Please find attached the completed MCS-150 form for the above-referenced " + f"motor carrier. This form is being submitted as a biennial update per " + f"49 CFR § 390.19.\n\n" + f"Attached documents:\n" + f"1. Completed MCS-150 Form (PDF)\n" + ) + if photo_id_path: + question_text += "2. Government-issued photo ID of authorized signer\n" + question_text += ( + f"\nPlease process this update at your earliest convenience. " + f"If you require any additional information, please contact us at " + f"{contact_email} or (888) 411-0383.\n\n" + f"Submitted by Performance West Inc. on behalf of {entity_name}.\n" + f"Reference: {dot_number}" + ) + + await page.fill("#rn_QuestionInput_21_Incident\\.Threads", + question_text, timeout=5000) + await page.wait_for_timeout(500) + + # Attach documents + LOG.info("[fmcsa] Attaching documents") + file_input = await page.query_selector("#rn_FileAttachmentUpload_22_FileInput") + if file_input: + files = [pdf_path] + if photo_id_path and os.path.exists(photo_id_path): + files.append(photo_id_path) + await file_input.set_input_files(files) + await page.wait_for_timeout(2000) + + # Screenshot before submit + pre_submit_path = os.path.join(screenshot_dir, f"fmcsa_pre_submit_{dot_number}.png") + await page.screenshot(path=pre_submit_path, full_page=True) + LOG.info("[fmcsa] Pre-submit screenshot: %s", pre_submit_path) + + # Click Continue (checkbox + submit link) + try: + # The "Continue..." is a link/button, not a regular submit + submit_el = await page.query_selector( + '#rn_FormSubmit_23_Submission, ' + 'a:has-text("Continue"), button:has-text("Continue"), ' + 'input[value="Continue"]' + ) + if submit_el: + tag = await submit_el.evaluate("el => el.tagName") + if tag == "INPUT" and await submit_el.get_attribute("type") == "checkbox": + # It's a checkbox — check it, then find the actual submit + await submit_el.check() + await page.wait_for_timeout(500) + real_submit = await page.query_selector( + 'a:has-text("Continue"), button:has-text("Submit")' + ) + if real_submit: + await real_submit.click() + else: + await submit_el.click() + await page.wait_for_timeout(5000) + await page.wait_for_load_state("networkidle", timeout=15000) + else: + LOG.warning("[fmcsa] Could not find submit element") + except Exception as e: + LOG.warning("[fmcsa] Submit click failed: %s", e) + + # Screenshot after submit (confirmation page) + post_submit_path = os.path.join(screenshot_dir, f"fmcsa_confirmation_{dot_number}.png") + await page.screenshot(path=post_submit_path, full_page=True) + + # Get confirmation text + confirmation_text = await page.inner_text("body") + + # Check for success indicators + success = any(kw in confirmation_text.lower() for kw in [ + "thank you", "submitted", "received", "confirmation", + "reference number", "ticket", "case" + ]) + + LOG.info("[fmcsa] Submission %s for DOT %s", + "succeeded" if success else "may have failed", dot_number) + + await browser.close() + + return { + "success": success, + "screenshot_path": post_submit_path, + "pre_submit_screenshot": pre_submit_path, + "submitted_at": submitted_at.isoformat(), + "confirmation_text": confirmation_text[:2000], + "error": None if success else "Could not confirm submission — check screenshot", + } + + except Exception as exc: + LOG.error("[fmcsa] Web submission failed: %s", exc) + return { + "success": False, + "screenshot_path": None, + "submitted_at": submitted_at.isoformat(), + "confirmation_text": None, + "error": str(exc), + } + + +async def test_submission(): + """Test with a dummy PDF to verify the form works.""" + import urllib.request + + # Download a dummy PDF + dummy_url = "https://www.w3.org/WAI/ER/tests/xhtml/testfiles/resources/pdf/dummy.pdf" + dummy_path = "/tmp/fmcsa_test_dummy.pdf" + urllib.request.urlretrieve(dummy_url, dummy_path) + + result = await submit_mcs150( + pdf_path=dummy_path, + photo_id_path=None, + dot_number="9999999", + mc_number="", + entity_name="TEST SUBMISSION — PLEASE DISREGARD", + contact_first="Test", + contact_last="Submission", + contact_email=FILING_EMAIL, + ) + + print(f"Success: {result['success']}") + print(f"Screenshot: {result['screenshot_path']}") + print(f"Error: {result['error']}") + if result['confirmation_text']: + print(f"Confirmation: {result['confirmation_text'][:500]}") + + return result + + +if __name__ == "__main__": + asyncio.run(test_submission())