Add Playwright automation for BOC-3 filing on processagent.com
- boc3_playwright.py: 4-step form automation (company info, contact, account, payment) using patchright/undetected Playwright - Payment with PW company card ($25/filing), credentials from env - CAPTCHA detection — falls back to admin todo if reCAPTCHA triggers - boc3_filing.py: process() tries Playwright first, falls back to manual admin todo on failure - Env vars needed: PW_CARD_NUMBER, PW_CARD_CVC, PW_CARD_EXP_MONTH, PW_CARD_EXP_YEAR, BOC3_ACCOUNT_PASSWORD Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
justin
parent
584f887f82
commit
46411c09c6
2 changed files with 489 additions and 5 deletions
|
|
@ -35,6 +35,7 @@ Filing flow:
|
|||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
|
|
@ -63,8 +64,53 @@ class BOC3FilingHandler:
|
|||
SERVICE_NAME = "BOC-3 Process Agent Filing"
|
||||
|
||||
async def process(self, order_data: dict) -> list[str]:
|
||||
"""Entry point called by job_server. Delegates to handle()."""
|
||||
"""Entry point called by job_server. Tries Playwright, falls back to handle()."""
|
||||
order_number = order_data.get("order_number", order_data.get("name", ""))
|
||||
|
||||
intake = order_data.get("intake_data") or {}
|
||||
if isinstance(intake, str):
|
||||
intake = json.loads(intake)
|
||||
|
||||
dot_number = intake.get("dot_number", "")
|
||||
customer_email = order_data.get("customer_email", "")
|
||||
|
||||
# Try Playwright automation if credentials are configured
|
||||
if dot_number and os.environ.get("PW_CARD_NUMBER") and os.environ.get("BOC3_ACCOUNT_PASSWORD"):
|
||||
try:
|
||||
from .boc3_playwright import BOC3ProcessAgent
|
||||
adapter = BOC3ProcessAgent()
|
||||
result = await adapter.file_boc3({
|
||||
"dot_number": dot_number,
|
||||
"docket_number": intake.get("docket_number", ""),
|
||||
"legal_name": intake.get("entity_name", order_data.get("customer_name", "")),
|
||||
"entity_type": intake.get("entity_type", "carrier"),
|
||||
"contact": {
|
||||
"first_name": (order_data.get("customer_name") or "").split()[0] if order_data.get("customer_name") else "",
|
||||
"last_name": " ".join((order_data.get("customer_name") or "").split()[1:]),
|
||||
"phone": intake.get("phone", ""),
|
||||
"street": intake.get("address_street", ""),
|
||||
"city": intake.get("address_city", ""),
|
||||
"state": intake.get("address_state", ""),
|
||||
"zip": intake.get("address_zip", ""),
|
||||
},
|
||||
"email": customer_email,
|
||||
})
|
||||
if result.success:
|
||||
LOG.info("[%s] BOC-3 filed via Playwright! Order: %s", order_number, result.order_id)
|
||||
self._send_confirmation_email(
|
||||
order_number,
|
||||
intake.get("entity_name", order_data.get("customer_name", "")),
|
||||
dot_number, customer_email,
|
||||
)
|
||||
return []
|
||||
elif result.captcha_hit:
|
||||
LOG.warning("[%s] CAPTCHA on processagent.com — falling back to admin todo", order_number)
|
||||
else:
|
||||
LOG.warning("[%s] Playwright filing failed: %s — admin todo", order_number, result.error)
|
||||
except Exception as exc:
|
||||
LOG.warning("[%s] Playwright error: %s — admin todo", order_number, exc)
|
||||
|
||||
# Fall back to manual admin todo
|
||||
return self.handle(order_data, order_number)
|
||||
|
||||
def handle(self, order_data: dict, order_number: str) -> list[str]:
|
||||
|
|
@ -111,10 +157,7 @@ class BOC3FilingHandler:
|
|||
"requested_at": datetime.utcnow().isoformat(),
|
||||
}
|
||||
|
||||
# TODO: Automate via Playwright on processagent.com/order
|
||||
# For now, create admin todo for manual filing
|
||||
|
||||
# Create admin todo
|
||||
# Create admin todo for manual filing (Playwright attempt already made in process())
|
||||
todo_data = {
|
||||
"order_number": order_number,
|
||||
"service": self.SERVICE_NAME,
|
||||
|
|
|
|||
441
scripts/workers/services/boc3_playwright.py
Normal file
441
scripts/workers/services/boc3_playwright.py
Normal file
|
|
@ -0,0 +1,441 @@
|
|||
"""
|
||||
BOC-3 Process Agent Filing — Playwright automation for processagent.com.
|
||||
|
||||
Process Agent LLC (subsidiary of Registered Agents Inc) provides blanket
|
||||
BOC-3 process agent coverage in all 50 states + DC for $25/year.
|
||||
|
||||
They have no API — orders must be placed through their website.
|
||||
This adapter automates the 4-step order form at processagent.com/order.
|
||||
|
||||
Form steps:
|
||||
1. Company Info — authority name, DOT#, MC#, business type
|
||||
2. Contact Info — name, phone, address
|
||||
3. Account Setup — email, password (one account per carrier)
|
||||
4. Payment — credit card (Performance West company card)
|
||||
|
||||
reCAPTCHA: Uses Google reCAPTCHA v2. We use patchright (undetected
|
||||
Playwright) to avoid triggering it. If CAPTCHA appears, the handler
|
||||
creates an admin todo for manual completion.
|
||||
|
||||
Usage:
|
||||
adapter = BOC3ProcessAgent()
|
||||
result = await adapter.file_boc3({
|
||||
"dot_number": "1234567",
|
||||
"docket_number": "MC-123456",
|
||||
"legal_name": "Acme Trucking Inc",
|
||||
"entity_type": "LLC",
|
||||
"contact": {
|
||||
"first_name": "John",
|
||||
"last_name": "Doe",
|
||||
"phone": "5551234567",
|
||||
"street": "123 Main St",
|
||||
"city": "Dallas",
|
||||
"state": "TX",
|
||||
"zip": "75201",
|
||||
},
|
||||
"email": "john@acmetrucking.com",
|
||||
})
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import os
|
||||
import random
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
LOG = logging.getLogger("workers.services.boc3_playwright")
|
||||
SCREENSHOTS_DIR = Path(os.getenv("SCREENSHOTS_DIR", "/tmp/boc3-screenshots"))
|
||||
SCREENSHOTS_DIR.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Performance West company card for $25 BOC-3 payments
|
||||
# Loaded from env — never hardcoded
|
||||
PW_CARD_FIRST = os.environ.get("PW_CARD_FIRST_NAME", "Justin")
|
||||
PW_CARD_LAST = os.environ.get("PW_CARD_LAST_NAME", "Hannah")
|
||||
PW_CARD_NUMBER = os.environ.get("PW_CARD_NUMBER", "")
|
||||
PW_CARD_CVC = os.environ.get("PW_CARD_CVC", "")
|
||||
PW_CARD_EXP_MONTH = os.environ.get("PW_CARD_EXP_MONTH", "")
|
||||
PW_CARD_EXP_YEAR = os.environ.get("PW_CARD_EXP_YEAR", "")
|
||||
|
||||
# Account password for carriers on processagent.com
|
||||
# We create a unique account per carrier with a standard password
|
||||
BOC3_ACCOUNT_PASSWORD = os.environ.get("BOC3_ACCOUNT_PASSWORD", "")
|
||||
|
||||
ORDER_URL = "https://www.processagent.com/order"
|
||||
|
||||
# Entity type mapping from our intake to their dropdown values
|
||||
ENTITY_MAP = {
|
||||
"llc": "LLC",
|
||||
"corporation": "Corp",
|
||||
"sole_proprietorship": "Sole Proprietorship",
|
||||
"individual": "Individual",
|
||||
"lp": "LP",
|
||||
"llp": "LLP",
|
||||
"trust": "Trust",
|
||||
"carrier": "LLC", # default for motor carriers
|
||||
"broker": "LLC",
|
||||
"freight_forwarder": "LLC",
|
||||
}
|
||||
|
||||
# Docket number type mapping
|
||||
DOCKET_TYPE_MAP = {
|
||||
"MC": "MC",
|
||||
"FF": "FF",
|
||||
"MX": "MX",
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class BOC3FilingResult:
|
||||
success: bool
|
||||
order_id: str = ""
|
||||
error: str = ""
|
||||
screenshot_path: str = ""
|
||||
captcha_hit: bool = False
|
||||
|
||||
|
||||
class BOC3ProcessAgent:
|
||||
"""Automate BOC-3 filing through processagent.com."""
|
||||
|
||||
async def file_boc3(self, data: dict) -> BOC3FilingResult:
|
||||
"""File a BOC-3 through processagent.com order form.
|
||||
|
||||
Args:
|
||||
data: dict with dot_number, docket_number, legal_name,
|
||||
entity_type, contact (first_name, last_name, phone,
|
||||
street, city, state, zip), email
|
||||
|
||||
Returns:
|
||||
BOC3FilingResult with success status and order ID.
|
||||
"""
|
||||
if not PW_CARD_NUMBER or not BOC3_ACCOUNT_PASSWORD:
|
||||
return BOC3FilingResult(
|
||||
success=False,
|
||||
error="BOC3 payment credentials not configured (PW_CARD_NUMBER, BOC3_ACCOUNT_PASSWORD)",
|
||||
)
|
||||
|
||||
dot = data.get("dot_number", "")
|
||||
docket = data.get("docket_number", "")
|
||||
legal_name = data.get("legal_name", "")
|
||||
entity_type = data.get("entity_type", "carrier")
|
||||
contact = data.get("contact", {})
|
||||
email = data.get("email", "")
|
||||
|
||||
if not dot or not legal_name:
|
||||
return BOC3FilingResult(success=False, error="DOT number and legal name required")
|
||||
|
||||
# Parse docket type and number
|
||||
docket_type = ""
|
||||
docket_num = docket
|
||||
for prefix in ("MC-", "FF-", "MX-"):
|
||||
if docket.upper().startswith(prefix):
|
||||
docket_type = prefix.rstrip("-")
|
||||
docket_num = docket[len(prefix):]
|
||||
break
|
||||
|
||||
timestamp = int(time.time())
|
||||
screenshot_base = SCREENSHOTS_DIR / f"boc3_{dot}_{timestamp}"
|
||||
|
||||
try:
|
||||
from scripts.workers.services.telecom.undetected_browser import launch_context
|
||||
except ImportError:
|
||||
from playwright.async_api import async_playwright
|
||||
launch_context = None
|
||||
|
||||
browser = None
|
||||
context = None
|
||||
page = None
|
||||
|
||||
try:
|
||||
if launch_context:
|
||||
context = await launch_context(headless=True)
|
||||
page = await context.new_page()
|
||||
else:
|
||||
pw = await async_playwright().start()
|
||||
browser = await pw.chromium.launch(headless=True)
|
||||
context = await browser.new_context(
|
||||
viewport={"width": 1280, "height": 900},
|
||||
user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
|
||||
)
|
||||
page = await context.new_page()
|
||||
|
||||
# Navigate to order page
|
||||
LOG.info("[BOC3] Navigating to %s for DOT %s", ORDER_URL, dot)
|
||||
await page.goto(ORDER_URL, wait_until="networkidle", timeout=30000)
|
||||
await asyncio.sleep(random.uniform(2, 4))
|
||||
|
||||
# Take initial screenshot
|
||||
await page.screenshot(path=f"{screenshot_base}_01_loaded.png")
|
||||
|
||||
# ── Step 1: Company Information ─────────────────────────────
|
||||
LOG.info("[BOC3] Step 1: Company info for %s", legal_name)
|
||||
|
||||
# Authority Name (company name)
|
||||
name_input = page.locator('input[name="companyName"], input[placeholder*="authority"], input[placeholder*="company"]').first
|
||||
await name_input.click()
|
||||
await asyncio.sleep(0.3)
|
||||
await name_input.fill(legal_name)
|
||||
await asyncio.sleep(random.uniform(0.5, 1))
|
||||
|
||||
# Entity Type dropdown
|
||||
mapped_type = ENTITY_MAP.get(entity_type.lower(), "LLC")
|
||||
entity_select = page.locator('select[name="entityType"], select:has(option:text("Corp"))').first
|
||||
try:
|
||||
await entity_select.select_option(label=mapped_type, timeout=3000)
|
||||
except Exception:
|
||||
# Try value-based selection
|
||||
try:
|
||||
await entity_select.select_option(value=mapped_type, timeout=3000)
|
||||
except Exception:
|
||||
LOG.warning("[BOC3] Could not select entity type %s, trying default", mapped_type)
|
||||
|
||||
await asyncio.sleep(random.uniform(0.5, 1))
|
||||
|
||||
# USDOT Number
|
||||
dot_input = page.locator('input[name="USDOTNumber"], input[placeholder*="USDOT"], input[placeholder*="DOT"]').first
|
||||
await dot_input.click()
|
||||
await dot_input.fill(dot)
|
||||
await asyncio.sleep(random.uniform(0.3, 0.7))
|
||||
|
||||
# Docket Number (MC/FF/MX)
|
||||
if docket_num:
|
||||
# Select docket type if available
|
||||
if docket_type:
|
||||
docket_type_select = page.locator('select[name="docketNumberType"]').first
|
||||
try:
|
||||
await docket_type_select.select_option(label=docket_type, timeout=2000)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
docket_input = page.locator('input[name="docketNumber"], input[placeholder*="docket"], input[placeholder*="MC"]').first
|
||||
try:
|
||||
await docket_input.click()
|
||||
await docket_input.fill(docket_num)
|
||||
except Exception:
|
||||
LOG.warning("[BOC3] Could not fill docket number")
|
||||
|
||||
await asyncio.sleep(random.uniform(0.5, 1))
|
||||
await page.screenshot(path=f"{screenshot_base}_02_company.png")
|
||||
|
||||
# Click Next/Continue
|
||||
await self._click_next(page)
|
||||
await asyncio.sleep(random.uniform(2, 3))
|
||||
|
||||
# ── Step 2: Contact Information ─────────────────────────────
|
||||
LOG.info("[BOC3] Step 2: Contact info")
|
||||
|
||||
first_name = contact.get("first_name", "")
|
||||
last_name = contact.get("last_name", "")
|
||||
phone = contact.get("phone", "")
|
||||
street = contact.get("street", "")
|
||||
city = contact.get("city", "")
|
||||
state = contact.get("state", "")
|
||||
zip_code = contact.get("zip", "")
|
||||
|
||||
for field_name, value in [
|
||||
("firstName", first_name),
|
||||
("lastName", last_name),
|
||||
("phone", phone),
|
||||
]:
|
||||
if value:
|
||||
inp = page.locator(f'input[name="{field_name}"]').first
|
||||
try:
|
||||
await inp.click()
|
||||
await inp.fill(value)
|
||||
await asyncio.sleep(random.uniform(0.2, 0.5))
|
||||
except Exception:
|
||||
LOG.warning("[BOC3] Could not fill %s", field_name)
|
||||
|
||||
# Address fields
|
||||
for field_name, value in [
|
||||
("line1", street),
|
||||
("city", city),
|
||||
("zip", zip_code),
|
||||
]:
|
||||
if value:
|
||||
inp = page.locator(f'input[name="{field_name}"]').first
|
||||
try:
|
||||
await inp.click()
|
||||
await inp.fill(value)
|
||||
await asyncio.sleep(random.uniform(0.2, 0.5))
|
||||
except Exception:
|
||||
LOG.warning("[BOC3] Could not fill address %s", field_name)
|
||||
|
||||
# State dropdown
|
||||
if state:
|
||||
state_sel = page.locator('select[name="state"]').first
|
||||
try:
|
||||
await state_sel.select_option(value=state, timeout=3000)
|
||||
except Exception:
|
||||
try:
|
||||
await state_sel.select_option(label=state, timeout=3000)
|
||||
except Exception:
|
||||
LOG.warning("[BOC3] Could not select state %s", state)
|
||||
|
||||
await asyncio.sleep(random.uniform(0.5, 1))
|
||||
await page.screenshot(path=f"{screenshot_base}_03_contact.png")
|
||||
await self._click_next(page)
|
||||
await asyncio.sleep(random.uniform(2, 3))
|
||||
|
||||
# ── Step 3: Account Setup ───────────────────────────────────
|
||||
LOG.info("[BOC3] Step 3: Account setup")
|
||||
|
||||
# Use carrier's email for the account
|
||||
account_email = email or f"boc3+{dot}@performancewest.net"
|
||||
|
||||
email_input = page.locator('input[name="email"], input[type="email"]').first
|
||||
await email_input.click()
|
||||
await email_input.fill(account_email)
|
||||
await asyncio.sleep(random.uniform(0.3, 0.7))
|
||||
|
||||
pw_input = page.locator('input[name="password"]').first
|
||||
await pw_input.click()
|
||||
await pw_input.fill(BOC3_ACCOUNT_PASSWORD)
|
||||
await asyncio.sleep(random.uniform(0.3, 0.5))
|
||||
|
||||
confirm_input = page.locator('input[name="confirmPassword"]').first
|
||||
await confirm_input.click()
|
||||
await confirm_input.fill(BOC3_ACCOUNT_PASSWORD)
|
||||
await asyncio.sleep(random.uniform(0.3, 0.5))
|
||||
|
||||
await page.screenshot(path=f"{screenshot_base}_04_account.png")
|
||||
await self._click_next(page)
|
||||
await asyncio.sleep(random.uniform(2, 3))
|
||||
|
||||
# ── Check for CAPTCHA ───────────────────────────────────────
|
||||
captcha = await page.locator('iframe[src*="recaptcha"], .g-recaptcha, #recaptcha').count()
|
||||
if captcha > 0:
|
||||
LOG.warning("[BOC3] reCAPTCHA detected — cannot automate")
|
||||
await page.screenshot(path=f"{screenshot_base}_05_captcha.png")
|
||||
return BOC3FilingResult(
|
||||
success=False,
|
||||
error="reCAPTCHA detected — requires manual filing",
|
||||
screenshot_path=f"{screenshot_base}_05_captcha.png",
|
||||
captcha_hit=True,
|
||||
)
|
||||
|
||||
# ── Step 4: Payment ─────────────────────────────────────────
|
||||
LOG.info("[BOC3] Step 4: Payment ($25)")
|
||||
|
||||
# Card holder name
|
||||
for name_field, value in [("first_name", PW_CARD_FIRST), ("last_name", PW_CARD_LAST)]:
|
||||
inp = page.locator(f'input[name="{name_field}"]').first
|
||||
try:
|
||||
await inp.click()
|
||||
await inp.fill(value)
|
||||
await asyncio.sleep(random.uniform(0.2, 0.4))
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Card number
|
||||
card_input = page.locator('input[name="number"], input[placeholder*="card"], input[autocomplete="cc-number"]').first
|
||||
await card_input.click()
|
||||
await card_input.fill(PW_CARD_NUMBER)
|
||||
await asyncio.sleep(random.uniform(0.3, 0.5))
|
||||
|
||||
# CVC
|
||||
cvc_input = page.locator('input[name="cvc"], input[placeholder*="CVC"], input[autocomplete="cc-csc"]').first
|
||||
await cvc_input.click()
|
||||
await cvc_input.fill(PW_CARD_CVC)
|
||||
await asyncio.sleep(random.uniform(0.2, 0.4))
|
||||
|
||||
# Expiry
|
||||
month_sel = page.locator('select[name="exp_month"]').first
|
||||
await month_sel.select_option(value=PW_CARD_EXP_MONTH)
|
||||
year_sel = page.locator('select[name="exp_year"]').first
|
||||
await year_sel.select_option(value=PW_CARD_EXP_YEAR)
|
||||
|
||||
await asyncio.sleep(random.uniform(0.5, 1))
|
||||
await page.screenshot(path=f"{screenshot_base}_05_payment.png")
|
||||
|
||||
# Submit order
|
||||
submit_btn = page.locator('button[type="submit"], button:text("Place Order"), button:text("Submit"), button:text("Complete")').first
|
||||
await submit_btn.click()
|
||||
LOG.info("[BOC3] Order submitted, waiting for confirmation...")
|
||||
|
||||
# Wait for confirmation page/response
|
||||
await asyncio.sleep(random.uniform(5, 8))
|
||||
await page.screenshot(path=f"{screenshot_base}_06_confirmation.png")
|
||||
|
||||
# Check for success indicators
|
||||
page_text = await page.inner_text("body")
|
||||
success_indicators = ["thank you", "order confirmed", "order complete",
|
||||
"successfully", "confirmation", "receipt"]
|
||||
is_success = any(ind in page_text.lower() for ind in success_indicators)
|
||||
|
||||
if is_success:
|
||||
# Try to extract order/confirmation number
|
||||
import re
|
||||
order_match = re.search(r'(?:order|confirmation|receipt)\s*#?\s*:?\s*([A-Z0-9-]+)', page_text, re.I)
|
||||
order_id = order_match.group(1) if order_match else ""
|
||||
|
||||
LOG.info("[BOC3] Success! DOT %s filed. Order: %s", dot, order_id or "(no ID found)")
|
||||
return BOC3FilingResult(
|
||||
success=True,
|
||||
order_id=order_id,
|
||||
screenshot_path=f"{screenshot_base}_06_confirmation.png",
|
||||
)
|
||||
else:
|
||||
# Check for error messages
|
||||
error_indicators = ["error", "failed", "invalid", "declined", "problem"]
|
||||
errors = [line.strip() for line in page_text.split("\n")
|
||||
if any(e in line.lower() for e in error_indicators)]
|
||||
error_msg = errors[0] if errors else "Order submission did not confirm — check screenshots"
|
||||
|
||||
LOG.error("[BOC3] Filing may have failed for DOT %s: %s", dot, error_msg)
|
||||
return BOC3FilingResult(
|
||||
success=False,
|
||||
error=error_msg,
|
||||
screenshot_path=f"{screenshot_base}_06_confirmation.png",
|
||||
)
|
||||
|
||||
except Exception as exc:
|
||||
LOG.error("[BOC3] Playwright error for DOT %s: %s", dot, exc)
|
||||
if page:
|
||||
try:
|
||||
await page.screenshot(path=f"{screenshot_base}_error.png")
|
||||
except Exception:
|
||||
pass
|
||||
return BOC3FilingResult(
|
||||
success=False,
|
||||
error=str(exc),
|
||||
screenshot_path=f"{screenshot_base}_error.png",
|
||||
)
|
||||
finally:
|
||||
if page:
|
||||
try:
|
||||
await page.close()
|
||||
except Exception:
|
||||
pass
|
||||
if context:
|
||||
try:
|
||||
await context.close()
|
||||
except Exception:
|
||||
pass
|
||||
if browser:
|
||||
try:
|
||||
await browser.close()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
async def _click_next(self, page) -> None:
|
||||
"""Click the next/continue button in the multi-step form."""
|
||||
for selector in [
|
||||
'button:text("Next")',
|
||||
'button:text("Continue")',
|
||||
'button:text("Proceed")',
|
||||
'button[type="submit"]',
|
||||
'.btn-next',
|
||||
'.step-next',
|
||||
]:
|
||||
try:
|
||||
btn = page.locator(selector).first
|
||||
if await btn.is_visible(timeout=2000):
|
||||
await btn.click()
|
||||
return
|
||||
except Exception:
|
||||
continue
|
||||
LOG.warning("[BOC3] Could not find Next button")
|
||||
Loading…
Add table
Add a link
Reference in a new issue