#!/usr/bin/env python3 """ Browser Base Implementation - Persistent Firefox Browser Management This module implements the base persistent Firefox browser that stays alive across multiple requests, eliminating the overhead of browser startup/shutdown. """ import os import sys import asyncio from typing import Optional from patchright.async_api import async_playwright, BrowserContext, Playwright class PersistentBrowserBase: """ Base class for persistent Firefox browser management. This class maintains a single Firefox browser context that stays alive throughout the application lifetime, providing significant performance benefits over opening/closing browsers for each request. """ def __init__(self, user_data_dir: str, slowmo: int = 500): """ Initialize the persistent browser. Args: user_data_dir (str): Path to Firefox profile directory for persistence slowmo (int): Delay in milliseconds between browser actions (default: 500) """ self.user_data_dir = user_data_dir self.slowmo = slowmo # Browser components - initialized in initialize() self.playwright: Optional[Playwright] = None self.browser_context: Optional[BrowserContext] = None self.browser = None # For CDP connection mode async def initialize(self): """Initialize the browser context asynchronously.""" if not self.is_initialized(): await self._initialize_browser_context() self.browser_context.add_init_script(""" Object.defineProperty(navigator, 'webdriver', { get: () => undefined }); window.chrome = { runtime: {} }; // More realistic plugins Object.defineProperty(navigator, 'plugins', { get: () => [ {name: 'Chrome PDF Plugin', filename: 'internal-pdf-viewer', description: 'Portable Document Format'}, {name: 'Chrome PDF Viewer', filename: 'mhjfbmdgcfjbbpaeojofohoefgiehjai', description: ''}, {name: 'Native Client', filename: 'internal-nacl-plugin', description: ''} ] }); // Languages Object.defineProperty(navigator, 'languages', { get: () => ['en-US', 'en'] }); // Platform Object.defineProperty(navigator, 'platform', { get: () => 'Linux x86_64' }); // Hardware concurrency (CPU cores) Object.defineProperty(navigator, 'hardwareConcurrency', { get: () => 4 }); // DeviceMemory Object.defineProperty(navigator, 'deviceMemory', { get: () => 8 }); const originalGetContext = HTMLCanvasElement.prototype.getContext; HTMLCanvasElement.prototype.getContext = function(type, attributes) { const context = originalGetContext.call(this, type, attributes); if (type === 'webgl' || type === 'webgl2' || type === 'experimental-webgl') { const getParameter = context.getParameter.bind(context); context.getParameter = function(parameter) { // Vendor if (parameter === 7936) { return 'WebKit'; } // Renderer if (parameter === 7937) { return 'WebKit WebGL'; } // Unmasked vendor if (parameter === 37445) { return 'Intel Inc.'; } // Unmasked renderer if (parameter === 37446) { return 'Intel(R) Iris(TM) Plus Graphics 640'; } return getParameter(parameter); }; // Also spoof the extension const getSupportedExtensions = context.getSupportedExtensions.bind(context); context.getSupportedExtensions = function() { const extensions = getSupportedExtensions(); // Optionally filter out or add extensions return extensions; }; } return context; }; """) def is_initialized(self) -> bool: """Check if the browser is already initialized.""" return self.playwright is not None and self.browser_context is not None async def _initialize_browser_context(self): """ Initialize the persistent Firefox browser context. Creates a Firefox browser with persistent profile that stays open across multiple operations. Uses slowmo for human-like behavior. Can either launch new browser or connect to existing one. """ try: # Start Playwright self.playwright = await async_playwright().start() # Launch new persistent browser await self._launch_new_browser() print(f"✅ Firefox browser context initialized successfully") print(f"🌐 Browser pages available: {len(self.browser_context.pages)}") except Exception as e: print(f"❌ Failed to initialize browser context: {e}") await self._cleanup_on_error() raise async def _launch_new_browser(self): """Launch a new persistent Chromium browser.""" print(f"🔥 Launching new Chromium browser...") print(f"📁 Profile directory: {self.user_data_dir}") print(f"⏱️ Slowmo delay: {self.slowmo}ms") # Create profile directory if it doesn't exist os.makedirs(self.user_data_dir, exist_ok=True) # Read Chrome flags from environment variable chrome_flags_env = os.getenv('CHROME_FLAGS', '') if chrome_flags_env: # Split the flags string into a list chrome_flags = chrome_flags_env.split() print(f"🚀 Using Chrome flags from env: {chrome_flags}") else: # Default flags for stability and consistency chrome_flags = [ '--no-sandbox', '--disable-dev-shm-usage', '--disable-gpu', "--disable-features=WebRtcHideLocalIpsWithMdns" ] print(f"🚀 Using default Chrome flags: {chrome_flags}") self.browser_context = await self.playwright.chromium.launch_persistent_context( user_data_dir=self.user_data_dir, headless=False, slow_mo=self.slowmo, #viewport=None, ignore_default_args=["--disable-extensions"], # keep everything else default # Viewport adapts to actual window size no_viewport=True, args=chrome_flags, # channel="chrome" ) async def _cleanup_on_error(self): """Clean up resources if initialization fails.""" if self.browser_context: try: await self.browser_context.close() except: pass self.browser_context = None if self.browser: try: await self.browser.close() except: pass self.browser = None if self.playwright: try: await self.playwright.stop() except: pass self.playwright = None async def close(self): """ Gracefully close the browser context and cleanup resources. """ print("🔒 Closing persistent browser context...") try: if self.browser_context: print("📄 Closing all pages...") # Close all pages first for page in self.browser_context.pages: try: await page.close() except: pass # Only close context if we launched it (not in connect mode) # if not self.connect_to_existing: # print("📄 Closing browser context...") # await asyncio.wait_for(self.browser_context.close(), timeout=10.0) # print("✅ Browser context closed") # else: # print("⚠️ Leaving connected browser open (connect mode)") self.browser_context = None if self.browser: # Disconnect from CDP browser print("🔌 Disconnecting from browser...") await asyncio.wait_for(self.browser.close(), timeout=10.0) self.browser = None print("✅ Browser disconnected") if self.playwright: print("🎭 Stopping Playwright...") await asyncio.wait_for(self.playwright.stop(), timeout=10.0) self.playwright = None print("✅ Playwright stopped") print("🎉 Browser cleanup completed successfully") except asyncio.TimeoutError: print("⚠️ Browser cleanup timed out, forcing cleanup...") self.browser_context = None self.browser = None self.playwright = None except Exception as e: print(f"⚠️ Error during browser cleanup: {e}") # Ensure resources are marked as cleaned up even if there were errors self.browser_context = None self.browser = None self.playwright = None def __del__(self): """ Destructor to ensure browser cleanup when object is destroyed. This is a safety net to prevent resource leaks if close() is not explicitly called. Note: In async context, this should not be relied upon. """ if self.browser_context is not None or self.browser is not None or self.playwright is not None: print("🚨 Browser cleanup via destructor (should call close() explicitly)") # Note: Cannot call async close() from __del__, so we just warn self.browser_context = None self.browser = None self.playwright = None async def __aenter__(self): """Async context manager entry.""" await self.initialize() return self async def __aexit__(self, exc_type, exc_val, exc_tb): """Async context manager exit.""" await self.close() return False async def is_alive(self) -> bool: """ Check if the browser context is alive and responsive. Returns: bool: True if browser context is alive, False otherwise """ try: if not self.browser_context: return False # Check if we can access browser context pages pages = self.browser_context.pages return len(pages) >= 0 # Should always work if context is alive except Exception as e: print(f"⚠️ Browser context health check failed: {e}") return False async def get_status(self) -> dict: """ Get current status information about the browser context. Returns: dict: Status information including context state and page count """ if not self.browser_context: return { "status": "not_initialized", "context": None, "pages": 0, "profile_dir": self.user_data_dir } try: pages = self.browser_context.pages return { "status": "active", "context": "Firefox persistent context", "pages": len(pages), "profile_dir": self.user_data_dir, "slowmo": self.slowmo } except Exception as e: return { "status": "error", "error": str(e), "profile_dir": self.user_data_dir }