feature!: добавлен механизм обхода антибот защиты

src/antibot/base.py

@@ -0,0 +1,81 @@
+"""
+абстракции: ChallengeInfo, Detector, Solver
+
+Пользователь передает нам:
+
+- селектор со страницы (достаточно общий чтобы покрыть весь сайт, но не страницу антибот защиты)
+- ключ сервиса распознавания капчи
+"""
+from copy import deepcopy
+from dataclasses import dataclass
+from enum import Enum
+from typing import Type
+
+from playwright.async_api import Page
+
+
+class Stage(str, Enum):
+	INTERSTITIAL = "interstitial"
+	CHECKBOX = "checkbox"
+	INTERACTIVE = "interactive"
+
+
+@dataclass
+class ChallengeInfo:
+	stage: Stage
+	provider: str  # "cloudflare_turnstile", "datadome", "hcaptcha", "recaptcha_v2", ...
+	success: bool = False
+	success_locator: str | None = None  # локатор на странице за защитой
+	site_key: str | None = None  # для стадии C, чтобы передать в 2captcha
+	meta: dict | None = None  # что-то специфичное для провайдера
+
+
+class Solver:
+	"""Алгоритм прохождения. Один солвер = один антибот провайдер"""
+
+	def __init__(self, **kwargs):
+		self._opt = kwargs
+
+	async def solve(self, page: Page, info: ChallengeInfo):
+		"""Возвращает True, если стадия пройдена (или больше не требуется действий)."""
+		raise NotImplementedError
+
+
+class Detector:
+	"""Один детектор = одна сигнатура одного провайдера."""
+
+	async def detect(self, page: Page, info: ChallengeInfo):
+		raise NotImplementedError
+
+
+class AntiBotProvider:
+	provider: str
+	detector: Detector
+	solver: Solver
+	_solver_cls: Type[Solver]
+	_detector_cls: Type[Detector]
+	_success_locator: str | None = None
+
+	def __init__(self):
+		self.detector = self._detector_cls()
+
+	async def detect(self, page: Page) -> ChallengeInfo:
+		info = ChallengeInfo(
+			stage = Stage.INTERSTITIAL,
+			provider = self.provider,
+			success_locator = self._success_locator,
+		)
+		# обновление info
+		await self.detector.detect(page, info)
+		return info
+
+	async def solve(self, page: Page, info: ChallengeInfo):
+		await self.solver.solve(page, info)
+
+	def configure(self, success_locator=None, **kwargs):
+		self.solver = self._solver_cls(**kwargs)
+		self._success_locator = success_locator
+
+	def clear(self):
+		del self.solver
+		del self.detector

src/antibot/exceptions.py

@@ -0,0 +1,6 @@
+class ProviderNotDetected(Exception):
+	pass
+
+
+class IframeLoadError(Exception):
+	pass

src/antibot/humanize.py

@@ -0,0 +1,62 @@
+"""
+человекоподобные движения мыши (безье+ускорение), тайминги кликов
+
+Camoufox — это форк Firefox, заточенный под антидетект-автоматизацию (спуфинг fingerprint'а, WebGL, canvas,
+шрифтов и т.д.), с Python-обёрткой поверх Playwright-протокола (Juggler).
+
+У него есть параметр humanize при запуске:
+
+```
+from camoufox.sync_api import Camoufox
+
+with Camoufox(humanize=True) as browser:
+    ...
+```
+
+Или с указанием максимального времени на движение:
+
+```
+Camoufox(humanize=1.5)  # курсор доберётся до цели не более чем за 1.5 сек
+```
+
+Когда humanize включён, при вызовах вроде page.click() / locator.click() библиотека:
+
+- строит траекторию курсора от текущей позиции до цели не по прямой, а по кривой (используется подход,
+похожий на ghost-cursor — Безье-путь с шумом),
+- разбивает движение на промежуточные точки с нелинейной скоростью (ускорение/замедление к цели, как у реальной руки),
+- добавляет небольшую случайную задержку перед самим кликом.
+
+Это реализовано на уровне их собственного форка драйвера, а не через голый Playwright API — обычный Playwright (
+page.click()) телепортирует курсор мгновенно и такого не делает.
+"""
+import random
+import string
+
+from playwright.async_api import Page
+
+
+async def typing(page: Page, locator: str, text: str):
+	element = page.locator(locator)
+
+	await element.scroll_into_view_if_needed()
+	assert await element.is_visible()
+	assert await element.is_enabled()
+	assert await element.is_editable()
+	await element.focus()
+	await element.click()
+
+	for char in text:
+		await page.keyboard.press(char)
+
+		if char in string.ascii_lowercase:
+			delays = (50, 100)
+		elif char in string.ascii_uppercase:
+			delays = (100, 150)
+		elif char in string.digits:
+			delays = (100, 150)
+		elif char in string.punctuation:
+			delays = (100, 200)
+		else:
+			raise SystemError("character not supported: {}".format(char))
+
+		await page.wait_for_timeout(random.uniform(*delays))

