"""Fast and optimized Selenium base scraper for maximum performance.""" import logging import random import time import json from abc import ABC, abstractmethod from typing import List, Dict, Optional, Any from selenium import webdriver 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.chrome.options import Options from selenium.webdriver.chrome.service import Service from selenium.common.exceptions import TimeoutException, NoSuchElementException, WebDriverException from selenium.webdriver.common.action_chains import ActionChains import multiprocessing as mp import threading from src.config.settings import REQUEST_TIMEOUT, RETRY_ATTEMPTS from src.models.schemas import SearchResult logger = logging.getLogger(__name__) class FastSeleniumBase(ABC): """Optimized base class for high-performance Selenium scrapers.""" def __init__(self, engine_name: str, headless: bool = True, fast_mode: bool = True): self.engine_name = engine_name self.headless = headless self.fast_mode = fast_mode self.driver = None self.wait = None self._lock = threading.Lock() # Performance-optimized user agents self.user_agents = [ 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36', 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36', 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36', ] # Initialize driver self._setup_fast_driver() def _setup_fast_driver(self): """Setup ultra-fast Chrome driver with maximum optimizations.""" try: chrome_options = Options() # Headless configuration if self.headless: chrome_options.add_argument('--headless=new') # Performance optimizations chrome_options.add_argument('--no-sandbox') chrome_options.add_argument('--disable-dev-shm-usage') chrome_options.add_argument('--disable-gpu') chrome_options.add_argument('--disable-extensions') chrome_options.add_argument('--disable-plugins') chrome_options.add_argument('--disable-images') chrome_options.add_argument('--disable-javascript') # Disable JS for faster loading chrome_options.add_argument('--disable-css') chrome_options.add_argument('--disable-web-security') chrome_options.add_argument('--allow-running-insecure-content') chrome_options.add_argument('--disable-blink-features=AutomationControlled') # Network optimizations chrome_options.add_argument('--aggressive-cache-discard') chrome_options.add_argument('--disable-background-timer-throttling') chrome_options.add_argument('--disable-renderer-backgrounding') chrome_options.add_argument('--disable-backgrounding-occluded-windows') chrome_options.add_argument('--disable-client-side-phishing-detection') chrome_options.add_argument('--disable-default-apps') chrome_options.add_argument('--disable-hang-monitor') chrome_options.add_argument('--disable-popup-blocking') chrome_options.add_argument('--disable-prompt-on-repost') chrome_options.add_argument('--disable-sync') chrome_options.add_argument('--disable-translate') chrome_options.add_argument('--disable-features=TranslateUI') # Memory optimizations chrome_options.add_argument('--memory-pressure-off') chrome_options.add_argument('--max_old_space_size=4096') # Fast loading chrome_options.add_argument('--aggressive') chrome_options.add_argument('--no-first-run') chrome_options.add_argument('--no-default-browser-check') # Disable unnecessary features chrome_options.add_experimental_option("excludeSwitches", [ "enable-automation", "enable-logging", "enable-blink-features" ]) chrome_options.add_experimental_option('useAutomationExtension', False) # Prefs for faster loading prefs = { "profile.default_content_setting_values": { "images": 2, # Block images "plugins": 2, # Block plugins "popups": 2, # Block popups "geolocation": 2, # Block location sharing "notifications": 2, # Block notifications "media_stream": 2, # Block media stream }, "profile.managed_default_content_settings": { "images": 2 }, "profile.default_content_settings": { "popups": 0 } } chrome_options.add_experimental_option("prefs", prefs) # Random user agent user_agent = random.choice(self.user_agents) chrome_options.add_argument(f'--user-agent={user_agent}') # Window size for faster rendering chrome_options.add_argument('--window-size=1280,720') # Disable logging for performance chrome_options.add_argument('--log-level=3') chrome_options.add_argument('--silent') chrome_options.add_argument('--disable-logging') # Language chrome_options.add_argument('--lang=en-US') # Service configuration service = Service('/usr/local/bin/chromedriver') service.creation_flags = 0 # No new console window # Initialize driver self.driver = webdriver.Chrome(service=service, options=chrome_options) # Set page load timeout for faster failures self.driver.set_page_load_timeout(15) # Set script timeout self.driver.set_script_timeout(10) # Minimal implicit wait self.driver.implicitly_wait(5) # Execute performance scripts self._execute_performance_scripts() # Setup optimized wait self.wait = WebDriverWait(self.driver, 10) logger.info(f"Fast Selenium driver initialized for {self.engine_name}") except Exception as e: logger.error(f"Failed to setup fast driver for {self.engine_name}: {e}") raise def _execute_performance_scripts(self): """Execute JavaScript for performance optimizations.""" try: # Remove automation indicators self.driver.execute_script("Object.defineProperty(navigator, 'webdriver', {get: () => undefined})") # Disable resource loading for faster page loads self.driver.execute_script(""" // Block images, CSS, and other resources document.addEventListener('DOMContentLoaded', function() { const style = document.createElement('style'); style.textContent = '* { animation-duration: 0s !important; transition-duration: 0s !important; }'; document.head.appendChild(style); }); """) except Exception as e: logger.debug(f"Performance script execution failed: {e}") @abstractmethod def search(self, query: str, max_results: int = 10) -> List[SearchResult]: """Search for results using the specific engine.""" pass def _fast_navigate(self, url: str, timeout: int = 15) -> bool: """Fast navigation with optimized timeout.""" try: with self._lock: # Thread safety self.driver.get(url) # Quick page load check start_time = time.time() while (time.time() - start_time) < timeout: if self.driver.execute_script("return document.readyState") == "complete": break time.sleep(0.1) # Minimal wait time.sleep(random.uniform(0.5, 1.0)) return True except Exception as e: logger.warning(f"Fast navigation failed for {url}: {e}") return False def _fast_find_elements(self, selector: str, timeout: int = 5) -> List[Any]: """Fast element finding with reduced timeout.""" try: wait = WebDriverWait(self.driver, timeout) elements = wait.until(EC.presence_of_all_elements_located((By.CSS_SELECTOR, selector))) return elements except TimeoutException: return [] except Exception as e: logger.debug(f"Element finding failed: {e}") return [] def _fast_find_element(self, selector: str, timeout: int = 5) -> Optional[Any]: """Fast single element finding.""" try: wait = WebDriverWait(self.driver, timeout) element = wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, selector))) return element except TimeoutException: return None except Exception as e: logger.debug(f"Single element finding failed: {e}") return None def _get_text_fast(self, element: Any) -> str: """Fast text extraction.""" try: return element.text.strip() if element else "" except: try: return element.get_attribute('textContent').strip() if element else "" except: return "" def _get_attribute_fast(self, element: Any, attribute: str) -> str: """Fast attribute extraction.""" try: return element.get_attribute(attribute) if element else "" except: return "" def _clean_text(self, text: str) -> str: """Fast text cleaning.""" if not text: return "" return ' '.join(text.split()) def _create_result(self, title: str, url: str, description: str, position: int) -> SearchResult: """Create search result quickly.""" return SearchResult( title=self._clean_text(title), url=url.strip() if url else "", description=self._clean_text(description), engine=self.engine_name, position=position ) def _handle_blocking(self) -> bool: """Quick blocking detection.""" try: page_source = self.driver.page_source[:5000].lower() # Only check first 5k chars blocking_indicators = ['captcha', 'blocked', 'unusual traffic', 'verify you are human'] return any(indicator in page_source for indicator in blocking_indicators) except: return False def close(self): """Close driver quickly.""" try: if self.driver: self.driver.quit() self.driver = None except: pass def __del__(self): """Cleanup.""" self.close()