Files
crawle-snapp/crawler/driver.py
T

150 lines
4.7 KiB
Python

"""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,
)
if not use_headless:
driver.maximize_window()
else:
driver.set_window_size(1920, 1080)
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))