browser.py

"""
browser.py — Playwright Browser Automation for Manus-Class Agent
Provides web browsing, scraping, and interaction capabilities
"""
from __future__ import annotations

import asyncio
import json
import logging
import os
import re
import time
from typing import Any, Dict, List, Optional, Callable
from dataclasses import dataclass, field
from enum import Enum

logger = logging.getLogger("browser")

# ─── Configuration ─────────────────────────────────────────────────────────────
PLAYWRIGHT_ENABLED = os.environ.get("PLAYWRIGHT_ENABLED", "false").lower() == "true"
PLAYWRIGHT_HEADLESS = os.environ.get("PLAYWRIGHT_HEADLESS", "true").lower() == "true"
PLAYWRIGHT_TIMEOUT = int(os.environ.get("PLAYWRIGHT_TIMEOUT", "30000"))


class BrowserType(Enum):
    CHROMIUM = "chromium"
    FIREFOX = "firefox"
    WEBKIT = "webkit"


@dataclass
class BrowserAction:
    """Represents a browser action to perform."""
    action_type: str  # navigate, click, type, screenshot, evaluate, etc.
    selector: Optional[str] = None
    value: Optional[str] = None
    timeout: int = 30000
    wait_for: Optional[str] = None  # load, domcontentloaded, networkidle


@dataclass
class BrowserResult:
    """Result of browser operation."""
    success: bool
    content: Optional[str] = None
    screenshot: Optional[str] = None
    elements: List[Dict] = field(default_factory=list)
    url: Optional[str] = None
    title: Optional[str] = None
    error: Optional[str] = None
    actions: List[Dict] = field(default_factory=list)


