diff --git a/scripts/workers/services/mcs150_update.py b/scripts/workers/services/mcs150_update.py index 689a6b1..319b76d 100644 --- a/scripts/workers/services/mcs150_update.py +++ b/scripts/workers/services/mcs150_update.py @@ -338,6 +338,15 @@ class MCS150UpdateHandler: order_number) return [minio_path] if minio_path else [] + # Step 4c: ADMIN-ASSISTED AUTO-FILING. Some admin-assisted services have + # a real automated filing path even though they don't produce an MCS-150 + # form. Once an admin has approved (admin_approved=True), run the + # automation here instead of just creating a manual todo. Currently: + # - ucr-registration -> ucr.gov National Registration System + if slug == "ucr-registration": + return self._file_ucr_registration( + order_number, entity_name, dot_number, intake, customer_email) + # Step 5: Submit electronically (3x web → fax fallback) # GUARD: Skip actual submission in dev/test environments is_production = os.environ.get("NODE_ENV") == "production" or os.environ.get("ENV") == "production" @@ -861,6 +870,169 @@ class MCS150UpdateHandler: except Exception as exc: # noqa: BLE001 LOG.warning("[%s] Failed to mark intake validated: %s", order_number, exc) + def _file_ucr_registration(self, order_number, entity_name, dot_number, + intake, customer_email) -> list: + """Run the UCR.gov Playwright automation for an approved UCR order, then + persist the result. On success -> completed (+ confirmation + screenshot + evidence). On CAPTCHA / fee-mismatch / failure -> ready_to_file with a + high-priority 'file manually' todo so a human takes over.""" + import asyncio + LOG.info("[%s] Auto-filing UCR registration via ucr.gov", order_number) + + # Resolve the customer's payment method so we charge the right card. + payment_method = "card" + try: + import psycopg2 + conn = psycopg2.connect(os.environ.get("DATABASE_URL", "")) + with conn.cursor() as cur: + cur.execute("SELECT payment_method FROM compliance_orders WHERE order_number=%s", + (order_number,)) + row = cur.fetchone() + if row and row[0]: + payment_method = row[0] + conn.close() + except Exception as exc: # noqa: BLE001 + LOG.warning("[%s] Could not read payment_method (default card): %s", order_number, exc) + + try: + from scripts.workers.services.ucr_playwright import UCRRegistration + except Exception as exc: # noqa: BLE001 + LOG.error("[%s] UCR adapter import failed: %s", order_number, exc) + self._set_fulfillment_status(order_number, "ready_to_file") + self._create_admin_review_todo( + order_number, entity_name, dot_number, "ucr-registration", None, + customer_email, client_signed=False) + return [] + + data = { + "dot_number": dot_number, + "legal_name": intake.get("legal_name", entity_name), + "power_units": intake.get("power_units", ""), + "email": intake.get("email") or customer_email, + "phone": intake.get("phone", ""), + "address_street": intake.get("address_street", ""), + "address_city": intake.get("address_city", ""), + "address_state": intake.get("address_state", ""), + "address_zip": intake.get("address_zip", ""), + } + try: + adapter = UCRRegistration() + loop = asyncio.new_event_loop() + try: + result = loop.run_until_complete( + adapter.file_ucr(data, order_number=order_number, payment_method=payment_method)) + finally: + loop.close() + except Exception as exc: # noqa: BLE001 + LOG.error("[%s] UCR automation crashed: %s", order_number, exc) + self._set_fulfillment_status(order_number, "ready_to_file") + self._create_admin_review_todo( + order_number, entity_name, dot_number, "ucr-registration", None, + customer_email, client_signed=False) + return [] + + # Upload the confirmation screenshot to MinIO as durable evidence. + evidence = {} + if result.screenshot_path and os.path.exists(result.screenshot_path): + try: + from minio import Minio + mc = Minio( + f"{os.environ.get('MINIO_ENDPOINT', 'minio')}:{os.environ.get('MINIO_PORT', '9000')}", + access_key=os.environ.get("MINIO_ACCESS_KEY", ""), + secret_key=os.environ.get("MINIO_SECRET_KEY", ""), + secure=os.environ.get("MINIO_SECURE", "false").lower() == "true", + ) + bucket = os.environ.get("MINIO_BUCKET", "performancewest") + key = f"filings/ucr-registration/{order_number}/evidence/ucr_confirmation.png" + mc.fput_object(bucket, key, result.screenshot_path, content_type="image/png") + evidence["confirmation_screenshot"] = key + except Exception as exc: # noqa: BLE001 + LOG.warning("[%s] Could not upload UCR screenshot: %s", order_number, exc) + + # Persist filing_status on the order. + filing_status = { + "filing_method": "ucr_gov_web", + "filing_success": bool(result.success), + "submitted_at": datetime.now(timezone.utc).isoformat() if result.success else None, + "manual_confirmation": result.confirmation_number or None, + "fee_paid_usd": result.fee_paid_usd or None, + "dry_run": bool(result.dry_run), + "evidence": evidence, + "error": result.error or None, + } + try: + import psycopg2 + conn = psycopg2.connect(os.environ.get("DATABASE_URL", "")) + with conn.cursor() as cur: + cur.execute( + "UPDATE compliance_orders SET intake_data = jsonb_set(" + "COALESCE(intake_data,'{}'::jsonb), '{filing_status}', %s::jsonb), " + "updated_at=now() WHERE order_number=%s", + (json.dumps(filing_status), order_number), + ) + conn.commit() + conn.close() + except Exception as exc: # noqa: BLE001 + LOG.error("[%s] Failed to persist UCR filing_status: %s", order_number, exc) + + svc_label = self._service_label("ucr-registration") + if result.success: + self._set_fulfillment_status(order_number, "completed") + conf = result.confirmation_number or "(see receipt screenshot)" + self._notify_todo( + order_number, "ucr-registration", + f"{svc_label} FILED — {entity_name} (DOT {dot_number})", + "low", + (f"{svc_label} for {entity_name} (DOT {dot_number}).\n" + f"Status: FILED on ucr.gov{' (DEV dry-run)' if result.dry_run else ''}.\n" + f"Confirmation: {conf}\n" + f"Fee paid: ${result.fee_paid_usd:.2f}\n" + f"Customer: {customer_email}"), + ) + self._send_status_email(order_number, entity_name, dot_number, customer_email, "ucr-registration") + LOG.info("[%s] UCR filed (completed). conf=%s", order_number, conf) + else: + # Could not auto-file (CAPTCHA, fee mismatch, error) -> back to the + # manual-filing queue with a clear, high-priority todo. + self._set_fulfillment_status(order_number, "ready_to_file") + reason = "CAPTCHA — needs manual filing" if result.captcha_hit else (result.error or "automation could not confirm") + self._notify_todo( + order_number, "ucr-registration", + f"{svc_label} — FILE MANUALLY — {entity_name} (DOT {dot_number})", + "high", + (f"{svc_label} for {entity_name} (DOT {dot_number}).\n" + f"Status: AUTO-FILE FAILED — {reason}.\n" + f"ACTION: file manually on ucr.gov, then mark the order completed " + f"in the admin dashboard (enter the confirmation #).\n" + f"Customer: {customer_email}"), + ) + LOG.warning("[%s] UCR auto-file failed: %s", order_number, reason) + return [] + + def _notify_todo(self, order_number, slug, title, priority, description): + """Create an admin_todos row + Telegram fulfillment notification.""" + try: + import psycopg2 + conn = psycopg2.connect(os.environ.get("DATABASE_URL", "")) + with conn.cursor() as cur: + cur.execute( + """INSERT INTO admin_todos (title, category, priority, order_number, + service_slug, description, data, status) + VALUES (%s, 'filing', %s, %s, %s, %s, %s, 'pending')""", + (title, priority, order_number, slug, description, + json.dumps({"order_number": order_number})), + ) + conn.commit() + conn.close() + except Exception as exc: # noqa: BLE001 + LOG.error("[%s] Failed to create todo: %s", order_number, exc) + try: + notify_fulfillment_todo(title=title, order_number=order_number, + service_slug=slug, priority=priority, + description=description) + except Exception as exc: # noqa: BLE001 + LOG.warning("[%s] Telegram notify failed: %s", order_number, exc) + def _create_admin_review_todo(self, order_number, entity_name, dot_number, slug, minio_path, customer_email, client_signed): """High-priority admin todo: verify the prepared filing BEFORE submission. diff --git a/scripts/workers/services/ucr_playwright.py b/scripts/workers/services/ucr_playwright.py new file mode 100644 index 0000000..4848684 --- /dev/null +++ b/scripts/workers/services/ucr_playwright.py @@ -0,0 +1,474 @@ +"""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())