""" Vision Engine - Browser Automation + Image Classification Pipeline =================================================================== Orchestrates web browser automation with ONNX grid classification. Flow: 1. Generate a local HTML page with embedded verification widget 2. Start local HTTP server for page delivery 3. Launch headless Chromium browser with domain routing 4. Render the widget and wait for user interaction prompt 5. If image grid challenge appears: a. Read the task description (target object label) b. Extract tile images from the grid c. Run ONNX inference for classification d. Select matching tiles e. Submit the response 6. Extract the resulting verification token 7. Cleanup all resources """ import asyncio import io import logging import socket import threading import time from http.server import HTTPServer, BaseHTTPRequestHandler from typing import Optional, List, Tuple import tldextract from PIL import Image from patchright.async_api import async_playwright from grid_classifier import GridClassifier logger = logging.getLogger(__name__) # ── Local HTTP Server ──────────────────────────────────────────────────────── class _PageHandler(BaseHTTPRequestHandler): """Minimal HTTP handler that serves the verification HTML page.""" html_content = "" def do_GET(self): self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() self.wfile.write(self.html_content.encode("utf-8")) def log_message(self, format, *args): pass # Suppress request logs def _find_free_port() -> int: """Find a free TCP port on localhost.""" with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.bind(("127.0.0.1", 0)) return s.getsockname()[1] def _start_http_server(html: str, port: Optional[int] = None) -> Tuple[HTTPServer, int]: """Start a local HTTP server serving the HTML page.""" if port is None: port = _find_free_port() _PageHandler.html_content = html server = HTTPServer(("127.0.0.1", port), _PageHandler) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() logger.info(f"Local HTTP server started on port {port}") return server, port def _stop_http_server(server: HTTPServer): """Stop the local HTTP server.""" try: server.shutdown() except Exception: pass # ── HTML Generation ────────────────────────────────────────────────────────── def generate_widget_html( sitekey: str, url: str, is_enterprise: bool = False, is_invisible: bool = False, ) -> str: """ Generate a self-contained HTML page that renders a verification widget. The callback stores the token in a hidden textarea and updates document.title so the engine can detect completion. """ api_domain = "recaptcha.net" if is_enterprise else "google.com" script_url = f"https://www.{api_domain}/recaptcha/api.js" size = "invisible" if is_invisible else "normal" return f""" Processing...
Loading...
""" # ── Main Vision Engine Class ──────────────────────────────────────────────── class VisionEngine: """ Web automation engine that combines: - Local page rendering (widget generation + local HTTP delivery) - Patchright Chromium browser (automated browser) - ONNX Runtime (grid image classification) """ def __init__(self, models_dir: str = "models"): self.models_dir = models_dir self.classifier = GridClassifier(models_dir) self.loaded = False def load_models(self): """Load ONNX models. Call once at startup.""" self.classifier.load_models() self.loaded = True # ── Public API ─────────────────────────────────────────────────────────── async def process( self, sitekey: str, url: str, proxy: Optional[str] = None, headless: bool = True, timeout: int = 120, is_enterprise: bool = False, is_invisible: bool = False, ) -> Optional[str]: """ Process a verification challenge and return the response token. Args: sitekey: The widget site key url: The page URL (used for domain matching) proxy: Optional proxy URL headless: Run browser in headless mode timeout: Maximum processing time in seconds is_enterprise: Enterprise widget mode is_invisible: Invisible widget mode Returns: The verification response token, or None if processing failed. """ # Extract domain for DNS routing extracted = tldextract.extract(url) domain = f"{extracted.domain}.{extracted.suffix}" www_domain = f"www.{domain}" logger.info(f"Starting process for domain={domain}, sitekey={sitekey[:10]}...") # Generate widget HTML html = generate_widget_html(sitekey, url, is_enterprise, is_invisible) # Start local server server, port = _start_http_server(html) try: # Launch browser token = await self._run_browser( domain=domain, www_domain=www_domain, port=port, proxy=proxy, headless=headless, timeout=timeout, is_invisible=is_invisible, ) return token finally: _stop_http_server(server) logger.info("Local server stopped") # ── Browser Automation ────────────────────────────────────────────────── async def _run_browser( self, domain: str, www_domain: str, port: int, proxy: Optional[str], headless: bool, timeout: int, is_invisible: bool, ) -> Optional[str]: """Launch Chromium browser and orchestrate the processing flow.""" # Build DNS routing: map target domain -> 127.0.0.1 dns_rules = f"MAP {domain} 127.0.0.1,MAP {www_domain} 127.0.0.1" for sub in ["m", "app", "login", "auth", "secure"]: dns_rules += f",MAP {sub}.{domain} 127.0.0.1" launch_args = [ f"--host-resolver-rules={dns_rules}", "--disable-blink-features=AutomationControlled", "--no-sandbox", "--disable-dev-shm-usage", "--disable-gpu", "--disable-software-rasterizer", "--disable-extensions", "--no-first-run", "--no-default-browser-check", "--disable-infobars", ] proxy_config = None if proxy: proxy_config = {"server": proxy} async with async_playwright() as p: logger.info("Launching Chromium browser...") browser = await p.chromium.launch( headless=headless, args=launch_args, proxy=proxy_config, ) context = await browser.new_context( ignore_https_errors=True, user_agent=( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/125.0.0.0 Safari/537.36" ), viewport={"width": 500, "height": 700}, locale="en-US", timezone_id="America/New_York", ) await context.route( "**/*.{png,jpg,jpeg,webp,svg,woff,woff2,ttf,eot}", lambda route: route.abort() if "recaptcha" not in (route.request.url or "") else route.continue_(), ) page = await context.new_page() try: target_url = f"http://{domain}:{port}/" logger.info(f"Navigating to {target_url}") await page.goto(target_url, timeout=30000, wait_until="domcontentloaded") logger.info("Waiting for widget to load...") await self._wait_for_widget_load(page, timeout=30) if is_invisible: token = await self._handle_invisible(page, timeout) else: token = await self._handle_checkbox_flow(page, timeout) return token except Exception as e: logger.error(f"Browser automation error: {e}") try: ss_path = f"/tmp/debug_{int(time.time())}.png" await page.screenshot(path=ss_path) logger.info(f"Debug screenshot saved: {ss_path}") except Exception: pass raise finally: await context.close() await browser.close() logger.info("Browser closed") # ── Widget Interaction ────────────────────────────────────────────────── async def _wait_for_widget_load(self, page, timeout: int = 30): """Wait until the verification widget is rendered.""" start = time.time() while time.time() - start < timeout: title = await page.title() if title.startswith("ERROR"): raise Exception(f"Widget load error: {title}") frames = page.frames for frame in frames: if "anchor" in (frame.url or "") or "recaptcha" in (frame.url or ""): logger.info(f"Widget frame detected: {frame.url[:60]}...") return await asyncio.sleep(0.5) raise TimeoutError("Widget failed to load") async def _handle_checkbox_flow(self, page, timeout: int) -> Optional[str]: """ Handle the checkbox verification flow: 1. Click the checkbox 2. If image challenge appears -> classify and select 3. Extract token """ start = time.time() checkbox_clicked = False while time.time() - start < timeout: token = await self._check_token(page) if token: return token title = await page.title() if title.startswith("ERROR"): raise Exception(f"Widget error: {title}") if not checkbox_clicked: clicked = await self._click_checkbox(page) if clicked: checkbox_clicked = True logger.info("Checkbox clicked, waiting for challenge...") await asyncio.sleep(2.0) continue challenge_frame = self._find_challenge_frame(page) if challenge_frame: logger.info("Image challenge detected, processing...") try: done = await self._process_image_challenge(challenge_frame, page) if done: logger.info("Image challenge processed, waiting for token...") await asyncio.sleep(2.0) continue except Exception as e: logger.warning(f"Image challenge error: {e}") await asyncio.sleep(1.0) await asyncio.sleep(0.5) logger.warning(f"Timeout after {timeout}s") return None async def _handle_invisible(self, page, timeout: int) -> Optional[str]: """Handle invisible verification widget (triggered programmatically).""" start = time.time() while time.time() - start < timeout: token = await self._check_token(page) if token: return token try: await page.evaluate(""" if (typeof grecaptcha !== 'undefined') { grecaptcha.execute(); } """) except Exception: pass challenge_frame = self._find_challenge_frame(page) if challenge_frame: try: await self._process_image_challenge(challenge_frame, page) await asyncio.sleep(2.0) except Exception: pass await asyncio.sleep(1.0) return None async def _check_token(self, page) -> Optional[str]: """Check if a verification token has been generated.""" try: token = await page.evaluate( "document.getElementById('verification-response').value || window._vt || ''" ) if token and len(token) > 20: logger.info(f"Token obtained (length={len(token)})") return str(token) except Exception: pass return None async def _click_checkbox(self, page) -> bool: """Find and click the verification checkbox.""" for frame in page.frames: url = frame.url or "" if "anchor" in url and "recaptcha" in url: try: checkbox = frame.locator("#recaptcha-anchor") count = await checkbox.count() if count > 0: visible = await checkbox.is_visible() if visible: await checkbox.click(timeout=5000) logger.info("Clicked checkbox") return True except Exception as e: logger.debug(f"Checkbox click error: {e}") return False def _find_challenge_frame(self, page): """Find the image challenge frame (bframe).""" for frame in page.frames: url = frame.url or "" if ("bframe" in url or "challenge" in url.lower()) and "recaptcha" in url: return frame return None # ── Image Challenge Processing ────────────────────────────────────────── async def _process_image_challenge(self, frame, page) -> bool: """ Process an image classification challenge: 1. Read task description 2. Determine grid type (3x3 or 4x4) 3. Extract images 4. Run ONNX classification 5. Select matching tiles 6. Submit """ await asyncio.sleep(1.5) # Read task description task_text = await frame.evaluate(""" (() => { const el = document.querySelector('.rc-imageselect-instructions'); if (!el) return ''; const spans = el.querySelectorAll('span'); return Array.from(spans).map(s => s.textContent.trim()).filter(Boolean).join(' '); })() """) if not task_text: logger.warning("Could not read task description") return False logger.info(f"Task: {task_text}") # Determine grid size tile_count = await frame.evaluate(""" document.querySelectorAll('.rc-imageselect-tile').length """) is_hard = await frame.evaluate(""" (() => { const rows = document.querySelectorAll('.rc-imageselect-instructions > span'); return rows.length >= 3; })() """) grid_size = 4 if tile_count == 16 else 3 logger.info(f"Grid: {grid_size}x{grid_size}, tiles={tile_count}, hard={is_hard}") # Extract tile images tiles: List[Image.Image] = [] background_img: Optional[Image.Image] = None if grid_size == 4: bg_src = await frame.evaluate(""" (() => { const table = document.querySelector('.rc-imageselect-table'); if (!table) return null; const img = table.querySelector('img'); return img ? img.src : null; })() """) if bg_src: try: background_img = await self._download_image(page, bg_src) logger.info(f"Background image downloaded: {background_img.size}") except Exception as e: logger.error(f"Failed to download background image: {e}") for i in range(tile_count): try: screenshot_bytes = await frame.evaluate(f""" (() => {{ const tile = document.querySelectorAll('.rc-imageselect-tile')[{i}]; if (!tile) return null; const img = tile.querySelector('img'); if (!img) return null; const canvas = document.createElement('canvas'); canvas.width = img.naturalWidth; canvas.height = img.naturalHeight; const ctx = canvas.getContext('2d'); ctx.drawImage(img, 0, 0); return canvas.toDataURL('image/png'); }})() """) if screenshot_bytes: import base64 b64_data = screenshot_bytes.split(",")[1] img_bytes = base64.b64decode(b64_data) img = Image.open(io.BytesIO(img_bytes)).convert("RGB") tiles.append(img) logger.debug(f"Tile {i}: {img.size}") except Exception as e: logger.warning(f"Failed to extract tile {i}: {e}") tiles.append(Image.new("RGB", (100, 100), (128, 128, 128))) if len(tiles) == 0: logger.error("No tiles extracted") return False # Run ONNX classification selected = self.classifier.classify( task_description=task_text, tiles=tiles, background_image=background_img, is_hard=is_hard, ) logger.info(f"Selected tiles: {selected}") if not selected: logger.info("No tiles selected, reloading challenge...") await self._reload_challenge(frame) return False # Click selected tiles for idx in selected: try: await frame.evaluate(f""" (() => {{ const tile = document.querySelectorAll('.rc-imageselect-tile')[{idx}]; if (tile) {{ const rect = tile.getBoundingClientRect(); const event = new MouseEvent('click', {{ bubbles: true, cancelable: true, clientX: rect.left + rect.width / 2, clientY: rect.top + rect.height / 2, }}); tile.dispatchEvent(event); }} }})() """) await asyncio.sleep(0.3) except Exception as e: logger.warning(f"Failed to select tile {idx}: {e}") await asyncio.sleep(1.0) # Click verify/submit button try: await frame.evaluate(""" (() => { const btn = document.querySelector('#recaptcha-verify-button'); if (btn) btn.click(); })() """) logger.info("Clicked verify button") except Exception as e: logger.warning(f"Failed to click verify: {e}") await asyncio.sleep(1.5) # Check for errors error = await frame.evaluate(""" (() => { const err = document.querySelector('.rc-imageselect-error'); if (!err) return null; return err.textContent.trim(); })() """) if error: logger.warning(f"Challenge error: {error}") if any(kw in error.lower() for kw in ["try again", "select all", "also check"]): await asyncio.sleep(1.0) await self._reload_challenge(frame) return False return True async def _download_image(self, page, url: str) -> Image.Image: """Download an image from a URL using the browser context.""" import base64 b64 = await page.evaluate(f""" (() => {{ return new Promise((resolve, reject) => {{ const img = new Image(); img.crossOrigin = 'anonymous'; img.onload = () => {{ const canvas = document.createElement('canvas'); canvas.width = img.naturalWidth; canvas.height = img.naturalHeight; const ctx = canvas.getContext('2d'); ctx.drawImage(img, 0, 0); resolve(canvas.toDataURL('image/png').split(',')[1]); }}; img.onerror = () => reject(new Error('Image load failed')); img.src = '{url}'; }}); }})() """) img_bytes = base64.b64decode(b64) return Image.open(io.BytesIO(img_bytes)).convert("RGB") async def _reload_challenge(self, frame): """Reload the image challenge (get new images).""" try: await frame.evaluate(""" (() => { const btn = document.querySelector('#recaptcha-reload-button'); if (btn) btn.click(); const reload = document.querySelector('.rc-imageselect-refresh'); if (reload) reload.click(); })() """) logger.info("Challenge reloaded") await asyncio.sleep(2.0) except Exception as e: logger.warning(f"Failed to reload challenge: {e}")