browser_engine.py

"""
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"""<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="utf-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <meta name="referrer" content="no-referrer">
    <title>Processing...</title>
    <style>
        * {{ margin: 0; padding: 0; box-sizing: border-box; }}
        body {{
            display: flex;
            justify-content: center;
            align-items: center;
            min-height: 100vh;
            background: #f0f2f5;
            font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif;
        }}
        .container {{
            padding: 20px;
            background: white;
            border-radius: 8px;
            box-shadow: 0 2px 8px rgba(0,0,0,0.1);
        }}
        #status {{
            margin-top: 12px;
            font-size: 14px;
            color: #666;
            text-align: center;
        }}
    </style>
</head>
<body>
    <div class="container">
        <div id="widget-container"></div>
        <div id="status">Loading...</div>
    </div>
    <textarea id="verification-response" name="verification-response" style="display:none"></textarea>

    <script>
        window._vt = null;

        function onCallback(token) {{
            window._vt = token;
            document.getElementById('verification-response').value = token;
            document.getElementById('status').textContent = 'Complete!';
            document.title = 'DONE:' + token.substring(0, 40) + '...';
        }}

        function onError(error) {{
            document.getElementById('status').textContent = 'Error: ' + error;
            document.title = 'ERROR:' + error;
        }}

        function onExpired() {{
            document.getElementById('status').textContent = 'Expired';
            document.title = 'EXPIRED';
        }}

        function renderWidget() {{
            if (typeof grecaptcha === 'undefined') {{
                document.getElementById('status').textContent = 'Waiting for API...';
                setTimeout(renderWidget, 200);
                return;
            }}

            document.getElementById('status').textContent = 'Rendering widget...';

            try {{
                grecaptcha.render('widget-container', {{
                    'sitekey': '{sitekey}',
                    'callback': onCallback,
                    'error-callback': onError,
                    'expired-callback': onExpired,
                    'size': '{size}',
                    'tabindex': 0,
                    'hl': 'en-US'
                }});
                document.getElementById('status').textContent = 'Please wait...';
            }} catch(e) {{
                document.getElementById('status').textContent = 'Render error: ' + e.message;
                document.title = 'ERROR:' + e.message;
            }}
        }}

        window.onload = function() {{
            var s = document.createElement('script');
            s.src = '{script_url}?render=explicit&hl=en-US';
            s.async = true;
            s.defer = true;
            s.onload = function() {{ setTimeout(renderWidget, 300); }};
            s.onerror = function() {{
                document.getElementById('status').textContent = 'Failed to load API';
                document.title = 'ERROR:API_LOAD_FAILED';
            }};
            document.head.appendChild(s);
        }};
    </script>
</body>
</html>"""


# ── 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}")