Upload hcaptcha.py

src/services/hcaptcha.py

@@ -49,6 +49,36 @@ _EXTRACT_HCAPTCHA_TOKEN_JS = """
 }
 """
 
+_EXTRACT_HCAPTCHA_META_JS = """
+() => {
+    const result = { token: null, respKey: null };
+    const textarea = document.querySelector('[name="h-captcha-response"]')
+        || document.querySelector('[name="g-recaptcha-response"]');
+    if (textarea && textarea.value && textarea.value.length > 20) {
+        result.token = textarea.value;
+    }
+    try {
+        if (window.hcaptcha) {
+            if (!result.token && typeof window.hcaptcha.getResponse === 'function') {
+                const response = window.hcaptcha.getResponse();
+                if (response && response.length > 20) {
+                    result.token = response;
+                }
+            }
+            if (typeof window.hcaptcha.getRespKey === 'function') {
+                const respKey = window.hcaptcha.getRespKey();
+                if (respKey) {
+                    result.respKey = String(respKey);
+                }
+            }
+        }
+    } catch (err) {
+        return result;
+    }
+    return result;
+}
+"""
+
 _QUESTION_JS = """
 () => {
     const prompt = document.querySelector('.prompt-text')
@@ -81,6 +111,28 @@ _VERIFY_BUTTON_SELECTORS = (
     'button[aria-label*="Verify"]',
 )
 
+_CHECKBOX_SELECTORS = (
+    "#checkbox",
+    '[id="checkbox"]',
+    'div[role="checkbox"]',
+    'input[type="checkbox"]',
+    '.checkbox',
+    '[aria-checked]',
+)
+
+_CHALLENGE_FRAME_HINTS = (
+    "frame=challenge",
+    "challenge",
+    "hcaptcha-inner",
+)
+
+_CHECKBOX_FRAME_HINTS = (
+    "frame=checkbox",
+    "checkbox",
+    "hcaptcha-invisible",
+    "hcaptcha-checkbox",
+)
+
 
 class HCaptchaSolver:
     """Solves ``HCaptchaTaskProxyless`` tasks via Playwright."""
@@ -123,12 +175,19 @@ class HCaptchaSolver:
     async def solve(self, params: dict[str, Any]) -> dict[str, Any]:
         website_url = params["websiteURL"]
         website_key = params["websiteKey"]
+        enterprise_payload = params.get("enterprisePayload") or {}
+        rqdata = str(enterprise_payload.get("rqdata") or "").strip()
+        if rqdata:
+            log.info("Received hCaptcha enterprisePayload.rqdata (len=%d)", len(rqdata))
 
         last_error: Exception | None = None
         for attempt in range(self._config.captcha_retries):
             try:
-                token = await self._solve_once(website_url, website_key)
-                return {"gRecaptchaResponse": token}
+                token, resp_key = await self._solve_once(website_url, website_key)
+                solution = {"gRecaptchaResponse": token}
+                if resp_key:
+                    solution["respKey"] = resp_key
+                return solution
             except Exception as exc:
                 last_error = exc
                 log.warning(
@@ -144,7 +203,7 @@ class HCaptchaSolver:
             f"HCaptcha failed after {self._config.captcha_retries} attempts: {last_error}"
         )
 
-    async def _solve_once(self, website_url: str, website_key: str) -> str:
+    async def _solve_once(self, website_url: str, website_key: str) -> tuple[str, str]:
         assert self._browser is not None
         target_url = self._prepare_target_url(website_url, website_key)
         if target_url != website_url:
@@ -173,8 +232,9 @@ class HCaptchaSolver:
             # 先给低风险会话一个直接出 token 的机会。
             token = await self._wait_for_token(page, seconds=4)
             if token:
+                resp_key = await self._wait_for_resp_key(page, seconds=2)
                 log.info("Got hCaptcha token directly after checkbox click (len=%d)", len(token))
-                return token
+                return token, resp_key
 
             # 无头环境常见路径：进入图片 challenge，然后走 classification fallback。
             log.info(
@@ -183,23 +243,27 @@ class HCaptchaSolver:
             fallback_handled = await self._solve_image_selection_challenge(page)
             if fallback_handled:
                 token = await self._wait_for_token(page)
+            resp_key = await self._wait_for_resp_key(page, seconds=2)
 
             if not isinstance(token, str) or len(token) < 20:
                 raise RuntimeError(f"Invalid hCaptcha token: {token!r}")
 
             log.info("Got hCaptcha token (len=%d)", len(token))
-            return token
+            return token, resp_key
         finally:
             await context.close()
 
     async def _click_checkbox(self, page: Page) -> None:
         frame = await self._find_frame(page, "checkbox", wait_seconds=10)
         if frame is None:
-            raise RuntimeError("Could not find hCaptcha checkbox frame")
+            raise RuntimeError(self._build_missing_frame_error(page, "checkbox"))
 
-        checkbox = await frame.query_selector("#checkbox")
+        checkbox = await self._find_checkbox_element(frame)
         if checkbox is None:
-            raise RuntimeError("Could not find hCaptcha checkbox element")
+            raise RuntimeError(
+                "Could not find hCaptcha checkbox element inside frame "
+                f"{getattr(frame, 'url', None) or '<empty>'}"
+            )
 
         await checkbox.click(timeout=10_000)
         log.info("Clicked hCaptcha checkbox")
@@ -207,24 +271,135 @@ class HCaptchaSolver:
     async def _wait_for_token(self, page: Page, *, seconds: int | None = None) -> str | None:
         remaining = max(1, seconds or self._config.captcha_timeout)
         for _ in range(remaining):
-            token = await page.evaluate(_EXTRACT_HCAPTCHA_TOKEN_JS)
+            token, _ = await self._extract_hcaptcha_meta(page)
             if isinstance(token, str) and len(token) > 20:
                 return token
             await asyncio.sleep(1)
         return None
 
+    async def _wait_for_resp_key(self, page: Page, *, seconds: int | None = None) -> str:
+        remaining = max(1, seconds or 2)
+        last_resp_key = ""
+        for _ in range(remaining):
+            _, resp_key = await self._extract_hcaptcha_meta(page)
+            if resp_key:
+                return resp_key
+            await asyncio.sleep(1)
+        return last_resp_key
+
+    async def _extract_hcaptcha_meta(self, page: Page) -> tuple[str | None, str]:
+        contexts: list[Page | Frame] = [page]
+        for frame in page.frames:
+            url = (getattr(frame, "url", None) or "").lower()
+            if self._is_hcaptcha_related_frame(url):
+                contexts.append(frame)
+
+        token: str | None = None
+        resp_key = ""
+        for context in contexts:
+            try:
+                payload = await context.evaluate(_EXTRACT_HCAPTCHA_META_JS)
+            except Exception:
+                continue
+            if not isinstance(payload, dict):
+                continue
+            current_token = payload.get("token")
+            current_resp_key = payload.get("respKey")
+            if not token and isinstance(current_token, str) and len(current_token) > 20:
+                token = current_token
+            if not resp_key and isinstance(current_resp_key, str) and current_resp_key.strip():
+                resp_key = current_resp_key.strip()
+            if token and resp_key:
+                break
+        return token, resp_key
+
     async def _find_frame(
         self, page: Page, frame_role: str, *, wait_seconds: int = 5
     ) -> Frame | None:
         attempts = max(1, wait_seconds * 2)
+        main_frame = getattr(page, "main_frame", None)
         for _ in range(attempts):
+            exact_match: Frame | None = None
+            hinted_match: Frame | None = None
+            dom_match: Frame | None = None
             for frame in page.frames:
-                url = frame.url or ""
-                if "hcaptcha" in url and f"frame={frame_role}" in url:
-                    return frame
+                if main_frame is not None and frame is main_frame:
+                    continue
+                url = (frame.url or "").lower()
+                if not self._is_hcaptcha_related_frame(url):
+                    continue
+
+                if frame_role == "checkbox":
+                    if await self._find_checkbox_element(frame) is not None:
+                        dom_match = dom_match or frame
+                    if any(hint in url for hint in _CHECKBOX_FRAME_HINTS):
+                        if "frame=checkbox" in url:
+                            exact_match = exact_match or frame
+                        else:
+                            hinted_match = hinted_match or frame
+                elif frame_role == "challenge":
+                    if await self._is_challenge_frame(frame):
+                        dom_match = dom_match or frame
+                    if any(hint in url for hint in _CHALLENGE_FRAME_HINTS):
+                        if "frame=challenge" in url:
+                            exact_match = exact_match or frame
+                        else:
+                            hinted_match = hinted_match or frame
+
+            if exact_match is not None:
+                return exact_match
+            if dom_match is not None:
+                return dom_match
+            if hinted_match is not None:
+                return hinted_match
             await asyncio.sleep(0.5)
         return None
 
+    @staticmethod
+    def _is_hcaptcha_related_frame(url: str) -> bool:
+        return (
+            "hcaptcha" in url
+            or "newassets.hcaptcha.com" in url
+            or "api.hcaptcha.com" in url
+            or "js.stripe.com/v3/hcaptcha" in url
+        )
+
+    async def _find_checkbox_element(self, frame: Frame) -> ElementHandle[Any] | None:
+        for selector in _CHECKBOX_SELECTORS:
+            element = await frame.query_selector(selector)
+            if element is not None:
+                return element
+        return None
+
+    async def _is_challenge_frame(self, frame: Frame) -> bool:
+        prompt = await frame.evaluate(_QUESTION_JS)
+        if isinstance(prompt, str) and prompt.strip():
+            return True
+
+        for selector in _CHALLENGE_TILE_SELECTORS:
+            elements = await frame.query_selector_all(selector)
+            if elements:
+                return True
+
+        if await frame.locator("canvas").count() > 0:
+            return True
+
+        for selector in _VERIFY_BUTTON_SELECTORS:
+            if await frame.query_selector(selector) is not None:
+                return True
+
+        return False
+
+    @staticmethod
+    def _build_missing_frame_error(page: Page, frame_role: str) -> str:
+        frame_urls = [
+            getattr(frame, "url", None) or "<empty>"
+            for frame in page.frames
+        ]
+        return (
+            f"Could not find hCaptcha {frame_role} frame; available frames={frame_urls}"
+        )
+
     @staticmethod
     def _prepare_target_url(website_url: str, website_key: str) -> str:
         """为官方 demo 自动补齐/对齐 sitekey，确保按请求参数测试真实行为。"""