"""Stealth browser factory — bypasses Cloudflare/ArvanCloud bot detection.""" from __future__ import annotations import os import random import re import shutil import signal import subprocess import sys import tempfile import time from contextlib import contextmanager from pathlib import Path from typing import Generator import undetected_chromedriver as uc from selenium.webdriver.chrome.options import Options from selenium_stealth import stealth from config import config from crawler.recording import SessionRecorder from logger import get_logger log = get_logger(__name__) # Selenium talks to ChromeDriver over localhost — must not go through any system proxy. for _var in ("http_proxy", "https_proxy", "HTTP_PROXY", "HTTPS_PROXY"): os.environ.pop(_var, None) _no_proxy = "localhost,127.0.0.1,::1" os.environ.setdefault("no_proxy", _no_proxy) os.environ.setdefault("NO_PROXY", _no_proxy) _USER_AGENTS = [ "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36", "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36", ] _CHROME_CANDIDATES = [ "google-chrome-stable", "google-chrome", "chromium", "chromium-browser", ] def _detect_chrome() -> tuple[str, int]: """Return (binary_path, major_version) for the first Chrome found.""" candidates = list(_CHROME_CANDIDATES) if config.chrome_binary: candidates.insert(0, config.chrome_binary) for binary in candidates: resolved = shutil.which(binary) or (binary if Path(binary).exists() else None) if not resolved: continue try: out = subprocess.check_output( [resolved, "--version"], stderr=subprocess.DEVNULL, text=True, timeout=5 ).strip() match = re.search(r"\b(\d+(?:\.\d+){3})\b", out) if match is None: continue major = int(match.group(1).split(".")[0]) log.debug("Detected Chrome %d via '%s'", major, resolved) return resolved, major except Exception: continue log.critical("No Chrome binary found. Install Chrome or set CHROME_BINARY in .env") sys.exit(1) def _local_driver_path() -> str | None: path = Path(config.chromedriver_path) if path.exists(): return str(path) log.warning( "No patched ChromeDriver at '%s'. Run 'make patch-driver' once. Falling back to auto-download.", path, ) return None def _build_options( user_agent: str, headless: bool, chrome_binary: str, profile_dir: str | None = None, display: str | None = None, ) -> Options: opts = Options() opts.binary_location = chrome_binary opts.page_load_strategy = "eager" if headless: opts.add_argument("--headless=new") if profile_dir: opts.add_argument(f"--user-data-dir={profile_dir}") if display: opts.add_argument(f"--display={display}") opts.add_argument("--ozone-platform=x11") opts.add_argument(f"--user-agent={user_agent}") opts.add_argument("--no-sandbox") opts.add_argument("--disable-gpu") opts.add_argument("--renderer-process-limit=2") opts.add_argument("--disable-blink-features=AutomationControlled") opts.add_argument("--disable-infobars") opts.add_argument("--window-size=1920,1080") return opts def make_driver( headless: bool | None = None, profile_dir: str | None = None, display: str | None = None, ) -> uc.Chrome: """Return a stealthed undetected Chrome instance.""" use_headless = config.headless if headless is None else headless chrome_binary, major = _detect_chrome() ua = random.choice(_USER_AGENTS) opts = _build_options(ua, use_headless, chrome_binary, profile_dir, display) log.debug("Starting ChromeDriver (headless=%s, Chrome %d)", use_headless, major) driver = uc.Chrome( options=opts, driver_executable_path=_local_driver_path(), version_main=major, use_subprocess=True, ) if display: driver.set_window_size(1920, 1080) elif not use_headless: driver.maximize_window() else: driver.set_window_size(1920, 1080) driver.set_page_load_timeout(45) stealth( driver, languages=["en-US", "en"], vendor="Google Inc.", platform="Win32", webgl_vendor="Intel Inc.", renderer="Intel Iris OpenGL Engine", fix_hairline=True, ) driver.execute_cdp_cmd( "Page.addScriptToEvaluateOnNewDocument", {"source": """ Object.defineProperty(navigator, 'webdriver', {get: () => undefined}); Object.defineProperty(navigator, 'plugins', {get: () => [1, 2, 3, 4, 5]}); Object.defineProperty(navigator, 'languages', {get: () => ['en-US', 'en']}); """}, ) log.info("Driver ready (Chrome %d)", major) return driver def _profile_processes(profile_dir: str) -> list[int]: marker = profile_dir.encode() pids: list[int] = [] for entry in Path("/proc").iterdir(): if not entry.name.isdigit(): continue try: if marker in (entry / "cmdline").read_bytes(): pids.append(int(entry.name)) except (FileNotFoundError, PermissionError, ProcessLookupError): continue return pids def _terminate_profile_processes(profile_dir: str) -> None: for sig in (signal.SIGTERM, signal.SIGKILL): pids = _profile_processes(profile_dir) for pid in pids: try: os.kill(pid, sig) except ProcessLookupError: pass if sig == signal.SIGTERM and pids: time.sleep(0.5) @contextmanager def driver_session( headless: bool | None = None, recording_name: str | None = None, ) -> Generator[uc.Chrome, None, None]: profile_dir = tempfile.mkdtemp(prefix="seed-chrome-") driver: uc.Chrome | None = None recorder: SessionRecorder | None = None display: str | None = None effective_headless = config.headless if headless is None else headless try: if config.record_sessions: recorder = SessionRecorder(recording_name) if recorder.start(): display = recorder.display effective_headless = False driver = make_driver( effective_headless, profile_dir=profile_dir, display=display, ) yield driver finally: _terminate_profile_processes(profile_dir) if driver is not None: try: driver.quit() except Exception as exc: log.warning("Driver shutdown error: %s", exc) _terminate_profile_processes(profile_dir) if recorder is not None: recorder.stop() shutil.rmtree(profile_dir, ignore_errors=True) log.debug("Driver session closed") def human_delay(lo: float = 0.8, hi: float = 2.5) -> None: time.sleep(random.uniform(lo, hi))