#!/usr/bin/env python3 """ Page Management System - Intelligent Page State and Pool Management This module provides intelligent page management for handling multiple tabs with state tracking, busy detection, and recovery mechanisms. """ import time import asyncio import random from enum import Enum from typing import Optional, Dict, List from patchright.async_api import Page class PageState(Enum): """Page state enumeration.""" IDLE = "idle" BUSY = "busy" ERROR = "error" CRASHED = "crashed" LOADING = "loading" STUCK = "stuck" class ManagedPage: """ Wrapper for Playwright Page with state management and monitoring. """ def __init__(self, page: Page, page_id: str): """ Initialize managed page. Args: page (Page): Playwright page instance page_id (str): Unique identifier for this page """ self.page = page self.page_id = page_id self.state = PageState.IDLE self.current_operation: Optional[str] = None self.operation_start_time: Optional[float] = None self.error_count = 0 self.last_activity = time.time() self.lock = asyncio.Lock() self.last_known_mouse_x = 0 self.last_known_mouse_y = 0 # Configure page timeouts self.page.set_default_timeout(60000) # 60 seconds self.page.set_default_navigation_timeout(60000) def is_busy(self) -> bool: """Check if page is currently busy.""" return self.state == PageState.BUSY def is_available(self) -> bool: """Check if page is available for new operations.""" return self.state == PageState.IDLE # Remove sync call to is_responsive async def is_responsive(self) -> bool: """ Check if page is responsive and ready for operations. Returns: bool: True if page is responsive, False otherwise """ try: if not self.page or self.state == PageState.CRASHED: return False # Quick responsiveness test await self.page.evaluate("() => document.readyState") return True except Exception as e: print(f"⚠️ Page {self.page_id} responsiveness check failed: {e}") await self._set_state(PageState.CRASHED) return False async def is_loading(self) -> bool: """Check if page is currently loading.""" try: if not self.page or self.state == PageState.CRASHED: return False loading_info = await self.page.evaluate(""" () => { return { readyState: document.readyState, loading: document.readyState !== 'complete' } } """) return loading_info.get('loading', False) except Exception: return False def is_stuck(self, max_operation_time: int = 300) -> bool: """ Check if current operation has been running too long. Args: max_operation_time (int): Maximum operation time in seconds (default: 5 minutes) Returns: bool: True if operation appears stuck """ if self.state == PageState.BUSY and self.operation_start_time: operation_duration = time.time() - self.operation_start_time return operation_duration > max_operation_time return False async def _set_state(self, new_state: PageState, operation: Optional[str] = None): """Thread-safe state change.""" async with self.lock: old_state = self.state self.state = new_state self.last_activity = time.time() if new_state == PageState.BUSY: self.current_operation = operation self.operation_start_time = time.time() print(f"📄 Page {self.page_id} is now BUSY with: {operation}") elif new_state == PageState.IDLE: self.current_operation = None self.operation_start_time = None if old_state == PageState.BUSY: print(f"📄 Page {self.page_id} is now IDLE") elif new_state == PageState.ERROR: self.error_count += 1 self.current_operation = f"ERROR: {operation}" if operation else "ERROR" print(f"📄 Page {self.page_id} is now in ERROR state: {operation}") elif new_state == PageState.CRASHED: self.current_operation = "CRASHED" print(f"📄 Page {self.page_id} is now CRASHED") async def set_busy(self, operation: str): """Mark page as busy with specific operation.""" await self._set_state(PageState.BUSY, operation) async def set_idle(self): """Mark page as idle and ready for new operations.""" await self._set_state(PageState.IDLE) async def set_error(self, error: str): """Mark page as in error state.""" await self._set_state(PageState.ERROR, error) async def set_crashed(self): """Mark page as crashed.""" await self._set_state(PageState.CRASHED) async def get_status(self) -> dict: """Get detailed page status information.""" try: current_url = self.page.url if self.page and await self.is_responsive() else None current_title = await self.page.title() if self.page and await self.is_responsive() else None except: current_url = None current_title = None operation_duration = None if self.operation_start_time: operation_duration = time.time() - self.operation_start_time return { "page_id": self.page_id, "state": self.state.value, "current_operation": self.current_operation, "operation_duration": operation_duration, "responsive": await self.is_responsive(), "loading": await self.is_loading(), "stuck": self.is_stuck(), "error_count": self.error_count, "last_activity": self.last_activity, "url": current_url, "title": current_title } async def wait_until_ready(self, timeout: int = 30) -> bool: """ Wait until page is ready for new operations. Args: timeout (int): Maximum time to wait in seconds Returns: bool: True if page became ready, False if timeout """ start_time = time.time() while time.time() - start_time < timeout: if self.is_available(): return True await asyncio.sleep(0.5) return False async def force_recovery(self): """Force page recovery by resetting state and clearing operations.""" print(f"🔧 Force recovering page {self.page_id}...") async with self.lock: old_state = self.state self.state = PageState.IDLE self.current_operation = None self.operation_start_time = None self.last_activity = time.time() print(f"🔄 Page {self.page_id} recovered from {old_state.value} to IDLE") def __repr__(self): return f"" class IntelligentPagePool: """ Intelligent page pool that manages multiple tabs with automatic recovery and load balancing. """ def __init__(self, browser_context, max_pages: int = 5): """ Initialize page pool. Args: browser_context: Playwright browser context initial_pages (int): Number of pages to create initially max_pages (int): Maximum number of pages allowed """ self.browser_context = browser_context self.max_pages = max_pages self.pages: Dict[str, ManagedPage] = {} self.page_counter = 0 self.pool_lock = asyncio.Lock() async def setup_listeners(self): """ Setup browser context event listeners. Listens for: - New pages created (manual user tabs, popups, etc) - Auto-registers orphaned pages with generated IDs """ # Listen for new pages created in browser (user clicks +, window.open, etc) self.browser_context.on("page", self._on_page_created) print("📡 Browser context listeners configured") def _on_page_created(self, page): """ Handle new page created in browser context. This fires when: - User clicks + button to create new tab - User opens a popup (window.open) - Links open in new tab We auto-register these orphaned pages so the pool stays in sync. Args: page: The newly created Playwright page """ # Check if this page is already tracked if any(mp.page == page for mp in self.pages.values()): print(f"📄 Page already tracked, ignoring duplicate") return # Auto-generate unique ID for this orphaned page # Format: auto_{timestamp_ms}_{random_hex} # This ensures uniqueness even if multiple pages created simultaneously timestamp = int(time.time() * 1000) random_suffix = ''.join(random.choices('abcdef0123456789', k=6)) auto_page_id = f"auto_{timestamp}_{random_suffix}" print(f"🆕 Detected new page created by user, auto-registering: {auto_page_id}") # Create managed page wrapper managed_page = ManagedPage(page, auto_page_id) # Add to pool self.pages[auto_page_id] = managed_page print(f"✅ Orphaned page registered: {auto_page_id} ({len(self.pages)}/{self.max_pages})") # Attach close event listener self._attach_close_handler(managed_page) async def initialize(self, initial_pages: int = 1): """Initialize pages asynchronously.""" # Setup listeners FIRST before creating pages await self.setup_listeners() # Create initial pages for _ in range(initial_pages): await self._create_page() async def _create_page(self, custom_page_id: Optional[str] = None) -> ManagedPage: """Create a new managed page, reusing the initial Firefox page if needed.""" async with self.pool_lock: # Determine the page ID to use if custom_page_id: if custom_page_id in self.pages: raise Exception(f"Page with ID '{custom_page_id}' already exists") page_id = custom_page_id else: self.page_counter += 1 page_id = f"page_{self.page_counter}" # Ensure the generated ID doesn't exist (very unlikely but safety first) while page_id in self.pages: self.page_counter += 1 page_id = f"page_{self.page_counter}" try: # Check page limit if len(self.pages) >= self.max_pages: raise Exception(f"Maximum page limit ({self.max_pages}) reached. Cannot create new page.") # If this is the first page and browser_context.pages already has one, reuse it if not custom_page_id and self.page_counter == 1 and hasattr(self.browser_context, "pages") and self.browser_context.pages: playwright_page = self.browser_context.pages[0] else: playwright_page = await self.browser_context.new_page() # vp_x = playwright_page.viewport_size.get("width", 1920) # vp_y = playwright_page.viewport_size.get("height", 1080) # mouse_initial_x = random.uniform(0, vp_x) # mouse_initial_y = random.uniform(0, vp_y) managed_page = ManagedPage(playwright_page, page_id) self.pages[page_id] = managed_page # Attach close event listener to track user-initiated closes self._attach_close_handler(managed_page) print(f"📄 Created new page: {page_id} ({len(self.pages)}/{self.max_pages})") return managed_page except Exception as e: print(f"❌ Failed to create page {page_id}: {e}") raise def _attach_close_handler(self, managed_page: ManagedPage): """ Attach a close event listener to a page. When user closes the tab in browser (clicks X button), this handler: 1. Removes page from pool immediately to prevent stale references 2. Marks page as CRASHED for any lingering API references 3. Logs the event for debugging Args: managed_page (ManagedPage): The managed page to attach listener to """ page_id = managed_page.page_id def on_close(): """Callback when page is closed by user in browser""" print(f"🔴 Page {page_id} was closed by user (browser X button)") # Remove from pool immediately - prevents stale references being used if page_id in self.pages: del self.pages[page_id] print(f"🗑️ Page {page_id} removed from pool") # Mark as crashed for debugging any lingering API references try: # Can't await in sync callback, but set state directly managed_page.state = PageState.CRASHED print(f"💥 Page {page_id} marked as CRASHED") except: pass # Attach the listener to the Playwright page object managed_page.page.on("close", on_close) print(f"📌 Close event listener attached to page {page_id}") async def get_available_page(self, wait_timeout: int = 10) -> Optional[ManagedPage]: """ Get an available page from the pool. """ start_time = time.time() request_id = f"req_{int(time.time() * 1000) % 10000}" # Add request tracking print(f"🔍 [{request_id}] Searching for available page... Pool size: {len(self.pages)}") while time.time() - start_time < wait_timeout: # Log current page states page_states = {pid: page.state.value for pid, page in self.pages.items()} print(f"🔍 [{request_id}] Current page states: {page_states}") # First, try to find an immediately available page for page in self.pages.values(): if page.is_available(): print(f"📄 [{request_id}] Using available page: {page.page_id}") return page # If no pages available, try to create a new one if len(self.pages) < self.max_pages: print(f"📄 [{request_id}] Creating new page ({len(self.pages)}/{self.max_pages})") try: return await self._create_page() except Exception as e: print(f"⚠️ [{request_id}] Failed to create new page: {e}") print(f"⏳ [{request_id}] No pages available, waiting... ({time.time() - start_time:.1f}s)") await asyncio.sleep(0.5) print(f"⚠️ [{request_id}] No pages available within timeout") return None def get_page_by_id(self, page_id: str) -> Optional[ManagedPage]: """Get specific page by ID.""" return self.pages.get(page_id) async def create_page(self, page_id: Optional[str] = None) -> ManagedPage: """ Create a new page with optional custom ID. Args: page_id (str, optional): Custom page ID. If None, auto-generated ID will be used. Returns: ManagedPage: The newly created managed page Raises: Exception: If page creation fails or page limit exceeded """ # Delegate to the private _create_page method which handles all the logic return await self._create_page(custom_page_id=page_id) async def recover_stuck_pages(self): """Recover pages that appear to be stuck.""" for page in self.pages.values(): if page.is_stuck(): print(f"🔧 Recovering stuck page: {page.page_id}") await page.force_recovery() async def cleanup_crashed_pages(self): """Remove crashed pages from the pool.""" crashed_pages = [ page_id for page_id, page in self.pages.items() if page.state == PageState.CRASHED ] for page_id in crashed_pages: print(f"🗑️ Removing crashed page: {page_id}") try: await self.pages[page_id].page.close() except: pass del self.pages[page_id] async def get_pool_status(self) -> dict: """Get comprehensive pool status.""" status_by_state = {} page_statuses = [] for page in self.pages.values(): status = await page.get_status() page_statuses.append(status) state = status["state"] status_by_state[state] = status_by_state.get(state, 0) + 1 return { "total_pages": len(self.pages), "max_pages": self.max_pages, "status_by_state": status_by_state, "pages": page_statuses } async def close_all_pages(self): """Close all pages in the pool.""" print("🔒 Closing all pages in pool...") for page_id, page in self.pages.items(): try: print(f"📄 Closing page: {page_id}") await page.page.close() except Exception as e: print(f"⚠️ Error closing page {page_id}: {e}") self.pages.clear() print("✅ All pages closed")