301 lines
12 KiB
Python
301 lines
12 KiB
Python
"""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 <fmcsa_ask@mailag.cx.usg.oraclecloud.com>
|
|
|
|
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(
|
|
'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())
|