robotex/src/antibot/known_providers/recaptcha.py

import logging

from patchright.async_api import Page

from captcha.base import Captcha

from ..base import Detector, ChallengeInfo, Solver, AntiBotProvider, click_and_race
from ..base import Stage, PWTimeoutError
from ..registry import register
from ..exceptions import IframeLoadError

log = logging.getLogger(__name__)

# iframe окна с картинками (bframe) вставляется в DOM вместе с чекбоксом, но его контейнер стоит за экраном:
# style="visibility: hidden; ... top: -10000px". Когда выпадает челлендж, контейнер переезжает на экран и
# становится видимым — локатор совпадает только с открытым окном
INTERACTIVE_MARKER_SELECTOR = (
	"xpath=//iframe[contains(@src, '/bframe')][not(ancestor::div[contains(@style, 'top: -10000px')])]"
)

# признак reCAPTCHA, который есть уже в исходном HTML, до загрузки iframe с чекбоксом (он грузится долго):
# скрипт api.js / enterprise.js, контейнер .g-recaptcha (у render=explicit его может не быть) или сам iframe.
# #recaptcha-token не подходит — он внутри iframe чекбокса, с основной страницы его не видно
DETECT_MARKER_SELECTOR = (
	"script[src*='/recaptcha/api.js'], script[src*='/recaptcha/enterprise.js'], "
	".g-recaptcha, iframe[title='reCAPTCHA']"
)

JS_CHALLENGE_SUBMIT = """(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;
}"""


class ReCaptchaSolver(Solver):
	page: Page
	# пауза на стадии INTERSTITIAL перед повторным detect
	INTERSTITIAL_WAIT_MS = 15_000

	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 form_submit(self, token: str) -> bool:
		# ___grecaptcha_cfg живёт в основном мире страницы, а patchright по умолчанию выполняет скрипт в изолированном
		return await self.page.evaluate(JS_CHALLENGE_SUBMIT, token, isolated_context=False)

	async def solve(self, page: Page, info: ChallengeInfo):
		self.page = page
		log.debug("Начинаю решение recaptcha антибот преграды")
		if info.stage == Stage.INTERSTITIAL:
			await page.wait_for_timeout(self.INTERSTITIAL_WAIT_MS)
			return
		elif info.stage == Stage.CHECKBOX:
			log.debug("Нужно тыкнуть чекбокс")
			checkbox_selector = "#recaptcha-anchor"
			element = page.frame_locator("iframe[title='reCAPTCHA']").locator(checkbox_selector)
			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("Нужно решать капчу. Закрою окошко с картинками.")
			# полупрозрачная подложка на весь экран (position: fixed), клик по ней закрывает окно. В её центре лежит
			# само окно с картинками и перехватывает клик — поэтому кликаем в угол
			element = page.locator("xpath=//div[@class='g-recaptcha-bubble-arrow']/../div[1]")
			if await element.is_visible():
				try:
					await element.click(position={"x": 1, "y": 1}, timeout=5_000)
				except PWTimeoutError:
					# не закрылось — не страшно, токен уходит через callback
					log.debug("Окошко с картинками закрыть не удалось")

			log.debug("Решаем капчу...")
			assert info.detect_locator
			token = await self.captcha_solver.solve(info.detect_locator, page)

			log.debug("Капча решена. Возбуждаем JS")
			async with page.expect_navigation(wait_until='load'):
				found = await self.form_submit(token)
				assert found, "Recaptcha callback function not found"

			log.debug("Все готово. Проверяем еще раз.")
			return


class ReCaptchaDetector(Detector):

	async def detect(self, page: Page, info: ChallengeInfo):
		log.debug("ReCaptchaDetector: попытка определить есть ли преграда")
		info.meta = {}
		checkbox_selector = "#recaptcha-anchor"

		# попытка определить что мы вообще на странице блокировки
		marker_selector = info.detect_locator or DETECT_MARKER_SELECTOR
		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']")
		# маркер есть — значит капча будет, ждём пока догрузится iframe с чекбоксом

		info.stage = Stage.INTERSTITIAL
		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).first.is_visible():
			log.debug("Нужно решать капчу")
			# нужно решение
			info.stage = Stage.INTERACTIVE
		else:
			log.debug("Попробую просто кликнуть")

		return


@register
class ReCaptchaProvider(AntiBotProvider):
	provider = 'recaptcha'
	_solver_cls = ReCaptchaSolver
	_detector_cls = ReCaptchaDetector