import time import os import subprocess import sys import shutil import urllib.request import tarfile from typing import Tuple, Optional from playwright.sync_api import Page, BrowserContext, Browser, Playwright from config import ( logger, DEFAULT_TIMEOUT, MAX_RETRIES, BROWSER_OPTIONS, USER_AGENT, IS_HF_SPACE ) class BrowserManager: """Handles browser session management and recovery for headless mode.""" def __init__(self, playwright: Playwright): self.playwright = playwright self.browser = None self.context = None self.page = None def connect(self) -> Tuple[Page, BrowserContext, Browser]: """Launch a headless browser and establish connection.""" try: return self.launch_headless_browser() except Exception as e: error_msg = str(e) if "Executable doesn't exist" in error_msg: logger.warning("Browser executable not found, attempting to install browsers") self._install_browsers() # Try again after installation return self.launch_headless_browser() elif "unexpected keyword argument 'executablePath'" in error_msg: # Fix old parameter name to new logger.warning("Detected incompatible executablePath parameter, adjusting to executable_path") if "executablePath" in BROWSER_OPTIONS: exec_path = BROWSER_OPTIONS.pop("executablePath") BROWSER_OPTIONS["executable_path"] = exec_path return self.launch_headless_browser() elif "error while loading shared libraries" in error_msg or "cannot open shared object file" in error_msg: # Missing system library dependencies logger.warning(f"Detected missing system libraries: {error_msg}") self._check_system_dependencies() # Try again after installing dependencies return self.launch_headless_browser() else: logger.error(f"Connection error (not related to missing browser): {e}") raise def _install_browsers(self): """Install Playwright browsers if they're missing.""" try: logger.info("Installing Playwright browsers") success = False # Ensure cache directory exists with proper permissions os.makedirs("/home/user/.cache/ms-playwright", exist_ok=True) os.system("chmod -R 777 /home/user/.cache") # Check and install system dependencies if needed self._check_system_dependencies() # Method 0: Download pre-compiled chromium for HF Space environment if IS_HF_SPACE: try: logger.info("Hugging Face Space detected, downloading pre-compiled Chromium") user_browser_path = "/home/user/.cache/ms-playwright" # Create the directory if it doesn't exist os.makedirs(user_browser_path, exist_ok=True) # Download a minimal version of Chromium that works with Playwright download_url = "https://playwright.azureedge.net/builds/chromium/1105/chromium-linux.zip" local_zip = "/tmp/chromium-linux.zip" logger.info(f"Downloading browser from {download_url}") urllib.request.urlretrieve(download_url, local_zip) # Extract to the proper location chromium_dir = os.path.join(user_browser_path, "chromium-1105") os.makedirs(chromium_dir, exist_ok=True) # Extract using unzip command which is usually available logger.info(f"Extracting browser to {chromium_dir}") extract_cmd = f"unzip -o {local_zip} -d {chromium_dir}" os.system(extract_cmd) # Ensure chrome executable has proper permissions chrome_path = os.path.join(chromium_dir, "chrome-linux", "chrome") if os.path.exists(chrome_path): os.chmod(chrome_path, 0o755) logger.info(f"Chrome executable found at {chrome_path} and made executable") # Set executable path in browser options - using the correct parameter name if "executablePath" in BROWSER_OPTIONS: BROWSER_OPTIONS.pop("executablePath") # Remove old parameter if exists BROWSER_OPTIONS["executable_path"] = chrome_path success = True else: logger.warning(f"Chrome executable not found at {chrome_path} after extraction") except Exception as e: logger.warning(f"Failed to download and extract browser: {e}") # Method 1: Using subprocess with playwright CLI if not success: try: logger.info("Attempting to install browsers using playwright CLI") result = subprocess.run(["playwright", "install", "--with-deps", "chromium"], check=False, stdout=subprocess.PIPE, stderr=subprocess.PIPE) if result.returncode == 0: logger.info("Playwright browsers installed successfully using CLI") success = True else: error_output = result.stderr.decode('utf-8', errors='ignore') logger.warning(f"Failed to install browsers using playwright CLI: {error_output}") except Exception as e: logger.warning(f"Failed to install browsers using playwright CLI: {e}") # Method 2: Using python -m if not success: try: logger.info("Attempting to install browsers using python -m playwright") result = subprocess.run([sys.executable, "-m", "playwright", "install", "--with-deps", "chromium"], check=False, stdout=subprocess.PIPE, stderr=subprocess.PIPE) if result.returncode == 0: logger.info("Playwright browsers installed successfully using python -m") success = True else: error_output = result.stderr.decode('utf-8', errors='ignore') logger.warning(f"Failed to install browsers using python -m: {error_output}") except Exception as e: logger.warning(f"Failed to install browsers using python -m: {e}") # Method 3: Using just install-deps if not success: try: logger.info("Attempting to install only browser dependencies") result = subprocess.run(["playwright", "install-deps", "chromium"], check=False, stdout=subprocess.PIPE, stderr=subprocess.PIPE) if result.returncode == 0: logger.info("Browser dependencies installed successfully") else: error_output = result.stderr.decode('utf-8', errors='ignore') logger.warning(f"Failed to install browser dependencies: {error_output}") except Exception as e: logger.warning(f"Failed to install browser dependencies: {e}") # Method 4: Using direct browser installation with fixed path if not success: try: logger.info("Attempting to install browsers using Python with explicit PLAYWRIGHT_BROWSERS_PATH") # Set explicit browsers path for installation os.environ["PLAYWRIGHT_BROWSERS_PATH"] = "/ms-playwright" result = subprocess.run([sys.executable, "-m", "playwright", "install", "chromium"], check=False, env=os.environ, stdout=subprocess.PIPE, stderr=subprocess.PIPE) if result.returncode == 0: logger.info("Browsers installed successfully with fixed path") success = True else: error_output = result.stderr.decode('utf-8', errors='ignore') logger.warning(f"Failed to install browsers with fixed path: {error_output}") except Exception as e: logger.warning(f"Failed to install browsers with fixed path: {e}") # Method 5: Copy from Docker image if not success: try: logger.info("Attempting to use pre-installed browser from Docker image") # The Docker image has browsers at /ms-playwright src_browser_path = "/ms-playwright" user_browser_path = "/home/user/.cache/ms-playwright" if os.path.exists(src_browser_path) and os.path.isdir(src_browser_path): if not os.path.exists(user_browser_path): os.makedirs(user_browser_path, exist_ok=True) # Find chromium directory in source chromium_dirs = [d for d in os.listdir(src_browser_path) if d.startswith("chromium-")] if chromium_dirs: chromium_dir = chromium_dirs[0] src_chromium_path = os.path.join(src_browser_path, chromium_dir) dest_chromium_path = os.path.join(user_browser_path, chromium_dir) if os.path.exists(src_chromium_path) and not os.path.exists(dest_chromium_path): logger.info(f"Copying browser from {src_chromium_path} to {dest_chromium_path}") shutil.copytree(src_chromium_path, dest_chromium_path) # Make chrome executable chrome_path = os.path.join(dest_chromium_path, "chrome-linux", "chrome") if os.path.exists(chrome_path): os.chmod(chrome_path, 0o755) logger.info("Browser copied and permissions set") success = True else: logger.warning(f"Source browser path {src_browser_path} not found") except Exception as e: logger.warning(f"Failed to copy browsers from Docker image: {e}") # Method 6: Check if we can use browser from a different location if not success: try: logger.info("Attempting to find browser in alternative locations") # Some containers might have browsers in different locations possible_paths = [ "/browser/chromium/chrome", "/browser/chrome-linux/chrome", "/usr/bin/chromium", "/usr/bin/chromium-browser", "/usr/bin/google-chrome", ] for browser_path in possible_paths: if os.path.exists(browser_path) and os.access(browser_path, os.X_OK): logger.info(f"Found executable browser at {browser_path}") # In this case, we'll set BROWSER_OPTIONS to use executable_path if "executablePath" in BROWSER_OPTIONS: BROWSER_OPTIONS.pop("executablePath") # Remove old parameter if exists BROWSER_OPTIONS["executable_path"] = browser_path success = True break except Exception as e: logger.warning(f"Failed to find alternative browser locations: {e}") # Method 7: Use the emergency backup browser if not success: try: logger.info("Attempting to use emergency backup browser") backup_zip = "/browser-backup/chromium-linux.zip" if os.path.exists(backup_zip): user_browser_path = "/home/user/.cache/ms-playwright" chromium_dir = os.path.join(user_browser_path, "chromium-1105") if not os.path.exists(chromium_dir): os.makedirs(chromium_dir, exist_ok=True) # Extract the emergency backup logger.info(f"Extracting emergency browser from {backup_zip} to {chromium_dir}") extract_cmd = f"unzip -o {backup_zip} -d {chromium_dir}" os.system(extract_cmd) # Ensure chrome executable has proper permissions chrome_path = os.path.join(chromium_dir, "chrome-linux", "chrome") if os.path.exists(chrome_path): os.chmod(chrome_path, 0o755) logger.info(f"Emergency browser found at {chrome_path} and made executable") # Set executable path in browser options if "executablePath" in BROWSER_OPTIONS: BROWSER_OPTIONS.pop("executablePath") # Remove old parameter if exists BROWSER_OPTIONS["executable_path"] = chrome_path # Create a configuration file for future runs try: os.makedirs(os.path.join(os.path.dirname(__file__), "browser_config"), exist_ok=True) with open(os.path.join(os.path.dirname(__file__), "browser_config", "system_browser.py"), "w") as f: f.write(f"""# Auto-generated browser config SYSTEM_BROWSER_PATH = "{chrome_path}" """) logger.info("Created browser config file for future runs") except Exception as e: logger.warning(f"Could not create browser config file: {e}") success = True else: logger.warning(f"Emergency browser not found at {chrome_path} after extraction") else: logger.warning(f"Emergency browser backup not found at {backup_zip}") except Exception as e: logger.warning(f"Failed to use emergency backup browser: {e}") if not success: # If we got here, all methods failed raise Exception("All browser installation methods failed") except Exception as e: logger.error(f"Failed to install browsers: {e}") raise def launch_headless_browser(self) -> Tuple[Page, BrowserContext, Browser]: """Launch a new browser instance (headless or non-headless based on config).""" try: # Make a copy of browser options to avoid modifying global config browser_options = BROWSER_OPTIONS.copy() # Check if we're in headless or non-headless mode is_headless = browser_options.get("headless", True) logger.info(f"Launching browser in {'headless' if is_headless else 'non-headless'} mode") # For Windows in non-headless mode, simplify arguments to avoid issues import platform is_windows = platform.system() == 'Windows' if not is_headless and is_windows: logger.info("Using simplified browser arguments for Windows non-headless mode") browser_options["args"] = [ "--disable-blink-features=AutomationControlled", "--disable-features=IsolateOrigins", ] else: # Default: Override some options for better evasion in headless mode # Make sure we're applying headless mode with additional arguments # Add additional evasion args if not already present additional_args = [ "--disable-blink-features=AutomationControlled", "--disable-features=IsolateOrigins,site-per-process", "--disable-web-security", "--no-default-browser-check" ] # Ensure existing args are not overwritten if "args" in browser_options: for arg in additional_args: if arg not in browser_options["args"]: browser_options["args"].append(arg) else: browser_options["args"] = additional_args # Ensure proper chromium flags browser_options["chromium_sandbox"] = False if is_windows else True # Add slowMo for non-headless mode to make actions more visible if not is_headless: browser_options["slow_mo"] = 50 # 50ms delay # Launch new browser with enhanced anti-detection options logger.info("Launching browser with options:") for key, value in browser_options.items(): if key != "args": # Skip printing all args to keep log clean logger.info(f" {key}: {value}") else: logger.info(f" args: {len(value)} arguments") self.browser = self.playwright.chromium.launch(**browser_options) # Create a context with more realistic profile context_options = { "user_agent": USER_AGENT, "viewport": {"width": 1920, "height": 1080}, "device_scale_factor": 1, "has_touch": False, "bypass_csp": True, "java_script_enabled": True, "locale": "en-US", "timezone_id": "America/New_York", "geolocation": {"longitude": -74.006, "latitude": 40.7128}, # NYC coordinates "permissions": ["geolocation"], "color_scheme": "light" } self.context = self.browser.new_context(**context_options) # Setup browser evasions self._set_evasions(self.context) self.page = self.context.new_page() logger.info("Browser launched successfully") return self.page, self.context, self.browser except Exception as e: logger.error(f"Browser launch failed: {e}") raise def ensure_alive(self, login_callback, combo_counter=None, revalidate_every=5) -> Tuple[Page, BrowserContext, Browser]: """ Ensure browser, context and page are alive. Implements a hierarchical recovery system. """ # Check if browser is connected try: if getattr(self.browser, "is_closed", lambda: True)(): logger.info("Browser connection lost, reconnecting") return self.reconnect_browser(login_callback) except Exception: logger.warning("Error checking browser connection") # Check if page is closed try: page_is_closed = self.page.is_closed() except Exception: page_is_closed = True # Check if context is valid try: context_is_valid = not getattr(self.context, "is_closed", lambda: True)() if not page_is_closed and context_is_valid: _ = self.context.pages # Test operation else: context_is_valid = False except Exception: context_is_valid = False # Recovery logic if page_is_closed and context_is_valid: logger.info("Page was closed, creating new page") try: self.page = self.context.new_page() except Exception: context_is_valid = False if not context_is_valid: logger.info("Context invalid, creating new context") try: self._clean_close(self.context) self.context = self.browser.new_context(user_agent=USER_AGENT) self._set_evasions(self.context) self.page = self.context.new_page() except Exception: logger.warning("Context creation failed, resetting browser") return self.reconnect_browser(login_callback) # Revalidate login if needed if combo_counter is not None and combo_counter % revalidate_every == 0: logger.info(f"Session checkpoint at combo #{combo_counter}") try: self.page, self.context = login_callback(self.page, self.context, self.browser) except Exception as e: logger.error(f"Login revalidation failed: {e}") return self.reconnect_browser(login_callback) return self.page, self.context, self.browser def reconnect_browser(self, login_callback) -> Tuple[Page, BrowserContext, Browser]: """Complete browser reconnection with login for headless mode.""" try: self._clean_close(self.browser) # Make a copy of browser options to avoid modifying global config browser_options = BROWSER_OPTIONS.copy() # Add additional evasion args if not already present additional_args = [ "--disable-blink-features=AutomationControlled", "--disable-features=IsolateOrigins,site-per-process", "--disable-web-security", "--no-default-browser-check" ] if "args" in browser_options: for arg in additional_args: if arg not in browser_options["args"]: browser_options["args"].append(arg) else: browser_options["args"] = additional_args # Launch a new headless browser with enhanced options self.browser = self.playwright.chromium.launch(**browser_options) # Create a context with more realistic profile context_options = { "user_agent": USER_AGENT, "viewport": {"width": 1920, "height": 1080}, "device_scale_factor": 1, "has_touch": False, "bypass_csp": True, "java_script_enabled": True, "locale": "en-US", "timezone_id": "America/New_York", "geolocation": {"longitude": -74.006, "latitude": 40.7128}, # NYC coordinates "permissions": ["geolocation"], "color_scheme": "light" } self.context = self.browser.new_context(**context_options) self._set_evasions(self.context) self.page = self.context.new_page() logger.info("Headless browser relaunched") # Perform login self.page, self.context = login_callback(self.page, self.context, self.browser) return self.page, self.context, self.browser except Exception as e: logger.critical(f"Complete browser reconnection failed: {e}") raise def _set_evasions(self, context): """Setup evasions to avoid bot detection.""" try: context.add_init_script("""() => { // Overwrite the navigator.webdriver property Object.defineProperty(navigator, 'webdriver', { get: () => false }); // Override permissions API if (window.navigator.permissions) { const originalQuery = window.navigator.permissions.query; window.navigator.permissions.query = (parameters) => ( parameters.name === 'notifications' || parameters.name === 'clipboard-read' || parameters.name === 'clipboard-write' ? Promise.resolve({ state: Notification.permission }) : originalQuery(parameters) ); } // Add fake plugins Object.defineProperty(navigator, 'plugins', { get: () => { return [ { 0: {type: "application/pdf", suffixes: "pdf", description: "Portable Document Format"}, name: "Chrome PDF Plugin", description: "Portable Document Format", filename: "internal-pdf-viewer", length: 1 }, { 0: {type: "application/pdf", suffixes: "pdf", description: "Portable Document Format"}, name: "Chrome PDF Viewer", description: "Portable Document Format", filename: "internal-pdf-viewer", length: 1 }, { 0: {type: "application/x-google-chrome-pdf", suffixes: "pdf", description: "Portable Document Format"}, name: "Chrome PDF Viewer", description: "Portable Document Format", filename: "internal-pdf-viewer", length: 1 } ]; } }); // Create a fake language list Object.defineProperty(navigator, 'languages', { get: () => ['en-US', 'en', 'es'] }); // Override user agent if needed if (!window.navigator.chrome) { // Add a chrome property to navigator Object.defineProperty(navigator, 'chrome', { get: () => ({ runtime: {}, loadTimes: function() {}, app: {}, csi: function() {}, browserInfo: {} }) }); } // Hide automation features const originalHasAttribute = Element.prototype.hasAttribute; Element.prototype.hasAttribute = function(name) { if (name === 'webdriver') return false; return originalHasAttribute.apply(this, arguments); }; // Spoof hairline feature if (window.chrome) { window.chrome.runtime = window.chrome.runtime || {}; window.chrome.runtime.sendMessage = (message, callback) => { setTimeout(callback, 100, { type: 'success' }); }; } // Add fake mouse movements const fakeMouseEvents = () => { // Create a fake mouse movement event const events = ['mousemove', 'mousedown', 'mouseup']; const randomEvent = events[Math.floor(Math.random() * events.length)]; const event = new MouseEvent(randomEvent, { 'view': window, 'bubbles': true, 'cancelable': true, 'clientX': Math.floor(Math.random() * window.innerWidth), 'clientY': Math.floor(Math.random() * window.innerHeight) }); document.body.dispatchEvent(event); }; // Occasionally send mouse events setInterval(fakeMouseEvents, Math.random() * 3000 + 1000); }""") logger.debug("Enhanced browser evasions applied") except Exception as e: logger.debug(f"Failed to set evasions: {e}") def _clean_close(self, resource): """Safely close a browser resource.""" if resource: try: resource.close() except Exception: pass # Ignore errors on close def _check_system_dependencies(self): """Check for common missing dependencies and attempt to install them.""" try: logger.info("Checking Chrome/Chromium system dependencies") # Only run on Linux if not sys.platform.startswith('linux'): logger.info("Not on Linux, skipping dependency check") return # Try to detect if we have permission to install packages has_apt = os.path.exists("/usr/bin/apt-get") and os.access("/usr/bin/apt-get", os.X_OK) if not has_apt: logger.warning("apt-get not available, skipping dependency check") return # Essential libraries to check for libraries = ["libnss3.so", "libnspr4.so", "libatk-1.0.so", "libatk-bridge-2.0.so", "libcups.so", "libdbus-1.so", "libxcomposite.so", "libxdamage.so"] # Check if libraries are present using ldconfig missing_libs = [] for lib in libraries: cmd = f"ldconfig -p | grep {lib}" result = subprocess.run(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) if result.returncode != 0: missing_libs.append(lib) if missing_libs: logger.warning(f"Missing libraries: {', '.join(missing_libs)}") # Try to install dependencies logger.info("Attempting to install missing Chrome dependencies") cmd = """apt-get update && apt-get install -y --no-install-recommends \ libnss3 libnspr4 libasound2 libatk1.0-0 libatk-bridge2.0-0 \ libcups2 libdbus-1-3 libdrm2 libgbm1 libgtk-3-0 libxkbcommon0 \ libxcomposite1 libxdamage1 libxfixes3 libxrandr2 libgbm-dev \ libxshmfence1 libgles2 libegl1 xvfb fonts-liberation""" try: # Try with sudo if available sudo_check = subprocess.run("which sudo", shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) if sudo_check.returncode == 0: logger.info("sudo is available, using it to install packages") os.system(f"sudo {cmd}") else: # Try without sudo (might work in container environments) logger.info("sudo not available, trying direct install") os.system(cmd) logger.info("Chrome dependencies installed") except Exception as e: logger.warning(f"Failed to install Chrome dependencies: {e}") else: logger.info("All required Chrome dependencies are present") except Exception as e: logger.warning(f"Error checking system dependencies: {e}")