#!/usr/bin/env python3 """ Browser Navigator Hub for Browser Operations This module provides the BrowserNavigator class that serves as a central orchestrator for all browser operations, hosting core browser management while providing access to specialized/scripted operations through primitive operations and executor methods. """ from typing import Optional, Dict, Any import base64 from datetime import datetime from patchright.async_api import BrowserContext from lib.browser_base import PersistentBrowserBase from lib.page_management import IntelligentPagePool, ManagedPage from lib.browser_primitives import PrimitiveBrowserOps from llm import llm class BrowserNavigator(PrimitiveBrowserOps, PersistentBrowserBase): """ Central hub for all browser operations. The BrowserNavigator consolidates browser management and provides primitive operations for scraping and fingerprinting. It serves as the main entry point for all browser automation tasks while maintaining a shared browser context and page pool. Core Responsibilities: - Browser context lifecycle management (initialize, maintain, cleanup) - Page pool management and coordination - Providing factory methods for specialized operations - Exposing primitive/atomic operations for custom workflows """ def __init__(self, user_data_dir: str, slowmo: int = 500, max_pages: int = 3): """ Initialize the BrowserNavigator with full browser setup. Args: user_data_dir (str): Path to Firefox profile directory for persistence slowmo (int): Delay in milliseconds between browser actions (default: 500) """ super().__init__(user_data_dir, slowmo) self.max_pages = max_pages self._page_pool: Optional[IntelligentPagePool] = None print(f"🧭 BrowserNavigator configured") print(f"📁 Profile directory: {user_data_dir}") print(f"⏱️ Slowmo delay: {slowmo}ms") print(f"📄 Max pages: {max_pages}") async def _get_page_to_use(self, page_id: Optional[str] = None) -> ManagedPage: """ Helper method to get a managed page to use. Args: page_id (str, optional): Specific page ID to use, or None for any available page Returns: ManagedPage: The managed page to use """ if not self._page_pool: raise Exception("Page pool not initialized") # Get page to use if page_id: managed_page = self._page_pool.pages.get(page_id) if not managed_page: raise Exception(f"Page {page_id} not found") else: managed_page = await self._page_pool.get_available_page(wait_timeout=30) if not managed_page: raise Exception("No pages available") return managed_page @classmethod async def from_existing_context(cls, browser_context: BrowserContext, slowmo: int = 1000, max_pages: int = 3): """ Create BrowserNavigator from an existing browser context. Args: browser_context (BrowserContext): Existing Playwright browser context slowmo (int): Delay in milliseconds between browser actions (default: 1000) max_pages (int): Maximum number of pages in the pool (default: 3) Returns: BrowserNavigator: Instance using the shared browser context """ # Create instance without calling __init__ instance = cls.__new__(cls) # Set up basic attributes without creating new browser instance.user_data_dir = "external_context" # Placeholder since we don't own the context instance.slowmo = slowmo instance.max_pages = max_pages instance.playwright = None # We don't own the playwright instance instance.browser_context = browser_context instance._page_pool = None # Initialize page pool await instance._initialize_page_pool() print(f"🧭 BrowserNavigator initialized from existing context") print(f"⏱️ Slowmo delay: {slowmo}ms") print(f"📄 Max pages: {max_pages}") return instance async def initialize(self): """Initialize the browser context and page pool.""" # First initialize the browser context from parent await super().initialize() # Then initialize the page pool await self._initialize_page_pool() async def _initialize_page_pool(self): """Initialize the page pool with the active browser context.""" if not self.browser_context: raise Exception("Browser context must be initialized before page pool") self._page_pool = IntelligentPagePool( browser_context=self.browser_context, max_pages=self.max_pages ) # Initialize pages asynchronously await self._page_pool.initialize(initial_pages=1) print(f"✅ Page pool initialized with {self.max_pages} max pages") def get_browser_context(self) -> Optional[BrowserContext]: """Get the current browser context.""" return self.browser_context def get_page_pool(self) -> Optional[IntelligentPagePool]: """Get the current page pool.""" return self._page_pool async def create_page(self, page_id: Optional[str] = None) -> Dict[str, Any]: """ Create a new page/tab in the browser. Args: page_id (str, optional): Custom page ID. If None, auto-generated ID will be used. Returns: Dict[str, Any]: Result containing page information """ if not self._page_pool: return {"error": "Page pool not initialized"} try: print(f"🆕 Creating new page with ID: {page_id or 'auto-generated'}") managed_page = await self._page_pool.create_page(page_id) result = { "success": True, "page_id": managed_page.page_id, "message": f"Page '{managed_page.page_id}' created successfully", "url": managed_page.page.url, "state": managed_page.state.value, "total_pages": len(self._page_pool.pages) } print(f"✅ Page created successfully: {managed_page.page_id}") return result except Exception as e: error_msg = f"Failed to create page: {str(e)}" print(f"❌ {error_msg}") return {"error": error_msg} async def destroy_page(self, page_id: str) -> Dict[str, Any]: """ Destroy/close a page/tab in the browser. Args: page_id (str): ID of the page to destroy Returns: Dict[str, Any]: Result containing destruction status """ if not self._page_pool: return {"error": "Page pool not initialized"} try: print(f"🗑️ Destroying page with ID: {page_id}") # Check if page exists if page_id not in self._page_pool.pages: return {"error": f"Page '{page_id}' not found"} managed_page = self._page_pool.pages[page_id] # Close the page await managed_page.page.close() # Remove from pool del self._page_pool.pages[page_id] result = { "success": True, "page_id": page_id, "message": f"Page '{page_id}' destroyed successfully", "total_pages": len(self._page_pool.pages) } print(f"✅ Page destroyed successfully: {page_id}") return result except Exception as e: error_msg = f"Failed to destroy page: {str(e)}" print(f"❌ {error_msg}") return {"error": error_msg} async def do_search_object(self, description: str, page_id: Optional[str] = None) -> Dict[str, Any]: """ Search for an object on the page based on description using vision AI. Args: description (str): Description of the object to search for page_id (str, optional): Specific page ID to use, or None for any available page Returns: Dict containing search result with bounding box and coordinates """ start_time = datetime.now() managed_page = await self._get_page_to_use(page_id) try: # Mark page as busy await managed_page.set_busy(f"searching for object: {description}") print(f"🔍 Searching for object: {description}") # Take screenshot print("📸 Taking screenshot") screenshot_bytes = await managed_page.page.screenshot() screenshot_base64 = base64.b64encode(screenshot_bytes).decode("utf-8") # Call Gemini vision API to detect bounding boxes print("🤖 Calling Gemini vision API") result = llm.detect_bounding_boxes( object_description=description, image_base64=screenshot_base64 ) # Calculate duration and prepare result end_time = datetime.now() duration = (end_time - start_time).total_seconds() print(f"✅ Object found at center: {result.center}, bounding box: {result.bounding_box}") return { "success": True, "description": description, "page_id": managed_page.page_id, "bounding_box": result.bounding_box, "center_x": result.center["x"], "center_y": result.center["y"], "tokens_used": result.token_total(), "costs_usd": result.cost_total(), "duration_seconds": duration, } except Exception as e: print(f"❌ Object search failed: {e}") await managed_page.set_error(str(e)) return { "success": False, "description": description, "page_id": managed_page.page_id, "error": str(e), } finally: # Set page back to idle state await managed_page.set_idle()