"""Stealth browser factory — bypasses Cloudflare/ArvanCloud bot detection.""" from __future__ import annotations import os import random import shutil import subprocess import sys 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 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() major = int(out.split()[-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) -> Options: opts = Options() opts.binary_location = chrome_binary if headless: opts.add_argument("--headless=new") opts.add_argument(f"--user-agent={user_agent}") opts.add_argument("--no-sandbox") opts.add_argument("--disable-dev-shm-usage") opts.add_argument("--disable-gpu") 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) -> 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) 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, ) 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 @contextmanager def driver_session(headless: bool | None = None) -> Generator[uc.Chrome, None, None]: driver = make_driver(headless) try: yield driver finally: driver.quit() log.debug("Driver session closed") def human_delay(lo: float = 0.8, hi: float = 2.5) -> None: time.sleep(random.uniform(lo, hi))