Upload 32 files

src/main.py

@@ -77,11 +77,13 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
         task_manager.register_solver(task_type, v2_solver)
     log.info("Registered reCAPTCHA v2 solver for types: %s", _RECAPTCHA_V2_TYPES)
 
-    hcaptcha_solver = HCaptchaSolver(config)
-    await hcaptcha_solver.start()
-    for task_type in _HCAPTCHA_TYPES:
-        task_manager.register_solver(task_type, hcaptcha_solver)
-    log.info("Registered hCaptcha solver for types: %s", _HCAPTCHA_TYPES)
+    classifier = ClassificationSolver(config)
+
+    hcaptcha_solver = HCaptchaSolver(config, classifier=classifier)
+    await hcaptcha_solver.start()
+    for task_type in _HCAPTCHA_TYPES:
+        task_manager.register_solver(task_type, hcaptcha_solver)
+    log.info("Registered hCaptcha solver for types: %s", _HCAPTCHA_TYPES)
 
     turnstile_solver = TurnstileSolver(config)
     await turnstile_solver.start()
@@ -94,10 +96,9 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
         task_manager.register_solver(task_type, recognizer)
     log.info("Registered image captcha recognizer for types: %s", _IMAGE_TEXT_TYPES)
 
-    classifier = ClassificationSolver(config)
-    for task_type in _CLASSIFICATION_TYPES:
-        task_manager.register_solver(task_type, classifier)
-    log.info("Registered classification solver for types: %s", _CLASSIFICATION_TYPES)
+    for task_type in _CLASSIFICATION_TYPES:
+        task_manager.register_solver(task_type, classifier)
+    log.info("Registered classification solver for types: %s", _CLASSIFICATION_TYPES)
 
     yield
     # ── shutdown ──

src/models/task.py

@@ -7,20 +7,21 @@ from pydantic import BaseModel, Field
 
 # ── createTask ──────────────────────────────────────────────
 
-class TaskObject(BaseModel):
-    type: str
-    websiteURL: str | None = None
-    websiteKey: str | None = None
+class TaskObject(BaseModel):
+    type: str
+    websiteURL: str | None = None
+    websiteKey: str | None = None
     pageAction: str | None = None
     minScore: float | None = None
     isInvisible: bool | None = None
     # Image captcha / classification fields
     body: str | None = None
-    image: str | None = None
-    images: list[str] | None = None
-    question: str | None = None
-    queries: list[str] | str | None = None
-    project_name: str | None = None
+    image: str | None = None
+    images: list[str] | None = None
+    examples: list[str] | None = None
+    question: str | None = None
+    queries: list[str] | str | None = None
+    project_name: str | None = None
 
 
 class CreateTaskRequest(BaseModel):

src/services/classification.py

@@ -23,23 +23,29 @@ from ..core.config import Config
 
 log = logging.getLogger(__name__)
 
-HCAPTCHA_SYSTEM_PROMPT = """\
-You are an image classification assistant for HCaptcha challenges.
-Given a question and one or more base64-encoded images, determine which images match the question.
-
-Return STRICT JSON only. No markdown, no extra text.
-
-For single-image questions (is this image X?):
-{"answer": true}  or  {"answer": false}
-
-For multi-image grid questions (select all images containing X):
-{"answer": [0, 2, 5]}
-where numbers are 0-indexed positions of matching images.
-
-Rules:
-- Return ONLY the JSON object, nothing else.
-- Be precise with your classification.
-"""
+HCAPTCHA_SYSTEM_PROMPT = """\
+You are an image classification assistant for HCaptcha challenges.
+
+You may receive:
+1. optional sample/example images that show the target object, and
+2. one or more candidate captcha images that must be classified.
+
+Determine which candidate images match the question or the sample images.
+
+Return STRICT JSON only. No markdown, no extra text.
+
+For single-image questions (is this image X?):
+{"answer": true}  or  {"answer": false}
+
+For multi-image selection questions:
+{"answer": [0, 2, 5]}
+where numbers are 0-indexed positions of matching candidate images.
+
+Rules:
+- Return ONLY the JSON object, nothing else.
+- Use example images only as references; do not include them in the returned indices.
+- Be precise with your classification.
+"""
 
 RECAPTCHA_V2_SYSTEM_PROMPT = """\
 You are an image classification assistant for reCAPTCHA v2 challenges.
@@ -85,7 +91,7 @@ Rules:
 """
 
 
