chore: добавлена поддержка решения капчи hcaptcha на формах

src/captcha/__init__.py

@@ -0,0 +1,2 @@
+from .hcaptcha import SolverHCaptcha
+from .base import load_captcha

src/captcha/base.py

@@ -0,0 +1,25 @@
+from playwright.async_api import Page
+from twocaptcha import TwoCaptcha
+
+
+class Captcha:
+	ctype = "captcha"
+
+	def __init__(self, api_key: str):
+		self._api_key: str = api_key
+		self._solver = TwoCaptcha(api_key)
+
+	def get_balance(self) -> float:
+		return self._solver.balance()
+
+	async def solve(self, selector: str, page: Page):
+		raise NotImplementedError()
+
+
+def load_captcha(key, captcha_type) -> Captcha:
+	for cls in Captcha.__subclasses__():
+		if cls.ctype == captcha_type:
+			break
+	else:
+		raise Exception("Invalid captcha type")
+	return cls(key)

src/captcha/hcaptcha.py

@@ -0,0 +1,60 @@
+"""
+
+dl.formRow:nth-child(3) > dd:nth-child(2) > div:nth-child(1)
+<div data-xf-init="h-captcha" data-sitekey="83433494-1c9f-446b-85c3-e27b8b7375e1" data-invisible="">
+
+iframe[data-hcaptcha-widget-id="0dm47zf0gn8t"]
+
+textarea[id="g-recaptcha-response-0dm47zf0gn8t"]
+textarea[id="h-captcha-response-0dm47zf0gn8t"]
+
+
+"""
+import logging
+
+from .base import Captcha, Page
+
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.DEBUG)
+logger.addHandler(logging.StreamHandler())
+
+
+class SolverHCaptcha(Captcha):
+	ctype = "hcaptcha"
+
+	async def solve(self, selector, page: Page):
+		div_container = page.locator(selector)
+		await div_container.wait_for()
+
+		if not div_container:
+			return None
+
+		site_key = await div_container.get_attribute("data-sitekey")
+		is_invisible = bool(await div_container.get_attribute("data-invisible"))
+
+		iframe = div_container.locator("iframe:nth-child(1)")
+		await iframe.wait_for()
+		textarea_id_postfix = await iframe.get_attribute("data-hcaptcha-widget-id")
+
+		textarea_g = await page.query_selector(f"#g-recaptcha-response-{textarea_id_postfix}")
+		textarea_h = await page.query_selector(f"#h-captcha-response-{textarea_id_postfix}")
+
+		code = await self.hcaptcha(page.url, site_key, is_invisible)
+		await textarea_g.evaluate((
+			"(el, value) => {"
+			"el.value = value;"
+			"el.dispatchEvent(new Event('input', {bubbles: true}));"
+			"}"
+		), code)
+		await textarea_h.evaluate((
+			"(el, value) => {"
+			"el.value = value;"
+			"el.dispatchEvent(new Event('input', {bubbles: true}));"
+			"}"
+		), code)
+		return None
+
+	async def hcaptcha(self, url, site_key, is_invisible):
+		resp = self._solver.hcaptcha(site_key, url, invisible=is_invisible)
+		logger.debug(f"hcaptcha response: {resp}")
+		return resp["code"]

src/engine/actions.py

@@ -2,6 +2,7 @@ import asyncio
 
 from playwright.async_api import Page
 
+from captcha import load_captcha
 from engine.schemas import ActionRequest
 
 _actions: dict = {}
@@ -13,54 +14,66 @@ def register(fun) -> None:
 
 
 @register
-async def fill(page: Page, act: ActionRequest) -> str | None:
+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 delay(page: Page, act: ActionRequest) -> str | None:
+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) -> str | None:
+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) -> str | None:
+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) -> str | None:
+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) -> str | None:
+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]) -> list[dict]:
+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)
+			result = await _actions[act.type](page, act, **ctx)
 		except Exception as e:
 			new_line["ok"] = False
 			new_line["result"] = str(e)

src/engine/schemas.py

