Парсинг сайтов как мы зделали rtfox-browser — форк undetected-chromedriver с поддержкой SOCKS5
⚠️ Статья носит познавательный и развлекательный характер. Автор не призывает к нарушению правил сервисов.
Всем привет! Недавно мы столкнулись с темой парсинга. Начали искать инструменты — BeautifulSoup, Selenium, Playwright, Puppeteer. Сразу скажу: мы не конкуренты этим библиотекам, далее объясним почему.
Нам нужно было обойти CloudFlare, запускать несколько процессов одновременно, и чтобы на каждый процесс был свой SOCKS5 прокси с авторизацией.
-
Как мы нашли отправную точку
Нам попалась библиотека undetected-chromedriver. Библиотека перестала поддерживаться разработчиком, но функционал сохранялся. При первом запуске сразу столкнулись с ошибкой — проблема была в патчинге ChromeDriver и его скачивании. После исправления запуск удался и капча была пройдена. В дальнейшем возникли проблемы с SOCKS5 и многопроцессорностью.
Тогда пришла мысль: почему бы не взять эту же библиотеку, прикрутить прокси и добавить многопроцессорность? Но чтобы это осуществить, нужно сначала понять как вообще всё работает.
-
Как устроена цепочка Chrome → ChromeDriver → Selenium
Прежде чем рассказывать что мы сделали — разберём архитектуру изнутри.
Звено 1 — Chrome.exe
Обычный браузер при запуске открывает WebSocket-сервер на каком-то порту и слушает команды DevTools Protocol (CDP). Через WebSocket можно отправить например:
{ "id": 1, "method": "Page.navigate", "params": {"url": "https://google.com"}
}
Звено 2 — ChromeDriver
ChromeDriver — это отдельная программа, которая работает как переводчик между вашим кодом (Selenium) и браузером Chrome. Вы пишете на Python driver.get("https://google.com"), Selenium превращает это в HTTP-запрос и отправляет ChromeDriver. А ChromeDriver уже объясняет Chrome на его родном языке (DevTools Protocol) что нужно открыть страницу.
ChromeDriver одновременно является:
- HTTP-сервером (слушает порт, принимает команды от Selenium)
- WebSocket-клиентом (подключается к Chrome, отправляет CDP-команды)
Звено 3 — Selenium
Selenium — это библиотека для Python которая позволяет программно управлять браузером. Selenium не управляет браузером напрямую — он отправляет команды ChromeDriver, а тот уже разбирается с Chrome.
Полная цепочка запуска
Ваш код: driver = webdriver.Chrome() ↓
Selenium запускает ChromeDriver.exe как отдельный процесс ↓
ChromeDriver.exe запускает Chrome.exe с флагами автоматизации
При запуске Chrome, ChromeDriver автоматически выставляет флаги автоматизации и сообщает что браузером управляет автоматическое ПО. Библиотека undetected-chromedriver как раз с этим и разбирается — она патчит ChromeDriver.exe, убирает эти флаги и патчит сигнатуру.
Полная цепочка запроса
Ваш код: driver.get("https://google.com") ↓
Selenium → ChromeDriver: HTTP POST /session/abc/url ↓
ChromeDriver → Chrome: WebSocket команда Page.navigate ↓
Chrome → Google: HTTP GET / ↓
Google → Chrome: HTML страница ↓
Chrome → ChromeDriver: WebSocket ответ (готово) ↓
ChromeDriver → Selenium: HTTP 200 OK ↓
Selenium: возвращает управление вашему коду
-
Что мы реализовали
1. Автоматическое определение платформы и версии Chrome
Первым делом нам нужно было автоматически определять платформу (Windows, Linux, macOS) чтобы корректно получать версию установленного Chrome.
Пример для Linux:
google-chrome --version
# Google Chrome 124.0.6367.91
Из этой строки извлекается major-версия (124).
Получение версии ChromeDriver:
- Для Chrome ≤ 114: обращается к старому CDN chromedriver.storage.googleapis.com
- Для Chrome ≥ 115: использует новый сервис chrome-for-testing, парсит JSON с актуальными версиями
Скачивание и распаковка:
- Формирует URL для скачивания ZIP-архива
- Скачивает во временную папку
- Распаковывает архив
- Ищет файл chromedriver (или .exe)
- Перемещает его в папку проекта /drivers
Патч бинарника — что такое cdc-сигнатура?
В оригинальном ChromeDriver есть строка {window.cdc_adoqpoas...} — её используют сайты для обнаружения ботов. Патч делает следующее:
- Ищет в бинарнике паттерн {window.cdc.*?;}
- Заменяет его на заглушку {console.log("ok")}
Благодаря этому сайты не могут определить что управление идёт через Selenium.
-
2. Многопроцессорность
С оригинальной библиотекой при запуске 10 процессов каждый запускал один и тот же chromedriver. Как мы уже знаем, Selenium запускает chromedriver на своём порту. Каждый новый воркер просто перезапускал chromedriver, создавал новый порт — и прошлый воркер падал.
Решение простое: мы не даём воркерам взаимодействовать с одним chromedriver. У нас есть эталонная версия (пропатченная). Каждый воркер получает только свою копию и работает с ней. При завершении копия удаляется.
Также реализована межпроцессорная блокировка:
- Если 100 воркеров одновременно вызовут ensure_driver(), они не будут скачивать драйвер 100 раз
- Первый процесс захватывает файловую блокировку (через fcntl на Linux/Mac или msvcrt на Windows)
- Остальные ждут пока первый не скачает и не пропатчит драйвер
- Затем все используют уже готовый файл
driver = uc.Chrome(worker_id="worker_1",
)
-
3. SOCKS5 прокси с авторизацией
driver = uc.Chrome(proxy={"host": "1.2.3.4",
"port": 1080,
"user": "my_login",
"pass": "my_password"})
Chrome из коробки не умеет работать с SOCKS5 и авторизацией — только HTTP/HTTPS, и при этом требует заполнять данные во всплывающем окне. Вот как мы решили эту проблему.
Мы запускаем локальный SOCKS5-сервер с двумя потоками:
- Поток-слушатель — принимает трафик от Chrome на localhost
- Поток-отправитель — подключается к внешнему прокси и отправляет логин/пароль
Chrome получает флаг --proxy-server=127.0.0.1:56789 и шлёт туда весь трафик, думая что это обычный прокси без авторизации. А наш локальный прокси сам добавляет логин и пароль и форвардит запросы на внешний SOCKS5-сервер.
Полная картина:
Chrome получает флаг --proxy-server=127.0.0.1:56789 ↓
Chrome думает: "Это обычный прокси без авторизации" ↓
Chrome отправляет HTTP-запрос на локальный порт 56789 ↓
Поток-слушатель принимает запрос ↓
Поток-отправитель: - подключается к внешнему SOCKS5 (1.2.3.4:1080) - отправляет логин и пароль - пересылает HTTP-запрос ↓
Получает ответ от внешнего прокси ↓
Возвращает ответ Chrome через поток-слушатель
-
4. Изолированное логирование через loguru
Loguru из коробки поддерживает глобальный логер. Нам это не подходило — для прокси нужен свой изолированный логер с возможностью включать через флаг, для ChromeDriver свой, для пользователя свой.
Решение — создаём новый экземпляр логера и добавляем фильтры:
_lib_logger = Logger(
core=Core(),
exception=None,
depth=0,
record=False,
lazy=False,
colors=False,
raw=False,
capture=True,
patchers=[],
extra={},
)
_lib_logger.add(sys.stderr, level="ERROR")
def get_logger(debug: bool = False, debug_proxy: bool = False):
_lib_logger.remove()
_lib_logger.add(
sys.stderr,
level="DEBUG" if debug else "ERROR",
format="{time:HH:mm:ss} | {level: <8} | "
"Chrome - {message}",
filter=lambda r: r["extra"].get("source") != "proxy"
)
_lib_logger.add(
sys.stderr,
level="DEBUG" if debug_proxy else "ERROR",
format="{time:HH:mm:ss} | {level: <8} | "
"PROXY - {message}",
filter=lambda r: r["extra"].get("source") == "proxy"
)
return _lib_logger
В Chrome и прокси привязываем источник:
self.logger = self.logger_str.bind(source="chrome")
-
5. Модуль капчи с автозагрузкой солверов
Мы реализовали автоматическую загрузку солверов из папки solvers/. Это значит: есть класс с логикой прохождения капчи — просто кладёшь .py файл в папку и метод автоматически появляется.
BaseCaptchaSolver — базовый класс:
from abc import ABC, abstractmethod
class BaseCaptchaSolver(ABC):
name: str = None
_service = None
@property
def driver(self):
return self._service.driver
@property
def api_key(self):
return self._service.api_key
@abstractmethod
def solve(self, **kwargs) -> bool:
...
Чтобы написать свой солвер достаточно:
- Унаследоваться от BaseCaptchaSolver
- Задать name
- Реализовать метод solve()
Loader — автоматический поиск солверов:
import importlib.util
import inspect
from pathlib import Path
from loguru import logger
def load_solvers_from_dir(directory: Path, registry: dict):
if not directory.exists():
logger.debug(f"Solvers directory not found: {directory}")
return
for py_file in directory.glob("*.py"):
if py_file.name.startswith("_"):
continue
try:
spec = importlib.util.spec_from_file_location(py_file.stem, py_file)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
for _, obj in inspect.getmembers(module, inspect.isclass):
if (hasattr(obj, 'name')
and obj.name
and obj.__name__ != 'BaseCaptchaSolver'):
registry[obj.name] = obj
logger.debug(f"Captcha solver loaded: '{obj.name}' ← {py_file.name}")
except Exception as e:
logger.warning(f"Failed to load solver {py_file.name}: {e}")
Loader проходит по всем .py файлам в папке solvers/, находит классы-солверы и регистрирует их по имени. Например: в папке лежит my_solver.py с классом MySolver(name="my_captcha") — Loader найдёт его и добавит в реестр под ключом "my_captcha".
CaptchaService — главный класс:
from pathlib import Path
from loguru import logger
from .loader import load_solvers_from_dir
from .base import BaseCaptchaSolver
class CaptchaService:
"""
Использование:
captcha = CaptchaService(api_key="KEY", driver=driver)
captcha.ebay_hcaptcha()
"""
def __init__(self, api_key: str, driver, solvers_dir=None):
self.api_key = api_key
self.driver = driver
self._solvers: dict[str, type[BaseCaptchaSolver]] = {}
# Ищем solvers/ рядом с main.py (cwd)
directory = Path(solvers_dir) if solvers_dir else Path.cwd() / "solvers"
load_solvers_from_dir(directory, self._solvers)
self._bind_methods()
def _instantiate(self, solver_cls):
instance = solver_cls()
instance._service = self
return instance
def _bind_methods(self):
for name, solver_cls in self._solvers.items():
self._create_method(name, solver_cls)
def _create_method(self, name: str, solver_cls):
def method(**kwargs) -> bool:
solver = self._instantiate(solver_cls)
return solver.solve(**kwargs)
method.__name__ = name
method.__doc__ = solver_cls.__doc__ or f"Solve '{name}' captcha."
setattr(self, name, method)
def register(self, solver_cls):
"""Ручная регистрация солвера."""
self._solvers[solver_cls.name] = solver_cls
self._create_method(solver_cls.name, solver_cls)
return self
def available(self) -> list[str]:
return list(self._solvers.keys())
Как использовать:
from rtfox_browser.captcha import CaptchaService
captcha = CaptchaService( api_key="YOUR_2CAPTCHA_KEY", driver=driver
)
# Посмотреть доступные солверы
print(captcha.available())
# ['ebay_hcaptcha', 'aws_image']
# Решить капчу
captcha.ebay_hcaptcha()
Свой солвер — просто бросить файл в папку:
from rtfox_browser.captcha import BaseCaptchaSolver
class MySolver(BaseCaptchaSolver):
name = "my_captcha"
def solve(self, **kwargs) -> bool:
# self.driver — браузер
# self.api_key — ключ API
return True
-
Итог
Три основные проблемы решены:
- ✅ Обход CloudFlare из коробки
- ✅ SOCKS5 прокси с авторизацией
- ✅ Многопроцессорность с изоляцией воркеров
Плюс добавили изолированное логирование и модуль капчи с плагин-архитектурой.
Библиотека доступна на GitHub и PyPI:
pip install rtfox-browser
-
Установка и использование
Установка
pip install rtfox-browser
Требования: Python >= 3.9
Базовый запуск
import rtfox_browser as uc
driver = uc.Chrome()
driver.get("https://example.com")
driver.quit()
Запуск с прокси
import rtfox_browser as uc
driver = uc.Chrome(proxy={"host": "1.2.3.4",
"port": 1080,
"user": "my_login",
"pass": "my_password"})
driver.get("https://example.com")
driver.quit()
Запуск нескольких процессов
from multiprocessing import Pool
import rtfox_browser as uc
def run_worker(worker_id):
driver = uc.Chrome(worker_id=worker_id,proxy={"host": "1.2.3.4",
"port": 1080,
"user": "my_login",
"pass": "my_password"})
driver.get("https://example.com")
driver.quit()
if __name__ == "__main__":
with Pool(4) as pool:
pool.map(run_worker, ["w1", "w2", "w3", "w4"])
Решение капчи
import rtfox_browser as uc
from rtfox_browser.captcha import CaptchaService
driver = uc.Chrome()
driver.get("https://example.com")
captcha = CaptchaService(api_key="YOUR_2CAPTCHA_KEY",
driver=driver
)
# Посмотреть доступные солверы
print(captcha.available())
# ['ebay_hcaptcha', 'aws_image']
# Решить капчу
captcha.ebay_hcaptcha()
Обработка ошибок прокси
from rtfox_browser.exceptions import ProxyError, ProxyAuthError, ProxyInvalidAddressError
try:
driver = uc.Chrome(proxy={...})
except ProxyInvalidAddressError:
print("Invalid proxy address")
except ProxyAuthError:
print("Wrong credentials")
except ProxyError:
print("Proxy connection failed")
Включение отладочных логов
import rtfox_browser as uc
# Логи ChromeDriver
driver = uc.Chrome(log_debug=True)
# Логи прокси
driver = uc.Chrome(log_debug_proxy=True)
# Все логи
driver = uc.Chrome(log_debug=True, log_debug_proxy=True)
-Источник
