import asyncio from playwright.async_api import Page 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) async def execute(page: Page, actions: list[ActionRequest], captcha: str | None = None) -> list[dict]: actions_result = [] ctx = { "captcha_key": captcha, } 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