robotex/src/api/actions.py

import asyncio
import time

from patchright.async_api import Page

from antibot.orchestrator import pass_challenges
from captcha import load_captcha
from api.schemas import ActionRequest

_actions: dict = {}


class ChallengeFailed(Exception):
	pass


def register(fun) -> None:
	_actions[fun.__name__] = fun
	return fun


async def _query(page: Page, selector: str):
	element = await page.query_selector(selector)
	if element is None:
		raise LookupError(f"selector not found: {selector}")
	return element


@register
async def fill(page: Page, act: ActionRequest, **kwargs) -> str | None:
	await page.fill(act.selector, act.value)


@register
async def captcha(page: Page, act: ActionRequest, captcha_key: str, **kwargs) -> str | None:
	captcha_solver = load_captcha(captcha_key, act.value)
	await captcha_solver.solve(act.selector, page)


@register
async def delay(page: Page, act: ActionRequest, **kwargs) -> str | None:
	await asyncio.sleep(int(act.value))


@register
async def mouse_move(page: Page, act: ActionRequest, **kwargs) -> str | None:
	element = await _query(page, act.selector)
	await element.hover(scroll="auto")


@register
async def click(page: Page, act: ActionRequest, **kwargs) -> str | None:
	element = page.locator(act.selector)
	await element.click()


@register
async def submit(page: Page, act: ActionRequest, **kwargs) -> str | None:
	element = page.locator(act.selector)

	async with page.expect_event(
		"framenavigated",
		predicate=lambda frame: frame == page.main_frame,
	):
		await element.click()


@register
async def inner_html(page: Page, act: ActionRequest, **kwargs) -> str | None:
	element = await _query(page, act.selector)
	return await element.inner_html()


@register
async def wait_for(page: Page, act: ActionRequest, **kwargs) -> str | None:
	await page.wait_for_selector(act.selector, timeout=act.timeout * 1000)


@register
async def pass_challenge(
	page: Page, act: ActionRequest, *,
	antibot_name, success_locator=None, captcha_key=None, detect_locator=None,
	**kwargs
) -> str | None:
	result = await pass_challenges(
		antibot_name, page,
		captcha=captcha_key,
		success_locator=act.selector or success_locator,
		detect_locator=detect_locator
	)
	if not result.cleared:
		raise Exception(f"challenge not passed: {result.reason}")
	return result.reason


async def execute(
	page: Page, actions: list[ActionRequest],
	captcha_key: str | None = None, antibot_name: str | None = None,
	success_locator: str | None = None,
	detect_locator: str | None = None) -> list[dict]:
	"""Выполнение очереди автоматизации.

	Если антибот не пройден — бросает ChallengeFailed, действия не выполняются: на странице-заглушке
	они бы упали с невнятной ошибкой селектора."""
	actions_result = []
	ctx = {
		"captcha_key": captcha_key,
		"antibot_name": antibot_name,
		"success_locator": success_locator,
		"detect_locator": detect_locator,
	}
	outcome = await pass_challenges(
		antibot_name, page,
		captcha=captcha_key,
		success_locator=success_locator,
		detect_locator=detect_locator,
	)
	if not outcome.cleared:
		stage = outcome.last_stage.value if outcome.last_stage else None
		raise ChallengeFailed(f"{antibot_name} not passed: {outcome.reason} (stage: {stage})")

	for act in actions:
		new_line = {
			"type": act.type,
		}
		started = time.monotonic()
		try:
			result = await _actions[act.type](page, act, **ctx)
		except Exception as e:
			new_line["ok"] = False
			new_line["result"] = str(e) or type(e).__name__
		else:
			new_line["ok"] = True
			if result:
				new_line["result"] = result
		new_line["duration_ms"] = int((time.monotonic() - started) * 1000)
		actions_result.append(new_line)
		if not new_line["ok"]:
			break
	return actions_result