feat: 添加 CapSolver 自动完成 Cloudflare Turnstile 的功能

report_video.py

@@ -8,7 +8,70 @@ from playwright.sync_api import (
     Browser,
 )
 
+def solve_turnstile_capsolver(page: Page,
+                              timeout: int = 120) -> bool:
+    """
+    使用 CapSolver 自动完成当前 Page 上的 Cloudflare Turnstile。
+    成功返回 True，失败/超时返回 False。
+    """
+    cap_key="CAP-A76C932D4C6CCB3CA748F77FDC07D996"
+    widget = page.query_selector("div.cf-turnstile[data-sitekey]")
+    if not widget:
+        return False
+    sitekey = widget.get_attribute("data-sitekey")
+    page_url = page.url
 
+    create_payload = {
+        "clientKey": cap_key,
+        "task": {
+            "type": "TurnstileTaskProxyLess",
+            "websiteURL": page_url,
+            "websiteKey": sitekey
+        }
+    }
+    create_resp = requests.post(
+        "https://api.capsolver.com/createTask",
+        json=create_payload, timeout=20
+    ).json()
+    if create_resp.get("errorId"):
+        print("[CapSolver] createTask 失败:", create_resp)
+        return False
+    task_id = create_resp["taskId"]
+
+
+    poll_payload = {"clientKey": cap_key, "taskId": task_id}
+    token = None
+    elapsed, step = 0, 3
+    while elapsed < timeout:
+        time.sleep(step)
+        elapsed += step
+        res = requests.post(
+            "https://api.capsolver.com/getTaskResult",
+            json=poll_payload, timeout=15
+        ).json()
+        if res.get("status") == "ready":
+            token = res["solution"]["token"]
+            break
+        if res.get("status") != "processing":
+            print("[CapSolver] getTaskResult 异常:", res)
+            return False
+
+    if not token:
+        print("[CapSolver] 超时未取到 token")
+        return False
+
+
+    page.evaluate(
+        """(tk) => {
+            const ta = document.querySelector('textarea[name="cf-turnstile-response"]');
+            if (ta) ta.value = tk;
+            if (window.turnstileCallback)
+                try { window.turnstileCallback(tk); } catch(e){}
+        }""",
+        token
+    )
+    page.wait_for_timeout(1500)
+    return True
 def require_login(func):
     @functools.wraps(func)
     def wrapper(self, *args, **kwargs):