class BrowserManager:
    """
    Manages Playwright browser instances for web automation.
    
    Features:
    - Multi-page browsing
    - Element interaction (click, type, hover)
    - Screenshot capture
    - JavaScript execution
    - Page content extraction
    - Form filling
    """
    
    def __init__(
        self,
        headless: bool = PLAYWRIGHT_HEADLESS,
        browser_type: BrowserType = BrowserType.CHROMIUM,
        timeout: int = PLAYWRIGHT_TIMEOUT,
    ):
        self.headless = headless
        self.browser_type = browser_type
        self.timeout = timeout
        self._browser = None
        self._context = None
        self._page = None
    
    async def start(self) -> bool:
        """Start browser and create context."""
        try:
            from playwright.async_api import async_playwright
            
            self._playwright = await async_playwright().start()
            
            # Launch browser
            browser_name = self.browser_type.value
            self._browser = await self._playwright.__dict__[browser_name].launch(
                headless=self.headless,
                args=[
                    '--disable-blink-features=AutomationControlled',
                    '--disable-dev-shm-usage',
                    '--no-sandbox',
                ]
            )
            
            # Create context with stealth settings
            self._context = await self._browser.new_context(
                viewport={"width": 1920, "height": 1080},
                user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
                ignore_https_errors=True,
            )
            
            # Add stealth scripts
            await self._context.add_init_script("""
                Object.defineProperty(navigator, 'webdriver', { get: () => false });
                Object.defineProperty(navigator, 'plugins', { get: () => [1, 2, 3, 4, 5] });
                Object.defineProperty(navigator, 'languages', { get: () => ['en-US', 'en'] });
            """)
            
            self._page = await self._context.new_page()
            await self._page.set_default_timeout(self.timeout)
            
            logger.info(f"Browser started: {self.browser_type.value} (headless={self.headless})")
            return True
            
        except Exception as e:
            logger.error(f"Failed to start browser: {e}")
            return False
    
    async def close(self):
        """Close browser and cleanup."""
        try:
            if self._page:
                await self._page.close()
            if self._context:
                await self._context.close()
            if self._browser:
                await self._browser.close()
            if hasattr(self, '_playwright'):
                await self._playwright.stop()
            logger.info("Browser closed")
        except Exception as e:
            logger.warning(f"Browser cleanup error: {e}")
    
    async def navigate(self, url: str, wait_until: str = "domcontentloaded") -> BrowserResult:
        """Navigate to URL."""
        if not self._page:
            return BrowserResult(success=False, error="Browser not started")
        
        try:
            response = await self._page.goto(url, wait_until=wait_until, timeout=self.timeout)
            
            return BrowserResult(
                success=response is not None and response.ok,
                url=self._page.url,
                title=await self._page.title(),
                content=await self._page.content(),
                actions=[{"action": "navigate", "url": url, "status": response.status if response else "failed"}],
            )
        except Exception as e:
            return BrowserResult(success=False, error=str(e))
    
    async def click(self, selector: str, timeout: int = None) -> BrowserResult:
        """Click element by selector."""
        if not self._page:
            return BrowserResult(success=False, error="Browser not started")
        
        try:
            await self._page.click(selector, timeout=timeout or self.timeout)
            return BrowserResult(
                success=True,
                url=self._page.url,
                title=await self._page.title(),
                actions=[{"action": "click", "selector": selector}],
            )
        except Exception as e:
            return BrowserResult(success=False, error=str(e))
    
    async def type_text(self, selector: str, text: str, delay: int = 100) -> BrowserResult:
        """Type text into element."""
        if not self._page:
            return BrowserResult(success=False, error="Browser not started")
        
        try:
            await self._page.fill(selector, text)
            # Alternative: await self._page.type(selector, text, delay=delay)
            return BrowserResult(
                success=True,
                url=self._page.url,
                title=await self._page.title(),
                actions=[{"action": "type", "selector": selector, "text_length": len(text)}],
            )
        except Exception as e:
            return BrowserResult(success=False, error=str(e))
    
    async def screenshot(self, full_page: bool = False) -> BrowserResult:
        """Take screenshot of current page."""
        if not self._page:
            return BrowserResult(success=False, error="Browser not started")
        
        try:
            screenshot_bytes = await self._page.screenshot(full_page=full_page)
            import base64
            screenshot_b64 = base64.b64encode(screenshot_bytes).decode()
            
            return BrowserResult(
                success=True,
                screenshot=screenshot_b64,
                url=self._page.url,
                title=await self._page.title(),
                actions=[{"action": "screenshot", "full_page": full_page}],
            )
        except Exception as e:
            return BrowserResult(success=False, error=str(e))
    
    async def evaluate(self, script: str) -> BrowserResult:
        """Execute JavaScript on page."""
        if not self._page:
            return BrowserResult(success=False, error="Browser not started")
        
        try:
            result = await self._page.evaluate(script)
            return BrowserResult(
                success=True,
                content=json.dumps(result),
                url=self._page.url,
                title=await self._page.title(),
                actions=[{"action": "evaluate", "script": script[:100]}],
            )
        except Exception as e:
            return BrowserResult(success=False, error=str(e))
    
    async def extract_text(self, selector: str = "body") -> BrowserResult:
        """Extract text content from element(s)."""
        if not self._page:
            return BrowserResult(success=False, error="Browser not started")
        
        try:
            elements = await self._page.query_selector_all(selector)
            texts = []
            for el in elements:
                text = await el.inner_text()
                texts.append(text)
            
            return BrowserResult(
                success=True,
                content="\n".join(texts),
                elements=[{"selector": selector, "count": len(elements)}],
                url=self._page.url,
                title=await self._page.title(),
            )
        except Exception as e:
            return BrowserResult(success=False, error=str(e))
    
    async def extract_links(self, selector: str = "a") -> BrowserResult:
        """Extract all links from page."""
        if not self._page:
            return BrowserResult(success=False, error="Browser not started")
        
        try:
            links = await self._page.query_selector_all(selector)
            results = []
            for link in links:
                href = await link.get_attribute("href")
                text = await link.inner_text()
                results.append({"href": href, "text": text.strip()})
            
            return BrowserResult(
                success=True,
                elements=results,
                url=self._page.url,
                title=await self._page.title(),
                actions=[{"action": "extract_links", "count": len(results)}],
            )
        except Exception as e:
            return BrowserResult(success=False, error=str(e))
    
    async def wait_for_selector(self, selector: str, state: str = "visible") -> BrowserResult:
        """Wait for element to appear."""
        if not self._page:
            return BrowserResult(success=False, error="Browser not started")
        
        try:
            await self._page.wait_for_selector(selector, state=state, timeout=self.timeout)
            return BrowserResult(
                success=True,
                url=self._page.url,
                title=await self._page.title(),
                actions=[{"action": "wait_for", "selector": selector, "state": state}],
            )
        except Exception as e:
            return BrowserResult(success=False, error=str(e))
    
    async def scroll(self, direction: str = "down", amount: int = 500) -> BrowserResult:
        """Scroll page."""
        if not self._page:
            return BrowserResult(success=False, error="Browser not started")
        
        try:
            if direction == "down":
                script = f"window.scrollBy(0, {amount})"
            else:
                script = f"window.scrollBy(0, -{amount})"
            
            await self._page.evaluate(script)
            return BrowserResult(
                success=True,
                url=self._page.url,
                title=await self._page.title(),
                actions=[{"action": "scroll", "direction": direction, "amount": amount}],
            )
        except Exception as e:
            return BrowserResult(success=False, error=str(e))
    
    async def fill_form(self, form_data: Dict[str, str]) -> BrowserResult:
        """Fill form fields."""
        if not self._page:
            return BrowserResult(success=False, error="Browser not started")
        
        try:
            for selector, value in form_data.items():
                await self._page.fill(selector, value)
            
            return BrowserResult(
                success=True,
                url=self._page.url,
                title=await self._page.title(),
                actions=[{"action": "fill_form", "fields": len(form_data)}],
            )
        except Exception as e:
            return BrowserResult(success=False, error=str(e))
    
    async def submit_form(self, selector: str = "form") -> BrowserResult:
        """Submit form."""
        if not self._page:
            return BrowserResult(success=False, error="Browser not started")
        
        try:
            await self._page.click(selector)
            await self._page.wait_for_load_state("networkidle")
            
            return BrowserResult(
                success=True,
                url=self._page.url,
                title=await self._page.title(),
                actions=[{"action": "submit_form", "selector": selector}],
            )
        except Exception as e:
            return BrowserResult(success=False, error=str(e))


