501 lines
20 KiB
Python
501 lines
20 KiB
Python
"""
|
|
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 ERPNext Sensitive ID at runtime, fallback to env vars
|
|
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", "")
|
|
|
|
|
|
def _load_matching_card(order_number: str) -> dict:
|
|
"""Load PW company card matching the customer's payment method.
|
|
|
|
We have 3 company cards tied to payment processors:
|
|
- PW-STRIPE → used when customer paid via card or Klarna
|
|
- PW-PAYPAL → used when customer paid via PayPal
|
|
- PW-CRYPTO → used when customer paid via crypto
|
|
|
|
Cards stored in ERPNext Sensitive ID documents.
|
|
Falls back to env vars if ERPNext lookup fails.
|
|
"""
|
|
card = {
|
|
"first_name": PW_CARD_FIRST, "last_name": PW_CARD_LAST,
|
|
"number": PW_CARD_NUMBER, "cvc": PW_CARD_CVC,
|
|
"exp_month": PW_CARD_EXP_MONTH, "exp_year": PW_CARD_EXP_YEAR,
|
|
}
|
|
|
|
try:
|
|
import psycopg2
|
|
conn = psycopg2.connect(os.environ.get("DATABASE_URL", ""))
|
|
cur = conn.cursor()
|
|
cur.execute(
|
|
"SELECT payment_method FROM compliance_orders WHERE order_number = %s",
|
|
(order_number,),
|
|
)
|
|
row = cur.fetchone()
|
|
conn.close()
|
|
payment_method = (row[0] or "card") if row else "card"
|
|
|
|
# Map customer payment method → our company card
|
|
card_map = {
|
|
"card": "PW-STRIPE",
|
|
"klarna": "PW-STRIPE",
|
|
"ach": "PW-STRIPE",
|
|
"paypal": "PW-PAYPAL",
|
|
"crypto": "PW-CRYPTO",
|
|
}
|
|
card_name = card_map.get(payment_method, "PW-STRIPE")
|
|
|
|
from scripts.workers.erpnext_client import ERPNextClient
|
|
erp = ERPNextClient()
|
|
doc = erp.get_resource("Sensitive ID", card_name)
|
|
|
|
card = {
|
|
"first_name": doc.get("custom_first_name", PW_CARD_FIRST),
|
|
"last_name": doc.get("custom_last_name", PW_CARD_LAST),
|
|
"number": doc.get("custom_card_number", ""),
|
|
"cvc": doc.get("custom_cvc", ""),
|
|
"exp_month": doc.get("custom_exp_month", ""),
|
|
"exp_year": doc.get("custom_exp_year", ""),
|
|
}
|
|
LOG.info("[boc3] Using %s for order %s (customer paid via %s)",
|
|
card_name, order_number, payment_method)
|
|
|
|
except Exception as exc:
|
|
LOG.warning("[boc3] Card lookup failed for %s: %s (using env fallback)", order_number, exc)
|
|
|
|
return card
|
|
|
|
# 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")
|