Method _get_recaptcha_token

src/services/flow_client.py:4512–4736

获取reCAPTCHA token - 支持多种打码方式 Args: project_id: 项目ID action: reCAPTCHA action类型 - IMAGE_GENERATION: 图片生成和2K/4K图片放大 (默认) - VIDEO_GENERATION: 视频生成和视频放大 token_id: 当前业务 token id（browser 模式下用于读取 token 级打码代理）

(
        self,
        project_id: str,
        action: str = "IMAGE_GENERATION",
        token_id: Optional[int] = None
    )

        return max(12, min(base_timeout, target_timeout))

    async def _get_recaptcha_token(
        self,
        project_id: str,
        action: str = "IMAGE_GENERATION",
        token_id: Optional[int] = None
    ) -> tuple[Optional[str], Optional[Union[int, str]]]:
        """获取reCAPTCHA token - 支持多种打码方式
        
        Args:
            project_id: 项目ID
            action: reCAPTCHA action类型
                - IMAGE_GENERATION: 图片生成和2K/4K图片放大 (默认)
                - VIDEO_GENERATION: 视频生成和视频放大
            token_id: 当前业务 token id（browser 模式下用于读取 token 级打码代理）
        
        Returns:
            (token, browser_id) 元组。
            - browser 模式: browser_id 为本地浏览器 ID
            - remote_browser 模式: browser_id 为远程 session_id
            - 其他模式: browser_id 为 None
        """
        captcha_method = config.captcha_method
        debug_logger.log_info(f"[reCAPTCHA] 开始获取 token: method={captcha_method}, project_id={project_id}, action={action}")

        if captcha_method == "extension":
            try:
                from .browser_captcha_extension import ExtensionCaptchaService
                service = await ExtensionCaptchaService.get_instance(self.db)
                extension_timeout = 45 if action == "VIDEO_GENERATION" else 25
                token = await service.get_token(
                    project_id,
                    action,
                    timeout=extension_timeout,
                    token_id=token_id
                )
                self._set_request_fingerprint(None)
                return token, None
            except Exception as e:
                debug_logger.log_error(f"[reCAPTCHA Extension] 错误: {str(e)}")
                self._set_request_fingerprint(None)
                return None, None

        # 内置浏览器打码 (nodriver)
        if captcha_method == "personal":
            debug_logger.log_info(f"[reCAPTCHA] 使用 personal 模式")
            try:
                from .browser_captcha_personal import BrowserCaptchaService
                debug_logger.log_info(f"[reCAPTCHA] 导入 BrowserCaptchaService 成功")
                service = await BrowserCaptchaService.get_instance(self.db)
                debug_logger.log_info(f"[reCAPTCHA] 获取服务实例成功，准备调用 get_token")
                solve_bundle = None
                get_token_bundle = getattr(service, "get_token_bundle", None)
                if callable(get_token_bundle):
                    solve_bundle = await get_token_bundle(
                        project_id,
                        action,
                        token_id=token_id,
                    )