Method login

src/services/protocol_login.py:192–321

(
        self,
        google_cookies_raw: str,
        proxy: Optional[str] = None,
        email: Optional[str] = None,
    )

    """Login to labs.google/fx through NextAuth Google OAuth using Google cookies."""

    async def login(
        self,
        google_cookies_raw: str,
        proxy: Optional[str] = None,
        email: Optional[str] = None,
    ) -> Dict[str, Any]:
        google_cookies = _parse_google_cookies(google_cookies_raw)
        if not any(name in google_cookies for name in GOOGLE_COOKIE_NAMES):
            return {
                "success": False,
                "error": "未找到有效的 Google cookie（需要 SID/HSID/SSID/APISID/SAPISID 中至少一个）",
            }

        session_kwargs: Dict[str, Any] = {"impersonate": IMPERSONATE, "trust_env": False}
        proxy_url = _normalize_proxy_url(proxy)
        if proxy_url:
            session_kwargs["proxy"] = proxy_url

        async with AsyncSession(**session_kwargs) as session:
            try:
                csrf_resp = await session.get(f"{LABS_BASE}/api/auth/csrf")
                if csrf_resp.status_code != 200:
                    return {"success": False, "error": f"CSRF 失败: HTTP {csrf_resp.status_code}"}
                csrf_token = (csrf_resp.json() or {}).get("csrfToken")
                if not csrf_token:
                    return {"success": False, "error": "CSRF 响应缺少 csrfToken"}

                labs_cookies: Dict[str, str] = {}
                _merge_cookies(labs_cookies, csrf_resp.headers)

                signin_resp = await session.post(
                    f"{LABS_BASE}/api/auth/signin/google",
                    data={
                        "csrfToken": csrf_token,
                        "callbackUrl": LABS_BASE,
                        "json": "true",
                    },
                    headers={
                        "Referer": LABS_BASE,
                        "Origin": "https://labs.google",
                        "Cookie": _build_cookie_header(labs_cookies),
                    },
                    allow_redirects=False,
                )
                if signin_resp.status_code != 200:
                    return {"success": False, "error": f"Signin 失败: HTTP {signin_resp.status_code}"}

                _merge_cookies(labs_cookies, signin_resp.headers)
                signin_data = signin_resp.json() or {}
                redirect_url = signin_data.get("redirect") or signin_data.get("url")
                if not redirect_url:
                    return {"success": False, "error": f"无重定向 URL: {json.dumps(signin_data)[:200]}"}
                redirect_url = _append_login_hint(redirect_url, email)

                google_cookie_header = _build_cookie_header(google_cookies)
                callback_url = ""
                current_url = redirect_url