@@ -20,8 +20,14 @@ class SolveRequest(BaseModel):
 	"""
 	url: str
 	proxy: str
-	actions: list[ActionRequest]
 	timeout: int | None = None
 	screen: str = "1280x720"
 	user_agent: str | None = None
+
+	actions: list[ActionRequest]
+
 	html: bool = False
+
+	captcha: str | None = None
+	captcha_selector: str | None = None
+	captcha_type: str | None = None

src/main.py

@@ -1,7 +1,6 @@
 import json
 import logging
 import uuid
-from datetime import datetime
 from pathlib import Path
 from typing import Annotated
 from urllib.parse import urlsplit
@@ -16,6 +15,7 @@ from fastapi import Response
 from playwright.async_api import ProxySettings
 from ua_parser import user_agent_parser
 
+import captcha
 from engine import actions
 from engine.schemas import SolveRequest
 
@@ -80,7 +80,9 @@ async def solve(
 	response: Response,
 	session_id: Annotated[str | None, Cookie()] = None
 ):
-	logger.info("Incoming payload: %s", payload)
+	api_key = request.headers.get('X-API-Key')
+
+	logger.info("Incoming payload: [%s] %s", session_id, payload)
 	global BROWSER_SLOTS
 
 	if BROWSER_SLOTS <= 0:
@@ -91,57 +93,65 @@ async def solve(
 	else:
 		BROWSER_SLOTS -= 1
 
-	api_key = request.headers.get('X-API-Key')
-
 	if not session_id:
 		session_id = str(uuid.uuid4())
 		response.set_cookie("session_id", session_id)
+		logger.info("Setup new session with session_id: %s", session_id)
 
 	scr_w, scr_h = payload.screen.split("x")
 	scr_size = int(scr_w), int(scr_h)
-	# TODO: распарс proxy-урл и заполнение объекта
-	# proxy = ProxySettings(socks5://host:port, username, password)
-	proxy = None
+	proxy = parse_proxy(payload.proxy)
 
 	user_data_dir = Path(__file__).parent / f"../extra/user_data_dir/{session_id}"
 	if not user_data_dir.exists():
 		user_data_dir.mkdir(parents=True)
 
-	ctx_opts = {
-		# "locale": payload.locale,
-		# "timezone": payload.time_zone
-	}
-
-	config = load_or_create_identity(user_data_dir, user_agent=payload.user_agent, **ctx_opts)
-
+	config = load_or_create_identity(user_data_dir, user_agent=payload.user_agent)
 	ctx_opts = {**config["context_options"], "config": config["config"]}
+
+	if payload.captcha:
+		# "dl.formRow:nth-child(3) > dd:nth-child(2) > div:nth-child(1)"
+		# hcaptcha
+		# rucaptcha api_key
+		captcha_solver = captcha.load_captcha(
+			payload.captcha,
+			payload.captcha_type
+		)
+	else:
+		captcha_solver = None
+
 	async with AsyncCamoufox(
 		persistent_context=True,
 		user_data_dir=user_data_dir,
 		proxy=proxy,
 		window=scr_size,
 		i_know_what_im_doing=True,
+		geoip=True,
 		**ctx_opts,
 	) as camoufo:
 		await camoufo.add_init_script(config["init_script"])
 		page = camoufo.pages[0] if camoufo.pages else await camoufo.new_page()
 		page.set_default_timeout(30_000)
 		await page.goto(url=payload.url)
-		actions_result = await actions.execute(page, payload.actions)
+		# if captcha_solver:
+		# 	assert payload.captcha_selector
+		# 	await captcha_solver.solve(payload.captcha_selector, page)
+		actions_result = await actions.execute(page, payload.actions, payload.captcha)
 		status = not "error" in actions_result[-1]
 		error = ["Something error", None][status]
 		cookies = await camoufo.cookies()
 		html = await page.inner_html('html')
-		await page.screenshot(
-			path=(
-				f"/home/sokol/PycharmProjects/WebRoboApi/extra/screenshots/"
-				f"{datetime.timestamp(datetime.now())}.png"
-			)
-		)
+		# await page.screenshot(
+		# 	path=(
+		# 		f"/home/sokol/PycharmProjects/WebRoboApi/extra/screenshots/"
+		# 		f"{datetime.timestamp(datetime.now())}.png"
+		# 	)
+		# )
 
 	BROWSER_SLOTS += 1
 	result = {
 		"status": ["err", "ok"][status],
+		"user_agent": ctx_opts["user_agent"],
 		"cookies": cookies,
 		"actions": actions_result,
 	}