120 lines
3.8 KiB
Python
120 lines
3.8 KiB
Python
"""
|
|
абстракции: ChallengeInfo, Detector, Solver
|
|
|
|
Пользователь передает нам:
|
|
|
|
- селектор со страницы (достаточно общий чтобы покрыть весь сайт, но не страницу антибот защиты)
|
|
- ключ сервиса распознавания капчи
|
|
"""
|
|
import asyncio
|
|
from dataclasses import dataclass
|
|
from enum import Enum
|
|
from typing import Type
|
|
|
|
from playwright.async_api import Page
|
|
from playwright.async_api import TimeoutError as PWTimeoutError
|
|
|
|
|
|
async def click_and_race(page, element, challenge_locator, timeout=30_000):
|
|
nav_task = asyncio.create_task(
|
|
page.wait_for_event(
|
|
"framenavigated",
|
|
predicate=lambda frame: frame == page.main_frame,
|
|
timeout=timeout,
|
|
)
|
|
)
|
|
challenge_task = asyncio.create_task(
|
|
challenge_locator.wait_for(state="attached", timeout=timeout)
|
|
)
|
|
|
|
await element.click()
|
|
|
|
done, pending = await asyncio.wait(
|
|
{nav_task, challenge_task}, return_when=asyncio.FIRST_COMPLETED
|
|
)
|
|
for t in pending:
|
|
t.cancel()
|
|
|
|
if nav_task in done and nav_task.exception() is None:
|
|
return "navigation"
|
|
if challenge_task in done and challenge_task.exception() is None:
|
|
return "challenge"
|
|
|
|
# оба протухли по таймауту / или упали с другой ошибкой
|
|
for t in done:
|
|
exc = t.exception()
|
|
if exc and not isinstance(exc, PWTimeoutError):
|
|
raise exc
|
|
raise PWTimeoutError("Ни навигация, ни challenge-iframe не появились вовремя")
|
|
|
|
|
|
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 # локатор на странице за защитой
|
|
detect_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
|
|
_detect_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,
|
|
detect_locator = self._detect_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, detect_locator=None, **kwargs):
|
|
self.solver = self._solver_cls(**kwargs)
|
|
self._success_locator = success_locator
|
|
self._detect_locator = detect_locator
|
|
|
|
def clear(self):
|
|
del self.solver
|
|
del self.detector
|