aadf9f5bc1
Adds scripts/workers/services/ucr_playwright.py — a UCR.gov National Registration System automation that, given a USDOT + fleet size, runs the register/pay flow, pays the federal UCR fee with the matched PW filing card (Relay/Stripe Issuing), and captures a confirmation screenshot + number. Conventions match boc3_playwright / fmcsa_web_submitter: dev-mode dry-run guard, undetected (patchright) browser, CAPTCHA detection, screenshot evidence, dataclass result. Safety: verifies the displayed fee against the federal schedule before paying and refuses to auto-charge a surprising amount (UCR_MAX_AUTO_FEE_USD) — falls back to manual filing instead. Wires it into MCS150UpdateHandler: when an approved (admin_approved) order has slug ucr-registration, _file_ucr_registration runs the automation, uploads the confirmation screenshot to MinIO, records filing_status + confirmation, and sets fulfillment_status=completed on success. On CAPTCHA / fee-mismatch / failure it reverts to ready_to_file with a high-priority 'file manually' todo. This replaces the old behavior where approving a UCR just sat at authorization_signed.
474 lines
22 KiB
Python
474 lines
22 KiB
Python
"""UCR Annual Registration — Playwright automation for the National Registration
|
|
System at https://www.ucr.gov.
|
|
|
|
Unified Carrier Registration (UCR) is an annual registration + fee that every
|
|
interstate for-hire carrier, broker, and freight forwarder must pay. There is no
|
|
public API; registration is done through the official National Registration
|
|
System website. This adapter automates the "Register / Pay" flow:
|
|
|
|
1. Start a registration for the given registration year by USDOT number.
|
|
2. Confirm the carrier's company information (pre-filled from the USDOT record).
|
|
3. Set / confirm the fleet size (total power units) — this determines the fee
|
|
bracket per the federal UCR fee schedule.
|
|
4. Pay the UCR fee with the Performance West filing card (Relay/Stripe Issuing,
|
|
matched to the customer's payment method).
|
|
5. Capture the confirmation screenshot + receipt / confirmation number.
|
|
|
|
reCAPTCHA: the site may use a CAPTCHA on the payment step. We launch with the
|
|
undetected (patchright) browser to avoid triggering it; if one appears we abort
|
|
cleanly and surface captcha_hit so the order falls back to manual filing.
|
|
|
|
Conventions mirror boc3_playwright.py / fmcsa_web_submitter.py:
|
|
- dev/test environments NEVER submit (NODE_ENV/ENV != production -> dry run).
|
|
- screenshots written under SCREENSHOTS_DIR for durable evidence.
|
|
- returns a dataclass with success / confirmation_number / screenshot_path.
|
|
|
|
Usage:
|
|
adapter = UCRRegistration()
|
|
result = await adapter.file_ucr({
|
|
"dot_number": "1167703",
|
|
"legal_name": "COMPOUND TECHNOLOGIES INC",
|
|
"power_units": "2",
|
|
"registration_year": 2026, # optional; defaults to current UCR year
|
|
"email": "carrier@example.com",
|
|
"phone": "5551234567",
|
|
"address_street": "72 RIVERSIDE DRIVE",
|
|
"address_city": "CARTERSVILLE",
|
|
"address_state": "GA",
|
|
"address_zip": "30120",
|
|
}, order_number="CO-...", payment_method="card")
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import os
|
|
import random
|
|
import re
|
|
import time
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
LOG = logging.getLogger("workers.services.ucr_playwright")
|
|
SCREENSHOTS_DIR = Path(os.getenv("SCREENSHOTS_DIR", "/tmp/ucr-screenshots"))
|
|
SCREENSHOTS_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
UCR_URL = "https://www.ucr.gov"
|
|
|
|
# Federal UCR fee schedule (2024/2025 plan year) by total power units.
|
|
# (low, high_inclusive, fee_usd). Used to sanity-check the fee the site shows
|
|
# before we authorize payment, so we never overpay on a mis-read bracket.
|
|
UCR_FEE_SCHEDULE = [
|
|
(0, 2, 46.00),
|
|
(3, 5, 138.00),
|
|
(6, 20, 276.00),
|
|
(21, 100, 963.00),
|
|
(101, 1000, 4592.00),
|
|
(1001, 10**9, 44836.00),
|
|
]
|
|
|
|
# Hard ceiling: never authorize a UCR charge above this without a human. A
|
|
# small-fleet carrier should be in the $46-$276 range; anything wildly above
|
|
# this for a 0-20 unit fleet means we mis-read the page.
|
|
UCR_MAX_AUTO_FEE_USD = float(os.getenv("UCR_MAX_AUTO_FEE_USD", "300"))
|
|
|
|
|
|
def expected_ucr_fee(power_units: int) -> float:
|
|
"""Return the federal UCR fee for a given power-unit count (0 if unknown)."""
|
|
for lo, hi, fee in UCR_FEE_SCHEDULE:
|
|
if lo <= power_units <= hi:
|
|
return fee
|
|
return 0.0
|
|
|
|
|
|
def current_ucr_year() -> int:
|
|
"""UCR registration opens in the fall for the NEXT calendar year. Use the
|
|
next year once we're past Oct 1, otherwise the current year."""
|
|
now = datetime.now(timezone.utc)
|
|
return now.year + 1 if now.month >= 10 else now.year
|
|
|
|
|
|
@dataclass
|
|
class UCRFilingResult:
|
|
success: bool
|
|
confirmation_number: str = ""
|
|
fee_paid_usd: float = 0.0
|
|
error: str = ""
|
|
screenshot_path: str = ""
|
|
captcha_hit: bool = False
|
|
dry_run: bool = False
|
|
|
|
|
|
def _load_filing_card(order_number: str, payment_method: str):
|
|
"""Load the PW filing card matching the customer's payment method, mirroring
|
|
boc3_playwright._load_matching_card. Returns a CardDetails or None."""
|
|
try:
|
|
from scripts.workers.relay_integration import load_card_from_erpnext
|
|
card = load_card_from_erpnext(payment_method or "card")
|
|
if card:
|
|
LOG.info("[ucr] Using %s card for %s (customer paid via %s)",
|
|
card.masked(), order_number, payment_method)
|
|
return card
|
|
except Exception as exc: # noqa: BLE001
|
|
LOG.warning("[ucr] Could not load filing card: %s", exc)
|
|
return None
|
|
|
|
|
|
class UCRRegistration:
|
|
"""Automate UCR annual registration + payment through ucr.gov."""
|
|
|
|
async def file_ucr(self, data: dict, order_number: str = "",
|
|
payment_method: str = "card") -> UCRFilingResult:
|
|
dot = str(data.get("dot_number", "")).strip()
|
|
legal_name = data.get("legal_name", "")
|
|
try:
|
|
power_units = int(str(data.get("power_units", "")).strip() or "0")
|
|
except ValueError:
|
|
power_units = 0
|
|
reg_year = int(data.get("registration_year") or current_ucr_year())
|
|
|
|
if not dot:
|
|
return UCRFilingResult(success=False, error="DOT number required for UCR")
|
|
if power_units <= 0:
|
|
return UCRFilingResult(
|
|
success=False,
|
|
error="power_units required to determine UCR fee bracket",
|
|
)
|
|
|
|
expected_fee = expected_ucr_fee(power_units)
|
|
LOG.info("[ucr] DOT %s — %s power units → expected fee $%.2f (year %s)",
|
|
dot, power_units, expected_fee, reg_year)
|
|
|
|
# ── Dev/test guard: never touch the live registration system ─────────
|
|
is_prod = os.getenv("NODE_ENV") == "production" or os.getenv("ENV") == "production"
|
|
if not is_prod:
|
|
LOG.info("[ucr] DEV MODE — skipping live UCR registration for DOT %s", dot)
|
|
return UCRFilingResult(
|
|
success=True, dry_run=True,
|
|
confirmation_number="DEV-SKIP",
|
|
fee_paid_usd=expected_fee,
|
|
error="",
|
|
)
|
|
|
|
card = _load_filing_card(order_number, payment_method)
|
|
if not card or not card.card_number:
|
|
return UCRFilingResult(
|
|
success=False,
|
|
error="No filing card configured (RELAY_FILING_CARD / PW card)",
|
|
)
|
|
|
|
timestamp = int(time.time())
|
|
shot_base = SCREENSHOTS_DIR / f"ucr_{dot}_{timestamp}"
|
|
|
|
try:
|
|
from patchright.async_api import async_playwright
|
|
except ImportError:
|
|
from playwright.async_api import async_playwright
|
|
|
|
try:
|
|
from scripts.workers.services.telecom.undetected_browser import launch_context
|
|
except ImportError:
|
|
launch_context = None
|
|
|
|
browser = context = page = None
|
|
try:
|
|
async with async_playwright() as pw:
|
|
if launch_context:
|
|
browser, context = await launch_context(pw, headless=True)
|
|
else:
|
|
browser = await pw.chromium.launch(headless=True)
|
|
context = await browser.new_context(
|
|
viewport={"width": 1366, "height": 900},
|
|
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"
|
|
),
|
|
locale="en-US",
|
|
timezone_id="America/New_York",
|
|
)
|
|
page = await context.new_page()
|
|
|
|
# ── Step 1: Open the registration flow ───────────────────────
|
|
LOG.info("[ucr] Navigating to %s for DOT %s", UCR_URL, dot)
|
|
await page.goto(UCR_URL, wait_until="domcontentloaded", timeout=45000)
|
|
await asyncio.sleep(random.uniform(2, 4))
|
|
await page.screenshot(path=f"{shot_base}_01_home.png", full_page=True)
|
|
|
|
# Click the primary "Register / File Now" CTA. Try several
|
|
# accessible-name variants the site has used.
|
|
started = await self._click_first(page, [
|
|
'a:has-text("Register Now")', 'a:has-text("File Now")',
|
|
'a:has-text("Begin Registration")', 'button:has-text("Register")',
|
|
'a:has-text("Start")', 'a[href*="register"]',
|
|
])
|
|
if not started:
|
|
return self._fail(page, shot_base, "Could not find the UCR Register/File CTA")
|
|
await asyncio.sleep(random.uniform(2, 4))
|
|
|
|
# ── Step 2: Identify the carrier (USDOT + year) ──────────────
|
|
await self._fill_first(page, [
|
|
'input[name*="dot" i]', 'input[id*="dot" i]',
|
|
'input[placeholder*="USDOT" i]', 'input[placeholder*="DOT" i]',
|
|
], dot)
|
|
# Registration year, when the site asks for it.
|
|
await self._select_first(page, [
|
|
'select[name*="year" i]', 'select[id*="year" i]',
|
|
], str(reg_year))
|
|
await page.screenshot(path=f"{shot_base}_02_identify.png", full_page=True)
|
|
await self._click_first(page, [
|
|
'button:has-text("Continue")', 'button:has-text("Next")',
|
|
'button:has-text("Search")', 'button:has-text("Lookup")',
|
|
'input[type="submit"]',
|
|
])
|
|
await asyncio.sleep(random.uniform(3, 5))
|
|
|
|
# ── CAPTCHA gate ─────────────────────────────────────────────
|
|
if await self._captcha_present(page):
|
|
LOG.warning("[ucr] CAPTCHA detected — aborting for manual filing")
|
|
await page.screenshot(path=f"{shot_base}_03_captcha.png", full_page=True)
|
|
return UCRFilingResult(
|
|
success=False, captcha_hit=True,
|
|
error="CAPTCHA detected — requires manual filing",
|
|
screenshot_path=f"{shot_base}_03_captcha.png",
|
|
)
|
|
|
|
# ── Step 3: Confirm company info + fleet size ────────────────
|
|
# The site pre-fills the carrier record from USDOT; we only set
|
|
# the power-unit count if it asks (it drives the fee bracket).
|
|
await self._fill_first(page, [
|
|
'input[name*="power" i]', 'input[id*="power" i]',
|
|
'input[name*="vehicle" i]', 'input[placeholder*="power unit" i]',
|
|
], str(power_units), required=False)
|
|
await self._fill_first(page, [
|
|
'input[name*="email" i]', 'input[type="email"]',
|
|
], data.get("email", ""), required=False)
|
|
await page.screenshot(path=f"{shot_base}_04_company.png", full_page=True)
|
|
await self._click_first(page, [
|
|
'button:has-text("Continue")', 'button:has-text("Next")',
|
|
'input[type="submit"]',
|
|
])
|
|
await asyncio.sleep(random.uniform(3, 5))
|
|
|
|
# ── Step 4: Verify the fee BEFORE paying ─────────────────────
|
|
shown_fee = await self._read_fee(page)
|
|
if shown_fee is not None:
|
|
LOG.info("[ucr] Site shows fee $%.2f (expected $%.2f)", shown_fee, expected_fee)
|
|
# Refuse to pay a surprising amount for a small fleet.
|
|
if shown_fee > max(expected_fee * 1.5, UCR_MAX_AUTO_FEE_USD):
|
|
await page.screenshot(path=f"{shot_base}_05_fee_mismatch.png", full_page=True)
|
|
return UCRFilingResult(
|
|
success=False,
|
|
error=(f"UCR fee ${shown_fee:.2f} exceeds expected "
|
|
f"${expected_fee:.2f} — held for manual review"),
|
|
screenshot_path=f"{shot_base}_05_fee_mismatch.png",
|
|
)
|
|
fee_to_record = shown_fee if shown_fee is not None else expected_fee
|
|
|
|
# ── Step 5: Payment ──────────────────────────────────────────
|
|
if await self._captcha_present(page):
|
|
await page.screenshot(path=f"{shot_base}_05_captcha_pay.png", full_page=True)
|
|
return UCRFilingResult(
|
|
success=False, captcha_hit=True,
|
|
error="CAPTCHA on payment step — requires manual filing",
|
|
screenshot_path=f"{shot_base}_05_captcha_pay.png",
|
|
)
|
|
|
|
await self._fill_card(page, card)
|
|
await page.screenshot(path=f"{shot_base}_06_payment.png", full_page=True)
|
|
|
|
await self._click_first(page, [
|
|
'button:has-text("Pay")', 'button:has-text("Submit Payment")',
|
|
'button:has-text("Place Order")', 'button:has-text("Complete")',
|
|
'button[type="submit"]', 'input[type="submit"]',
|
|
])
|
|
LOG.info("[ucr] Payment submitted — waiting for confirmation")
|
|
await asyncio.sleep(random.uniform(6, 9))
|
|
try:
|
|
await page.wait_for_load_state("networkidle", timeout=20000)
|
|
except Exception:
|
|
pass
|
|
await page.screenshot(path=f"{shot_base}_07_confirmation.png", full_page=True)
|
|
|
|
# ── Step 6: Parse the confirmation ───────────────────────────
|
|
body = (await page.inner_text("body")).lower()
|
|
ok_words = ["thank you", "confirmation", "successfully", "receipt",
|
|
"registration complete", "payment received", "approved"]
|
|
err_words = ["declined", "error", "failed", "invalid card", "try again"]
|
|
success = any(w in body for w in ok_words) and not any(w in body for w in err_words)
|
|
|
|
conf = self._extract_confirmation(await page.inner_text("body"))
|
|
if success:
|
|
LOG.info("[ucr] SUCCESS DOT %s — conf=%s fee=$%.2f", dot, conf or "(none)", fee_to_record)
|
|
return UCRFilingResult(
|
|
success=True, confirmation_number=conf,
|
|
fee_paid_usd=fee_to_record,
|
|
screenshot_path=f"{shot_base}_07_confirmation.png",
|
|
)
|
|
LOG.error("[ucr] Could not confirm UCR registration for DOT %s", dot)
|
|
return UCRFilingResult(
|
|
success=False,
|
|
error="Could not confirm UCR registration — check screenshot",
|
|
screenshot_path=f"{shot_base}_07_confirmation.png",
|
|
)
|
|
except Exception as exc: # noqa: BLE001
|
|
LOG.error("[ucr] UCR automation failed for DOT %s: %s", dot, exc)
|
|
try:
|
|
if page:
|
|
await page.screenshot(path=f"{shot_base}_99_error.png", full_page=True)
|
|
except Exception:
|
|
pass
|
|
return UCRFilingResult(success=False, error=str(exc),
|
|
screenshot_path=f"{shot_base}_99_error.png")
|
|
finally:
|
|
try:
|
|
if context:
|
|
await context.close()
|
|
if browser:
|
|
await browser.close()
|
|
except Exception:
|
|
pass
|
|
|
|
# ── helpers ──────────────────────────────────────────────────────────────
|
|
async def _click_first(self, page, selectors: list[str]) -> bool:
|
|
for sel in selectors:
|
|
try:
|
|
el = page.locator(sel).first
|
|
if await el.count() and await el.is_visible():
|
|
await el.click(timeout=4000)
|
|
return True
|
|
except Exception:
|
|
continue
|
|
return False
|
|
|
|
async def _fill_first(self, page, selectors: list[str], value: str,
|
|
required: bool = True) -> bool:
|
|
if not value and not required:
|
|
return False
|
|
for sel in selectors:
|
|
try:
|
|
el = page.locator(sel).first
|
|
if await el.count() and await el.is_visible():
|
|
await el.click(timeout=4000)
|
|
await el.fill(value, timeout=4000)
|
|
await asyncio.sleep(random.uniform(0.2, 0.5))
|
|
return True
|
|
except Exception:
|
|
continue
|
|
if required:
|
|
LOG.warning("[ucr] Could not fill any of: %s", selectors)
|
|
return False
|
|
|
|
async def _select_first(self, page, selectors: list[str], value: str) -> bool:
|
|
for sel in selectors:
|
|
try:
|
|
el = page.locator(sel).first
|
|
if await el.count() and await el.is_visible():
|
|
await el.select_option(value=value, timeout=4000)
|
|
return True
|
|
except Exception:
|
|
continue
|
|
return False
|
|
|
|
async def _captcha_present(self, page) -> bool:
|
|
try:
|
|
return bool(await page.locator(
|
|
'iframe[src*="recaptcha"], .g-recaptcha, #recaptcha, '
|
|
'iframe[src*="hcaptcha"], .h-captcha'
|
|
).count())
|
|
except Exception:
|
|
return False
|
|
|
|
async def _read_fee(self, page) -> float | None:
|
|
"""Best-effort scrape of the displayed fee/total dollar amount."""
|
|
try:
|
|
text = await page.inner_text("body")
|
|
except Exception:
|
|
return None
|
|
# Prefer an amount near a fee/total label.
|
|
for pat in (
|
|
r'(?:total|amount due|fee|registration fee)[^$]{0,40}\$\s*([0-9][0-9,]*\.?[0-9]{0,2})',
|
|
r'\$\s*([0-9][0-9,]*\.[0-9]{2})',
|
|
):
|
|
m = re.search(pat, text, re.I)
|
|
if m:
|
|
try:
|
|
return float(m.group(1).replace(",", ""))
|
|
except ValueError:
|
|
continue
|
|
return None
|
|
|
|
def _extract_confirmation(self, text: str) -> str:
|
|
for pat in (
|
|
r'(?:confirmation|registration|receipt|transaction)\s*(?:number|no\.?|#)\s*:?\s*([A-Z0-9-]{4,})',
|
|
r'\b(UCR[-\s]?[A-Z0-9]{4,})\b',
|
|
):
|
|
m = re.search(pat, text, re.I)
|
|
if m:
|
|
return m.group(1).strip()
|
|
return ""
|
|
|
|
async def _fill_card(self, page, card) -> None:
|
|
"""Fill the payment fields. Tries common name/autocomplete attributes
|
|
and handles split or single expiry fields."""
|
|
await self._fill_first(page, [
|
|
'input[autocomplete="cc-number"]', 'input[name*="card" i][name*="num" i]',
|
|
'input[name="number"]', 'input[placeholder*="card number" i]',
|
|
], card.card_number, required=False)
|
|
# Cardholder name
|
|
await self._fill_first(page, [
|
|
'input[autocomplete="cc-name"]', 'input[name*="cardholder" i]',
|
|
'input[name*="name_on_card" i]', 'input[placeholder*="name on card" i]',
|
|
], card.cardholder_name, required=False)
|
|
# CVV
|
|
await self._fill_first(page, [
|
|
'input[autocomplete="cc-csc"]', 'input[name*="cvc" i]',
|
|
'input[name*="cvv" i]', 'input[placeholder*="CVC" i]', 'input[placeholder*="CVV" i]',
|
|
], card.cvv, required=False)
|
|
# Expiry: try a single MM/YY field first, then split selects/inputs.
|
|
single = await self._fill_first(page, [
|
|
'input[autocomplete="cc-exp"]', 'input[placeholder*="MM/YY" i]',
|
|
'input[name*="exp" i][name*="date" i]',
|
|
], f"{card.exp_month}/{str(card.exp_year)[-2:]}", required=False)
|
|
if not single:
|
|
await self._select_first(page, [
|
|
'select[autocomplete="cc-exp-month"]', 'select[name*="exp" i][name*="month" i]',
|
|
], card.exp_month) or await self._fill_first(page, [
|
|
'input[name*="exp" i][name*="month" i]',
|
|
], card.exp_month, required=False)
|
|
yr_full = str(card.exp_year) if len(str(card.exp_year)) == 4 else f"20{card.exp_year}"
|
|
await self._select_first(page, [
|
|
'select[autocomplete="cc-exp-year"]', 'select[name*="exp" i][name*="year" i]',
|
|
], yr_full) or await self._fill_first(page, [
|
|
'input[name*="exp" i][name*="year" i]',
|
|
], yr_full, required=False)
|
|
# Billing ZIP (often required by the payment processor).
|
|
if card.billing_zip:
|
|
await self._fill_first(page, [
|
|
'input[autocomplete="postal-code"]', 'input[name*="zip" i]',
|
|
'input[name*="postal" i]',
|
|
], card.billing_zip, required=False)
|
|
|
|
def _fail(self, page, shot_base, msg: str) -> UCRFilingResult:
|
|
LOG.error("[ucr] %s", msg)
|
|
return UCRFilingResult(success=False, error=msg,
|
|
screenshot_path=f"{shot_base}_fail.png")
|
|
|
|
|
|
async def _selftest():
|
|
"""Dry-run self test (no live submission unless NODE_ENV=production)."""
|
|
res = await UCRRegistration().file_ucr({
|
|
"dot_number": "1167703",
|
|
"legal_name": "COMPOUND TECHNOLOGIES INC",
|
|
"power_units": "2",
|
|
"email": "test@performancewest.net",
|
|
"address_state": "GA",
|
|
}, order_number="CO-SELFTEST", payment_method="card")
|
|
print("dry_run:", res.dry_run, "success:", res.success,
|
|
"fee:", res.fee_paid_usd, "conf:", res.confirmation_number, "err:", res.error)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(_selftest())
|