browser_utils.py

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}")