138 lines
5.3 KiB
Python
138 lines
5.3 KiB
Python
import json
|
|
import logging
|
|
|
|
from playwright.async_api import Page
|
|
|
|
from captcha.base import Captcha
|
|
|
|
from ..base import Detector, ChallengeInfo, Solver, AntiBotProvider, click_and_race
|
|
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: Captcha, **kwargs) -> None:
|
|
super().__init__(captcha_solver=captcha_solver, **kwargs)
|
|
self.captcha_solver: Captcha = self._opt["captcha_solver"]
|
|
|
|
async def solve(self, page: Page, info: ChallengeInfo):
|
|
log.debug("Начинаю решение recaptcha антибот преграды")
|
|
if info.stage == Stage.INTERSTITIAL:
|
|
# TODO: вытащить
|
|
await page.wait_for_timeout(15)
|
|
return
|
|
elif info.stage == Stage.CHECKBOX:
|
|
log.debug("Нужно тыкнуть чекбокс")
|
|
checkbox_selector = "#recaptcha-anchor"
|
|
element = page.frame_locator("iframe[title='reCAPTCHA']").locator(checkbox_selector)
|
|
interactive_marker_selector = "//div[@class='g-recaptcha-bubble-arrow']/..//iframe"
|
|
challenge_locator = page.locator(interactive_marker_selector)
|
|
result = await click_and_race(page, element, challenge_locator)
|
|
log.debug(f"Ожидание завершено: {result}")
|
|
log.debug("Возврат к детекту")
|
|
return
|
|
else:
|
|
log.debug("Нужно решать капчу. Закрою окошко с картинками.")
|
|
element = page.locator("xpath=//div[@class='g-recaptcha-bubble-arrow']/../div[1]")
|
|
if await element.is_visible():
|
|
await element.click()
|
|
|
|
log.debug("Решаем капчу...")
|
|
assert info.detect_locator
|
|
token = await self.captcha_solver.solve(info.detect_locator, page)
|
|
|
|
log.debug("Капча решена. Возбуждаем JS")
|
|
try:
|
|
async with page.expect_navigation(wait_until='load'):
|
|
js = """(token) => {
|
|
const clients = window.___grecaptcha_cfg && window.___grecaptcha_cfg.clients;
|
|
if (!clients) return false;
|
|
for (const cid in clients) {
|
|
for (const key in clients[cid]) {
|
|
const obj = clients[cid][key];
|
|
for (const prop in obj) {
|
|
const cb = obj[prop] && obj[prop].callback;
|
|
if (typeof cb === 'function') { cb(token); return true; }
|
|
if (typeof cb === 'string' && typeof window[cb] === 'function') {
|
|
window[cb](token); return true;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
return false;
|
|
}"""
|
|
found = await page.evaluate(f"mw:({js})({json.dumps(token)})")
|
|
assert found, "Recaptcha callback function not found"
|
|
except Exception as e:
|
|
log.warning("JS navigation error", exc_info=True)
|
|
|
|
log.debug("Все готово. Проверяем еще раз.")
|
|
return
|
|
|
|
|
|
class ReCaptchaDetector(Detector):
|
|
|
|
async def detect(self, page: Page, info: ChallengeInfo):
|
|
log.debug("ReCaptchaDetector: попытка определить есть ли преграда")
|
|
info.meta = {}
|
|
checkbox_selector = "#recaptcha-anchor"
|
|
interactive_marker_selector = "//div[@class='g-recaptcha-bubble-arrow']/..//iframe"
|
|
|
|
# попытка определить что мы вообще на странице блокировки
|
|
marker_selector = info.detect_locator or "#captcha-form"
|
|
info.detect_locator = marker_selector
|
|
|
|
log.debug(f"ищем по detect: {marker_selector} success: {info.success_locator}")
|
|
if await page.locator(marker_selector).count() == 0:
|
|
log.debug("На странице нет маркера преграды")
|
|
if info.success_locator:
|
|
info.success = bool(await page.query_selector(info.success_locator))
|
|
info.meta["reason"] = [
|
|
"Нет маркера что тут преграда и нет маркера что это искомая страница",
|
|
"Нет маркера что это преграда, зато есть маркер что это нужная страница"
|
|
][info.success]
|
|
else:
|
|
info.success = True
|
|
# noinspection unresolved-references
|
|
info.meta["reason"] = "нет маркера на это указывающего"
|
|
log.debug(f"Решение не требуется, потому что: {info.meta['reason']}")
|
|
return
|
|
|
|
log.debug("Вижу что это страница с проверкой. Определяю состояние.")
|
|
iframe = page.frame_locator("iframe[title='reCAPTCHA']")
|
|
# ожидаем когда будет загружен чекбокс
|
|
|
|
checkbox = iframe.locator(checkbox_selector)
|
|
try:
|
|
await checkbox.wait_for(state="attached")
|
|
except Exception:
|
|
log.debug("Checkbox not present")
|
|
# noinspection unresolved-references
|
|
info.meta["reason"] = "Checkbox not present"
|
|
raise IframeLoadError("Checkbox not present")
|
|
else:
|
|
log.debug("Чек бокс найден")
|
|
info.stage = Stage.CHECKBOX
|
|
|
|
if await page.locator(interactive_marker_selector).count() > 0:
|
|
log.debug("Нужно решать капчу")
|
|
# нужно решение
|
|
info.stage = Stage.INTERACTIVE
|
|
else:
|
|
log.debug("Попробую просто кликнуть")
|
|
|
|
return
|
|
|
|
|
|
@register
|
|
class ReCaptchaProvider(AntiBotProvider):
|
|
provider = 'recaptcha'
|
|
_solver_cls = ReCaptchaSolver
|
|
_detector_cls = ReCaptchaDetector
|