src/antibot/known_providers/cloudflare.py

@@ -0,0 +1,20 @@
+from playwright.async_api import Page
+
+from ..base import Detector, ChallengeInfo
+from ..base import Stage
+from ..registry import register_detector
+
+
+@register_detector
+class CloudflareDetector(Detector):
+	provider = 'cloudflare_turnstile'
+	stage = Stage.INTERSTITIAL
+
+	async def detect(self, page: Page) -> ChallengeInfo:
+		result = ChallengeInfo(
+			stage=self.stage,
+			provider=self.provider,
+		)
+		# TODO: добавить функцию определения что страница с проверкой показана
+		raise NotImplementedError
+		return result

src/antibot/known_providers/recaptcha.py

@@ -0,0 +1,81 @@
+import logging
+
+from playwright.async_api import Page
+
+from ..base import Detector, ChallengeInfo, Solver, AntiBotProvider
+from ..base import Stage
+from ..registry import register
+from ..exceptions import IframeLoadError
+
+log = logging.getLogger(__name__)
+log.setLevel(logging.DEBUG)
+log.addHandler(logging.StreamHandler())
+
+
+class ReCaptchaSolver(Solver):
+
+	def __init__(self, captcha_solver) -> None:
+		super().__init__(captcha_solver=captcha_solver)
+		self.captcha_solver = self._opt["captcha_solver"]
+
+	async def solve(self, page: Page, info: ChallengeInfo):
+		if info.stage == Stage.INTERSTITIAL:
+			await page.wait_for_timeout(15)
+		elif info.stage == Stage.CHECKBOX:
+			checkbox_selector = "#recaptcha-anchor"
+			element = page.locator(checkbox_selector)
+			async with page.expect_response("**/recaptcha/enterprise/reload**", timeout=30_000):
+				await element.click()
+		else:
+			element = page.locator("body > div:nth-child(2) > div:nth-child(1)")
+			if await element.is_visible():
+				await element.click()
+
+			self.captcha_solver.recaptcha(page)
+			# TODO: execute js
+
+class ReCaptchaDetector(Detector):
+
+	async def detect(self, page: Page, info: ChallengeInfo):
+		info.meta = {}
+		checkbox_selector = "#recaptcha-anchor"
+		interactive_marker_selector = "body > div:nth-child(2) > div:nth-child(4) > iframe:nth-child(1)"
+
+		# попытка определить что мы вообще на странице блокировки
+		marker_selector = "#captcha-form"
+		if not await page.query_selector(marker_selector):
+			log.debug("Page is not a challenge form")
+			if info.success_locator:
+				info.success = bool(await page.query_selector(info.success_locator))
+				info.meta["reason"] = [
+					"Success locator not present and marker too.",
+					"Marker not present but success locator is it!"
+				][info.success]
+			else:
+				info.success = True
+				# noinspection unresolved-references
+				info.meta["reason"] = "Page is not a challenge form"
+			return
+
+		# ожидаем когда будет загружен чекбокс
+		if not await page.wait_for_selector(checkbox_selector, timeout=30_000):
+			log.debug("Checkbox not present")
+			# noinspection unresolved-references
+			info.meta["reason"] = "Checkbox not present"
+			raise IframeLoadError("Checkbox not present")
+		else:
+			# нужно сделать клик
+			info.stage = Stage.CHECKBOX
+
+		if await page.wait_for_selector(interactive_marker_selector, timeout=15_000):
+			# нужно решение
+			info.stage = Stage.INTERACTIVE
+
+		return
+
+
+@register
+class ReCaptchaProvider(AntiBotProvider):
+	provider = 'recaptcha'
+	_solver_cls = ReCaptchaSolver
+	_detector_cls = ReCaptchaDetector

src/antibot/orchestrator.py

