chore: настроен обход антибот защиты

This commit is contained in:
protokey committed 2026-09-21 19:33:49 +04:00
1 parent 4cc068a616
commit 1cfd9bafef
6 files changed
+118 -66

No files matched your search

+30 -30
View File
@@ -16,36 +16,36 @@ 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)
)
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()
await element.click()
done, pending = await asyncio.wait(
{nav_task, challenge_task}, return_when=asyncio.FIRST_COMPLETED
)
for t in pending:
t.cancel()
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"
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 не появились вовремя")
# оба протухли по таймауту / или упали с другой ошибкой
for t in done:
exc = t.exception()
if exc and not isinstance(exc, PWTimeoutError):
raise exc
raise PWTimeoutError("Ни навигация, ни challenge-iframe не появились вовремя")
class Stage(str, Enum):
@@ -97,10 +97,10 @@ class AntiBotProvider:
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,
stage=Stage.INTERSTITIAL,
provider=self.provider,
success_locator=self._success_locator,
detect_locator=self._detect_locator,
)
# обновление info
await self.detector.detect(page, info)