Files
crawle-snapp/crawler/driver.py
T

154 lines
4.8 KiB
Python

"""Stealth browser factory — bypasses Cloudflare/ArvanCloud bot detection."""
from __future__ import annotations
import os
import random
import re
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()
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) -> 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))