diff --git a/pyproject.toml b/pyproject.toml index 36297891..8175c19c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -66,6 +66,9 @@ keywords = [ "captcha", "turnstile", "amazon", "amazon_waf", + "vk-captcha", + "fox-captcha", + "temu-captcha", "friendly-captcha" ] license = "MIT" @@ -78,6 +81,7 @@ classifiers = [ "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", "Framework :: AsyncIO", "Operating System :: Unix", "Operating System :: Microsoft :: Windows", diff --git a/src/python_rucaptcha/__version__.py b/src/python_rucaptcha/__version__.py index f0a00b37..b61ecefd 100644 --- a/src/python_rucaptcha/__version__.py +++ b/src/python_rucaptcha/__version__.py @@ -1 +1 @@ -__version__ = "6.4.0" +__version__ = "6.5.0" diff --git a/src/python_rucaptcha/core/base.py b/src/python_rucaptcha/core/base.py index 44f32be5..af5ff582 100644 --- a/src/python_rucaptcha/core/base.py +++ b/src/python_rucaptcha/core/base.py @@ -3,7 +3,7 @@ import uuid import base64 import asyncio -from typing import Optional +from typing import Any, Optional from pathlib import Path import aiohttp @@ -30,16 +30,32 @@ def __init__( rucaptcha_key: str, method: str, sleep_time: int = 10, - service_type: str = ServiceEnm.TWOCAPTCHA.value, - **kwargs, + service_type: ServiceEnm | str = ServiceEnm.TWOCAPTCHA, + **kwargs: dict[str, Any], ): """ - :param rucaptcha_key: User API key - :param method: Captcha type - :param sleep_time: Time to wait for captcha solution - :param service_type: URL with which the program will work, "2captcha" option is possible (standard) - and "rucaptcha" - :param kwargs: Designed to pass OPTIONAL parameters to the payload for a request to RuCaptcha + Base class for interacting with CAPTCHA-solving services such as 2Captcha and RuCaptcha. + + This class handles the setup of request payloads, session configuration, and service-specific + parameters required to submit CAPTCHA tasks and retrieve their results. It supports optional + customization of task parameters via keyword arguments and includes retry logic for HTTP requests. + + Args: + rucaptcha_key (str): + API key provided by the CAPTCHA-solving service. + method (str): + Type of CAPTCHA to solve (e.g., "ImageToText", "ReCaptchaV2"). + sleep_time (int, optional): + Time in seconds to wait between polling attempts. Defaults to 10. + service_type (ServiceEnm | str, optional): + Service provider to use. Accepts `ServiceEnm.TWOCAPTCHA` or `"rucaptcha"`. Defaults to TWOCAPTCHA. + **kwargs (dict[str, Any]): + Optional parameters to be injected into the task payload (e.g., `websiteURL`, `siteKey`, `proxy`). + + Example: + >>> captcha = BaseCaptcha("your-api-key", method="ReCaptchaV2", websiteURL="https://example.com", siteKey="abc123") + >>> captcha.create_task_payload + {'clientKey': 'your-api-key', 'task': {'type': 'ReCaptchaV2', 'websiteURL': 'https://example.com', 'siteKey': 'abc123'}} """ self.result = GetTaskResultResponseSer() # assign args to validator @@ -48,7 +64,7 @@ def __init__( # prepare create task payload self.create_task_payload = CreateTaskBaseSer( - clientKey=rucaptcha_key, task=TaskSer(type=method).to_dict() + clientKey=rucaptcha_key, task=TaskSer(type=method) ).to_dict() # prepare get task result data payload self.get_task_payload = GetTaskResultRequestSer(clientKey=rucaptcha_key) @@ -61,7 +77,7 @@ def __init__( self.session.mount("http://", HTTPAdapter(max_retries=RETRIES)) self.session.mount("https://", HTTPAdapter(max_retries=RETRIES)) - def _processing_response(self, **kwargs: dict) -> dict: + def _processing_response(self, **kwargs: dict[str, Any]) -> dict[str, Any]: """ Method processing captcha solving task creation result :param kwargs: additional params for Requests library @@ -90,13 +106,13 @@ def _processing_response(self, **kwargs: dict) -> dict: url_response=self.params.url_response, ) - def url_open(self, url: str, **kwargs): + def url_open(self, url: str, **kwargs: dict[str, Any]): """ Method open links """ return self.session.get(url=url, **kwargs) - async def aio_url_read(self, url: str, **kwargs) -> bytes: + async def aio_url_read(self, url: str, **kwargs: dict[str, Any]) -> bytes | None: """ Async method read bytes from link """ @@ -106,7 +122,7 @@ async def aio_url_read(self, url: str, **kwargs) -> bytes: async with session.get(url=url, **kwargs) as resp: return await resp.content.read() - async def _aio_processing_response(self) -> dict: + async def _aio_processing_response(self) -> dict[str, Any]: """ Method processing async captcha solving task creation result """ @@ -167,23 +183,24 @@ def _file_const_saver(self, content: bytes, file_path: str, file_extension: str def _body_file_processing( self, - save_format: SaveFormatsEnm, + save_format: SaveFormatsEnm | str, file_path: str, file_extension: str = "png", - captcha_link: Optional[str] = None, - captcha_file: Optional[str] = None, - captcha_base64: Optional[bytes] = None, - **kwargs, + image_key: str = "body", + captcha_link: str | None = None, + captcha_file: str | None = None, + captcha_base64: bytes | None = None, + **kwargs: dict[str, Any], ): # if a local file link is passed if captcha_file: self.create_task_payload["task"].update( - {"body": base64.b64encode(self._local_file_captcha(captcha_file)).decode("utf-8")} + {image_key: base64.b64encode(self._local_file_captcha(captcha_file)).decode("utf-8")} ) # if the file is transferred in base64 encoding elif captcha_base64: self.create_task_payload["task"].update( - {"body": base64.b64encode(captcha_base64).decode("utf-8")} + {image_key: base64.b64encode(captcha_base64).decode("utf-8")} ) # if a URL is passed elif captcha_link: @@ -192,7 +209,9 @@ def _body_file_processing( # according to the value of the passed parameter, select the function to save the image if save_format == SaveFormatsEnm.CONST.value: self._file_const_saver(content, file_path, file_extension=file_extension) - self.create_task_payload["task"].update({"body": base64.b64encode(content).decode("utf-8")}) + self.create_task_payload["task"].update( + {image_key: base64.b64encode(content).decode("utf-8")} + ) except Exception as error: self.result.errorId = 12 self.result.errorCode = self.NO_CAPTCHA_ERR @@ -204,23 +223,24 @@ def _body_file_processing( async def _aio_body_file_processing( self, - save_format: SaveFormatsEnm, + save_format: SaveFormatsEnm | str, file_path: str, file_extension: str = "png", + image_key: str = "body", captcha_link: Optional[str] = None, captcha_file: Optional[str] = None, captcha_base64: Optional[bytes] = None, - **kwargs, + **kwargs: dict[str, Any], ): # if a local file link is passed if captcha_file: self.create_task_payload["task"].update( - {"body": base64.b64encode(self._local_file_captcha(captcha_file)).decode("utf-8")} + {image_key: base64.b64encode(self._local_file_captcha(captcha_file)).decode("utf-8")} ) # if the file is transferred in base64 encoding elif captcha_base64: self.create_task_payload["task"].update( - {"body": base64.b64encode(captcha_base64).decode("utf-8")} + {image_key: base64.b64encode(captcha_base64).decode("utf-8")} ) # if a URL is passed elif captcha_link: @@ -229,7 +249,9 @@ async def _aio_body_file_processing( # according to the value of the passed parameter, select the function to save the image if save_format == SaveFormatsEnm.CONST.value: self._file_const_saver(content, file_path, file_extension=file_extension) - self.create_task_payload["task"].update({"body": base64.b64encode(content).decode("utf-8")}) + self.create_task_payload["task"].update( + {image_key: base64.b64encode(content).decode("utf-8")} + ) except Exception as error: self.result.errorId = 12 self.result.errorCode = self.NO_CAPTCHA_ERR diff --git a/src/python_rucaptcha/core/config.py b/src/python_rucaptcha/core/config.py index 4f194704..763fd2d6 100644 --- a/src/python_rucaptcha/core/config.py +++ b/src/python_rucaptcha/core/config.py @@ -1,3 +1,5 @@ +from typing import Generator + from tenacity import AsyncRetrying, wait_fixed, stop_after_attempt from requests.adapters import Retry @@ -8,7 +10,7 @@ # Connection retry generator -def attempts_generator(amount: int = 20): +def attempts_generator(amount: int = 20) -> Generator[int, None, None]: """ Function generates a generator of length equal to `amount` diff --git a/src/python_rucaptcha/core/enums.py b/src/python_rucaptcha/core/enums.py index ab9a0feb..544be6e6 100644 --- a/src/python_rucaptcha/core/enums.py +++ b/src/python_rucaptcha/core/enums.py @@ -168,3 +168,11 @@ class ProsopoEnm(str, MyEnum): class CaptchaFoxEnm(str, MyEnum): CaptchaFoxTask = "CaptchaFoxTask" + + +class VKCaptchaEnm(str, MyEnum): + VKCaptchaTask = "VKCaptchaTask" + + +class TemuCaptchaEnm(str, MyEnum): + TemuCaptchaTask = "TemuCaptchaTask" diff --git a/src/python_rucaptcha/core/result_handler.py b/src/python_rucaptcha/core/result_handler.py index f7114291..adc5cfb3 100644 --- a/src/python_rucaptcha/core/result_handler.py +++ b/src/python_rucaptcha/core/result_handler.py @@ -1,7 +1,7 @@ import time import asyncio import logging -from typing import Union +from typing import Any import aiohttp import requests @@ -12,38 +12,80 @@ def get_sync_result( get_payload: GetTaskResultRequestSer, sleep_time: int, url_response: str -) -> Union[dict, Exception]: +) -> dict[str, str]: """ - Function periodically send the SYNC request to service and wait for captcha solving result + Periodically sends a synchronous request to a remote service to retrieve the result + of a CAPTCHA-solving task. + + This function polls the service using blocking HTTP requests until the CAPTCHA is solved, + an error is returned, or the task times out. It handles intermediate states and retries + with a configurable sleep interval between attempts. + + Args: + get_payload (GetTaskResultRequestSer): + Serialized request object containing the task ID and payload data. + sleep_time (int): + Time in seconds to wait between polling attempts when the task is still processing. + url_response (str): + Endpoint URL to query for the CAPTCHA-solving result. + + Returns: + dict[str, str]: + A dictionary containing the final task result. If the task fails or an exception + occurs, the dictionary includes error details such as status, errorId, errorCode, + and errorDescription. """ + response_ser = GetTaskResultResponseSer(taskId=get_payload.taskId) # generator for repeated attempts to connect to the server attempts = attempts_generator() for _ in attempts: try: # send a request for the result of solving the captcha - captcha_response = GetTaskResultResponseSer( - **requests.post(url_response, json=get_payload.to_dict()).json(), taskId=get_payload.taskId - ) - logging.warning(f"{captcha_response = }") + result: dict[str, Any] = requests.post(url_response, json=get_payload.to_dict()).json() + logging.info(f"Received captcha sync result - {result = }") + response_ser = GetTaskResultResponseSer(**result, taskId=get_payload.taskId) # if the captcha has not been resolved yet, wait - if captcha_response.status == "processing": + if response_ser.status == "processing": time.sleep(sleep_time) continue - elif captcha_response.status == "ready": + elif response_ser.status == "ready": break - elif captcha_response.errorId != 0: - return captcha_response.to_dict() + elif response_ser.errorId != 0: + return response_ser.to_dict() except Exception as error: - return error - return captcha_response.to_dict() + response_ser.status = "failed" + response_ser.errorId = 12 + response_ser.errorCode = "System error" + response_ser.errorDescription = str(error) + return response_ser.to_dict() async def get_async_result( get_payload: GetTaskResultRequestSer, sleep_time: int, url_response: str -) -> Union[dict, Exception]: +) -> dict[str, str]: """ - Function periodically send the ASYNC request to service and wait for captcha solving result + Periodically sends an asynchronous request to a remote service to retrieve the result + of a CAPTCHA-solving task. + + This function polls the service at regular intervals until the CAPTCHA is solved, + an error occurs, or the task times out. It uses aiohttp for asynchronous HTTP requests + and handles various response states including 'processing', 'ready', and error conditions. + + Args: + get_payload (GetTaskResultRequestSer): + Serialized request object containing the task ID and payload data. + sleep_time (int): + Time in seconds to wait between polling attempts when the task is still processing. + url_response (str): + Endpoint URL to query for the CAPTCHA-solving result. + + Returns: + dict[str, str]: + A dictionary containing the final task result if successful or partially failed. + If an exception occurs during the request, returns a dictionary with error details + including status, errorId, errorCode, and errorDescription. """ + response_ser = GetTaskResultResponseSer(taskId=get_payload.taskId) # generator for repeated attempts to connect to the server attempts = attempts_generator() async with aiohttp.ClientSession() as session: @@ -53,17 +95,21 @@ async def get_async_result( async with session.post( url_response, json=get_payload.to_dict(), raise_for_status=True ) as resp: - captcha_response = await resp.json(content_type=None) - captcha_response = GetTaskResultResponseSer(**captcha_response, taskId=get_payload.taskId) + result: dict[str, Any] = await resp.json(content_type=None) + logging.info(f"Received captcha async result - {result = }") + response_ser = GetTaskResultResponseSer(**result, taskId=get_payload.taskId) # if the captcha has not been resolved yet, wait - if captcha_response.status == "processing": + if response_ser.status == "processing": await asyncio.sleep(sleep_time) continue - elif captcha_response.status == "ready": + elif response_ser.status == "ready": break - elif captcha_response.errorId != 0: - return captcha_response.to_dict() + elif response_ser.errorId != 0: + return response_ser.to_dict() except Exception as error: - return error - return captcha_response.to_dict() + response_ser.status = "failed" + response_ser.errorId = 12 + response_ser.errorCode = "System error" + response_ser.errorDescription = str(error) + return response_ser.to_dict() diff --git a/src/python_rucaptcha/core/serializer.py b/src/python_rucaptcha/core/serializer.py index b74df0eb..a12b4b14 100644 --- a/src/python_rucaptcha/core/serializer.py +++ b/src/python_rucaptcha/core/serializer.py @@ -1,4 +1,6 @@ -from typing import Literal, Optional +from typing import Any, Literal +from decimal import Decimal +from datetime import date, datetime from msgspec import Struct @@ -7,8 +9,27 @@ class MyBaseModel(Struct): - def to_dict(self): - return {f: getattr(self, f) for f in self.__struct_fields__} + def to_dict(self) -> dict[str, Any]: + result = {} + for field in self.__struct_fields__: + value = getattr(self, field) + + if isinstance(value, MyBaseModel): + result[field] = value.to_dict() + + elif isinstance(value, (list, tuple)) and all(isinstance(el, Struct) for el in value): + result[field] = [el.to_dict() for el in value] + + elif isinstance(value, (date, datetime)): + result[field] = value.isoformat() + + elif isinstance(value, Decimal): + result[field] = str(value) + + else: + result[field] = value + + return result """ @@ -22,23 +43,23 @@ class TaskSer(MyBaseModel): class CreateTaskBaseSer(MyBaseModel): clientKey: str - task: TaskSer = {} + task: TaskSer languagePool: str = "en" - callbackUrl: str = None + callbackUrl: str | None = None soft_id: Literal[APP_KEY] = APP_KEY class GetTaskResultRequestSer(MyBaseModel): clientKey: str - taskId: int = None + taskId: int | None = None class CaptchaOptionsSer(MyBaseModel): sleep_time: int = 10 - service_type: enums.ServiceEnm = enums.ServiceEnm.TWOCAPTCHA.value + service_type: enums.ServiceEnm | str = enums.ServiceEnm.TWOCAPTCHA - url_request: Optional[str] = None - url_response: Optional[str] = None + url_request: str = f"http://api.{enums.ServiceEnm.TWOCAPTCHA.value}.com/2captcha/in.php" + url_response: str = f"http://api.{enums.ServiceEnm.TWOCAPTCHA.value}.com/2captcha/res.php" def urls_set(self): """ @@ -59,16 +80,16 @@ def urls_set(self): class GetTaskResultResponseSer(MyBaseModel): status: str = "ready" - solution: dict = None - cost: float = None - ip: str = None - createTime: int = None - endTime: int = None - solveCount: int = None - taskId: int = None + solution: dict[str, str] | None = None + cost: float = 0.0 + ip: str | None = None + createTime: int | None = None + endTime: int | None = None + solveCount: int | None = None + taskId: int | None = None # control method params - balance: float = None + balance: float | None = None # error info errorId: int = 0 - errorCode: str = None - errorDescription: str = None + errorCode: str | None = None + errorDescription: str | None = None diff --git a/src/python_rucaptcha/image_captcha.py b/src/python_rucaptcha/image_captcha.py index 7f78bb92..18a3e7e0 100644 --- a/src/python_rucaptcha/image_captcha.py +++ b/src/python_rucaptcha/image_captcha.py @@ -1,5 +1,5 @@ import shutil -from typing import Union, Optional +from typing import Any, Union, Optional from .core.base import BaseCaptcha from .core.enums import SaveFormatsEnm, ImageCaptchaEnm @@ -181,8 +181,8 @@ def captcha_handler( captcha_link: Optional[str] = None, captcha_file: Optional[str] = None, captcha_base64: Optional[bytes] = None, - **kwargs, - ) -> dict: + **kwargs: dict[str, Any], + ) -> dict[str, Any]: """ Sync solving method @@ -215,8 +215,8 @@ async def aio_captcha_handler( captcha_link: Optional[str] = None, captcha_file: Optional[str] = None, captcha_base64: Optional[bytes] = None, - **kwargs, - ) -> dict: + **kwargs: dict[str, Any], + ) -> dict[str, Any]: """ Async solving method diff --git a/src/python_rucaptcha/temu_captcha.py b/src/python_rucaptcha/temu_captcha.py new file mode 100644 index 00000000..d2e9deb7 --- /dev/null +++ b/src/python_rucaptcha/temu_captcha.py @@ -0,0 +1,158 @@ +import shutil +from typing import Any, Union + +from .core.base import BaseCaptcha +from .core.enums import SaveFormatsEnm, TemuCaptchaEnm + + +class TemuCaptcha(BaseCaptcha): + def __init__( + self, + save_format: Union[str, SaveFormatsEnm] = SaveFormatsEnm.TEMP, + img_clearing: bool = True, + img_path: str = "PythonRuCaptchaImages", + *args, + **kwargs: dict[str, Any], + ): + """ + Solve TemuImageTask CAPTCHA via 2Captcha/RuCaptcha API. + + This class creates and monitors TemuImageTask jobs, which require + a base64‐encoded background image plus an array of movable image + pieces (parts) in base64 format. It extends BaseCaptcha to handle + the low‐level request/response workflow. + + Args: + save_format (str | SaveFormatsEnm): Where to save temporary images. + - SaveFormatsEnm.TEMP: use system temp directory + - SaveFormatsEnm.CONST: keep files in img_path until deletion + img_clearing (bool): If True and save_format is CONST, delete the + img_path directory when this instance is destroyed. + img_path (str): Directory under which to store downloaded or decoded + images before sending to the API. + *args: Positional args forwarded to BaseCaptcha constructor (e.g. + client_key, method override). + **kwargs: Keyword args forwarded to BaseCaptcha for task creation. + Common params include: + - redirectUri: URL to confirm CAPTCHA resolution + - any other API‐supported parameters + Examples: + >>> captcha = TemuCaptcha( + ... clientKey="YOUR_API_KEY", + ... userAgent="Mozilla/5.0 ...", + ... proxyType="socks5", + ... proxyAddress="1.2.3.4", + ... proxyPort="1080" + ... ) + >>> response = captcha.captcha_handler( + ... parts=["part1_b64", "part2_b64", "part3_b64"], + ... captcha_base64=b"full_image_b64", + ... timeout=120 + ... ) + >>> print(response) + { + "errorId": 0, + "status": "ready", + "solution": { + "coordinates": [{"x":155,"y":358}, {"x":152,"y":153}, {"x":251,"y":333}] + }, + "cost": "0.0012", + "createTime": 1754563182, + "endTime": 1754563190, + "taskId": 80306543329, + "ip": "46.53.232.76", + "solveCount": 1 + } + + Notes: + https://2captcha.com/api-docs/temu-captcha + + https://rucaptcha.com/api-docs/temu-captcha + """ + super().__init__(method=TemuCaptchaEnm.TemuCaptchaTask, *args, **kwargs) + + self.save_format = save_format + self.img_clearing = img_clearing + self.img_path = img_path + + def captcha_handler( + self, + parts: list[str], + captcha_link: str | None = None, + captcha_file: str | None = None, + captcha_base64: bytes | None = None, + **kwargs: dict[str, Any], + ) -> dict[str, Any]: + """ + Synchronously solve a TemuImageTask. + + Args: + parts (list[str]): List of base64‐encoded strings for each + movable image piece. + captcha_link (str | None): URL to background image. Overrides + captcha_file and captcha_base64 if provided. + captcha_file (str | None): Path to an image file to read & send. + captcha_base64 (bytes | None): Raw bytes or base64 string of + the background image. + **kwargs: Passed through to the HTTP request call (e.g. timeout, + headers). + + Returns: + dict[str, Any]: Full JSON response from the 2Captcha/RuCaptcha API, + including errorId, taskId, status, solution, cost, times, etc. + """ + self.create_task_payload["task"].update({"parts": parts}) + self._body_file_processing( + save_format=self.save_format, + file_path=self.img_path, + captcha_link=captcha_link, + captcha_file=captcha_file, + captcha_base64=captcha_base64, + image_key="image", + **kwargs, + ) + if not self.result.errorId: + return self._processing_response(**kwargs) + return self.result.to_dict() + + async def aio_captcha_handler( + self, + parts: list[str], + captcha_link: str | None = None, + captcha_file: str | None = None, + captcha_base64: bytes | None = None, + ) -> dict[str, Any]: + """ + Asynchronously solve a TemuImageTask. + + Args: + parts (list[str]): List of base64‐encoded strings for each + movable image piece. + captcha_link (str | None): URL to background image. + captcha_file (str | None): Path to an image file to read & send. + captcha_base64 (bytes | None): Raw bytes or base64 string of image. + **kwargs: Passed through to the async HTTP request call. + + Returns: + dict[str, Any]: API response containing task status and solution. + """ + self.create_task_payload["task"].update({"parts": parts}) + await self._aio_body_file_processing( + save_format=self.save_format, + file_path=self.img_path, + captcha_link=captcha_link, + captcha_file=captcha_file, + captcha_base64=captcha_base64, + image_key="image", + **kwargs, + ) + if not self.result.errorId: + return await self._aio_processing_response() + return self.result.to_dict() + + def __del__(self): + """ + Cleanup saved images folder if configured to do so. + """ + if self.save_format == SaveFormatsEnm.CONST.value and self.img_clearing: + shutil.rmtree(self.img_path) diff --git a/src/python_rucaptcha/vk_captcha.py b/src/python_rucaptcha/vk_captcha.py new file mode 100644 index 00000000..23d9f6a8 --- /dev/null +++ b/src/python_rucaptcha/vk_captcha.py @@ -0,0 +1,120 @@ +from typing import Any + +from .core.base import BaseCaptcha +from .core.enums import VKCaptchaEnm + + +class VKCaptcha(BaseCaptcha): + def __init__( + self, + redirectUri: str, + userAgent: str, + proxyType: str, + proxyAddress: str, + proxyPort: str, + *args, + **kwargs: dict[str, Any], + ): + """ + The class is used to work with VKCaptchaTask. + + Args: + rucaptcha_key: User API key + redirectUri: The URL that is returned on requests to the captcha API. + userAgent: User-Agent of your browser will be used to load the captcha. + Use only modern browser's User-Agents + proxyType: Proxy type - `http`, `socks4`, `socks5` + proxyAddress: Proxy IP address or hostname + proxyPort: Proxy port + method: Captcha type + kwargs: Not required params for task creation request + + Examples: + >>> VKCaptcha(rucaptcha_key="aa9011f31111181111168611f1151122", + ... redirectUri="https://id.vk.com/not_robot_captcha?domain=vk.com...", + ... userAgent="Mozilla/5.0 .....", + ... proxyType="socks5", + ... proxyAddress="1.2.3.4", + ... proxyPort="445", + ... ).captcha_handler() + { + "errorId":0, + "status":"ready", + "solution":{ + "token":"142000f.....er" + }, + "cost":"0.002", + "ip":"1.2.3.4", + "createTime":1692863536, + "endTime":1692863556, + "solveCount":0, + "taskId": 73243152973, + } + + >>> await VKCaptcha(rucaptcha_key="aa9011f31111181111168611f1151122", + ... redirectUri="https://id.vk.com/not_robot_captcha?domain=vk.com...", + ... userAgent="Mozilla/5.0 .....", + ... proxyType="socks5", + ... proxyAddress="1.2.3.4", + ... proxyPort="445", + ... ).aio_captcha_handler() + { + "errorId":0, + "status":"ready", + "solution":{ + "token":"142000f.....er" + }, + "cost":"0.002", + "ip":"1.2.3.4", + "createTime":1692863536, + "endTime":1692863556, + "solveCount":0, + "taskId": 73243152973, + } + + Returns: + Dict with full server response + + Notes: + https://2captcha.com/api-docs/vk-captcha + + https://rucaptcha.com/api-docs/vk-captcha + """ + super().__init__(method=VKCaptchaEnm.VKCaptchaTask, *args, **kwargs) + + self.create_task_payload["task"].update( + { + "websiteURL": redirectUri, + "userAgent": userAgent, + "proxyType": proxyType, + "proxyAddress": proxyAddress, + "proxyPort": proxyPort, + } + ) + + def captcha_handler(self, **kwargs: dict[str, Any]) -> dict[str, Any]: + """ + Sync solving method + + Args: + kwargs: additional params for `requests` library + + Returns: + Dict with full server response + + Notes: + Check class docstirng for more info + """ + return self._processing_response(**kwargs) + + async def aio_captcha_handler(self) -> dict[str, Any]: + """ + Async solving method + + Returns: + Dict with full server response + + Notes: + Check class docstirng for more info + """ + return await self._aio_processing_response()