backend/browser_controller.py

import asyncio
import subprocess
import os
import logging
import json
import base64
from typing import Optional, Dict, List, Any, Tuple
import hashlib
from dataclasses import dataclass, asdict
from pydantic import BaseModel
from pathlib import Path

# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

from playwright.async_api import async_playwright, Page, CDPSession

@dataclass
class ElementInfo:
    """DOM element information compatible with browser-use"""
    index: int
    id: str
    tag_name: str
    xpath: str
    css_selector: str
    text: str
    attributes: Dict[str, str]
    is_clickable: bool
    is_input: bool
    is_visible: bool = True
    is_in_viewport: bool = True
    input_type: Optional[str] = None
    placeholder: Optional[str] = None
    bounding_box: Optional[Dict[str, float]] = None
    center_coordinates: Optional[Dict[str, float]] = None
    viewport_coordinates: Optional[Dict[str, float]] = None

class PageState:
    """Page state compatible with browser-use"""
    def __init__(self, url: str, title: str, elements: List[ElementInfo], selector_map: Dict[int, ElementInfo], screenshot: Optional[str] = None):
        self.url = url
        self.title = title
        self.elements = elements
        self.selector_map = selector_map
        self.screenshot = screenshot
        self.clickable_elements = [e for e in elements if e.is_clickable]
        self.input_elements = [e for e in elements if e.is_input]

