protokeyandClaude Opus 5.5
4d90dc4a21
- Маркер интерактивного челленджа: iframe bframe, контейнер которого не стоит за экраном (top: -10000px); click_and_race ждёт видимости, а не появления в DOM. - Маркер наличия капчи берётся из исходного HTML (api.js / enterprise.js, .g-recaptcha, iframe чекбокса), проверка мгновенная, без ожидания iframe. - SolverReCaptcha ждёт detect_locator в DOM, а не видимости (script невидим). - test_main.http: секреты вынесены в http-client.private.env.json, запрос на стенд zennolab, исправлены XPath с @class. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
123 lines
4.0 KiB
Python
123 lines
4.0 KiB
Python
"""
|
|
абстракции: ChallengeInfo, Detector, Solver
|
|
|
|
Пользователь передает нам:
|
|
|
|
- селектор со страницы (достаточно общий чтобы покрыть весь сайт, но не страницу антибот защиты)
|
|
- ключ сервиса распознавания капчи
|
|
"""
|
|
import asyncio
|
|
from dataclasses import dataclass
|
|
from enum import Enum
|
|
from typing import Type
|
|
|
|
from patchright.async_api import Page
|
|
from patchright.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_locator должен совпадать только с открытым окном челленджа: элементы челленджа могут быть в DOM
|
|
# заранее (reCAPTCHA держит их за экраном), поэтому ждём видимости, а не появления
|
|
challenge_task = asyncio.create_task(
|
|
challenge_locator.first.wait_for(state="visible", 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=None, # noqa
|
|
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):
|
|
assert info.stage, "Unknown stage"
|
|
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
|