Парсинг сайтов как мы зделали 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="<green>{time:HH:mm:ss}</green> | <level>{level: <8}</level> | " "<cyan>Chrome</cyan> - <level>{message}</level>", filter=lambda r: r["extra"].get("source") != "proxy" ) _lib_logger.add( sys.stderr, level="DEBUG" if debug_proxy else "ERROR", format="<blue>{time:HH:mm:ss}</blue> | <level>{level: <8}</level> | " "<magenta>PROXY</magenta> - <level>{message}</level>", 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, abstractmethodclass 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.utilimport inspectfrom pathlib import Pathfrom loguru import loggerdef 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 Pathfrom loguru import loggerfrom .loader import load_solvers_from_dirfrom .base import BaseCaptchaSolverclass 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 CaptchaServicecaptcha = CaptchaService( api_key="YOUR_2CAPTCHA_KEY", driver=driver)# Посмотреть доступные солверыprint(captcha.available())# ['ebay_hcaptcha', 'aws_image']# Решить капчуcaptcha.ebay_hcaptcha()
Свой солвер — просто бросить файл в папку:
from rtfox_browser.captcha import BaseCaptchaSolverclass 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
-
💬 Telegram: https://t.me/rtf_labs_studio
-
🎥 YouTube: https://www.youtube.com/@RTF_Labs_Studio
Установка и использование
Установка
pip install rtfox-browser
Требования: Python >= 3.9
Базовый запуск
import rtfox_browser as ucdriver = uc.Chrome()driver.get("https://example.com")driver.quit()
Запуск с прокси
import rtfox_browser as ucdriver = 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 Poolimport rtfox_browser as ucdef 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 ucfrom rtfox_browser.captcha import CaptchaServicedriver = 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, ProxyInvalidAddressErrortry: 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# Логи ChromeDriverdriver = uc.Chrome(log_debug=True)# Логи проксиdriver = uc.Chrome(log_debug_proxy=True)# Все логиdriver = uc.Chrome(log_debug=True, log_debug_proxy=True)
ссылка на оригинал статьи https://habr.com/ru/articles/1030550/