from selenium import webdriver from selenium.webdriver.chrome.options import Options from selenium.webdriver.chrome.service import Service from selenium.webdriver.common.by import By from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC from selenium.webdriver.common.action_chains import ActionChains from selenium.common.exceptions import TimeoutException, WebDriverException from webdriver_manager.chrome import ChromeDriverManager import asyncio import logging from typing import Optional, List, Dict, Any from contextlib import asynccontextmanager import random import time import json logger = logging.getLogger(__name__) class AdvancedSeleniumManager: def __init__(self, headless: bool = True, proxy: Optional[str] = None, stealth_mode: bool = True): self.headless = headless self.proxy = proxy self.stealth_mode = stealth_mode self.driver: Optional[webdriver.Chrome] = None self.wait: Optional[WebDriverWait] = None def _get_stealth_chrome_options(self) -> Options: """Get Chrome options with advanced anti-detection""" options = Options() if self.headless: options.add_argument('--headless=new') # Use new headless mode # Basic options options.add_argument('--no-sandbox') options.add_argument('--disable-dev-shm-usage') options.add_argument('--disable-gpu') options.add_argument('--disable-extensions') options.add_argument('--disable-plugins') options.add_argument('--disable-blink-features=AutomationControlled') # Window size with random variation width = random.randint(1200, 1920) height = random.randint(800, 1080) options.add_argument(f'--window-size={width},{height}') # User agent rotation user_agents = [ 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36', 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36', 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36', 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36', 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36' ] options.add_argument(f'--user-agent={random.choice(user_agents)}') # Proxy configuration if self.proxy: options.add_argument(f'--proxy-server={self.proxy}') # Advanced anti-detection if self.stealth_mode: options.add_experimental_option("excludeSwitches", ["enable-automation"]) options.add_experimental_option('useAutomationExtension', False) options.add_argument('--disable-web-security') options.add_argument('--allow-running-insecure-content') options.add_argument('--disable-features=VizDisplayCompositor') # Additional stealth options prefs = { "profile.default_content_setting_values": { "notifications": 2, "media_stream": 2, }, "profile.managed_default_content_settings": { "images": 2 # Block images for faster loading } } options.add_experimental_option("prefs", prefs) return options async def start_driver(self): """Start Chrome driver with stealth configuration""" try: options = self._get_stealth_chrome_options() service = Service(ChromeDriverManager().install()) self.driver = webdriver.Chrome(service=service, options=options) if self.stealth_mode: # Execute stealth scripts await self._apply_stealth_scripts() self.wait = WebDriverWait(self.driver, 10) logger.info("Advanced Chrome driver started successfully") except Exception as e: logger.error(f"Error starting Chrome driver: {e}") raise async def _apply_stealth_scripts(self): """Apply JavaScript stealth scripts""" stealth_scripts = [ # Remove webdriver property "Object.defineProperty(navigator, 'webdriver', {get: () => undefined})", # Mock plugins """ Object.defineProperty(navigator, 'plugins', { get: () => [1, 2, 3, 4, 5] }); """, # Mock languages """ Object.defineProperty(navigator, 'languages', { get: () => ['en-US', 'en'] }); """, # Mock permissions """ const originalQuery = window.navigator.permissions.query; window.navigator.permissions.query = (parameters) => ( parameters.name === 'notifications' ? Promise.resolve({ state: Notification.permission }) : originalQuery(parameters) ); """, # Mock chrome runtime """ window.chrome = { runtime: {} }; """ ] for script in stealth_scripts: try: self.driver.execute_script(script) except Exception as e: logger.warning(f"Error executing stealth script: {e}") async def human_like_navigation(self, url: str, wait_for_element: Optional[str] = None): """Navigate with human-like behavior""" try: # Random delay before navigation await asyncio.sleep(random.uniform(0.5, 2.0)) self.driver.get(url) # Random mouse movements await self._random_mouse_movements() # Random scroll await self._random_scroll() # Wait for element if specified if wait_for_element: self.wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, wait_for_element))) # Additional random delay await asyncio.sleep(random.uniform(1.0, 3.0)) logger.debug(f"Human-like navigation to: {url}") except Exception as e: logger.error(f"Error in human-like navigation: {e}") raise async def _random_mouse_movements(self): """Perform random mouse movements""" try: actions = ActionChains(self.driver) # Random mouse movements for _ in range(random.randint(2, 5)): x = random.randint(100, 800) y = random.randint(100, 600) actions.move_by_offset(x, y) await asyncio.sleep(random.uniform(0.1, 0.3)) actions.perform() except Exception as e: logger.warning(f"Error in random mouse movements: {e}") async def _random_scroll(self): """Perform random scrolling""" try: # Random scroll amount scroll_amount = random.randint(100, 500) self.driver.execute_script(f"window.scrollBy(0, {scroll_amount});") await asyncio.sleep(random.uniform(0.5, 1.5)) # Sometimes scroll back up if random.random() < 0.3: scroll_back = random.randint(50, scroll_amount // 2) self.driver.execute_script(f"window.scrollBy(0, -{scroll_back});") await asyncio.sleep(random.uniform(0.3, 0.8)) except Exception as e: logger.warning(f"Error in random scroll: {e}") async def smart_find_elements(self, selector: str, timeout: int = 10, retry_count: int = 3) -> List[Any]: """Find elements with retry logic and error handling""" for attempt in range(retry_count): try: wait = WebDriverWait(self.driver, timeout) elements = wait.until(EC.presence_of_all_elements_located((By.CSS_SELECTOR, selector))) if elements: logger.debug(f"Found {len(elements)} elements for selector: {selector}") return elements except TimeoutException: if attempt < retry_count - 1: logger.warning(f"Timeout finding elements (attempt {attempt + 1}), retrying...") await asyncio.sleep(random.uniform(1, 3)) continue else: logger.warning(f"No elements found for selector after {retry_count} attempts: {selector}") return [] except Exception as e: logger.error(f"Error finding elements: {e}") if attempt < retry_count - 1: await asyncio.sleep(random.uniform(1, 3)) continue return [] return [] async def bypass_cloudflare(self): """Attempt to bypass Cloudflare protection""" try: # Check if Cloudflare challenge is present cloudflare_selectors = [ "#cf-wrapper", ".cf-browser-verification", "#challenge-form" ] for selector in cloudflare_selectors: element = await self.smart_find_elements(selector, timeout=5) if element: logger.info("Cloudflare challenge detected, waiting...") # Wait for challenge to complete (up to 30 seconds) for _ in range(30): await asyncio.sleep(1) # Check if challenge is completed if not await self.smart_find_elements(selector, timeout=1): logger.info("Cloudflare challenge completed") return True logger.warning("Cloudflare challenge timeout") return False return True # No Cloudflare detected except Exception as e: logger.error(f"Error handling Cloudflare: {e}") return False async def handle_cookie_consent(self): """Handle cookie consent popups""" try: cookie_selectors = [ "button[id*='accept']", "button[class*='accept']", "button[id*='consent']", "button[class*='consent']", ".cookie-accept", "#cookie-accept", "[data-testid*='accept']" ] for selector in cookie_selectors: elements = await self.smart_find_elements(selector, timeout=2) if elements: try: elements[0].click() logger.debug("Clicked cookie consent button") await asyncio.sleep(1) return True except Exception as e: logger.warning(f"Error clicking cookie consent: {e}") continue return False except Exception as e: logger.warning(f"Error handling cookie consent: {e}") return False async def stop_driver(self): """Stop the Chrome driver""" if self.driver: try: self.driver.quit() logger.info("Advanced Chrome driver stopped") except Exception as e: logger.error(f"Error stopping Chrome driver: {e}") @asynccontextmanager async def driver_context(self): """Context manager for driver lifecycle""" await self.start_driver() try: yield self finally: await self.stop_driver()