chore: перетащил обертку над браузером из scrapling

src/api/actions.py

@@ -0,0 +1,134 @@
+import asyncio
+
+from patchright.async_api import Page
+
+from antibot.base import click_and_race
+from antibot.orchestrator import pass_challenges
+from captcha import load_captcha
+from api.schemas import ActionRequest
+
+_actions: dict = {}
+
+
+def register(fun) -> None:
+	_actions[fun.__name__] = fun
+	return fun
+
+
+@register
+async def fill(page: Page, act: ActionRequest, **kwargs) -> str | None:
+	assert act.selector
+	assert act.value
+	await page.fill(act.selector, act.value)
+
+
+@register
+async def captcha(page: Page, act: ActionRequest, captcha_key: str, **kwargs) -> str | None:
+	assert act.selector
+	assert act.value
+	assert captcha_key
+	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:
+	assert act.value
+	await asyncio.sleep(int(act.value))
+
+
+@register
+async def mouse_move(page: Page, act: ActionRequest, **kwargs) -> str | None:
+	assert act.selector
+	element = await page.query_selector(act.selector)
+	await element.hover(scroll="auto")
+
+
+@register
+async def click(page: Page, act: ActionRequest, **kwargs) -> str | None:
+	assert act.selector
+	element = page.locator(act.selector)
+	await element.click()
+
+
+@register
+async def submit(page: Page, act: ActionRequest, **kwargs) -> str | None:
+	assert act.selector
+	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:
+	assert act.selector
+	element = await page.query_selector(act.selector)
+	return await element.inner_html()
+
+
+@register
+async def wait_for(page: Page, act: ActionRequest, **kwargs) -> str | None:
+	assert act.selector
+	assert 1 <=  act.timeout <= 7200
+	try:
+		await page.wait_for_selector(act.selector, timeout=act.timeout * 1000)
+	except Exception as e:
+		return str(e)
+
+
+@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
+	)
+	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]:
+	"""Выполнение очереди автоматизации"""
+	actions_result = []
+	ctx = {
+		"captcha_key": captcha_key,
+		"antibot_name": antibot_name,
+		"success_locator": success_locator,
+		"detect_locator": detect_locator,
+	}
+	await pass_challenges(
+		antibot_name, page,
+		captcha=captcha_key,
+		success_locator=success_locator,
+		detect_locator=detect_locator,
+	)
+	for act in actions:
+		new_line = {
+			"type": act.type,
+		}
+		try:
+			# TODO: добавить время выполнения
+			result = await _actions[act.type](page, act, **ctx)
+		except Exception as e:
+			new_line["ok"] = False
+			new_line["result"] = str(e)
+			actions_result.append(new_line)
+			break
+		else:
+			new_line["ok"] = True
+			if result:
+				new_line["result"] = result
+			actions_result.append(new_line)
+	return actions_result if len(actions_result) else [{"status": "pass"}]