Merge pull request #32 from ZennoLab/grk717/RecognitionComplexImageTask

Grk717/recognition complex image task

Author: azeriker

capmonstercloud_client/CapMonsterCloudClient.py

@@ -29,7 +29,8 @@
     ((BasiliskCustomTaskRequest, BasiliskCustomTaskProxylessRequest), getBasiliskTimeouts),
     ((AmazonWafRequest, AmazonWafProxylessRequest), getAmazonWafTimeouts),
     ((BinanceTaskRequest, BinanceTaskProxylessRequest), getBinanceTimeouts),
-    ((ImpervaCustomTaskRequest, ImpervaCustomTaskProxylessRequest), getImpervaTimeouts)
+    ((ImpervaCustomTaskRequest, ImpervaCustomTaskProxylessRequest), getImpervaTimeouts),
+    ((RecognitionComplexImageTaskRequest), getCITTimeouts)
 )
 
 
@@ -82,7 +83,8 @@ async def solve_captcha(self, request: Union[RecaptchaV2EnterpriseProxylessReque
                                                  BinanceTaskRequest,
                                                  BinanceTaskProxylessRequest,
                                                  ImpervaCustomTaskRequest,
-                                                 ImpervaCustomTaskProxylessRequest],
+                                                 ImpervaCustomTaskProxylessRequest,
+                                                 RecognitionComplexImageTaskRequest],
                             ) -> Dict[str, str]:
         '''
         Non-blocking method for captcha solving. 

capmonstercloud_client/GetResultTimeouts.py

@@ -49,3 +49,6 @@ def getBinanceTimeouts() -> GetResultTimeouts:
 
 def getImpervaTimeouts() -> GetResultTimeouts:
     return GetResultTimeouts(1, 0, 1, 20)
+
+def getCITTimeouts() -> GetResultTimeouts:
+    return GetResultTimeouts(0.35, 0, 0.2, 10)

capmonstercloud_client/requests/RecaptchaComplexImageTask.py

@@ -71,7 +71,7 @@ def getTaskDict(self) -> Dict[str, Union[str, int, bool]]:
 
         if self.websiteUrl is not None:
             task["websiteUrl"] = self.websiteUrl
-        
+
         return task
 
 

capmonstercloud_client/requests/RecognitionComplexImageTaskRequest.py

@@ -0,0 +1,45 @@
+from typing import Optional, List, Dict, Union
+from pydantic import validator
+from .ComplexImageTaskBase import ComplexImageTaskRequestBase
+from ..exceptions import NumbersImagesErrors, ZeroImagesErrors, TaskNotDefinedError
+
+
+class RecognitionComplexImageTaskRequest(ComplexImageTaskRequestBase):
+    captchaClass: str = 'recognition'
+    metadata: Dict[str, str]
+
+    @validator('metadata')
+    def validate_metadata(cls, value):
+        if value.get('Task') is None:
+            raise TaskNotDefinedError(f'Expect that Task will be defined.')
+        else:
+            if not isinstance(value.get('Task'), str):
+                raise TypeError(f'Expect that Task will be str.')
+        if not set(value.keys()).issubset(set(["Task", "TaskArgument"])):
+            raise TypeError(f'Allowed keys for metadata are "Task" and "TaskArgument"')
+        return value
+    
+    @validator('imagesBase64')
+    def validate_images_array(cls, value):
+        if value is not None:
+            if not isinstance(value, (list, tuple)):
+                raise TypeError(f'Expect that type imagesBase64 array will be <list> or <tuple>, got {type(value)}')
+            elif not len(value):
+                raise ZeroImagesErrors(f'At least one image base64 expected, got {len(value)}')
+            # Check for each element type
+            contain_types = [isinstance(x, str) for x in value]
+            if not all(contain_types):
+                raise TypeError(f'Next images from imagesBase64 array are not string: {contain_types}')
+        else:
+            raise ZeroImagesErrors(f'At least one image base64 expected, got {len(value)}')
+        return value
+    
+    def getTaskDict(self) -> Dict[str, Union[str, int, bool]]:
+        task = {}
+        task['type'] = self.taskType
+        task['class'] = self.captchaClass
+        task['imagesBase64'] = self.imagesBase64
+        task['metadata'] = self.metadata
+        if self.userAgent is not None:
+            task['userAgent'] = self.userAgent
+        return task

capmonstercloud_client/requests/__init__.py

@@ -28,6 +28,7 @@
 from .BinanceTaskProxylessRequest import BinanceTaskProxylessRequest
 from .ImpervaCustomTaskRequest import ImpervaCustomTaskRequest
 from .ImpervaCustomTaskProxylessRequest import ImpervaCustomTaskProxylessRequest
+from .RecognitionComplexImageTaskRequest import RecognitionComplexImageTaskRequest
 
 
 REQUESTS = ['RecaptchaV2EnterpiseRequest', 'RecaptchaV2EnterpriseProxylessRequest', 
@@ -38,4 +39,4 @@
             'TenDiCustomTaskRequest', 'TenDiCustomTaskProxylessRequest', 'BasiliskCustomTaskRequest',
             'BasiliskCustomTaskProxylessRequest', 'AmazonWafRequest', 'AmazonWafProxylessRequest',
             'BinanceTaskRequest', 'BinanceTaskProxylessRequest', 'ImpervaCustomTaskProxylessRequest',
-            'ImpervaCustomTaskRequest']
+            'ImpervaCustomTaskRequest', 'RecognitionComplexImageTaskRequest']

capmonstercloud_client/version.txt

@@ -1 +1 @@
-1.6.0
+2.0.0

examples/imperva.py

@@ -7,10 +7,10 @@
 import json
 
 async def solve_captcha_sync(num_requests):
-    return [await cap_monster_client.solve_captcha(datadome_request) for _ in range(num_requests)]
+    return [await cap_monster_client.solve_captcha(imperva_request) for _ in range(num_requests)]
 
 async def solve_captcha_async(num_requests):
-    tasks = [asyncio.create_task(cap_monster_client.solve_captcha(datadome_request)) 
+    tasks = [asyncio.create_task(cap_monster_client.solve_captcha(imperva_request)) 
              for _ in range(num_requests)]
     return await asyncio.gather(*tasks, return_exceptions=True)
 
@@ -21,7 +21,7 @@ async def solve_captcha_async(num_requests):
     metadata = {"incapsulaScriptBase64": "",
 			"incapsulaSessionCookie": "SAyLRzdYgUntD6v0r7nFBmxTYGcAAAAArkznhRMmVs/cBynTg3r6YA==",
 			"reese84UrlEndpoint": "Alarums-Exeunter-Hath-Brese-Banq-Wheth-frangerd-"}
-    datadome_request = ImpervaCustomTaskProxylessRequest(
+    imperva_request = ImpervaCustomTaskProxylessRequest(
         websiteUrl='https://example.com/login',
         metadata=metadata
     )

examples/recognitionCIT.py

@@ -0,0 +1,42 @@
+import os
+import time
+import asyncio
+import base64
+from capmonstercloudclient.requests import RecognitionComplexImageTaskRequest
+from capmonstercloudclient import ClientOptions, CapMonsterClient
+
+async def solve_captcha_sync(num_requests):
+    return [await cap_monster_client.solve_captcha(oocl_request) for _ in range(num_requests)]
+
+async def solve_captcha_async(num_requests):
+    tasks = [asyncio.create_task(cap_monster_client.solve_captcha(oocl_request)) 
+             for _ in range(num_requests)]
+    return await asyncio.gather(*tasks, return_exceptions=True)
+
+if __name__ == '__main__':
+    key = os.getenv('API_KEY')
+    client_options = ClientOptions(api_key=key)
+    cap_monster_client = CapMonsterClient(options=client_options)
+    with open("/path/to/img/0_2.png", 'rb') as f:
+        bg = base64.b64encode(f.read()).decode("utf-8")
+    with open("/path/to/img/0_0.png", 'rb') as f:
+        ring = base64.b64encode(f.read()).decode("utf-8")
+    with open("/path/to/img/0_1.png", 'rb') as f:
+        circle = base64.b64encode(f.read()).decode("utf-8")
+    oocl_request = RecognitionComplexImageTaskRequest(
+        metadata={"Task": "oocl_rotate_double_new"},
+        imagesBase64=[bg, ring, circle]
+    )
+    nums = 3
+
+    # Sync test
+    sync_start = time.time()
+    sync_responses = asyncio.run(solve_captcha_sync(nums))
+    print(f'average execution time sync {1/((time.time()-sync_start)/nums):0.2f} ' \
+          f'resp/sec\nsolution: {sync_responses[0]}')
+
+    # Async test
+    async_start = time.time()
+    async_responses = asyncio.run(solve_captcha_async(nums))
+    print(f'average execution time async {1/((time.time()-async_start)/nums):0.2f} ' \
+          f'resp/sec\nsolution: {async_responses[0]}')

requirements.txt

@@ -1,2 +1,18 @@
-aiohttp>=3.7.4
-pydantic==2.1.*
+aiohappyeyeballs==2.4.4 ; python_version >= "3.9" and python_version <= "3.13"
+aiohttp==3.11.10 ; python_version >= "3.9" and python_version <= "3.13"
+aiosignal==1.3.1 ; python_version >= "3.9" and python_version <= "3.13"
+annotated-types==0.7.0 ; python_version >= "3.9" and python_version <= "3.13"
+async-timeout==5.0.1 ; python_version >= "3.9" and python_version <= "3.11"
+attrs==24.2.0 ; python_version >= "3.9" and python_version <= "3.13"
+frozenlist==1.5.0 ; python_version >= "3.9" and python_version <= "3.13"
+idna==3.10 ; python_version >= "3.9" and python_version <= "3.13"
+multidict==6.1.0 ; python_version >= "3.9" and python_version <= "3.13"
+propcache==0.2.1 ; python_version >= "3.9" and python_version <= "3.13"
+pydantic-core==2.27.1 ; python_version >= "3.9" and python_version <= "3.13"
+pydantic==2.10.3 ; python_version >= "3.9" and python_version <= "3.13"
+typing-extensions==4.12.2 ; python_version >= "3.9" and python_version <= "3.13"
+yarl==1.18.3 ; python_version >= "3.9" and python_version <= "3.13"
+aiohttp==3.7.4 ; python_version >= "3.6" and python_version <= "3.8"
+pydantic==2.10.3 ; python_version >= "3.8" and python_version < "3.9"
+pydantic==2.5.3 ; python_version >= "3.7" and python_version < "3.8"
+pydantic==2.1.* ; python_version < "3.7"

test/cit_test.py

@@ -0,0 +1,71 @@
+import unittest
+import urllib
+import base64
+from pydantic.error_wrappers import ValidationError
+from capmonstercloudclient.requests import RecognitionComplexImageTaskRequest
+from capmonstercloudclient.exceptions import NumbersImagesErrors, TaskNotDefinedError, ZeroImagesErrors, \
+    UserAgentNotDefinedError
+
+def read_image(image_url: str,):
+    image_bytes = urllib.request.urlopen(image_url).read()
+    return base64.b64encode(image_bytes).decode('utf-8')
+
+class RecognitionCITImageRequestTest(unittest.TestCase):
+    
+    websiteUrlExample = 'https://lessons.zennolab.com/captchas/recaptcha/v2_simple.php?level=middle'
+    imageUrlsExamples = ['https://i.postimg.cc/H8yBD5FJ/0-2.png', 'https://i.postimg.cc/rz0hrXz8/0-0.png',
+                         'https://i.postimg.cc/qgP1cbC2/0-1.png']
+    imageBase64Examples = [read_image(i) for i in imageUrlsExamples]
+    metadataExample = {"Task": "oocl_rotate_double_new"}
+    
+    def testImagesTypes(self):
+        
+        with self.assertRaises(ValidationError):
+            request = RecognitionComplexImageTaskRequest(metadata=RecognitionCITImageRequestTest.metadataExample,
+                                                       imagesBase64='[]')
+        
+    
+    def testImagesFilling(self):
+         
+        with self.assertRaises(ZeroImagesErrors, 
+                               msg='Empty array imagesBase64 must be cause ZeroImagesErrors'):
+            request = RecognitionComplexImageTaskRequest(metadata=RecognitionCITImageRequestTest.metadataExample,
+                                                       imagesBase64=[])
+                
+    def testAllRequiredFieldsFilling(self):
+        required_fields = ['class', 'type', 'metadata']
+        metadata_fields = ['Task']
+        request = RecognitionComplexImageTaskRequest(
+                                                   metadata=RecognitionCITImageRequestTest.metadataExample,
+                                                   imagesBase64=RecognitionCITImageRequestTest.imageBase64Examples)
+        request_dict = request.getTaskDict()
+        for i in required_fields:
+            self.assertTrue(i in list(request_dict.keys()), 
+                            msg=f'Required field {i} not in {request_dict}')
+            
+        metadata_dict = request_dict['metadata']
+        for i in metadata_fields:
+            self.assertTrue(i in list(metadata_dict.keys()), 
+                            msg=f'Required field {i} not in {request_dict}')
+            
+        self.assertEqual(request_dict['class'], 'recognition')
+        self.assertEqual(request_dict['type'], 'ComplexImageTask')
+            
+    def testTaskDefined(self):
+        with self.assertRaises(TaskNotDefinedError, 
+                               msg='Expect that empty "Task" field will be cause TaskNotDefinedError'):
+            request = RecognitionComplexImageTaskRequest(metadata={},
+                                                         imagesBase64=RecognitionCITImageRequestTest.imageBase64Examples)
+        
+        with self.assertRaises(TaskNotDefinedError):
+            request = RecognitionComplexImageTaskRequest(metadata={'dsfsdf': 'sdfsdf'},
+                                                       imagesUrls=RecognitionCITImageRequestTest.imageUrlsExamples)
+
+    def testExtraMetadata(self):
+        with self.assertRaises(TypeError, 
+                               msg='Expect that extra metadata fields will be cause TypeError'):
+            request = RecognitionComplexImageTaskRequest(metadata={"Task": "oocl_rotate_new", 'asd': 'asd'},
+                                                         imagesBase64=RecognitionCITImageRequestTest.imageBase64Examples)
+        
+if __name__ == '__main__':
+    unittest.main()