"""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) # Guard: don't submit to FMCSA in dev/test is_prod = os.getenv("NODE_ENV") == "production" or os.getenv("ENV") == "production" if not is_prod: LOG.info("[fmcsa] DEV MODE — skipping web submission for DOT %s", dot_number) return { "success": True, "screenshot_path": None, "pre_submit_screenshot": None, "submitted_at": submitted_at.isoformat(), "confirmation_text": "DEV MODE — submission skipped", "error": None, } 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( 'select[name="Incident.Product_lvl1"]', value="175", timeout=5000, ) await page.wait_for_timeout(2000) # Sub-category — "Change/update my USDOT company information" (value=226) # Wait for the lvl2 dropdown to appear after lvl1 selection try: await page.wait_for_selector( 'select[name="Incident.Product_lvl2"]', state="visible", timeout=5000, ) await page.select_option( 'select[name="Incident.Product_lvl2"]', value="226", timeout=5000, ) except Exception: # Fallback: try by partial ID match lvl2 = await page.query_selector('select[id*="Product_lvl2"]') if lvl2: await lvl2.select_option(value="226") else: LOG.warning("[fmcsa] Could not find/select lvl2 dropdown") 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 # Note: after successful submission, FMCSA redirects to www.fmcsa.dot.gov # which may return 403 (Akamai WAF). The submission still went through — # FMCSA sends a confirmation email to the address we provided. # So a redirect away from ask.fmcsa.dot.gov IS a success signal. redirected_away = "ask.fmcsa.dot.gov" not in page.url has_confirmation_keywords = any(kw in confirmation_text.lower() for kw in [ "thank you", "submitted", "received", "confirmation", "reference number", "ticket", "case", ]) has_error_keywords = any(kw in confirmation_text.lower() for kw in [ "please select an item", "required field", "invalid", ]) success = (redirected_away or has_confirmation_keywords) and not has_error_keywords 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())