-class ClassificationSolver:
+class ClassificationSolver:
     """Solves image classification captchas using a vision model."""
 
     def __init__(self, config: Config) -> None:
@@ -95,18 +101,28 @@ class ClassificationSolver:
             api_key=config.local_api_key,
         )
 
-    async def solve(self, params: dict[str, Any]) -> dict[str, Any]:
-        task_type = params.get("type", "")
-        system_prompt = self._get_system_prompt(task_type)
-        question = params.get("question", "") or params.get("queries", "")
-
-        # Handle different image field names across task types
-        images = self._extract_images(params)
-        if not images:
-            raise ValueError("No image data provided")
-
-        result = await self._classify(system_prompt, question, images)
-        return result
+    async def solve(self, params: dict[str, Any]) -> dict[str, Any]:
+        task_type = params.get("type", "")
+        system_prompt = self._get_system_prompt(task_type)
+        question = params.get("question", "") or params.get("queries", "")
+
+        # Handle different image field names across task types
+        images = self._extract_images(params)
+        if not images:
+            raise ValueError("No image data provided")
+
+        examples = self._extract_examples(params)
+        log.info(
+            "Classification request: task_type=%s model=%s images=%d examples=%d question=%r",
+            task_type or "unknown",
+            self._config.captcha_multimodal_model,
+            len(images),
+            len(examples),
+            question[:120] if isinstance(question, str) else question,
+        )
+        result = await self._classify(system_prompt, question, images, examples=examples)
+        log.info("Classification parsed result: %s", result)
+        return result
 
     @staticmethod
     def _get_system_prompt(task_type: str) -> str:
@@ -119,7 +135,7 @@ class ClassificationSolver:
         return prompts.get(task_type, RECAPTCHA_V2_SYSTEM_PROMPT)
 
     @staticmethod
-    def _extract_images(params: dict[str, Any]) -> list[str]:
+    def _extract_images(params: dict[str, Any]) -> list[str]:
         """Extract base64 image(s) from various param formats."""
         images: list[str] = []
 
@@ -137,10 +153,19 @@ class ClassificationSolver:
             images.append(params["body"])
 
         # HCaptcha queries format: list of base64 strings
-        if "queries" in params and isinstance(params["queries"], list):
-            images.extend(params["queries"])
-
-        return images
+        if "queries" in params and isinstance(params["queries"], list):
+            images.extend(params["queries"])
+
+        return images
+
+    @staticmethod
+    def _extract_examples(params: dict[str, Any]) -> list[str]:
+        examples = params.get("examples")
+        if isinstance(examples, list):
+            return [item for item in examples if isinstance(item, str)]
+        if isinstance(examples, str):
+            return [examples]
+        return []
 
     @staticmethod
     def _prepare_image(b64_data: str) -> str:
@@ -156,17 +181,55 @@ class ClassificationSolver:
         except Exception:
             return f"data:image/png;base64,{b64_data}"
 
-    async def _classify(
-        self, system_prompt: str, question: str, images: list[str]
-    ) -> dict[str, Any]:
-        content: list[dict[str, Any]] = []
-
-        for img_b64 in images:
-            data_url = self._prepare_image(img_b64)
-            content.append({
-                "type": "image_url",
-                "image_url": {"url": data_url, "detail": "high"},
-            })
+    async def _classify(
+        self,
+        system_prompt: str,
+        question: str,
+        images: list[str],
+        *,
+        examples: list[str] | None = None,
+    ) -> dict[str, Any]:
+        content: list[dict[str, Any]] = []
+
+        prepared_examples = examples or []
+        if prepared_examples:
+            content.append(
+                {
+                    "type": "text",
+                    "text": (
+                        "Sample images showing the target object. "
+                        "Do not classify these; use them only as references."
+                    ),
+                }
+            )
+        for example_b64 in prepared_examples:
+            data_url = self._prepare_image(example_b64)
+            content.append(
+                {
+                    "type": "image_url",
+                    "image_url": {"url": data_url, "detail": "high"},
+                }
+            )
+
+        if len(images) > 1:
+            content.append(
+                {
+                    "type": "text",
+                    "text": (
+                        "Candidate images to classify. "
+                        "Indices are 0-based in display order."
+                    ),
+                }
+            )
+
+        for img_b64 in images:
+            data_url = self._prepare_image(img_b64)
+            content.append(
+                {
+                    "type": "image_url",
+                    "image_url": {"url": data_url, "detail": "high"},
+                }
+            )
 
         user_text = question if question else "Classify this captcha image."
         content.append({"type": "text", "text": user_text})
