117 lines
2.9 KiB
Python
117 lines
2.9 KiB
Python
import asyncio
|
|
|
|
from playwright.async_api import Page
|
|
|
|
from antibot.orchestrator import pass_challenges
|
|
from captcha import load_captcha
|
|
from engine.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 = await page.query_selector(act.selector)
|
|
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
|
|
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:
|
|
await pass_challenges(
|
|
antibot_name, page,
|
|
captcha=captcha_key,
|
|
success_locator=act.selector or success_locator,
|
|
detect_locator=detect_locator
|
|
)
|
|
|
|
|
|
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:
|
|
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"}]
|