@@ -0,0 +1,57 @@
+"""
+цикл: detect -> solve -> verify -> retry
+"""
+import logging
+import time
+from dataclasses import dataclass
+
+from captcha import load_captcha
+
+from .base import Stage
+from .registry import select_one
+from .exceptions import ProviderNotDetected
+
+log = logging.getLogger(__name__)
+log.setLevel(logging.DEBUG)
+log.addHandler(logging.StreamHandler())
+
+
+@dataclass
+class ChallengeOutcome:
+	cleared: bool
+	last_stage: Stage | None = None  # на чём застряли, если cleared=False
+	last_provider: str | None = None
+	reason: str | None = None  # "no captcha_key" / "timeout" / ...
+
+
+async def pass_challenges(
+	antibot_name, page, *, captcha=None, success_locator: str | None = None, timeout_ms=120_000) -> ChallengeOutcome:
+	# установка дедлайна
+	deadline = time.monotonic() + timeout_ms / 1000
+
+	# 1. определить кто блокирует
+	try:
+		if captcha:
+			captcha_solver = load_captcha(captcha, antibot_name)
+		else:
+			captcha_solver = None
+		provider = await select_one(
+			page, antibot_name, captcha_solver=captcha_solver, success_locator=success_locator)
+	except ProviderNotDetected:
+		return ChallengeOutcome(False, None, None, "Provider not detected")
+
+	info = None
+	while time.monotonic() < deadline:
+		info = await provider.detect(page)  # noqa
+		# 3. проверяем что решение успешно
+		if info.success:  # noqa
+			return ChallengeOutcome(True, reason=(info.meta or {}).get("reason"))
+		# 2. передаем страницу в класс и ждем решения
+		await provider.solve(page, info)  # noqa
+	else:
+		return ChallengeOutcome(
+			False,
+			info.stage if info else None,
+			info.provider if info else None,
+			"timeout"
+		)

src/antibot/registry.py

@@ -0,0 +1,32 @@
+"""
+реестр детекторов, порядок проверки
+"""
+from playwright.async_api import Page
+
+from .base import AntiBotProvider
+from .exceptions import ProviderNotDetected
+
+_detectors: list[AntiBotProvider] = []
+
+
+def register(provider_cls) -> AntiBotProvider:
+	_detectors.append(provider_cls())
+	return provider_cls
+
+
+async def select_any(page: Page, **kwargs) -> AntiBotProvider:
+	for provider in _detectors:
+		if await provider.detect(page):
+			provider.configure(**kwargs)
+			return provider
+	else:
+		raise ProviderNotDetected()
+
+
+async def select_one(page: Page, provider_name: str, **kwargs) -> AntiBotProvider:
+	for provider in _detectors:
+		if provider.provider == provider_name:
+			provider.configure(**kwargs)
+			return provider
+	else:
+		raise ProviderNotDetected()

src/antibot/stages/checkbox.py

@@ -0,0 +1,20 @@
+"""
+стадия B: универсальный click-solver + сигнатуры провайдеров
+"""
+
+from ..registry import detect_any
+from ..base import Stage
+
+
+async def solve_checkbox(page, info, max_attempts=3):
+	for attempt in range(max_attempts):
+		frame = page.locator(page, info.frame_locator)
+
+
+
+		# после клика — либо чисто, либо эскалация в картинки (стадия C)
+		if await detect_any(page, Stage.CHECKBOX) is None:
+			return True
+		if await detect_any(page, Stage.INTERACTIVE):
+			return False  # передаём выше, пусть оркестратор переключит стадию
+	return False

src/antibot/stages/interactive.py

@@ -0,0 +1,4 @@
+# стадия C: мост к src/captcha (hcaptcha/recaptcha/...)
+async def solve_interactive(page, info, captcha):
+	await captcha.solve(info.frame_locator or info.meta["container_selector"], page)
+	return True

src/antibot/stages/interstitial.py

@@ -0,0 +1,15 @@
+"""
+стадия A: универсальный wait-solver + сигнатуры провайдеров
+"""
+from ..base import Stage
+from ..registry import detect_any
+
+
+async def solve_interstitial(page, info, timeout_ms=15000, poll_ms=500):
+	elapsed = 0
+	while elapsed < timeout_ms:
+		if await detect_any(page, Stage.INTERSTITIAL) is None:
+			return True
+		await page.wait_for_timeout(poll_ms)
+		elapsed += poll_ms
+	return False