@@ -174,17 +237,18 @@ class ClassificationSolver:
         last_error: Exception | None = None
         for attempt in range(self._config.captcha_retries):
             try:
-                response = await self._client.chat.completions.create(
-                    model=self._config.captcha_multimodal_model,
-                    temperature=0.05,
-                    max_tokens=512,
+                response = await self._client.chat.completions.create(
+                    model=self._config.captcha_multimodal_model,
+                    temperature=0.05,
+                    max_tokens=512,
                     messages=[
                         {"role": "system", "content": system_prompt},
                         {"role": "user", "content": content},
                     ],
-                )
-                raw = response.choices[0].message.content or ""
-                return self._parse_json(raw)
+                )
+                raw = response.choices[0].message.content or ""
+                log.info("Classification raw response: %s", raw[:300])
+                return self._parse_json(raw)
             except Exception as exc:
                 last_error = exc
                 log.warning("Classification attempt %d failed: %s", attempt + 1, exc)

src/services/hcaptcha.py

@@ -1,142 +1,477 @@
-"""HCaptcha solver using Playwright browser automation.
-
-Supports HCaptchaTaskProxyless task type.
-Visits the target page, interacts with the hCaptcha widget, and extracts the response token.
-"""
-
-from __future__ import annotations
-
-import asyncio
-import logging
-from typing import Any
-
-from playwright.async_api import Browser, Playwright, async_playwright
-
-from ..core.config import Config
-
-log = logging.getLogger(__name__)
-
-_STEALTH_JS = """
-Object.defineProperty(navigator, 'webdriver', {get: () => undefined});
-Object.defineProperty(navigator, 'languages', {get: () => ['en-US', 'en']});
-Object.defineProperty(navigator, 'plugins', {get: () => [1, 2, 3, 4, 5]});
-window.chrome = {runtime: {}, loadTimes: () => {}, csi: () => {}};
-"""
-
-_EXTRACT_HCAPTCHA_TOKEN_JS = """
-() => {
-    const textarea = document.querySelector('[name="h-captcha-response"]')
-        || document.querySelector('[name="g-recaptcha-response"]');
-    if (textarea && textarea.value && textarea.value.length > 20) {
-        return textarea.value;
-    }
-    if (window.hcaptcha && typeof window.hcaptcha.getResponse === 'function') {
-        const resp = window.hcaptcha.getResponse();
-        if (resp && resp.length > 20) return resp;
-    }
-    return null;
-}
-"""
-
-
-class HCaptchaSolver:
-    """Solves HCaptchaTaskProxyless tasks via headless Chromium."""
-
-    def __init__(self, config: Config, browser: Browser | None = None) -> None:
-        self._config = config
-        self._playwright: Playwright | None = None
-        self._browser: Browser | None = browser
-        self._owns_browser = browser is None
-
-    async def start(self) -> None:
-        if self._browser is not None:
-            return
-        self._playwright = await async_playwright().start()
-        self._browser = await self._playwright.chromium.launch(
-            headless=self._config.browser_headless,
-            args=[
-                "--disable-blink-features=AutomationControlled",
-                "--no-sandbox",
-                "--disable-dev-shm-usage",
-                "--disable-gpu",
-            ],
-        )
-        log.info("HCaptchaSolver browser started")
-
-    async def stop(self) -> None:
-        if self._owns_browser:
-            if self._browser:
-                await self._browser.close()
-            if self._playwright:
-                await self._playwright.stop()
-        log.info("HCaptchaSolver stopped")
-
-    async def solve(self, params: dict[str, Any]) -> dict[str, Any]:
-        website_url = params["websiteURL"]
-        website_key = params["websiteKey"]
-
-        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}
-            except Exception as exc:
-                last_error = exc
-                log.warning(
-                    "HCaptcha attempt %d/%d failed: %s",
-                    attempt + 1,
-                    self._config.captcha_retries,
-                    exc,
-                )
-                if attempt < self._config.captcha_retries - 1:
-                    await asyncio.sleep(2)
-
-        raise RuntimeError(
-            f"HCaptcha failed after {self._config.captcha_retries} attempts: {last_error}"
-        )
-
-    async def _solve_once(self, website_url: str, website_key: str) -> str:
-        assert self._browser is not None
-
-        context = await self._browser.new_context(
-            user_agent=(
-                "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
-                "AppleWebKit/537.36 (KHTML, like Gecko) "
-                "Chrome/131.0.0.0 Safari/537.36"
-            ),
-            viewport={"width": 1920, "height": 1080},
-            locale="en-US",
-        )
-        page = await context.new_page()
-        await page.add_init_script(_STEALTH_JS)
-
-        try:
-            timeout_ms = self._config.browser_timeout * 1000
-            await page.goto(website_url, wait_until="networkidle", timeout=timeout_ms)
-
-            await page.mouse.move(400, 300)
-            await asyncio.sleep(1)
-
-            # Click only the checkbox iframe — match by specific title to avoid the challenge iframe
-            iframe_element = page.frame_locator(
-                'iframe[title="Widget containing checkbox for hCaptcha security challenge"]'
-            )
-            checkbox = iframe_element.locator("#checkbox")
-            await checkbox.click(timeout=10_000)
-
-            # Wait for token — may require challenge completion; poll up to 30s
-            for _ in range(6):
-                await asyncio.sleep(5)
-                token = await page.evaluate(_EXTRACT_HCAPTCHA_TOKEN_JS)
-                if isinstance(token, str) and len(token) > 20:
-                    break
-            else:
-                token = None
-
-            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
-        finally:
-            await context.close()
+"""HCaptcha solver using Playwright browser automation.
+
+Supports ``HCaptchaTaskProxyless`` task type.
+
+Strategy:
+  1. Visit the target page with a realistic browser context.
+  2. Click the hCaptcha checkbox.
+  3. If a token is issued immediately, return it.
+  4. If an image-selection challenge appears, extract the prompt + tile images,
+     call ``ClassificationSolver`` for ``HCaptchaClassification``-style
+     reasoning, click the matching tiles, submit the challenge, and continue
+     polling for the token.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import base64
+import logging
+from typing import Any
+from urllib.parse import parse_qs, urlencode, urlsplit, urlunsplit
+
+from playwright.async_api import Browser, ElementHandle, Frame, Page, Playwright, async_playwright
+
+from ..core.config import Config
+from .classification import ClassificationSolver
+
+log = logging.getLogger(__name__)
+
+_STEALTH_JS = """
+Object.defineProperty(navigator, 'webdriver', {get: () => undefined});
+Object.defineProperty(navigator, 'languages', {get: () => ['en-US', 'en']});
+Object.defineProperty(navigator, 'plugins', {get: () => [1, 2, 3, 4, 5]});
+window.chrome = {runtime: {}, loadTimes: () => {}, csi: () => {}};
+"""
+
+_EXTRACT_HCAPTCHA_TOKEN_JS = """
+() => {
+    const textarea = document.querySelector('[name="h-captcha-response"]')
+        || document.querySelector('[name="g-recaptcha-response"]');
+    if (textarea && textarea.value && textarea.value.length > 20) {
+        return textarea.value;
+    }
+    if (window.hcaptcha && typeof window.hcaptcha.getResponse === 'function') {
+        const resp = window.hcaptcha.getResponse();
+        if (resp && resp.length > 20) return resp;
+    }
+    return null;
+}
+"""
+
+_QUESTION_JS = """
+() => {
+    const prompt = document.querySelector('.prompt-text')
+        || document.querySelector('h2.prompt-text')
+        || document.querySelector('.challenge-prompt')
+        || document.querySelector('[class*="prompt"]');
+    return prompt?.textContent?.trim() || null;
+}
+"""
+
+_CHALLENGE_TILE_SELECTORS = (
+    ".task-grid .task-image",
+    ".task-grid .task",
+    ".task-grid .image",
+    ".challenge-container .task-image",
+    ".challenge-view .task-image",
+    ".task-image",
+    ".task",
+)
+
+_EXAMPLE_IMAGE_SELECTORS = (
+    ".challenge-example .image",
+    ".challenge-example",
+    ".example-wrapper .image",
+)
+
+_VERIFY_BUTTON_SELECTORS = (
+    ".button-submit",
+    'button[type="submit"]',
+    'button[aria-label*="Verify"]',
+)
+
+
+class HCaptchaSolver:
+    """Solves ``HCaptchaTaskProxyless`` tasks via Playwright."""
+
+    def __init__(
+        self,
+        config: Config,
+        browser: Browser | None = None,
+        classifier: ClassificationSolver | None = None,
+    ) -> None:
+        self._config = config
+        self._playwright: Playwright | None = None
+        self._browser: Browser | None = browser
+        self._owns_browser = browser is None
+        self._classifier = classifier
+
+    async def start(self) -> None:
+        if self._browser is not None:
+            return
+        self._playwright = await async_playwright().start()
+        self._browser = await self._playwright.chromium.launch(
+            headless=self._config.browser_headless,
+            args=[
+                "--disable-blink-features=AutomationControlled",
+                "--no-sandbox",
+                "--disable-dev-shm-usage",
+                "--disable-gpu",
+            ],
+        )
+        log.info("HCaptchaSolver browser started")
+
+    async def stop(self) -> None:
+        if self._owns_browser:
+            if self._browser:
+                await self._browser.close()
+            if self._playwright:
+                await self._playwright.stop()
+        log.info("HCaptchaSolver stopped")
+
+    async def solve(self, params: dict[str, Any]) -> dict[str, Any]:
+        website_url = params["websiteURL"]
+        website_key = params["websiteKey"]
+
+        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}
+            except Exception as exc:
+                last_error = exc
+                log.warning(
+                    "HCaptcha attempt %d/%d failed: %s",
+                    attempt + 1,
+                    self._config.captcha_retries,
+                    exc,
+                )
+                if attempt < self._config.captcha_retries - 1:
+                    await asyncio.sleep(2)
+
+        raise RuntimeError(
+            f"HCaptcha failed after {self._config.captcha_retries} attempts: {last_error}"
+        )
+
+    async def _solve_once(self, website_url: str, website_key: str) -> str:
+        assert self._browser is not None
+        target_url = self._prepare_target_url(website_url, website_key)
+        if target_url != website_url:
+            log.info("Normalized hCaptcha target URL to honor requested sitekey: %s", target_url)
+
+        context = await self._browser.new_context(
+            user_agent=(
+                "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+                "AppleWebKit/537.36 (KHTML, like Gecko) "
+                "Chrome/131.0.0.0 Safari/537.36"
+            ),
+            viewport={"width": 1920, "height": 1080},
+            locale="en-US",
+        )
+        page = await context.new_page()
+        await page.add_init_script(_STEALTH_JS)
+
+        try:
+            timeout_ms = self._config.browser_timeout * 1000
+            await page.goto(target_url, wait_until="networkidle", timeout=timeout_ms)
+            await page.mouse.move(400, 300)
+            await asyncio.sleep(1)
+
+            await self._click_checkbox(page)
+
+            # 先给低风险会话一个直接出 token 的机会。
+            token = await self._wait_for_token(page, seconds=4)
+            if token:
+                log.info("Got hCaptcha token directly after checkbox click (len=%d)", len(token))
+                return token
+
+            # 无头环境常见路径：进入图片 challenge，然后走 classification fallback。
+            log.info(
+                "No direct hCaptcha token after checkbox click, entering classification fallback"
+            )
+            fallback_handled = await self._solve_image_selection_challenge(page)
+            if fallback_handled:
+                token = await self._wait_for_token(page)
+
+            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
+        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")
+
+        checkbox = await frame.query_selector("#checkbox")
+        if checkbox is None:
+            raise RuntimeError("Could not find hCaptcha checkbox element")
+
+        await checkbox.click(timeout=10_000)
+        log.info("Clicked hCaptcha checkbox")
+
+    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)
+            if isinstance(token, str) and len(token) > 20:
+                return token
+            await asyncio.sleep(1)
+        return None
+
+    async def _find_frame(
+        self, page: Page, frame_role: str, *, wait_seconds: int = 5
+    ) -> Frame | None:
+        attempts = max(1, wait_seconds * 2)
+        for _ in range(attempts):
+            for frame in page.frames:
+                url = frame.url or ""
+                if "hcaptcha" in url and f"frame={frame_role}" in url:
+                    return frame
+            await asyncio.sleep(0.5)
+        return None
+
+    @staticmethod
+    def _prepare_target_url(website_url: str, website_key: str) -> str:
+        """为官方 demo 自动补齐/对齐 sitekey，确保按请求参数测试真实行为。"""
+        if not website_key:
+            return website_url
+
+        parsed = urlsplit(website_url)
+        host = parsed.netloc.lower()
+        path = parsed.path.rstrip("/")
+        is_official_demo = host in {"accounts.hcaptcha.com", "demo.hcaptcha.com"} and path == "/demo"
+        if not is_official_demo:
+            return website_url
+
+        query = parse_qs(parsed.query, keep_blank_values=True)
+        changed = False
+
+        current_sitekey = query.get("sitekey", [None])[0]
+        if current_sitekey != website_key:
+            query["sitekey"] = [website_key]
+            changed = True
+
+        if "hl" not in query:
+            query["hl"] = ["en"]
+            changed = True
+
+        if not changed:
+            return website_url
+
+        return urlunsplit(
+            (
+                parsed.scheme,
+                parsed.netloc,
+                parsed.path,
+                urlencode(query, doseq=True),
+                parsed.fragment,
+            )
+        )
+
+    async def _solve_image_selection_challenge(self, page: Page) -> bool:
+        if self._classifier is None:
+            raise RuntimeError(
+                "Classification fallback is unavailable because no ClassificationSolver was injected"
+            )
+
+        rounds = max(1, self._config.captcha_retries)
+        for round_index in range(rounds):
+            token = await self._wait_for_token(page, seconds=1)
+            if token:
+                return True
+
+            challenge = await self._collect_selection_challenge(page)
+            if challenge is None:
+                unsupported_reason = await self._describe_unsupported_challenge(page)
+                log.warning(
+                    "Could not collect hCaptcha image-selection challenge in round %d: %s",
+                    round_index + 1,
+                    unsupported_reason,
+                )
+                if round_index == 0:
+                    raise RuntimeError(unsupported_reason)
+                return False
+
+            log.info(
+                "Collected hCaptcha image-selection challenge in round %d: question=%r tiles=%d examples=%d",
+                round_index + 1,
+                challenge["question"],
+                len(challenge["tiles"]),
+                len(challenge["examples"]),
+            )
+            payload = self._build_classification_payload(
+                question=challenge["question"],
+                tile_images=challenge["tile_images"],
+                examples=challenge["examples"],
+            )
+            result = await self._classifier.solve(payload)
+            log.info("Classification solver returned raw result: %s", result)
+            indices = self._extract_selection_indices(
+                result=result,
+                tile_count=len(challenge["tiles"]),
+            )
+
+            await self._click_selected_tiles(challenge["tiles"], indices)
+            await self._click_verify_button(challenge["frame"])
+
+            token = await self._wait_for_token(page, seconds=6)
+            if token:
+                return True
+
+            log.info(
+                "hCaptcha challenge round %d submitted without immediate token, retrying",
+                round_index + 1,
+            )
+
+        return False
+
+    async def _collect_selection_challenge(self, page: Page) -> dict[str, Any] | None:
+        frame = await self._find_frame(page, "challenge", wait_seconds=10)
+        if frame is None:
+            return None
+
+        await asyncio.sleep(1)
+        question = await frame.evaluate(_QUESTION_JS)
+        if not isinstance(question, str) or not question.strip():
+            return None
+
+        tiles = await self._find_clickable_tiles(frame)
+        if not tiles:
+            return None
+
+        tile_entries: list[tuple[ElementHandle[Any], str]] = []
+        for tile in tiles:
+            encoded = await self._capture_element_base64(tile)
+            if encoded:
+                tile_entries.append((tile, encoded))
+
+        if not tile_entries:
+            return None
+
+        return {
+            "frame": frame,
+            "question": question.strip(),
+            "tiles": [tile for tile, _ in tile_entries],
+            "tile_images": [encoded for _, encoded in tile_entries],
+            "examples": await self._extract_example_images(frame),
+        }
+
+    async def _find_clickable_tiles(self, frame: Frame) -> list[ElementHandle[Any]]:
+        for selector in _CHALLENGE_TILE_SELECTORS:
+            elements = await frame.query_selector_all(selector)
+            if elements:
+                return elements
+        return []
+
+    async def _extract_example_images(self, frame: Frame) -> list[str]:
+        examples: list[str] = []
+        for selector in _EXAMPLE_IMAGE_SELECTORS:
+            elements = await frame.query_selector_all(selector)
+            if not elements:
+                continue
+            for element in elements:
+                encoded = await self._capture_element_base64(element)
+                if encoded:
+                    examples.append(encoded)
+            if examples:
+                break
+        return examples
+
+    async def _describe_unsupported_challenge(self, page: Page) -> str:
+        """给出更贴近真实 challenge 类型的错误信息，避免把 canvas/puzzle 误报成网格 DOM 问题。"""
+        frame = await self._find_frame(page, "challenge", wait_seconds=2)
+        if frame is None:
+            return (
+                "hCaptcha challenge iframe disappeared before the built-in fallback "
+                "could inspect it"
+            )
+
+        prompt = await frame.evaluate(_QUESTION_JS)
+        prompt_text = prompt.strip().lower() if isinstance(prompt, str) else ""
+        has_canvas = await frame.locator("canvas").count() > 0
+        submit_text = (
+            await frame.locator(".button-submit").first.inner_text()
+            if await frame.locator(".button-submit").count() > 0
+            else ""
+        )
+
+        if "puzzle piece" in prompt_text or (has_canvas and "skip" in submit_text.lower()):
+            log.warning(
+                "Detected unsupported hCaptcha canvas/puzzle challenge: prompt=%r submit=%r has_canvas=%s",
+                prompt,
+                submit_text,
+                has_canvas,
+            )
+            return (
+                "hCaptcha presented a canvas/puzzle challenge, which is not supported "
+                "by the built-in HCaptchaClassification fallback"
+            )
+
+        log.warning(
+            "Detected unsupported hCaptcha challenge layout: prompt=%r submit=%r has_canvas=%s",
+            prompt,
+            submit_text,
+            has_canvas,
+        )
+        return (
+            "hCaptcha image challenge detected, but the current DOM layout is not "
+            "supported by the built-in classification fallback"
+        )
+
+    async def _capture_element_base64(self, element: ElementHandle[Any]) -> str | None:
+        try:
+            image_bytes = await element.screenshot(type="png")
+        except Exception:
+            return None
+        return base64.b64encode(image_bytes).decode("ascii")
+
+    @staticmethod
+    def _build_classification_payload(
+        *, question: str, tile_images: list[str], examples: list[str]
+    ) -> dict[str, Any]:
+        payload: dict[str, Any] = {
+            "type": "HCaptchaClassification",
+            "question": question,
+            "images": tile_images,
+        }
+        if examples:
+            payload["examples"] = examples
+        return payload
+
+    @staticmethod
+    def _extract_selection_indices(
+        *, result: dict[str, Any], tile_count: int
+    ) -> list[int]:
+        raw_answer = result.get("answer")
+        if isinstance(raw_answer, bool):
+            indices = [0] if raw_answer and tile_count == 1 else []
+        elif isinstance(raw_answer, list):
+            indices = [int(idx) for idx in raw_answer if isinstance(idx, int | float)]
+        else:
+            raw_objects = result.get("objects")
+            if isinstance(raw_objects, list):
+                indices = [int(idx) for idx in raw_objects if isinstance(idx, int | float)]
+            else:
+                indices = []
+
+        deduped: list[int] = []
+        for idx in indices:
+            if 0 <= idx < tile_count and idx not in deduped:
+                deduped.append(idx)
+        return deduped
+
+    async def _click_selected_tiles(
+        self,
+        tiles: list[ElementHandle[Any]],
+        indices: list[int],
+    ) -> None:
+        for idx in indices:
+            await tiles[idx].click(timeout=10_000)
+            await asyncio.sleep(0.2)
+        log.info("Clicked %d hCaptcha tile(s): %s", len(indices), indices)
+
+    async def _click_verify_button(self, frame: Frame) -> None:
+        for selector in _VERIFY_BUTTON_SELECTORS:
+            button = await frame.query_selector(selector)
+            if button is None:
+                continue
+            await button.click(timeout=10_000)
+            await asyncio.sleep(1)
+            log.info("Submitted hCaptcha challenge with selector %s", selector)
+            return
+        raise RuntimeError("Could not find hCaptcha verify/submit button")