Feature new methods

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",

src/python_rucaptcha/__version__.py

@@ -1 +1 @@
-__version__ = "6.4.0"
+__version__ = "6.5.0"

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

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`
 

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"

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()