Upload hcaptcha_solver.py

hcaptcha_solver.py

@@ -1,7 +1,7 @@
 import asyncio
 import base64
 import random
-import chardet
+import re
 import nopecha
 import requests
 from hcaptcha_challenger.agents import Malenia
@@ -92,38 +92,56 @@ def url_to_base64(url):
         return None
 
 
+async def handle_response(page):
+    await asyncio.sleep(2)
+    await page.wait_for_selector("//iframe[contains(@title, 'hCaptcha challenge')]")
+    frame_challenge = page.frame_locator("//iframe[contains(@title, 'hCaptcha challenge')]")
+    question_text = frame_challenge.locator("//div[@class='challenge-prompt']")
+    element_handle = await question_text.element_handle()
+    prompt_padding_element = await element_handle.query_selector("div.prompt-padding")
+    if prompt_padding_element is not None:
+        span_element = await element_handle.query_selector("h2 > span")
+        span_text = await span_element.inner_text()
+        examples0 = frame_challenge.locator('//div[@class="challenge-example"]')
+        examples = examples0.locator('div.image[aria-hidden]')
+        aria_hidden_value = await examples.get_attribute("aria-hidden")
+        if aria_hidden_value == "true":
+            examples = None
+            print("no examples")
+        else:
+            examples1 = examples.locator('//div[@class="image"]')
+            await examples1.wait_for()
+            style_attribute = await examples1.get_attribute("style")
+            url_match = re.search(r'url\("(.+?)"\)', style_attribute)
+            if url_match:
+                url = url_match.group(1)
+                print(url)
+                examples = url
+        task0 = frame_challenge.locator("//div[@class='task-image']")
+        await frame_challenge.locator("//div[@tabindex='0']").nth(0).wait_for()
+        count = await task0.count()
+        print(span_text)
+        print("done tasks for getting examples")
+        data = []
+        for i in range(count):
+            sample = task0.nth(i)
+            img = sample.locator('//div[@class="image"]')
+            style_attribute = await img.get_attribute("style")
+            url_match = re.search(r'url\("(.+?)"\)', style_attribute)
+            if url_match:
+                url = url_match.group(1)
+                data.append(url)
+        print("done tasks for getting url")
+        await classify_click(page, data, 0, examples, span_text)
+    else:
+        await asyncio.sleep(random.uniform(0.1, 0.3))
+        await frame_challenge.locator("//div[@class='refresh button']").click()
+        print("refresh")
+
+
 async def on_response(response, page):
     if response.url.startswith("https://api.hcaptcha.com/getcaptcha"):
-        # 获取响应内容
-        content_type = response.headers.get('Content-Type')
-        print(response.headers)
-        print(content_type)
-        if content_type == "application/json":
-            data = await response.json()
-        else:
-            print("content type:", content_type)
-            content = await response.body()
-            print(content)
-            detected_encoding = chardet.detect(content)['encoding']
-            print(detected_encoding)
-            data = content.decode(detected_encoding)
-        # 打印响应内容
-        print(data)
-        examples_urls = data.get("requester_question_example")
-        examples = [url_to_base64(i) for i in examples_urls] if examples_urls else None
-        if data.get("request_type") == "image_label_binary":
-            await classify_click(page, data, 0, examples)
-        elif data.get("request_type") == "image_label_area_select":
-            # await asyncio.sleep(random.uniform(0.1, 0.3))
-            # await page.wait_for_selector("//iframe[contains(@title, 'hCaptcha challenge')]")
-            # frame_challenge = page.frame_locator("//iframe[contains(@title, 'hCaptcha challenge')]")
-            # await frame_challenge.locator("//div[@class='refresh button']").click()
-            await area_click(page, data, 0, examples)
-        else:
-            await asyncio.sleep(random.uniform(0.1, 0.3))
-            await page.wait_for_selector("//iframe[contains(@title, 'hCaptcha challenge')]")
-            frame_challenge = page.frame_locator("//iframe[contains(@title, 'hCaptcha challenge')]")
-            await frame_challenge.locator("//div[@class='refresh button']").click()
+        asyncio.create_task(handle_response(page))
     elif response.url.startswith("https://api.hcaptcha.com/checkcaptcha"):
         data0 = await response.json()
         if data0.get("pass"):
@@ -134,7 +152,7 @@ async def on_response(response, page):
         print(data0)
 
 
-async def classify_click(page, data, round0, examples):
+async def classify_click(page, data, round0, examples, quetsion0):
     try:
         await page.wait_for_selector("//iframe[contains(@title, 'hCaptcha challenge')]")
         frame_challenge = page.frame_locator("//iframe[contains(@title, 'hCaptcha challenge')]")
@@ -142,12 +160,19 @@ async def classify_click(page, data, round0, examples):
         await frame_challenge.locator("//div[@tabindex='0']").nth(0).wait_for()
         count = await samples.count()
         print(count)
-        clicks = nopecha.Recognition.solve(
-            type='hcaptcha',
-            task=data.get("requester_question").get("en"),
-            image_urls=[data.get("tasklist")[i]["datapoint_uri"] for i in (range(9) if round0 == 0 else range(9, 18))],
-            image_examples=examples
-        )
+        if examples:
+            clicks = nopecha.Recognition.solve(
+                type='hcaptcha',
+                task=quetsion0,
+                image_urls=data,
+                image_examples=examples
+            )
+        else:
+            clicks = nopecha.Recognition.solve(
+                type='hcaptcha',
+                task=quetsion0,
+                image_urls=data
+            )
         for i in range(count):
             sample = samples.nth(i)
             await sample.wait_for()
@@ -159,10 +184,26 @@ async def classify_click(page, data, round0, examples):
         fl = frame_challenge.locator("//div[@class='button-submit button']")
         await fl.click()
         if round0 == 0:
-            await classify_click(page, data, 1, examples)
+            await asyncio.sleep(2)
+            await page.wait_for_selector("//iframe[contains(@title, 'hCaptcha challenge')]")
+            frame_challenge = page.frame_locator("//iframe[contains(@title, 'hCaptcha challenge')]")
+            task0 = frame_challenge.locator("//div[@class='task-image']")
+            await frame_challenge.locator("//div[@tabindex='0']").nth(0).wait_for()
+            count = await task0.count()
+            data = []
+            for i in range(count):
+                sample = task0.nth(i)
+                img = sample.locator('//div[@class="image"]')
+                style_attribute = await img.get_attribute("style")
+                url_match = re.search(r'url\("(.+?)"\)', style_attribute)
+                if url_match:
+                    url = url_match.group(1)
+                    data.append(url)
+            print("done tasks for getting url")
+            await classify_click(page, data, 1, examples,quetsion0)
     except Exception as e:
         print(e)
-        await classify_click(page, data, round0, examples)
+        await classify_click(page, data, round0, examples,quetsion0)
 
 
 async def area_click(page, data, round0, examples):
@@ -216,7 +257,7 @@ async def hit_challenge(context: ASyncContext, host, sitekey, times: int = 8):
 
 async def bytedance(host, sitekey):
     async with async_playwright() as p:
-        browser = await p.firefox.launch(headless=True)
+        browser = await p.firefox.launch(headless=False)
         context = await browser.new_context(
             locale="en-US"
         )
@@ -243,3 +284,5 @@ async def main(host, key):
         print(e)
         return token
 
+
+asyncio.run(main("free.vps.vc", "3bae0a5b-f2b8-43ef-98b7-76865a8a3997"))