class BrowserController:
    def __init__(self, headless: bool, proxy: dict | None, enable_streaming: bool = False):
        self.headless = headless
        self.proxy = proxy
        self.enable_streaming = enable_streaming
        self.play = None
        self.browser = None
        self.page = None
        self.cdp_session = None
        self.streaming_active = False
        self.stream_clients = set()
        self._cached_page_state = None
        self._cached_url = None
        self._last_action_timestamp = None
        self.input_enabled = False  # Track if Input domain is available
        self._original_display = os.environ.get("DISPLAY")
        self._display_was_set = False
        self._xvfb_process: subprocess.Popen | None = None
        self._xvfb_display: str | None = None

        # Load the robust DOM extraction JavaScript
        self.dom_js = self._get_dom_extraction_js()

    def _find_free_display(self, start: int = 99, end: int = 110) -> int:
        """Locate a free X display number for Xvfb."""
        for display in range(start, end):
            lock_file = Path(f"/tmp/.X{display}-lock")
            if not lock_file.exists():
                return display
        # Fall back to the starting display even if locked (Xvfb will fail clearly)
        return start

    def _terminate_xvfb(self):
        """Stop the Xvfb process if it was started."""
        if not self._xvfb_process:
            return

        self._xvfb_process.terminate()
        try:
            self._xvfb_process.wait(timeout=5)
        except subprocess.TimeoutExpired:
            self._xvfb_process.kill()
        finally:
            self._xvfb_process = None
            self._xvfb_display = None

    async def _ensure_display(self):
        """Start a virtual X server when running in headful mode without DISPLAY."""
        if self.headless or os.environ.get("DISPLAY"):
            return

        display_number = self._find_free_display()
        display = f":{display_number}"
        xvfb_cmd = [
            "Xvfb",
            display,
            "-screen",
            "0",
            "1280x800x24",
            "-nolisten",
            "tcp",
        ]

        try:
            self._xvfb_process = subprocess.Popen(
                xvfb_cmd,
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
            )
            self._xvfb_display = display
            logger.info("🖥️ Started Xvfb on display %s for headful browser session", display)
        except FileNotFoundError:
            logger.warning("⚠️ Xvfb not available; falling back to headless mode")
            self.headless = True
            return

        # Wait briefly for Xvfb to be ready
        ready = False
        for _ in range(30):
            if self._xvfb_process.poll() is not None:
                logger.error(
                    "❌ Xvfb exited prematurely with code %s", self._xvfb_process.returncode
                )
                self._terminate_xvfb()
                self.headless = True
                return
            if Path(f"/tmp/.X{display_number}-lock").exists():
                ready = True
                break
            await asyncio.sleep(0.1)

        if not ready:
            logger.warning("⚠️ Timed out waiting for Xvfb; falling back to headless mode")
            self._terminate_xvfb()
            self.headless = True
            return

        os.environ["DISPLAY"] = display
        self._display_was_set = True

    def _restore_display(self):
        """Restore the DISPLAY environment variable and stop Xvfb if needed."""
        self._terminate_xvfb()

        if not self._display_was_set:
            return

        if self._original_display is None:
            os.environ.pop("DISPLAY", None)
        else:
            os.environ["DISPLAY"] = self._original_display

        self._display_was_set = False

    async def __aenter__(self):
        """Initialize browser with CDP streaming support"""
        await self._ensure_display()

        self.play = await async_playwright().start()

        launch_options = {
            "headless": self.headless,
            "args": [
                "--no-sandbox",
                "--disable-dev-shm-usage",
                "--disable-gpu",
                "--disable-web-security",
                "--disable-features=VizDisplayCompositor",
                "--window-size=1280,800",
                "--window-position=0,0",
                "--disable-blink-features=AutomationControlled",
                "--disable-extensions",
                "--no-first-run",
                "--disable-default-apps",
                # Enable remote debugging for CDP
                "--remote-debugging-port=0"  # Use random port
            ]
        }
        
        if self.proxy:
            launch_options["proxy"] = self.proxy
            
        self.browser = await self.play.chromium.launch(**launch_options)
        self.page = await self.browser.new_page(viewport={"width": 1280, "height": 800})
        
        # Set up CDP session for streaming
        if self.enable_streaming:
            await self._setup_cdp_streaming()
            
        await self.page.set_extra_http_headers({
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
        })
        
        return self

    async def __aexit__(self, exc_type, exc, tb):
        """Cleanup browser and CDP session"""
        if self.streaming_active:
            await self._stop_cdp_streaming()
        if self.browser:
            await self.browser.close()
        if self.play:
            await self.play.stop()
        self._restore_display()

    async def _setup_cdp_streaming(self):
        """Setup CDP session for real-time streaming with proper error handling"""
        try:
            # Get CDP session - ensure it's for the page target
            self.cdp_session = await self.page.context.new_cdp_session(self.page)
            
            # Enable required domains with error handling
            await self._enable_cdp_domain('Runtime')
            await self._enable_cdp_domain('Page')
            
            # Try to enable Input domain (optional)
            self.input_enabled = await self._enable_cdp_domain('Input', optional=True)
            
            if self.input_enabled:
                logger.info("✅ Input domain enabled - full interaction support available")
            else:
                logger.warning("⚠️ Input domain not available - using Playwright for interactions")
            
            logger.info("✅ CDP session established for streaming")
            
        except Exception as e:
            logger.error(f"❌ Failed to setup CDP streaming: {e}")
            raise

    async def _enable_cdp_domain(self, domain: str, optional: bool = False) -> bool:
        """Enable a CDP domain with proper error handling"""
        try:
            await self.cdp_session.send(f'{domain}.enable')
            logger.info(f"✅ {domain} domain enabled")
            return True
        except Exception as e:
            if optional:
                logger.warning(f"⚠️ {domain} domain not available: {e}")
                return False
            else:
                logger.error(f"❌ Required {domain} domain failed: {e}")
                raise

    async def start_streaming(self, quality: int = 80):
        """Start CDP screencast streaming with enhanced error handling"""
        if not self.cdp_session:
            raise RuntimeError("CDP session not initialized")
            
        try:
            # Check if Page.startScreencast is available
            await self.cdp_session.send('Page.startScreencast', {
                'format': 'jpeg',
                'quality': quality,
                'maxWidth': 1280,
                'maxHeight': 800,
                'everyNthFrame': 1  # Stream every frame for real-time
            })
            
            # Set up frame listener
            self.cdp_session.on('Page.screencastFrame', self._handle_screencast_frame)
            
            self.streaming_active = True
            logger.info("🎥 CDP streaming started successfully")
            
        except Exception as e:
            logger.error(f"❌ Failed to start CDP streaming: {e}")
            # Try alternative approach with screenshots
            await self._start_screenshot_polling()

    async def _start_screenshot_polling(self):
        """Fallback: Use screenshot polling if screencast not available"""
        logger.info("🔄 Starting screenshot polling as fallback")
        self.streaming_active = True
        
        async def screenshot_loop():
            while self.streaming_active:
                try:
                    screenshot_bytes = await self.page.screenshot(type='jpeg', quality=80)
                    screenshot_b64 = base64.b64encode(screenshot_bytes).decode('utf-8')
                    
                    frame_data = {
                        'type': 'frame',
                        'data': screenshot_b64,
                        'timestamp': asyncio.get_event_loop().time(),
                        'method': 'polling'
                    }
                    
                    await self._broadcast_to_clients(frame_data)
                    await asyncio.sleep(0.1)  # 10 FPS
                    
                except Exception as e:
                    logger.error(f"Screenshot polling error: {e}")
                    await asyncio.sleep(1)
        
        # Start screenshot polling in background
        asyncio.create_task(screenshot_loop())

    async def stop_streaming(self):
        """Stop CDP screencast streaming"""
        if self.cdp_session and self.streaming_active:
            try:
                await self.cdp_session.send('Page.stopScreencast')
                logger.info("🛑 CDP streaming stopped")
            except Exception as e:
                logger.warning(f"⚠️ Error stopping screencast (may not have been active): {e}")
            finally:
                self.streaming_active = False

    async def _stop_cdp_streaming(self):
        """Internal cleanup for CDP streaming"""
        await self.stop_streaming()
        if self.cdp_session:
            try:
                await self.cdp_session.detach()
            except Exception as e:
                logger.warning(f"⚠️ Error detaching CDP session: {e}")

    async def _handle_screencast_frame(self, params):
        """Handle incoming screencast frames"""
        try:
            # Acknowledge frame immediately
            await self.cdp_session.send('Page.screencastFrameAck', {
                'sessionId': params['sessionId']
            })
            
            # Broadcast frame to all connected clients
            frame_data = {
                'type': 'frame',
                'data': params['data'],  # Base64 encoded JPEG
                'timestamp': params.get('timestamp'),
                'method': 'screencast',
                'metadata': {
                    'sessionId': params['sessionId']
                }
            }
            
            # Send to all connected streaming clients
            await self._broadcast_to_clients(frame_data)
            
        except Exception as e:
            logger.error(f"❌ Error handling screencast frame: {e}")

    async def _broadcast_to_clients(self, data):
        """Broadcast data to all connected streaming clients"""
        if not self.stream_clients:
            return
            
        disconnected_clients = []
        for client in self.stream_clients:
            try:
                await client.send_text(json.dumps(data))
            except Exception:
                disconnected_clients.append(client)
                
        # Remove disconnected clients
        for client in disconnected_clients:
            self.stream_clients.discard(client)

    def add_stream_client(self, websocket):
        """Add a new streaming client"""
        self.stream_clients.add(websocket)
        logger.info(f"🔗 Stream client connected. Total clients: {len(self.stream_clients)}")

    def remove_stream_client(self, websocket):
        """Remove a streaming client"""
        self.stream_clients.discard(websocket)
        logger.info(f"🔌 Stream client disconnected. Total clients: {len(self.stream_clients)}")

    async def handle_mouse_event(self, event_data):
        """Handle mouse events with fallback support"""
        try:
            if self.input_enabled and self.cdp_session:
                # Use CDP Input domain if available
                await self.cdp_session.send('Input.dispatchMouseEvent', {
                    'type': event_data['eventType'],
                    'x': event_data['x'],
                    'y': event_data['y'],
                    'button': event_data.get('button', 'left'),
                    'clickCount': event_data.get('clickCount', 1)
                })
            else:
                # Fallback to Playwright mouse actions
                if event_data['eventType'] == 'mousePressed':
                    await self.page.mouse.click(event_data['x'], event_data['y'])
                elif event_data['eventType'] == 'mouseMoved':
                    await self.page.mouse.move(event_data['x'], event_data['y'])
                    
        except Exception as e:
            logger.error(f"❌ Error handling mouse event: {e}")

    async def handle_keyboard_event(self, event_data):
        """Handle keyboard events with fallback support"""
        try:
            if self.input_enabled and self.cdp_session:
                # Use CDP Input domain if available
                await self.cdp_session.send('Input.dispatchKeyEvent', {
                    'type': event_data['eventType'],
                    'text': event_data.get('text', ''),
                    'key': event_data.get('key', ''),
                    'code': event_data.get('code', ''),
                    'keyCode': event_data.get('keyCode', 0)
                })
            else:
                # Fallback to Playwright keyboard actions
                if event_data['eventType'] == 'keyDown' and event_data.get('key'):
                    await self.page.keyboard.press(event_data['key'])
                elif event_data.get('text'):
                    await self.page.keyboard.type(event_data['text'])
                    
        except Exception as e:
            logger.error(f"❌ Error handling keyboard event: {e}")

    def get_streaming_info(self):
        """Get streaming connection information"""
        if self.enable_streaming:
            return {
                "enabled": True,
                "active": self.streaming_active,
                "clients": len(self.stream_clients),
                "websocket_url": "ws://localhost:8000/stream",
                "input_enabled": self.input_enabled,
                "method": "screencast" if self.input_enabled else "polling"
            }
        return {"enabled": False}

    # Keep all your existing methods from the original code
    def _get_dom_extraction_js(self) -> str:
        """Get the robust DOM extraction JavaScript similar to browser-use"""
        return """
        (args) => {
            const { doHighlightElements = true, debugMode = false } = args || {};
            
            // Performance tracking
            const startTime = performance.now();
            let nodeCount = 0;
            let processedCount = 0;
            
            // Results
            const elementMap = new Map();
            const selectorMap = {};
            let highlightIndex = 0;
            
            // Helper functions
            function getClassName(element) {
                if (!element.className) return '';
                if (typeof element.className === 'string') return element.className;
                if (element.className.toString) return element.className.toString();
                if (element.classList && element.classList.length > 0) {
                    return Array.from(element.classList).join(' ');
                }
                return '';
            }
            
            function isInteractive(element) {
                const tagName = element.tagName.toLowerCase();
                const interactiveTags = ['a', 'button', 'input', 'select', 'textarea', 'label'];
                if (interactiveTags.includes(tagName)) return true;
                if (element.onclick || element.getAttribute('onclick')) return true;
                if (element.getAttribute('role') === 'button') return true;
                if (element.getAttribute('role') === 'link') return true;
                if (element.hasAttribute('tabindex')) return true;
                if (element.contentEditable === 'true') return true;
                const style = window.getComputedStyle(element);
                if (style.cursor === 'pointer') return true;
                return false;
            }
            
            function isInput(element) {
                const tagName = element.tagName.toLowerCase();
                return ['input', 'textarea', 'select'].includes(tagName) ||
                       element.contentEditable === 'true';
            }
            
            function getTextContent(element) {
                let text = '';
                if (element.textContent) {
                    text = element.textContent.trim();
                }
                if (element.value) {
                    text = element.value;
                } else if (element.placeholder) {
                    text = element.placeholder;
                }
                if (element.tagName === 'IMG' && element.alt) {
                    text = element.alt;
                }
                return text.substring(0, 200);
            }
            
            function isVisibleAndInViewport(element) {
                const rect = element.getBoundingClientRect();
                const style = window.getComputedStyle(element);
                const hasDimensions = rect.width > 0 && rect.height > 0;
                const isVisible = style.visibility !== 'hidden' &&
                                style.display !== 'none' &&
                                style.opacity !== '0';
                const isInViewport = rect.top < window.innerHeight &&
                                   rect.bottom > 0 &&
                                   rect.left < window.innerWidth &&
                                   rect.right > 0;
                return hasDimensions && isVisible && isInViewport;
            }
            
            // Process elements
            const allElements = document.querySelectorAll('*');
            const elements = [];
            
            allElements.forEach(element => {
                nodeCount++;
                if (!element || element.nodeType !== 1) return;
                
                const isElementVisible = isVisibleAndInViewport(element);
                const isElementInteractive = isInteractive(element);
                const isElementInput = isInput(element);
                
                if (!isElementVisible && !isElementInteractive) return;
                
                processedCount++;
                const rect = element.getBoundingClientRect();
                const elementId = `element_${processedCount}`;
                let currentHighlightIndex = null;
                
                if (isElementInteractive || isElementInput) {
                    currentHighlightIndex = highlightIndex++;
                    
                    if (doHighlightElements) {
                        element.style.outline = '2px solid red';
                        element.style.outlineOffset = '1px';
                        
                        const label = document.createElement('div');
                        label.textContent = currentHighlightIndex.toString();
                        label.style.cssText = `
                            position: absolute;
                            top: ${rect.top + window.scrollY - 20}px;
                            left: ${rect.left + window.scrollX}px;
                            background: red;
                            color: white;
                            padding: 2px 6px;
                            font-size: 12px;
                            font-weight: bold;
                            z-index: 10000;
                            border-radius: 3px;
                            pointer-events: none;
                        `;
                        document.body.appendChild(label);
                    }
                }
                
                const elementData = {
                    index: currentHighlightIndex,
                    id: elementId,
                    tagName: element.tagName.toLowerCase(),
                    xpath: '',
                    cssSelector: '',
                    text: getTextContent(element),
                    attributes: {},
                    isClickable: isElementInteractive,
                    isInput: isElementInput,
                    isVisible: isElementVisible,
                    isInViewport: isElementVisible,
                    inputType: element.type || null,
                    placeholder: element.placeholder || null,
                    boundingBox: {
                        x: rect.x,
                        y: rect.y,
                        width: rect.width,
                        height: rect.height,
                        top: rect.top,
                        bottom: rect.bottom,
                        left: rect.left,
                        right: rect.right
                    },
                    centerCoordinates: {
                        x: rect.left + rect.width / 2,
                        y: rect.top + rect.height / 2
                    }
                };
                
                if (element.attributes) {
                    for (let attr of element.attributes) {
                        elementData.attributes[attr.name] = attr.value;
                    }
                }
                
                elements.push(elementData);
                
                if (currentHighlightIndex !== null) {
                    selectorMap[currentHighlightIndex] = elementData;
                }
            });
            
            const endTime = performance.now();
            return {
                elements: elements,
                selectorMap: selectorMap,
                stats: {
                    totalNodes: nodeCount,
                    processedNodes: processedCount,
                    interactiveElements: Object.keys(selectorMap).length,
                    executionTime: endTime - startTime
                }
            };
        }
        """

    # Add all your existing methods here (goto, get_page_state, click_element_by_index, etc.)
    async def goto(self, url: str, wait_until: str = "domcontentloaded", timeout: int = 30000):
        """Navigate to a URL with proper waiting"""
        try:
            logger.info(f"Navigating to: {url}")
            await self.page.goto(url, wait_until=wait_until, timeout=timeout)
            await asyncio.sleep(2)
            logger.info(f"Successfully navigated to: {url}")
        except Exception as e:
            logger.error(f"Failed to navigate to {url}: {e}")
            raise

    async def get_page_state(self, include_screenshot: bool = True, highlight_elements: bool = True) -> PageState:
        """Get current page state with elements"""
        try:
            await self.page.wait_for_load_state("domcontentloaded", timeout=10000)
            await asyncio.sleep(1)
            
            url = self.page.url
            title = await self.page.title()
            
            screenshot = None
            if include_screenshot:
                screenshot_bytes = await self.page.screenshot(
                    full_page=False,
                    clip={'x': 0, 'y': 0, 'width': 1250, 'height': 800}
                )
                screenshot = base64.b64encode(screenshot_bytes).decode('utf-8')
            
            # Extract DOM elements
            try:
                dom_result = await self.page.evaluate(self.dom_js, {"doHighlightElements": highlight_elements})
                logger.info(f"Extracted {len(dom_result.get('elements', []))} interactive elements")
            except Exception as e:
                logger.error(f"DOM extraction failed: {e}")
                return PageState(url, title, [], {}, screenshot)
            
            elements = []
            selector_map = {}
            
            for elem_data in dom_result.get('elements', []):
                element_info = ElementInfo(
                    index=elem_data.get('index'),
                    id=elem_data.get('id', ''),
                    tag_name=elem_data.get('tagName', ''),
                    xpath=elem_data.get('xpath', ''),
                    css_selector=elem_data.get('cssSelector', ''),
                    text=elem_data.get('text', ''),
                    attributes=elem_data.get('attributes', {}),
                    is_clickable=elem_data.get('isClickable', False),
                    is_input=elem_data.get('isInput', False),
                    center_coordinates=elem_data.get('centerCoordinates')
                )
                
                elements.append(element_info)
                if element_info.index is not None:
                    selector_map[element_info.index] = element_info
            
            return PageState(url, title, elements, selector_map, screenshot)
            
        except Exception as e:
            logger.error(f"Failed to get page state: {e}")
            return PageState("", "", [], {}, None)

    async def click_element_by_index(self, index: int, page_state: PageState = None) -> bool:
        """Click element by index"""
        try:
            if page_state is None:
                page_state = await self.get_page_state(include_screenshot=False, highlight_elements=False)
            
            if index not in page_state.selector_map:
                logger.error(f"Element with index {index} not found")
                return False
            
            element = page_state.selector_map[index]
            if not element.center_coordinates:
                logger.error(f"Element at index {index} has no coordinates")
                return False
            
            x = element.center_coordinates['x']
            y = element.center_coordinates['y']
            
            logger.info(f"Clicking element {index}: {element.text[:50]}... at ({x}, {y})")
            
            await self.page.mouse.click(x, y)
            await asyncio.sleep(1)
            
            logger.info(f"Successfully clicked element {index}")
            return True
            
        except Exception as e:
            logger.error(f"Failed to click element at index {index}: {e}")
            return False

    async def input_text_by_index(self, index: int, text: str, page_state: PageState = None) -> bool:
        """Input text into element by index"""
        try:
            if page_state is None:
                page_state = await self.get_page_state(include_screenshot=False, highlight_elements=False)
            
            if index not in page_state.selector_map:
                logger.error(f"Element with index {index} not found")
                return False
            
            element = page_state.selector_map[index]
            if not element.center_coordinates:
                logger.error(f"Element at index {index} has no coordinates")
                return False
            
            x = element.center_coordinates['x']
            y = element.center_coordinates['y']
            
            logger.info(f"Typing '{text}' into element {index}")
            
            await self.page.mouse.click(x, y)
            await asyncio.sleep(0.5)
            await self.page.keyboard.press('Control+a')
            await self.page.keyboard.type(text)
            
            logger.info(f"Successfully typed text into element {index}")
            return True
            
        except Exception as e:
            logger.error(f"Failed to input text into element at index {index}: {e}")
            return False

    async def scroll_page(self, direction: str = "down", amount: int = 500):
        """Scroll the page"""
        if direction == "down":
            await self.page.mouse.wheel(0, amount)
        elif direction == "up":
            await self.page.mouse.wheel(0, -amount)
        await asyncio.sleep(1)

    async def press_key(self, key: str) -> bool:
        """Press a keyboard key"""
        try:
            await self.page.keyboard.press(key)
            logger.info(f"Pressed key: {key}")
            return True
        except Exception as e:
            logger.error(f"Failed to press key {key}: {e}")
            return False