add FMCSA web submitter: Playwright automation for ask.fmcsa.dot.gov ticket form
This commit is contained in:
justin
parent
1f1113d63c
commit
99a53ad970
1 changed files with 273 additions and 0 deletions
273
scripts/workers/services/fmcsa_web_submitter.py
Normal file
273
scripts/workers/services/fmcsa_web_submitter.py
Normal file
|
|
@ -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 <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(
|
||||
"#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())
|
||||
Loading…
Add table
Add a link
Reference in a new issue