# ─── Web Search Tool ──────────────────────────────────────────────────────────
async def web_search(query: str, num_results: int = 5) -> Dict[str, Any]:
    """
    Perform web search using browser automation.
    Falls back to DuckDuckGo or direct URL access.
    """
    browser = BrowserManager(headless=True)
    
    try:
        await browser.start()
        
        # Navigate to search engine
        search_url = f"https://duckduckgo.com/?q={query.replace(' ', '+')}"
        result = await browser.navigate(search_url)
        
        if not result.success:
            return {"error": f"Search failed: {result.error}"}
        
        # Wait for results
        await asyncio.sleep(2)
        await browser.wait_for_selector(".result", state="visible")
        
        # Extract links
        result = await browser.extract_links(".result__a")
        
        links = result.elements[:num_results]
        
        return {
            "query": query,
            "results": links,
            "url": browser._page.url if browser._page else None,
        }
        
    except Exception as e:
        return {"error": str(e)}
    finally:
        await browser.close()


async def read_url(url: str, extract: str = "text") -> Dict[str, Any]:
    """
    Read content from a URL using browser automation.
    
    Args:
        url: URL to read
        extract: What to extract - "text", "links", "screenshot", "html"
    """
    browser = BrowserManager(headless=True)
    
    try:
        await browser.start()
        
        # Navigate to URL
        result = await browser.navigate(url)
        
        if not result.success:
            return {"error": f"Navigation failed: {result.error}"}
        
        # Wait for content
        await asyncio.sleep(1)
        
        if extract == "text":
            result = await browser.extract_text()
            return {
                "url": url,
                "title": result.title,
                "content": result.content,
            }
        elif extract == "links":
            result = await browser.extract_links()
            return {
                "url": url,
                "title": result.title,
                "links": result.elements,
            }
        elif extract == "screenshot":
            result = await browser.screenshot()
            return {
                "url": url,
                "title": result.title,
                "screenshot": result.screenshot,
            }
        elif extract == "html":
            content = await browser._page.content()
            return {
                "url": url,
                "title": await browser._page.title(),
                "html": content,
            }
        else:
            return {"error": f"Unknown extract type: {extract}"}
            
    except Exception as e:
        return {"error": str(e)}
    finally:
        await browser.close()


# ─── Interactive Browser Session ───────────────────────────────────────────────
class BrowserSession:
    """
    Interactive browser session for agent use.
    Maintains state across multiple operations.
    """
    
    def __init__(self, session_id: str = None):
        self.session_id = session_id or str(time.time())
        self.browser = BrowserManager()
        self.history: List[BrowserResult] = []
    
    async def __aenter__(self):
        await self.browser.start()
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.browser.close()
    
    async def do(self, action: str, **kwargs) -> BrowserResult:
        """Execute a browser action."""
        method_map = {
            "navigate": self.browser.navigate,
            "click": self.browser.click,
            "type": self.browser.type_text,
            "screenshot": self.browser.screenshot,
            "evaluate": self.browser.evaluate,
            "extract": self.browser.extract_text,
            "links": self.browser.extract_links,
            "wait": self.browser.wait_for_selector,
            "scroll": self.browser.scroll,
            "fill": self.browser.fill_form,
            "submit": self.browser.submit_form,
        }
        
        method = method_map.get(action)
        if not method:
            return BrowserResult(success=False, error=f"Unknown action: {action}")
        
        result = await method(**kwargs)
        self.history.append(result)
        return result


# ─── Health Check ─────────────────────────────────────────────────────────────
def check_browser_health() -> Dict[str, Any]:
    """Check if browser automation is available."""
    try:
        from playwright.sync_api import sync_playwright
        return {"available": True, "status": "healthy"}
    except ImportError:
        return {"available": False, "status": "playwright not installed"}
    except Exception as e:
        return {"available": False, "status": f"error: {e}"}