protokeyandClaude Opus 5.5
437ea27be7
- слоты браузеров: asyncio.Semaphore, размер из BROWSER_SLOTS, при нехватке — 503 + Retry-After; устранена утечка счётчика при отказе (#12, #49) - session_id принимается только как UUID, путь профиля проверяется (#5) - секреты клиента в логах маскируются, полный дамп — только на DEBUG (#6) - ответ: status всегда "ok"/"err", поле code (no_slots / request_failed / challenge_failed / action_failed), response_model (#7) - непройденный антибот → challenge_failed, действия не выполняются (#8) - таймауты провайдеров в мс вынесены в константы; detect переживает навигацию посреди проверки; пауза между неудачными solve вместо холостого цикла (#9) - обязательные поля действий и формат прокси проверяются схемой → 422 до запуска браузера (#13) - прокси необязателен (#15) - user_agent сохраняется в профиле сессии и реально передаётся в браузер (#16) - duration_ms у действий и у запроса (#18), опция screenshot (#19) - общий конфиг логирования (LOG_LEVEL), print → логгер, CloudflareProvider, удалён cached_persistent, GET /health (#22) Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
44 lines
1.1 KiB
Python
44 lines
1.1 KiB
Python
import asyncio
|
|
|
|
from patchright.async_api import Page
|
|
from twocaptcha import TwoCaptcha
|
|
|
|
|
|
class Captcha:
|
|
ctype = "captcha"
|
|
|
|
def __init__(self, api_key: str):
|
|
self._api_key: str = api_key
|
|
self._solver = TwoCaptcha(api_key)
|
|
|
|
def get_balance(self) -> float:
|
|
return self._solver.balance()
|
|
|
|
async def solve(self, selector: str, page: Page) -> str:
|
|
raise NotImplementedError()
|
|
|
|
async def _run(self, fun, /, *args, **kwargs):
|
|
resp = await asyncio.to_thread(fun, *args, **kwargs)
|
|
return resp["code"]
|
|
|
|
async def hcaptcha(self, url, site_key, is_invisible):
|
|
resp = await self._run(self._solver.hcaptcha, site_key, url, invisible=is_invisible)
|
|
return resp
|
|
|
|
async def recaptcha(self, url, site_key):
|
|
resp = await self._run(self._solver.recaptcha, site_key, url)
|
|
return resp
|
|
|
|
async def turnstile(self, url, site_key):
|
|
resp = await self._run(self._solver.turnstile, site_key, url)
|
|
return resp
|
|
|
|
|
|
def load_captcha(key, captcha_type) -> Captcha:
|
|
for cls in Captcha.__subclasses__():
|
|
if cls.ctype == captcha_type:
|
|
break
|
|
else:
|
|
raise Exception("Invalid captcha type")
|
|
return cls(key)
|