diff --git "a/src/core/register.py" "b/src/core/register.py" deleted file mode 100644--- "a/src/core/register.py" +++ /dev/null @@ -1,3299 +0,0 @@ -""" -注册流程引擎 -从 main.py 中提取并重构的注册流程 -""" - -import re -import json -import time -import logging -import secrets -import string -import uuid -from typing import Optional, Dict, Any, Tuple, Callable, List -from dataclasses import dataclass -from datetime import datetime - -from curl_cffi import requests as cffi_requests - -from .anyauto.register_flow import AnyAutoRegistrationEngine -from .anyauto.sentinel_browser import get_sentinel_token_via_browser -from .anyauto.sentinel_token import build_sentinel_token -from .anyauto.utils import generate_datadog_trace -from .openai.oauth import OAuthManager, OAuthStart -from .http_client import OpenAIHTTPClient, HTTPClientError -from ..services import EmailServiceFactory, BaseEmailService, EmailServiceType -from ..database import crud -from ..database.session import get_db -from ..config.constants import ( - OPENAI_API_ENDPOINTS, - OPENAI_PAGE_TYPES, - generate_random_user_info, - OTP_CODE_PATTERN, - DEFAULT_PASSWORD_LENGTH, - PASSWORD_SPECIAL_CHARSET, - PASSWORD_CHARSET, - AccountStatus, - TaskStatus, -) -from ..config.settings import get_settings - - -logger = logging.getLogger(__name__) - -TOKEN_COMPLETENESS_COMPLETE = "complete" -TOKEN_COMPLETENESS_ONLY_ACCESS = "only_access_token" -TOKEN_COMPLETENESS_PARTIAL = "partial" -TOKEN_COMPLETENESS_MISSING_ACCESS = "missing_access_token" - - -def build_token_completeness_metadata( - access_token: str = "", - refresh_token: str = "", - id_token: str = "", - account_id: str = "", -) -> Dict[str, Any]: - """构建统一的 token 完整度元数据。""" - has_access_token = bool(str(access_token or "").strip()) - has_refresh_token = bool(str(refresh_token or "").strip()) - has_id_token = bool(str(id_token or "").strip()) - has_account_id = bool(str(account_id or "").strip()) - - if not has_access_token: - token_completeness = TOKEN_COMPLETENESS_MISSING_ACCESS - elif has_refresh_token and has_id_token and has_account_id: - token_completeness = TOKEN_COMPLETENESS_COMPLETE - elif not has_refresh_token: - token_completeness = TOKEN_COMPLETENESS_ONLY_ACCESS - else: - token_completeness = TOKEN_COMPLETENESS_PARTIAL - - return { - "has_access_token": has_access_token, - "has_refresh_token": has_refresh_token, - "has_id_token": has_id_token, - "has_account_id": has_account_id, - "token_completeness": token_completeness, - } - - -class RegistrationCancelled(Exception): - """Raised when a registration task is cancelled cooperatively.""" - - -@dataclass -class RegistrationResult: - """注册结果""" - success: bool - email: str = "" - password: str = "" # 注册密码 - account_id: str = "" - workspace_id: str = "" - access_token: str = "" - refresh_token: str = "" - id_token: str = "" - session_token: str = "" # 会话令牌 - device_id: str = "" # oai-did - error_message: str = "" - logs: list = None - metadata: dict = None - source: str = "register" # 'register' 或 'login'，区分账号来源 - - def to_dict(self) -> Dict[str, Any]: - """转换为字典""" - return { - "success": self.success, - "email": self.email, - "password": self.password, - "account_id": self.account_id, - "workspace_id": self.workspace_id, - "access_token": self.access_token[:20] + "..." if self.access_token else "", - "refresh_token": self.refresh_token[:20] + "..." if self.refresh_token else "", - "id_token": self.id_token[:20] + "..." if self.id_token else "", - "session_token": self.session_token[:20] + "..." if self.session_token else "", - "device_id": self.device_id, - "error_message": self.error_message, - "logs": self.logs or [], - "metadata": self.metadata or {}, - "source": self.source, - } - - -@dataclass -class SignupFormResult: - """提交注册表单的结果""" - success: bool - page_type: str = "" # 响应中的 page.type 字段 - is_existing_account: bool = False # 是否为已注册账号 - response_data: Dict[str, Any] = None # 完整的响应数据 - error_message: str = "" - - -class RegistrationEngine: - """ - 注册引擎 - 负责协调邮箱服务、OAuth 流程和 OpenAI API 调用 - """ - - def __init__( - self, - email_service: BaseEmailService, - proxy_url: Optional[str] = None, - callback_logger: Optional[Callable[[str], None]] = None, - task_uuid: Optional[str] = None, - cancel_checker: Optional[Callable[[], bool]] = None, - ): - """ - 初始化注册引擎 - - Args: - email_service: 邮箱服务实例 - proxy_url: 代理 URL - callback_logger: 日志回调函数 - task_uuid: 任务 UUID（用于数据库记录） - """ - self.email_service = email_service - self.proxy_url = proxy_url - self.callback_logger = callback_logger or (lambda msg: logger.info(msg)) - self.task_uuid = task_uuid - self.cancel_checker = cancel_checker or (lambda: False) - - # 创建 HTTP 客户端 - self.http_client = OpenAIHTTPClient(proxy_url=proxy_url) - - # 创建 OAuth 管理器 - settings = get_settings() - self.oauth_manager = OAuthManager( - client_id=settings.openai_client_id, - auth_url=settings.openai_auth_url, - token_url=settings.openai_token_url, - redirect_uri=settings.openai_redirect_uri, - scope=settings.openai_scope, - proxy_url=proxy_url # 传递代理配置 - ) - entry_flow = str(getattr(settings, "registration_entry_flow", "native") or "native").strip().lower() - # 配置层仅保留 native/abcard；Outlook 邮箱在执行时自动切换 outlook 链路。 - self.registration_entry_flow: str = entry_flow if entry_flow in {"native", "abcard"} else "native" - self.anyauto_browser_mode: str = str( - getattr(settings, "registration_anyauto_browser_mode", "headless") or "headless" - ).strip().lower() - - # 状态变量 - self.email: Optional[str] = None - self.inbox_email: Optional[str] = None # 邮箱服务原始地址（用于收件） - self.password: Optional[str] = None # 注册密码 - self.email_info: Optional[Dict[str, Any]] = None - self.oauth_start: Optional[OAuthStart] = None - self.session: Optional[cffi_requests.Session] = None - self.session_token: Optional[str] = None # 会话令牌 - self.device_id: Optional[str] = None # oai-did - self.logs: list = [] - self._otp_sent_at: Optional[float] = None # OTP 发送时间戳 - self._is_existing_account: bool = False # 是否为已注册账号（用于自动登录） - self._token_acquisition_requires_login: bool = False # 新注册账号需要二次登录拿 token - self._create_account_continue_url: Optional[str] = None # create_account 返回的 continue_url（ABCard链路兜底） - self._create_account_workspace_id: Optional[str] = None - self._create_account_account_id: Optional[str] = None - self._create_account_refresh_token: Optional[str] = None - self._last_validate_otp_continue_url: Optional[str] = None - self._last_validate_otp_workspace_id: Optional[str] = None - self._last_register_password_error: Optional[str] = None - self._last_otp_validation_code: Optional[str] = None - self._last_otp_validation_status_code: Optional[int] = None - self._last_otp_validation_outcome: str = "" # success/http_non_200/network_timeout/network_error - - def _check_cancelled(self, stage: str = "") -> None: - """Raise if the current registration task has been cancelled.""" - if not self.cancel_checker: - return - try: - cancelled = bool(self.cancel_checker()) - except Exception: - cancelled = False - if cancelled: - stage_text = f"（{stage}）" if stage else "" - self._log(f"检测到取消请求{stage_text}，当前任务准备收工", "warning") - raise RegistrationCancelled("任务已取消") - - def _log(self, message: str, level: str = "info"): - """记录日志""" - timestamp = datetime.now().strftime("%H:%M:%S") - log_message = f"[{timestamp}] {message}" - - # 添加到日志列表 - self.logs.append(log_message) - - # 调用回调函数 - if self.callback_logger: - self.callback_logger(log_message) - - # 记录到数据库（如果有关联任务） - if self.task_uuid: - try: - with get_db() as db: - crud.append_task_log(db, self.task_uuid, log_message) - except Exception as e: - logger.warning(f"记录任务日志失败: {e}") - - # 根据级别记录到日志系统 - if level == "error": - logger.error(message) - elif level == "warning": - logger.warning(message) - else: - logger.info(message) - - def _dump_session_cookies(self) -> str: - """导出当前会话 cookies（用于后续支付/绑卡自动化）。""" - if not self.session: - return "" - try: - cookie_map: dict[str, str] = {} - order: list[str] = [] - - def _push(name: Optional[str], value: Optional[str]): - key = str(name or "").strip() - val = str(value or "").strip() - if not key: - return - if key not in cookie_map: - cookie_map[key] = val - order.append(key) - return - # 同名 cookie 可能来自不同域/路径：优先保留非空且更长值，避免空值覆盖有效分片。 - prev = str(cookie_map.get(key) or "").strip() - if (not prev and val) or (val and len(val) > len(prev)): - cookie_map[key] = val - - # 1) 常规 requests/curl_cffi 字典接口 - try: - for key, value in self.session.cookies.items(): - _push(key, value) - except Exception: - pass - - # 2) CookieJar 接口（可拿到分片 cookie） - try: - jar = getattr(self.session.cookies, "jar", None) - if jar is not None: - for cookie in jar: - _push(getattr(cookie, "name", ""), getattr(cookie, "value", "")) - except Exception: - pass - - # 3) 关键 cookie 兜底读取 - for key in ( - "oai-did", - "oai-client-auth-session", - "__Secure-next-auth.session-token", - "_Secure-next-auth.session-token", - ): - try: - _push(key, self.session.cookies.get(key)) - except Exception: - continue - - pairs = [(k, cookie_map.get(k, "")) for k in order if k] - return "; ".join(f"{k}={v}" for k, v in pairs if k) - except Exception: - return "" - - @staticmethod - def _extract_session_token_from_cookie_jar(cookie_jar) -> str: - """ - 从 CookieJar 中提取 next-auth session token（兼容分片 + 重复域名）。 - """ - if not cookie_jar: - return "" - - entries: list[tuple[str, str]] = [] - try: - for key, value in cookie_jar.items(): - entries.append((str(key or "").strip(), str(value or "").strip())) - except Exception: - pass - - try: - jar = getattr(cookie_jar, "jar", None) - if jar is not None: - for cookie in jar: - entries.append( - ( - str(getattr(cookie, "name", "") or "").strip(), - str(getattr(cookie, "value", "") or "").strip(), - ) - ) - except Exception: - pass - - direct_candidates = [ - val - for name, val in entries - if name in ("__Secure-next-auth.session-token", "_Secure-next-auth.session-token") and val - ] - if direct_candidates: - return max(direct_candidates, key=len) - - chunk_map: dict[int, str] = {} - for name, value in entries: - if not ( - name.startswith("__Secure-next-auth.session-token.") - or name.startswith("_Secure-next-auth.session-token.") - ): - continue - if not value: - continue - try: - idx = int(name.rsplit(".", 1)[-1]) - except Exception: - continue - prev = chunk_map.get(idx, "") - if not prev or len(value) > len(prev): - chunk_map[idx] = value - - if chunk_map: - return "".join(chunk_map[i] for i in sorted(chunk_map.keys())) - return "" - - @staticmethod - def _flatten_set_cookie_headers(response) -> str: - """ - 合并多条 Set-Cookie（包含分片 cookie）。 - """ - try: - headers = getattr(response, "headers", None) - if headers is None: - return "" - if hasattr(headers, "get_list"): - values = headers.get_list("set-cookie") - if values: - return " | ".join(str(v or "") for v in values if v is not None) - if hasattr(headers, "get_all"): - values = headers.get_all("set-cookie") - if values: - return " | ".join(str(v or "") for v in values if v is not None) - return str(headers.get("set-cookie") or "") - except Exception: - return "" - - @staticmethod - def _extract_request_cookie_header(response) -> str: - """ - 从响应对象关联的请求头中提取 Cookie。 - 对齐 F12 Network -> Request Headers -> Cookie 的观测路径。 - """ - try: - request_obj = getattr(response, "request", None) - if request_obj is None: - return "" - headers = getattr(request_obj, "headers", None) - if headers is None: - return "" - - if hasattr(headers, "get"): - value = headers.get("cookie") or headers.get("Cookie") - if value: - return str(value) - - try: - for key, value in dict(headers).items(): - if str(key or "").strip().lower() == "cookie" and value: - return str(value) - except Exception: - pass - except Exception: - pass - return "" - - def _generate_password(self, length: int = DEFAULT_PASSWORD_LENGTH) -> str: - """生成随机密码""" - length = max(8, int(length or DEFAULT_PASSWORD_LENGTH)) - password_chars = [ - secrets.choice(string.ascii_lowercase), - secrets.choice(string.ascii_uppercase), - secrets.choice(string.digits), - secrets.choice(PASSWORD_SPECIAL_CHARSET), - ] - password_chars.extend(secrets.choice(PASSWORD_CHARSET) for _ in range(length - len(password_chars))) - secrets.SystemRandom().shuffle(password_chars) - return ''.join(password_chars) - - def _check_ip_location(self) -> Tuple[bool, Optional[str]]: - """检查 IP 地理位置""" - try: - return self.http_client.check_ip_location() - except Exception as e: - self._log(f"检查 IP 地理位置失败: {e}", "error") - return False, None - - def _create_email(self) -> bool: - """创建邮箱""" - try: - self._log(f"正在创建 {self.email_service.service_type.value} 邮箱，先给新账号整个收件箱...") - self.email_info = self.email_service.create_email() - - if not self.email_info or "email" not in self.email_info: - self._log("创建邮箱失败: 返回信息不完整", "error") - return False - - raw_email = str(self.email_info["email"] or "").strip() - normalized_email = raw_email.lower() - - # 保留原始收件地址，注册链路统一使用规范化邮箱，规避 "Failed to register username"。 - self.inbox_email = raw_email - self.email = normalized_email - self.email_info["email"] = normalized_email - - if raw_email and raw_email != normalized_email: - self._log(f"邮箱规范化: {raw_email} -> {normalized_email}") - - self._log(f"邮箱已就位，地址新鲜出炉: {self.email}") - return True - - except Exception as e: - self._log(f"创建邮箱失败: {e}", "error") - return False - - def _start_oauth(self) -> bool: - """开始 OAuth 流程""" - try: - self._log("开始 OAuth 授权流程，去门口刷个脸...") - self.oauth_start = self.oauth_manager.start_oauth() - self._log(f"OAuth URL 已备好，通道已经打开: {self.oauth_start.auth_url[:80]}...") - return True - except Exception as e: - self._log(f"生成 OAuth URL 失败: {e}", "error") - return False - - def _init_session(self) -> bool: - """初始化会话""" - try: - self.session = self.http_client.session - return True - except Exception as e: - self._log(f"初始化会话失败: {e}", "error") - return False - - def _get_device_id(self) -> Optional[str]: - """获取 Device ID""" - if not self.oauth_start: - return None - - max_attempts = 3 - for attempt in range(1, max_attempts + 1): - try: - if not self.session: - self.session = self.http_client.session - - response = self.session.get( - self.oauth_start.auth_url, - timeout=20 - ) - did = self.session.cookies.get("oai-did") - - if not did: - # 对齐 ABCard：部分环境 cookie 不落盘，尝试从 HTML 文本提取 - try: - m = re.search(r'oai-did["\s:=]+([a-f0-9-]{36})', str(response.text or ""), re.IGNORECASE) - if m: - did = str(m.group(1) or "").strip() - if did: - try: - self.session.cookies.set("oai-did", did, domain=".chatgpt.com", path="/") - except Exception: - pass - except Exception: - pass - - if did: - self._log(f"Device ID: {did}") - return did - - self._log( - f"获取 Device ID 失败: 未返回 oai-did Cookie (HTTP {response.status_code}, 第 {attempt}/{max_attempts} 次)", - "warning" if attempt < max_attempts else "error" - ) - except Exception as e: - self._log( - f"获取 Device ID 失败: {e} (第 {attempt}/{max_attempts} 次)", - "warning" if attempt < max_attempts else "error" - ) - - if attempt < max_attempts: - time.sleep(attempt) - self.http_client.close() - self.session = self.http_client.session - - # 对齐 ABCard：无法从响应拿到 did 时，优先复用上次成功 did，再使用 UUID 兜底。 - fallback_did = str(self.device_id or "").strip() or str(uuid.uuid4()) - try: - if self.session: - self.session.cookies.set("oai-did", fallback_did, domain=".chatgpt.com", path="/") - except Exception: - pass - self._log(f"未获取到 oai-did，使用兜底 Device ID: {fallback_did}", "warning") - return fallback_did - - def _default_user_agent(self) -> str: - """获取当前会话 User-Agent（用于 Sentinel token 生成）。""" - try: - if self.session: - ua = str(self.session.headers.get("User-Agent") or "").strip() - if ua: - return ua - except Exception: - pass - return ( - "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " - "AppleWebKit/537.36 (KHTML, like Gecko) " - "Chrome/145.0.0.0 Safari/537.36" - ) - - def _build_json_headers( - self, - *, - referer: str, - include_device_id: bool = False, - did: Optional[str] = None, - include_datadog: bool = True, - ) -> Dict[str, str]: - """构造 OpenAI Auth JSON 请求头。""" - headers = { - "referer": str(referer or "https://auth.openai.com/").strip(), - "accept": "application/json", - "content-type": "application/json", - "origin": "https://auth.openai.com", - "accept-language": "en-US,en;q=0.9", - "user-agent": self._default_user_agent(), - "sec-fetch-dest": "empty", - "sec-fetch-mode": "cors", - "sec-fetch-site": "same-origin", - } - if include_device_id: - resolved_did = str(did or self.device_id or "").strip() - if resolved_did: - headers["oai-device-id"] = resolved_did - if include_datadog: - headers.update(generate_datadog_trace()) - return headers - - def _build_navigation_headers(self, *, referer: str) -> Dict[str, str]: - """构造页面跳转请求头。""" - return { - "accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", - "accept-language": "en-US,en;q=0.9", - "referer": str(referer or "https://auth.openai.com/").strip(), - "user-agent": self._default_user_agent(), - "sec-fetch-dest": "document", - "sec-fetch-mode": "navigate", - "sec-fetch-site": "same-origin", - "sec-fetch-user": "?1", - "upgrade-insecure-requests": "1", - } - - def _check_sentinel(self, did: str, *, flow: str = "authorize_continue") -> Optional[str]: - """为指定 flow 获取完整的 openai-sentinel-token。""" - resolved_did = str(did or "").strip() - if not resolved_did: - return None - - try: - if not self.session: - self.session = self.http_client.session - except Exception: - pass - - page_url_map = { - "authorize_continue": "https://auth.openai.com/create-account", - "username_password_create": "https://auth.openai.com/create-account/password", - "password_verify": "https://auth.openai.com/log-in/password", - "email_otp_validate": "https://auth.openai.com/email-verification", - "oauth_create_account": "https://auth.openai.com/about-you", - } - prefer_browser = flow in {"username_password_create", "oauth_create_account"} - - if prefer_browser: - try: - browser_token = get_sentinel_token_via_browser( - flow=flow, - proxy=self.proxy_url, - page_url=page_url_map.get(flow), - headless=self.anyauto_browser_mode != "headed", - device_id=resolved_did, - log_fn=lambda msg: self._log(msg), - ) - if browser_token: - self._log(f"Sentinel Browser token 获取成功 ({flow})") - return browser_token - except Exception as e: - self._log(f"Sentinel Browser token 异常 ({flow}): {e}", "warning") - - try: - if self.session: - token = build_sentinel_token( - self.session, - resolved_did, - flow=flow, - user_agent=self._default_user_agent(), - ) - if token: - self._log(f"Sentinel token 获取成功 ({flow})") - return token - except Exception as e: - self._log(f"Sentinel token 生成异常 ({flow}): {e}", "warning") - - try: - fallback = self.http_client.check_sentinel(resolved_did) - fallback_text = str(fallback or "").strip() - if not fallback_text: - self._log(f"Sentinel 检查失败：未获取到 token ({flow})", "warning") - return None - try: - parsed = json.loads(fallback_text) - if isinstance(parsed, dict) and parsed.get("c"): - self._log(f"Sentinel fallback 获取成功 ({flow})") - return fallback_text - except Exception: - pass - self._log(f"Sentinel fallback 不是完整 token，已忽略 ({flow})", "warning") - return None - except Exception as e: - self._log(f"Sentinel 检查异常 ({flow}): {e}", "warning") - return None - - def _submit_auth_start( - self, - did: str, - sen_token: Optional[str], - *, - screen_hint: str, - referer: str, - log_label: str, - record_existing_account: bool = True, - ) -> SignupFormResult: - """ - 提交授权入口表单 - - Returns: - SignupFormResult: 提交结果，包含账号状态判断 - """ - max_attempts = 3 - current_did = str(did or "").strip() - current_sen_token = str(sen_token or "").strip() if sen_token else None - for attempt in range(1, max_attempts + 1): - try: - request_body = json.dumps({ - "username": { - "value": self.email, - "kind": "email", - }, - "screen_hint": screen_hint, - }) - - headers = self._build_json_headers( - referer=referer, - include_device_id=True, - did=current_did, - ) - if current_sen_token: - headers["openai-sentinel-token"] = current_sen_token - - response = self.session.post( - OPENAI_API_ENDPOINTS["signup"], - headers=headers, - data=request_body, - ) - - self._log(f"{log_label}状态: {response.status_code}") - - if response.status_code == 429 and attempt < max_attempts: - wait_seconds = min(18, 5 * attempt) - self._log( - f"{log_label}命中限流 429（第 {attempt}/{max_attempts} 次），{wait_seconds}s 后自动重试...", - "warning", - ) - time.sleep(wait_seconds) - continue - - # 部分网络/会话边界情况下会返回 409，做自愈重试而非直接失败。 - if response.status_code == 409 and attempt < max_attempts: - wait_seconds = min(10, 2 * attempt) - self._log( - f"{log_label}命中 409（第 {attempt}/{max_attempts} 次），" - f"会话上下文可能冲突，{wait_seconds}s 后自动重试...", - "warning", - ) - # 尝试刷新 sentinel，避免 token 过期导致冲突。 - try: - refreshed = self._check_sentinel(current_did) - if refreshed: - current_sen_token = refreshed - except Exception: - pass - # 预热一次授权页，帮助服务端重建登录上下文。 - try: - if self.oauth_start and getattr(self.oauth_start, "auth_url", None): - self.session.get(str(self.oauth_start.auth_url), timeout=12) - except Exception: - pass - time.sleep(wait_seconds) - continue - - if response.status_code != 200: - return SignupFormResult( - success=False, - error_message=f"HTTP {response.status_code}: {response.text[:200]}" - ) - - # 解析响应判断账号状态 - try: - response_data = response.json() - page_type = response_data.get("page", {}).get("type", "") - self._log(f"响应页面类型: {page_type}") - - is_existing = page_type == OPENAI_PAGE_TYPES["EMAIL_OTP_VERIFICATION"] - - if is_existing: - self._otp_sent_at = time.time() - if record_existing_account: - self._log(f"检测到已注册账号，将自动切换到登录流程") - self._is_existing_account = True - else: - self._log("登录流程已触发，等待系统自动发送的验证码") - - return SignupFormResult( - success=True, - page_type=page_type, - is_existing_account=is_existing, - response_data=response_data - ) - - except Exception as parse_error: - self._log(f"解析响应失败: {parse_error}", "warning") - # 无法解析，默认成功 - return SignupFormResult(success=True) - - except Exception as e: - if attempt < max_attempts: - self._log( - f"{log_label}异常（第 {attempt}/{max_attempts} 次）: {e}，准备重试...", - "warning", - ) - time.sleep(2 * attempt) - continue - self._log(f"{log_label}失败: {e}", "error") - return SignupFormResult(success=False, error_message=str(e)) - - return SignupFormResult(success=False, error_message=f"{log_label}失败: 超过最大重试次数") - - def _submit_signup_form( - self, - did: str, - sen_token: Optional[str], - *, - record_existing_account: bool = True, - ) -> SignupFormResult: - """提交注册入口表单。""" - return self._submit_auth_start( - did, - sen_token, - screen_hint="signup", - referer="https://auth.openai.com/create-account", - log_label="提交注册表单", - record_existing_account=record_existing_account, - ) - - def _submit_login_start(self, did: str, sen_token: Optional[str]) -> SignupFormResult: - """提交登录入口表单。""" - return self._submit_auth_start( - did, - sen_token, - screen_hint="login", - referer="https://auth.openai.com/log-in", - log_label="提交登录入口", - record_existing_account=False, - ) - - def _submit_login_password(self) -> SignupFormResult: - """提交登录密码，进入邮箱验证码页面。""" - max_attempts = 3 - password_text = str(self.password or "").strip() - if not password_text and self.email: - try: - with get_db() as db: - account = crud.get_account_by_email(db, self.email) - db_password = str(getattr(account, "password", "") or "").strip() if account else "" - if db_password: - self.password = db_password - password_text = db_password - self._log("登录阶段未发现内存密码，已从账号库回填密码") - except Exception as e: - self._log(f"登录阶段尝试回填密码失败: {e}", "warning") - - if not password_text: - return SignupFormResult( - success=False, - error_message="登录密码为空：该邮箱可能是已存在账号但当前任务未持有密码", - ) - - for attempt in range(1, max_attempts + 1): - try: - did = str(self.device_id or self.session.cookies.get("oai-did") or "").strip() - sen_token = self._check_sentinel(did, flow="password_verify") if did else None - headers = self._build_json_headers( - referer="https://auth.openai.com/log-in/password", - include_device_id=True, - did=did, - ) - if sen_token: - headers["openai-sentinel-token"] = sen_token - response = self.session.post( - OPENAI_API_ENDPOINTS["password_verify"], - headers=headers, - data=json.dumps({"password": self.password}), - ) - - self._log(f"提交登录密码状态: {response.status_code}") - - if response.status_code == 429 and attempt < max_attempts: - wait_seconds = min(18, 5 * attempt) - self._log( - f"提交登录密码命中限流 429（第 {attempt}/{max_attempts} 次），{wait_seconds}s 后自动重试...", - "warning", - ) - time.sleep(wait_seconds) - continue - - if response.status_code == 401 and attempt < max_attempts: - body = str(response.text or "") - if "invalid_username_or_password" in body: - wait_seconds = min(12, 3 * attempt) - self._log( - f"提交登录密码命中 401（第 {attempt}/{max_attempts} 次），" - f"疑似密码尚未生效或历史账号密码不一致，{wait_seconds}s 后自动重试...", - "warning", - ) - time.sleep(wait_seconds) - continue - - if response.status_code != 200: - return SignupFormResult( - success=False, - error_message=f"HTTP {response.status_code}: {response.text[:200]}" - ) - - response_data = response.json() - page_type = response_data.get("page", {}).get("type", "") - self._log(f"登录密码响应页面类型: {page_type}") - - is_existing = page_type == OPENAI_PAGE_TYPES["EMAIL_OTP_VERIFICATION"] - if is_existing: - self._otp_sent_at = time.time() - self._log("登录密码校验通过，等待系统自动发送的验证码") - - return SignupFormResult( - success=True, - page_type=page_type, - is_existing_account=is_existing, - response_data=response_data, - ) - - except Exception as e: - if attempt < max_attempts: - self._log( - f"提交登录密码异常（第 {attempt}/{max_attempts} 次）: {e}，准备重试...", - "warning", - ) - time.sleep(2 * attempt) - continue - self._log(f"提交登录密码失败: {e}", "error") - return SignupFormResult(success=False, error_message=str(e)) - - return SignupFormResult(success=False, error_message="提交登录密码失败: 超过最大重试次数") - - def _reset_auth_flow(self) -> None: - """重置会话，准备重新发起 OAuth 流程。""" - self.http_client.close() - self.session = None - self.oauth_start = None - self.session_token = None - self._otp_sent_at = None - - def _prepare_authorize_flow(self, label: str) -> Tuple[Optional[str], Optional[str]]: - """初始化当前阶段的授权流程，返回 device id 和 sentinel token。""" - self._log(f"{label}: 先把会话热热身...") - if not self._init_session(): - return None, None - - self._log(f"{label}: OAuth 流程准备开跑，系好鞋带...") - if not self._start_oauth(): - return None, None - - self._log(f"{label}: 领取 Device ID 通行证...") - did = str(self._get_device_id() or "").strip() - if not did: - return None, None - - self.device_id = did - - self._log(f"{label}: 解一道 Sentinel POW 小题，答对才给进...") - sen_token = self._check_sentinel(did) - if not sen_token: - return did, None - - self._log(f"{label}: Sentinel 点头放行，继续前进") - return did, sen_token - - @staticmethod - def _extract_session_token_from_cookie_text(cookie_text: str) -> str: - """从 Cookie 文本中提取 next-auth session token（兼容分片）。""" - text = str(cookie_text or "") - if not text: - return "" - - direct = re.search(r"(?:^|[;,]\s*)(?:__|_)Secure-next-auth\.session-token=([^;,]*)", text) - if direct: - direct_val = str(direct.group(1) or "").strip().strip('"').strip("'") - if direct_val: - return direct_val - - parts = re.findall(r"(?:__|_)Secure-next-auth\.session-token\.(\d+)=([^;,]*)", text) - if not parts: - return "" - - chunk_map = {} - for idx, value in parts: - try: - clean_value = str(value or "").strip().strip('"').strip("'") - if clean_value: - chunk_map[int(idx)] = clean_value - except Exception: - continue - if not chunk_map: - return "" - return "".join(chunk_map[i] for i in sorted(chunk_map.keys())) - - def _warmup_chatgpt_session(self) -> None: - """ - 仅预热 chatgpt 首页，避免提前消费一次性 continue_url。 - """ - try: - self.session.get( - "https://chatgpt.com/", - headers={ - "accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", - "referer": "https://auth.openai.com/", - "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" - ), - }, - timeout=20, - ) - except Exception as e: - self._log(f"chatgpt 首页预热异常: {e}", "warning") - - def _capture_auth_session_tokens(self, result: RegistrationResult, access_hint: Optional[str] = None) -> bool: - """ - 直接通过 /api/auth/session 捕获 session_token + access_token。 - 这是 ABCard Phase 1 的关键路径。 - """ - access_token = str(access_hint or "").strip() - set_cookie_text = "" - request_cookie_text = "" - try: - headers = { - "accept": "application/json", - "referer": "https://chatgpt.com/", - "origin": "https://chatgpt.com", - "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" - ), - "cache-control": "no-cache", - "pragma": "no-cache", - } - if access_token: - headers["authorization"] = f"Bearer {access_token}" - response = self.session.get( - "https://chatgpt.com/api/auth/session", - headers=headers, - timeout=20, - ) - set_cookie_text = self._flatten_set_cookie_headers(response) - request_cookie_text = self._extract_request_cookie_header(response) - if response.status_code == 200: - try: - data = response.json() or {} - access_from_json = str(data.get("accessToken") or "").strip() - if access_from_json: - access_token = access_from_json - except Exception: - pass - else: - self._log(f"/api/auth/session 返回异常状态: {response.status_code}", "warning") - except Exception as e: - self._log(f"获取 auth/session 失败: {e}", "warning") - - # 1) 直接从 cookie jar 拿 - session_token = self._extract_session_token_from_cookie_jar(self.session.cookies) - - # 2) 从完整 cookies 文本兜底（含分片） - if not session_token: - session_token = self._extract_session_token_from_cookie_text(self._dump_session_cookies()) - - # 3) 从 set-cookie 兜底（含分片） - if not session_token and set_cookie_text: - session_token = self._extract_session_token_from_cookie_text(set_cookie_text) - - # 4) 从请求 Cookie 头兜底（对齐 F12 Network 观测） - if not session_token and request_cookie_text: - session_token = self._extract_session_token_from_cookie_text(request_cookie_text) - - # 兜底：已有 access_token 但无 session_token 时，带 Bearer 再请求一次 auth/session - if (not session_token) and access_token: - try: - retry_response = self.session.get( - "https://chatgpt.com/api/auth/session", - headers={ - "accept": "application/json", - "referer": "https://chatgpt.com/", - "origin": "https://chatgpt.com", - "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" - ), - "authorization": f"Bearer {access_token}", - "cache-control": "no-cache", - "pragma": "no-cache", - }, - timeout=20, - ) - retry_set_cookie = self._flatten_set_cookie_headers(retry_response) - retry_request_cookie = self._extract_request_cookie_header(retry_response) - if not session_token: - session_token = self._extract_session_token_from_cookie_jar(self.session.cookies) - if not session_token: - session_token = self._extract_session_token_from_cookie_text(self._dump_session_cookies()) - if not session_token and retry_set_cookie: - session_token = self._extract_session_token_from_cookie_text(retry_set_cookie) - if not session_token and retry_request_cookie: - session_token = self._extract_session_token_from_cookie_text(retry_request_cookie) - except Exception as e: - self._log(f"Bearer 兜底换 session_token 失败: {e}", "warning") - - if not session_token: - cookies_text = self._dump_session_cookies() - raw_direct_match = re.search( - r"(?:^|[;,]\s*)(?:__|_)Secure-next-auth\.session-token=([^;,]*)", - cookies_text, - ) - raw_direct_len = len(str(raw_direct_match.group(1) or "").strip()) if raw_direct_match else 0 - chunk_count = len(re.findall(r"(?:__|_)Secure-next-auth\.session-token\.(\d+)=", cookies_text)) - req_cookie_len = len(str(request_cookie_text or "").strip()) - self._log( - f"auth/session 仍未命中 session_token（raw_direct_len={raw_direct_len}, chunks={chunk_count}, req_cookie_len={req_cookie_len}）", - "warning", - ) - - # 设备 ID 同步 - did = "" - try: - did = str(self.session.cookies.get("oai-did") or "").strip() - except Exception: - did = "" - if did: - self.device_id = did - result.device_id = did - - if session_token: - self.session_token = session_token - result.session_token = session_token - if access_token: - result.access_token = access_token - - self._log( - "Auth Session 捕获结果: session_token=" - + ("有" if bool(result.session_token) else "无") - + ", access_token=" - + ("有" if bool(result.access_token) else "无") - ) - return bool(result.session_token and result.access_token) - - def _bootstrap_chatgpt_signin_for_session(self, result: RegistrationResult) -> bool: - """ - 对齐 ABCard 的补会话路径： - csrf -> signin/openai -> 跟随跳转 -> auth/session，目标是拿到 session_token。 - """ - self._log("Session Token 还没就位，尝试 ABCard 同款会话桥接...") - self._warmup_chatgpt_session() - csrf_token = "" - auth_url = "" - try: - csrf_resp = self.session.get( - "https://chatgpt.com/api/auth/csrf", - headers={ - "accept": "application/json", - "referer": "https://chatgpt.com/auth/login", - "origin": "https://chatgpt.com", - "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" - ), - }, - timeout=20, - ) - if csrf_resp.status_code == 200: - csrf_token = str((csrf_resp.json() or {}).get("csrfToken") or "").strip() - else: - self._log(f"csrf 获取失败: HTTP {csrf_resp.status_code}", "warning") - except Exception as e: - self._log(f"csrf 获取异常: {e}", "warning") - - if not csrf_token: - self._log("csrf token 为空，跳过会话桥接", "warning") - return False - - try: - signin_resp = self.session.post( - "https://chatgpt.com/api/auth/signin/openai", - headers={ - "accept": "application/json", - "content-type": "application/x-www-form-urlencoded", - "origin": "https://chatgpt.com", - "referer": "https://chatgpt.com/auth/login", - "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" - ), - }, - data={ - "csrfToken": csrf_token, - "callbackUrl": "https://chatgpt.com/", - "json": "true", - }, - timeout=20, - ) - if signin_resp.status_code == 200: - auth_url = str((signin_resp.json() or {}).get("url") or "").strip() - else: - self._log(f"signin/openai 失败: HTTP {signin_resp.status_code}", "warning") - except Exception as e: - self._log(f"signin/openai 异常: {e}", "warning") - - if not auth_url: - self._log("signin/openai 未返回 auth_url，跳过会话桥接", "warning") - return False - - callback_url = "" - final_url = auth_url - try: - callback_url, final_url = self._follow_chatgpt_auth_redirects(auth_url) - except Exception as e: - self._log(f"会话桥接重定向跟踪异常: {e}", "warning") - callback_url = "" - final_url = auth_url - - # 若已拿到 callback，补打一跳确保 next-auth callback 被完整执行。 - if callback_url and "error=" not in callback_url: - try: - self.session.get( - callback_url, - headers={ - "accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", - "referer": "https://chatgpt.com/auth/login", - "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" - ), - }, - allow_redirects=True, - timeout=25, - ) - except Exception as e: - self._log(f"会话桥接 callback 补跳异常: {e}", "warning") - elif callback_url and "error=" in callback_url: - self._log(f"会话桥接回调返回错误参数: {callback_url[:140]}...", "warning") - else: - self._log(f"会话桥接未命中 callback，final_url={final_url[:120]}...", "warning") - # 命中 auth.openai 登录页时，尝试自动登录补会话（对齐 ABCard 的登录态建立思路）。 - if "auth.openai.com/log-in" in str(final_url or "").lower(): - self._log("会话桥接进入登录页，尝试自动登录后继续抓取 session_token...") - if self._bridge_login_for_session_token(result): - return True - - self._warmup_chatgpt_session() - cookie_text = self._dump_session_cookies() - direct_token = self._extract_session_token_from_cookie_text(cookie_text) - has_direct = bool(direct_token) - chunk_count = len(re.findall(r"(?:__|_)Secure-next-auth\.session-token\.(\d+)=", cookie_text)) - if direct_token and not result.session_token: - self.session_token = direct_token - result.session_token = direct_token - self._log(f"会话桥接已缓存 session_token（len={len(direct_token)}）") - self._log( - f"会话桥接后 cookie 概览: direct={'有' if has_direct else '无'}, chunks={chunk_count}" - ) - return self._capture_auth_session_tokens(result, access_hint=result.access_token) - - def _bridge_login_for_session_token(self, result: RegistrationResult) -> bool: - """ - 当 chatgpt signin/openai 跳回 auth.openai 登录页时，自动补一次登录流程： - login -> password -> email otp -> workspace -> auth/session。 - """ - try: - if not self.email or not self.password: - self._log("会话桥接自动登录缺少邮箱或密码，无法继续", "warning") - return False - - did = "" - try: - did = str(self.session.cookies.get("oai-did") or "").strip() - except Exception: - did = "" - if not did: - did = str(uuid.uuid4()) - try: - self.session.cookies.set("oai-did", did, domain=".chatgpt.com", path="/") - except Exception: - pass - self.device_id = did - result.device_id = result.device_id or did - - sen_token = self._check_sentinel(did) - login_start_result = self._submit_login_start(did, sen_token) - if not login_start_result.success: - self._log( - f"会话桥接自动登录入口失败: {login_start_result.error_message}", - "warning", - ) - return False - page_type = str(login_start_result.page_type or "").strip() - if page_type == OPENAI_PAGE_TYPES["EMAIL_OTP_VERIFICATION"]: - self._log("会话桥接自动登录已直达邮箱验证码页，跳过密码提交") - elif page_type == OPENAI_PAGE_TYPES["LOGIN_PASSWORD"]: - password_result = self._submit_login_password() - if not password_result.success: - self._log( - f"会话桥接自动登录提交密码失败: {password_result.error_message}", - "warning", - ) - return False - if not password_result.is_existing_account: - self._log( - f"会话桥接自动登录未进入邮箱验证码页: {password_result.page_type or 'unknown'}", - "warning", - ) - return False - else: - self._log( - f"会话桥接自动登录入口返回未知页面: {page_type or 'unknown'}", - "warning", - ) - return False - - if not self._verify_email_otp_with_retry(stage_label="会话桥接登录验证码", max_attempts=3): - self._log("会话桥接自动登录验证码校验失败", "warning") - return False - - # OTP 成功后先直接抓一次 auth/session，避免无谓依赖 workspace 流程。 - self._warmup_chatgpt_session() - if self._capture_auth_session_tokens(result, access_hint=result.access_token): - self._log("会话桥接自动登录在 OTP 后已命中 session_token") - return True - - workspace_id = self._get_workspace_id() - if not workspace_id: - workspace_id = str(result.workspace_id or "").strip() - if workspace_id: - self._log(f"会话桥接自动登录复用已知 workspace_id: {workspace_id}") - if not workspace_id: - self._log("会话桥接自动登录未获取到 workspace_id", "warning") - return False - result.workspace_id = workspace_id - - continue_url = self._select_workspace(workspace_id) - if not continue_url: - cached_continue = str(self._create_account_continue_url or "").strip() - if cached_continue: - continue_url = cached_continue - self._log("会话桥接自动登录未获取到 continue_url，改用 create_account 缓存 continue_url", "warning") - else: - self._log("会话桥接自动登录未获取到 continue_url", "warning") - return False - - callback_url, final_url = self._follow_redirects(continue_url) - self._log( - f"会话桥接自动登录重定向完成: callback={'有' if callback_url else '无'}, final={str(final_url or '')[:100]}..." - ) - - self._warmup_chatgpt_session() - return self._capture_auth_session_tokens(result, access_hint=result.access_token) - except Exception as e: - self._log(f"会话桥接自动登录异常: {e}", "warning") - return False - - def _follow_chatgpt_auth_redirects(self, start_url: str) -> Tuple[str, str]: - """ - 对齐 ABCard 的 next-auth 重定向跟踪： - - 手动跟踪 30x - - 识别 /api/auth/callback/openai - Returns: - (callback_url, final_url) - """ - import urllib.parse - - current_url = str(start_url or "").strip() - callback_url = "" - bridged_header_token = "" - if not current_url: - return "", "" - - max_redirects = 12 - ua = ( - "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " - "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36" - ) - for i in range(max_redirects): - self._log(f"会话桥接重定向 {i+1}/{max_redirects}: {current_url[:120]}...") - if "/api/auth/callback/openai" in current_url and not callback_url: - callback_url = current_url - - resp = self.session.get( - current_url, - headers={ - "accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", - "referer": "https://chatgpt.com/", - "user-agent": ua, - }, - timeout=25, - allow_redirects=False, - ) - - # 直接从每一跳响应头 Set-Cookie 抓 session_token（对齐 F12 Network 视角） - set_cookie_text = self._flatten_set_cookie_headers(resp) - token_from_header = self._extract_session_token_from_cookie_text(set_cookie_text) - if token_from_header: - bridged_header_token = token_from_header - # 同时写入两种命名兼容，避免库在不同平台下键名差异。 - for name in ("__Secure-next-auth.session-token", "_Secure-next-auth.session-token"): - for domain in (".chatgpt.com", "chatgpt.com"): - try: - self.session.cookies.set(name, token_from_header, domain=domain, path="/") - except Exception: - continue - self._log( - f"会话桥接命中 Set-Cookie session_token（len={len(token_from_header)}）" - ) - - if resp.status_code not in (301, 302, 303, 307, 308): - break - - location = str(resp.headers.get("Location") or "").strip() - if not location: - break - current_url = urllib.parse.urljoin(current_url, location) - - if callback_url and not str(current_url or "").startswith("https://chatgpt.com/"): - try: - self.session.get( - "https://chatgpt.com/", - headers={ - "accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", - "referer": current_url, - "user-agent": ua, - }, - timeout=20, - ) - except Exception: - pass - - self._log( - f"会话桥接重定向结束: callback={'有' if callback_url else '无'}, " - f"set_cookie_token={'有' if bool(bridged_header_token) else '无'}, final={current_url[:120]}..." - ) - return callback_url, current_url - - def _complete_token_exchange(self, result: RegistrationResult, require_login_otp: bool = True) -> bool: - """在登录态已建立后，补齐 session/access，并尽量获取 OAuth token。""" - if require_login_otp: - self._log("等待登录验证码到场，最后这位嘉宾还在路上...") - self._log("核对登录验证码，验明正身一下...") - if not self._verify_email_otp_with_retry(stage_label="登录验证码", max_attempts=3): - result.error_message = "验证码校验失败" - return False - else: - self._log("ABCard 入口链路：跳过二次登录验证码，直接进入 workspace + redirect + auth/session 抓取") - - self._log("摸一下 Workspace ID，看看该坐哪桌...") - workspace_id = self._get_workspace_id() - continue_url = "" - if workspace_id: - result.workspace_id = workspace_id - - self._log("选择 Workspace，安排个靠谱座位...") - continue_url = self._select_workspace(workspace_id) - if not continue_url: - cached_continue = str(self._create_account_continue_url or "").strip() - if cached_continue: - continue_url = cached_continue - self._log("workspace/select 未返回 continue_url，改用 create_account 缓存 continue_url", "warning") - else: - result.error_message = "选择 Workspace 失败" - return False - else: - cached_continue = str(self._create_account_continue_url or "").strip() - if cached_continue: - continue_url = cached_continue - self._log("未获取到 Workspace ID，改用 create_account 缓存 continue_url 继续链路", "warning") - else: - result.error_message = "获取 Workspace ID 失败" - return False - - self._log("顺着重定向面包屑往前走，别跟丢了...") - callback_url, final_url = self._follow_redirects(continue_url) - self._log( - f"重定向链完成，callback={'有' if callback_url else '无'}，final={final_url[:100]}..." - ) - self._log("重定向链结束，直接请求 /api/auth/session 抓取 session/access...") - captured = self._capture_auth_session_tokens(result, access_hint=result.access_token) - if not captured: - self._log("直抓未命中，补一次 chatgpt 预热后再抓取...", "warning") - self._warmup_chatgpt_session() - captured = self._capture_auth_session_tokens(result, access_hint=result.access_token) - final_url_lower = str(final_url or "").lower() - add_phone_gate = ("auth.openai.com/add-phone" in final_url_lower) - - # ABCard 入口常见失败点：被 add-phone 风控页截断，导致拿不到 callback/session。 - if add_phone_gate and (not callback_url) and (not captured): - self._log("检测到 auth.openai.com/add-phone 风控页，当前链路未完成 OAuth 回调", "warning") - if (not require_login_otp) and (not self._is_existing_account): - self._log("ABCard 入口命中 add-phone，回退原生重登链路再试一次...", "warning") - login_ready, login_error = self._restart_login_flow() - if not login_ready: - result.error_message = f"ABCard 回退原生链路失败: {login_error}" - return False - return self._complete_token_exchange(result, require_login_otp=True) - result.error_message = "命中 add-phone 风控页，未获取到 session_token" - return False - - callback_has_error = bool( - callback_url and ("error=" in callback_url) and ("code=" not in callback_url) - ) - if callback_url: - if callback_has_error: - self._log(f"回调返回错误参数，跳过 OAuth 回调: {callback_url[:140]}...", "warning") - if not captured: - result.error_message = "OAuth 回调返回 access_denied，且未获取到 auth/session" - return False - else: - self._log("处理 OAuth 回调，准备把 token 请出来...") - token_info = self._handle_oauth_callback(callback_url) - if token_info: - result.account_id = token_info.get("account_id", "") - result.access_token = token_info.get("access_token", "") or result.access_token - result.refresh_token = token_info.get("refresh_token", "") - result.id_token = token_info.get("id_token", "") - elif captured: - self._log("OAuth 回调失败，但 session/access 已拿到，继续后续流程", "warning") - else: - result.error_message = "处理 OAuth 回调失败" - return False - else: - if captured: - self._log("未拿到 callback_url，但 session/access 已拿到，继续后续流程", "warning") - else: - result.error_message = "跟随重定向链失败" - return False - - result.password = self.password or "" - result.source = "login" if self._is_existing_account else "register" - result.device_id = result.device_id or str(self.device_id or "") - - session_cookie = self.session.cookies.get("__Secure-next-auth.session-token") - if session_cookie: - self.session_token = session_cookie - result.session_token = session_cookie - self._log("Session Token 也捞到了，今天这网没白连") - - if not result.access_token or not result.session_token: - # 再捞一次，避免某些链路里 session 建立稍慢 - self._capture_auth_session_tokens(result, access_hint=result.access_token) - if not result.session_token: - # 对齐 ABCard：尝试走 csrf + signin/openai 的会话桥接。 - self._bootstrap_chatgpt_signin_for_session(result) - if not result.session_token: - result.session_token = self._extract_session_token_from_cookie_text(self._dump_session_cookies()) - if not result.device_id: - result.device_id = str(self.device_id or self.session.cookies.get("oai-did") or "") - - if not result.access_token: - result.error_message = "未获取到 access_token" - return False - if not result.session_token: - native_register_flow = (self.registration_entry_flow == "native") and (not self._is_existing_account) - if native_register_flow: - # 对齐 K:\1\2 备份：原生注册流程里 session_token 不做阻断。 - self._log( - "当前链路未拿到 session_token，先保存账号并标记待补会话（可在账号详情/支付页一键补全）", - "warning", - ) - else: - # 非原生注册入口仍保持强制，避免后续流程不可用。 - if not self._ensure_session_token_strict(result, max_rounds=2): - result.error_message = "未获取到 session_token（强制要求）" - self._log( - "强制模式未拿到 session_token，本次注册判定失败，请检查网络/代理与登录回调链路", - "error", - ) - return False - - return True - - def _complete_token_exchange_native_backup(self, result: RegistrationResult) -> bool: - """ - 原生入口对齐备份版收尾链路： - 登录验证码 -> Workspace -> redirect -> OAuth callback -> token 入袋。 - """ - def _is_registration_gate_url(url: str) -> bool: - u = str(url or "").strip().lower() - if not u: - return False - return ("auth.openai.com/about-you" in u) or ("auth.openai.com/add-phone" in u) - - self._log("等待登录验证码到场，最后这位嘉宾还在路上...") - self._log("核对登录验证码，验明正身一下...") - login_otp_tried_codes: set[str] = set() - login_otp_ok = self._verify_email_otp_with_retry( - stage_label="登录验证码", - max_attempts=1, - fetch_timeout=120, - attempted_codes=login_otp_tried_codes, - ) - if not login_otp_ok: - self._log("登录验证码首轮未命中，尝试在当前会话原地重发 OTP 后再校验...", "warning") - resent = self._send_verification_code(referer="https://auth.openai.com/email-verification") - if resent: - login_otp_ok = self._verify_email_otp_with_retry( - stage_label="登录验证码(原地重发)", - max_attempts=2, - fetch_timeout=120, - attempted_codes=login_otp_tried_codes, - ) - - if not login_otp_ok: - self._log("登录验证码仍未命中，尝试重触发登录 OTP 后再校验...", "warning") - if not self._retrigger_login_otp(): - self._log("重触发登录 OTP 失败，尝试完整重登链路后再校验一次...", "warning") - login_ready, login_error = self._restart_login_flow() - if not login_ready: - result.error_message = f"登录验证码重触发失败，且完整重登失败: {login_error}" - return False - login_otp_ok = self._verify_email_otp_with_retry( - stage_label="登录验证码(重发)", - max_attempts=3, - fetch_timeout=120, - attempted_codes=login_otp_tried_codes, - ) - if not login_otp_ok: - result.error_message = "验证码校验失败" - return False - - self._log("摸一下 Workspace ID，看看该坐哪桌...") - workspace_id = str(self._last_validate_otp_workspace_id or "").strip() - if workspace_id: - self._log(f"使用 OTP 返回的 Workspace ID: {workspace_id}") - if not workspace_id: - workspace_id = str(self._get_workspace_id() or "").strip() - if workspace_id: - result.workspace_id = workspace_id - - continue_url = "" - otp_continue = str(self._last_validate_otp_continue_url or "").strip() - if otp_continue and _is_registration_gate_url(otp_continue): - self._log("OTP 返回 continue_url 指向注册门页（about-you/add-phone），本轮收尾忽略该地址", "warning") - otp_continue = "" - - cached_continue = str(self._create_account_continue_url or "").strip() - if cached_continue and _is_registration_gate_url(cached_continue): - self._log("create_account 缓存 continue_url 指向注册门页（about-you/add-phone），本轮收尾忽略该地址", "warning") - cached_continue = "" - - if workspace_id: - self._log("选择 Workspace，安排个靠谱座位...") - continue_url = str(self._select_workspace(workspace_id) or "").strip() - if not continue_url: - self._log("workspace/select 未返回 continue_url，尝试 OAuth authorize 兜底", "warning") - - if not continue_url and otp_continue: - continue_url = otp_continue - self._log("使用 OTP 返回 continue_url 继续授权链路", "warning") - - if not continue_url and cached_continue: - continue_url = cached_continue - self._log("使用 create_account 缓存 continue_url 作为兜底", "warning") - - if not continue_url: - oauth_start_url = str( - ( - getattr(self.oauth_start, "auth_url", "") - or getattr(self.oauth_start, "url", "") - if self.oauth_start - else "" - ) - or "" - ).strip() - if oauth_start_url: - continue_url = oauth_start_url - self._log("使用 OAuth authorize URL 作为最后兜底 continue_url", "warning") - - if not continue_url: - result.error_message = "获取 continue_url 失败" - return False - - self._log("顺着重定向面包屑往前走，别跟丢了...") - callback_url, _final_url = self._follow_redirects(continue_url) - if not callback_url: - self._log("未命中 OAuth 回调，尝试 auth/session 兜底抓取 token...", "warning") - self._capture_auth_session_tokens(result, access_hint=result.access_token) - if not result.account_id: - result.account_id = str(self._create_account_account_id or "").strip() - if not result.workspace_id: - result.workspace_id = str(workspace_id or self._create_account_workspace_id or "").strip() - if not result.refresh_token: - result.refresh_token = str(self._create_account_refresh_token or "").strip() - if result.access_token: - result.password = self.password or "" - result.source = "login" if self._is_existing_account else "register" - result.device_id = result.device_id or str(self.device_id or "") - self._log("未命中 callback，已通过 auth/session 兜底拿到 Access Token，继续完成注册", "warning") - return True - - # 对新注册账号放宽：账号已创建成功时允许“注册成功、token 待补” - if (not self._is_existing_account) and self._create_account_account_id: - result.account_id = result.account_id or str(self._create_account_account_id or "").strip() - result.workspace_id = result.workspace_id or str(workspace_id or self._create_account_workspace_id or "").strip() - result.refresh_token = result.refresh_token or str(self._create_account_refresh_token or "").strip() - result.password = self.password or "" - result.source = "register" - result.device_id = result.device_id or str(self.device_id or "") - self._log("回调链路未命中且未抓到 Access Token，但账号已创建成功；按注册成功收尾（token 待后续补齐）", "warning") - return True - - result.error_message = "跟随重定向链失败" - return False - - self._log("处理 OAuth 回调，准备把 token 请出来...") - token_info = self._handle_oauth_callback(callback_url) - if not token_info: - if (not self._is_existing_account) and self._create_account_account_id: - result.account_id = result.account_id or str(self._create_account_account_id or "").strip() - result.workspace_id = result.workspace_id or str(workspace_id or self._create_account_workspace_id or "").strip() - result.refresh_token = result.refresh_token or str(self._create_account_refresh_token or "").strip() - result.password = self.password or "" - result.source = "register" - result.device_id = result.device_id or str(self.device_id or "") - self._log("OAuth 回调处理失败，但账号已创建成功；按注册成功收尾（token 待后续补齐）", "warning") - return True - result.error_message = "处理 OAuth 回调失败" - return False - - result.account_id = token_info.get("account_id", "") - result.access_token = token_info.get("access_token", "") - result.refresh_token = token_info.get("refresh_token", "") - result.id_token = token_info.get("id_token", "") - result.password = self.password or "" - result.source = "login" if self._is_existing_account else "register" - result.device_id = result.device_id or str(self.device_id or "") - - session_cookie = self.session.cookies.get("__Secure-next-auth.session-token") - if session_cookie: - self.session_token = session_cookie - result.session_token = session_cookie - self._log("Session Token 也捞到了，今天这网没白连") - - return True - - def _complete_token_exchange_outlook(self, result: RegistrationResult) -> bool: - """ - Outlook 入口链路（迁移版）： - 对齐 codex-console-main-clean 的收尾流程， - 走「登录 OTP -> Workspace -> OAuth callback」主干，避免 ABCard/native 增强链路干扰。 - 同时补齐“第二封验证码”重试链路，避免 Outlook 轮询卡死。 - """ - self._log("等待登录验证码到场，最后这位嘉宾还在路上...") - self._log("核对登录验证码，验明正身一下...") - login_otp_tried_codes: set[str] = set() - login_otp_ok = self._verify_email_otp_with_retry( - stage_label="登录验证码", - max_attempts=1, - fetch_timeout=90, - attempted_codes=login_otp_tried_codes, - ) - if not login_otp_ok: - self._log("登录验证码首轮未命中，先尝试当前会话原地重发 OTP 后再校验...", "warning") - resent = self._send_verification_code(referer="https://auth.openai.com/email-verification") - if resent: - login_otp_ok = self._verify_email_otp_with_retry( - stage_label="登录验证码(原地重发)", - max_attempts=2, - fetch_timeout=90, - attempted_codes=login_otp_tried_codes, - ) - - if not login_otp_ok: - self._log("登录验证码仍未命中，尝试重触发登录 OTP 后再校验...", "warning") - if not self._retrigger_login_otp(): - self._log("重触发登录 OTP 失败，尝试完整重登链路后再校验一次...", "warning") - login_ready, login_error = self._restart_login_flow() - if not login_ready: - result.error_message = f"登录验证码重触发失败，且完整重登失败: {login_error}" - return False - - login_otp_ok = self._verify_email_otp_with_retry( - stage_label="登录验证码(重发)", - max_attempts=3, - fetch_timeout=120, - attempted_codes=login_otp_tried_codes, - ) - if not login_otp_ok: - result.error_message = "验证码校验失败" - return False - - self._log("摸一下 Workspace ID，看看该坐哪桌...") - workspace_id = str(self._last_validate_otp_workspace_id or "").strip() - if workspace_id: - self._log(f"使用 OTP 返回的 Workspace ID: {workspace_id}") - if not workspace_id: - workspace_id = str(self._get_workspace_id() or "").strip() - if not workspace_id: - workspace_id = str(self._last_validate_otp_workspace_id or self._create_account_workspace_id or "").strip() - if workspace_id: - self._log(f"Workspace ID（缓存）: {workspace_id}", "warning") - - continue_url = "" - if workspace_id: - result.workspace_id = workspace_id - self._log("选择 Workspace，安排个靠谱座位...") - continue_url = str(self._select_workspace(workspace_id) or "").strip() - if not continue_url: - self._log("workspace/select 未返回 continue_url，尝试使用缓存 continue_url", "warning") - else: - self._log("未获取到 Workspace ID，尝试直接使用缓存 continue_url", "warning") - - if not continue_url: - continue_url = str(self._last_validate_otp_continue_url or self._create_account_continue_url or "").strip() - if continue_url: - self._log("使用缓存 continue_url 继续授权链路", "warning") - - if not continue_url: - result.error_message = "获取 Workspace ID 失败" - return False - - self._log("顺着重定向面包屑往前走，别跟丢了...") - callback_url, _final_url = self._follow_redirects(continue_url) - if not callback_url: - result.error_message = "跟随重定向链失败" - return False - - self._log("处理 OAuth 回调，准备把 token 请出来...") - token_info = self._handle_oauth_callback(callback_url) - if not token_info: - result.error_message = "处理 OAuth 回调失败" - return False - - result.account_id = str(token_info.get("account_id") or result.account_id or "").strip() - result.access_token = str(token_info.get("access_token") or result.access_token or "").strip() - result.refresh_token = str(token_info.get("refresh_token") or result.refresh_token or "").strip() - result.id_token = str(token_info.get("id_token") or result.id_token or "").strip() - result.password = self.password or "" - result.source = "login" if self._is_existing_account else "register" - result.device_id = result.device_id or str(self.device_id or "") - - if not result.account_id: - result.account_id = str(self._create_account_account_id or "").strip() - if not result.workspace_id: - result.workspace_id = str(self._create_account_workspace_id or "").strip() - if not result.refresh_token: - result.refresh_token = str(self._create_account_refresh_token or "").strip() - - session_cookie = self.session.cookies.get("__Secure-next-auth.session-token") - if session_cookie: - self.session_token = session_cookie - result.session_token = session_cookie - self._log("Session Token 也捞到了，今天这网没白连") - - if not result.access_token: - result.error_message = "未获取到 access_token" - return False - - return True - - def _ensure_session_token_strict(self, result: RegistrationResult, max_rounds: int = 2) -> bool: - """ - 强制确保 session_token 可用。 - - 先走 auth/session 直抓 - - 再走 ABCard 同款会话桥接 - 连续多轮失败则返回 False。 - """ - if result.session_token: - return True - - rounds = max(int(max_rounds), 1) - for idx in range(rounds): - self._log(f"强制补会话 round {idx + 1}/{rounds}：尝试补抓 session_token ...") - - self._warmup_chatgpt_session() - self._capture_auth_session_tokens(result, access_hint=result.access_token) - if result.session_token: - self._log("强制补会话成功：auth/session 已拿到 session_token") - return True - - self._bootstrap_chatgpt_signin_for_session(result) - if result.session_token: - self._log("强制补会话成功：桥接链路已拿到 session_token") - return True - - fallback_token = self._extract_session_token_from_cookie_text(self._dump_session_cookies()) - if fallback_token: - result.session_token = fallback_token - self.session_token = fallback_token - self._log("强制补会话成功：cookie 文本兜底命中 session_token") - return True - - self._log("强制补会话本轮未命中 session_token", "warning") - - return False - - def _capture_native_core_tokens(self, result: RegistrationResult) -> bool: - """ - 原生注册入口的轻量 token 抓取： - - 不做二次登录 - - 不强依赖 session_token - - 尽量补齐 account/workspace/access/refresh - """ - try: - client_id = str(getattr(self.oauth_manager, "client_id", "") or "").strip() - if client_id: - self._log(f"原生入口 token 抓取: Client ID: {client_id}") - - if (not result.account_id) and self._create_account_account_id: - result.account_id = str(self._create_account_account_id or "").strip() - self._log(f"原生入口 token 抓取: 复用 create_account Account ID: {result.account_id}") - if (not result.refresh_token) and self._create_account_refresh_token: - result.refresh_token = str(self._create_account_refresh_token or "").strip() - self._log("原生入口 token 抓取: 复用 create_account Refresh Token") - - workspace_id = str(result.workspace_id or "").strip() - if not workspace_id: - workspace_id = str(self._create_account_workspace_id or "").strip() - if not workspace_id: - workspace_id = str(self._get_workspace_id() or "").strip() - if workspace_id: - result.workspace_id = workspace_id - self._log(f"原生入口 token 抓取: Workspace ID: {workspace_id}") - else: - self._log("原生入口 token 抓取: 未获取到 Workspace ID", "warning") - - continue_url = "" - if workspace_id: - continue_url = str(self._select_workspace(workspace_id) or "").strip() - if not continue_url: - cached_continue = str(self._create_account_continue_url or "").strip() - if cached_continue: - continue_url = cached_continue - self._log("原生入口 token 抓取: 使用 create_account 缓存 continue_url", "warning") - - callback_url: Optional[str] = None - final_url = "" - if continue_url: - self._log("原生入口 token 抓取: 跟随重定向链获取 OAuth callback...") - callback_url, final_url = self._follow_redirects(continue_url) - self._log( - f"原生入口 token 抓取: 重定向完成，callback={'有' if callback_url else '无'}，final={str(final_url)[:100]}..." - ) - else: - self._log("原生入口 token 抓取: 未获得 continue_url，跳过 callback 交换", "warning") - - callback_has_error = bool( - callback_url and ("error=" in callback_url) and ("code=" not in callback_url) - ) - if callback_url and (not callback_has_error): - token_info = self._handle_oauth_callback(callback_url) - if token_info: - result.account_id = str(token_info.get("account_id") or result.account_id or "").strip() - result.access_token = str(token_info.get("access_token") or result.access_token or "").strip() - result.refresh_token = str(token_info.get("refresh_token") or result.refresh_token or "").strip() - result.id_token = str(token_info.get("id_token") or result.id_token or "").strip() - self._log( - "原生入口 token 抓取结果: " - f"account_id={'有' if bool(result.account_id) else '无'}, " - f"access={'有' if bool(result.access_token) else '无'}, " - f"refresh={'有' if bool(result.refresh_token) else '无'}" - ) - else: - self._log("原生入口 token 抓取: OAuth 回调处理失败", "warning") - elif callback_has_error: - self._log(f"原生入口 token 抓取: callback 含 error，跳过 token 交换: {callback_url[:140]}...", "warning") - else: - self._log("原生入口 token 抓取: 未命中 callback_url", "warning") - - # 不走重登，仅轻量探测 auth/session 里的 accessToken（不依赖 session_token）。 - if not result.access_token: - self._capture_access_token_light(result) - - if (not result.account_id) and result.id_token: - try: - account_info = self.oauth_manager.extract_account_info(result.id_token) - result.account_id = str(account_info.get("account_id") or "").strip() - except Exception: - pass - if (not result.account_id) and result.access_token: - token_acc = self._extract_account_id_from_access_token(result.access_token) - if token_acc: - result.account_id = token_acc - self._log(f"原生入口 token 抓取: 从 access_token 解析 Account ID: {token_acc}") - if not result.workspace_id: - try: - workspace_id_after = str(self._get_workspace_id() or "").strip() - if workspace_id_after: - result.workspace_id = workspace_id_after - self._log(f"原生入口 token 抓取: 二次获取 Workspace ID 成功: {workspace_id_after}") - except Exception: - pass - - missing = [] - if not result.account_id: - missing.append("Account ID") - if not result.workspace_id: - missing.append("Workspace ID") - if not result.access_token: - missing.append("Access Token") - if not result.refresh_token: - missing.append("Refresh Token") - if missing: - self._log(f"原生入口 token 抓取: 未获取字段 -> {', '.join(missing)}", "warning") - - return bool(result.access_token and result.refresh_token) - except Exception as e: - self._log(f"原生入口 token 抓取异常: {e}", "warning") - return False - - def _capture_access_token_light(self, result: RegistrationResult) -> bool: - """轻量从 /api/auth/session 抓 accessToken（不依赖 session_token）。""" - try: - response = self.session.get( - "https://chatgpt.com/api/auth/session", - headers={ - "accept": "application/json", - "referer": "https://chatgpt.com/", - }, - timeout=20, - ) - if response.status_code != 200: - self._log(f"原生入口轻量 auth/session 状态异常: {response.status_code}", "warning") - return False - data = response.json() or {} - access_token = str(data.get("accessToken") or "").strip() - if access_token: - result.access_token = access_token - self._log("原生入口轻量 auth/session 命中 Access Token") - return True - self._log("原生入口轻量 auth/session 未命中 Access Token", "warning") - return False - except Exception as e: - self._log(f"原生入口轻量 auth/session 异常: {e}", "warning") - return False - - def _extract_account_id_from_access_token(self, access_token: str) -> str: - """从 access_token 的 JWT payload 尝试解析 chatgpt_account_id。""" - try: - raw = str(access_token or "").strip() - if raw.count(".") < 2: - return "" - payload = raw.split(".")[1] - import base64 - pad = "=" * ((4 - (len(payload) % 4)) % 4) - decoded = base64.urlsafe_b64decode((payload + pad).encode("ascii")) - claims = json.loads(decoded.decode("utf-8")) - if not isinstance(claims, dict): - return "" - auth_claims = claims.get("https://api.openai.com/auth") or {} - account_id = str( - auth_claims.get("chatgpt_account_id") - or claims.get("chatgpt_account_id") - or "" - ).strip() - return account_id - except Exception: - return "" - - def _ensure_native_required_tokens(self, result: RegistrationResult) -> bool: - """ - 原生注册入口要求拿齐： - Account ID / Workspace ID / Client ID / Access Token / Refresh Token - """ - try: - if (not result.account_id) and result.id_token: - try: - account_info = self.oauth_manager.extract_account_info(result.id_token) - result.account_id = str(account_info.get("account_id") or "").strip() - except Exception: - pass - if (not result.account_id) and result.access_token: - result.account_id = self._extract_account_id_from_access_token(result.access_token) - - if not result.workspace_id: - result.workspace_id = str(self._get_workspace_id() or "").strip() - if (not result.refresh_token) and self._create_account_refresh_token: - result.refresh_token = str(self._create_account_refresh_token or "").strip() - - settings = get_settings() - client_id = str( - getattr(settings, "openai_client_id", "") - or getattr(self.oauth_manager, "client_id", "") - or "" - ).strip() - - missing = [] - if not result.account_id: - missing.append("Account ID") - if not result.workspace_id: - missing.append("Workspace ID") - if not client_id: - missing.append("Client ID") - if not result.access_token: - missing.append("Access Token") - if not result.refresh_token: - missing.append("Refresh Token") - - if missing: - self._log(f"原生入口关键参数缺失: {', '.join(missing)}", "error") - return False - - self._log( - "原生入口关键参数校验通过: " - f"Account ID={result.account_id}, Workspace ID={result.workspace_id}, " - f"Client ID={client_id}, Access=有, Refresh=有" - ) - return True - except Exception as e: - self._log(f"原生入口关键参数校验异常: {e}", "error") - return False - - def _restart_login_flow(self) -> Tuple[bool, str]: - """新注册账号完成建号后，重新发起一次登录流程拿 token。""" - self._token_acquisition_requires_login = True - self._log("注册这边忙完了，再走一趟登录把 token 请出来，收个尾...") - self._reset_auth_flow() - - did, sen_token = self._prepare_authorize_flow("重新登录") - if not did: - return False, "重新登录时获取 Device ID 失败" - if not sen_token: - return False, "重新登录时 Sentinel POW 验证失败" - - login_start_result = self._submit_login_start(did, sen_token) - if not login_start_result.success: - return False, f"重新登录提交邮箱失败: {login_start_result.error_message}" - if login_start_result.page_type != OPENAI_PAGE_TYPES["LOGIN_PASSWORD"]: - return False, f"重新登录未进入密码页面: {login_start_result.page_type or 'unknown'}" - - password_result = self._submit_login_password() - if not password_result.success: - return False, f"重新登录提交密码失败: {password_result.error_message}" - if not password_result.is_existing_account: - return False, f"重新登录未进入验证码页面: {password_result.page_type or 'unknown'}" - return True, "" - - def _retrigger_login_otp(self) -> bool: - """ - 在“登录验证码”阶段重触发 OTP 发送。 - 优先复用登录链路（login_start -> login_password），避免误走注册 OTP 流程。 - """ - try: - did = str(self.device_id or self.session.cookies.get("oai-did") or "").strip() - if not did: - did = str(uuid.uuid4()) - try: - self.session.cookies.set("oai-did", did, domain=".chatgpt.com", path="/") - except Exception: - pass - self.device_id = did - - sen_token = self._check_sentinel(did) - login_start_result = self._submit_login_start(did, sen_token) - if not login_start_result.success: - self._log( - f"重触发登录 OTP 失败：提交登录入口失败: {login_start_result.error_message}", - "warning", - ) - return False - - page_type = str(login_start_result.page_type or "").strip() - if page_type == OPENAI_PAGE_TYPES["EMAIL_OTP_VERIFICATION"]: - self._log("重触发登录 OTP 成功：已直达邮箱验证码页") - return True - - if page_type != OPENAI_PAGE_TYPES["LOGIN_PASSWORD"]: - self._log(f"重触发登录 OTP 失败：未进入密码页（{page_type or 'unknown'}）", "warning") - return False - - password_result = self._submit_login_password() - if not password_result.success: - self._log(f"重触发登录 OTP 失败：提交登录密码失败: {password_result.error_message}", "warning") - return False - if not password_result.is_existing_account: - self._log( - f"重触发登录 OTP 失败：密码后未进入验证码页（{password_result.page_type or 'unknown'}）", - "warning", - ) - return False - - self._log("重触发登录 OTP 成功：已进入邮箱验证码页") - return True - except Exception as e: - self._log(f"重触发登录 OTP 异常: {e}", "warning") - return False - - def _register_password(self, did: Optional[str] = None, sen_token: Optional[str] = None) -> Tuple[bool, Optional[str]]: - """注册密码""" - try: - self._last_register_password_error = None - # 生成密码 - password = self._generate_password() - self.password = password # 保存密码到实例变量 - self._log(f"生成密码: {password}") - - # 提交密码注册 - register_body = json.dumps({ - "password": password, - "username": self.email - }) - - resolved_did = str(did or self.device_id or self.session.cookies.get("oai-did") or "").strip() - register_token = self._check_sentinel(resolved_did, flow="username_password_create") if resolved_did else None - headers = self._build_json_headers( - referer="https://auth.openai.com/create-account/password", - include_device_id=True, - did=resolved_did, - ) - if register_token: - headers["openai-sentinel-token"] = register_token - response = self.session.post( - OPENAI_API_ENDPOINTS["register"], - headers=headers, - data=register_body, - ) - - self._log(f"提交密码状态: {response.status_code}") - - if response.status_code != 200: - error_text = response.text[:500] - self._log(f"密码注册失败: {error_text}", "warning") - - # 解析错误信息，判断是否是邮箱已注册 - try: - error_json = response.json() - error_msg = error_json.get("error", {}).get("message", "") - error_code = error_json.get("error", {}).get("code", "") - normalized_error_msg = str(error_msg or "").strip() - normalized_error_code = str(error_code or "").strip() - - # 检测邮箱已注册的情况 - if "already" in normalized_error_msg.lower() or "exists" in normalized_error_msg.lower() or normalized_error_code == "user_exists": - self._log(f"邮箱 {self.email} 可能已在 OpenAI 注册过", "error") - # 标记此邮箱为已注册状态 - self._mark_email_as_registered() - self._last_register_password_error = "该邮箱可能已在 OpenAI 注册，建议更换邮箱或改走登录流程" - elif "failed to register username" in normalized_error_msg.lower(): - self._last_register_password_error = ( - "OpenAI 拒绝当前邮箱用户名（可能已占用或触发风控），建议更换邮箱后重试" - ) - if did: - self._log("检测到用户名注册失败，尝试登录入口探测邮箱是否已存在...", "warning") - try: - probe = self._submit_login_start(did, sen_token) - if probe.success and probe.page_type in ( - OPENAI_PAGE_TYPES["LOGIN_PASSWORD"], - OPENAI_PAGE_TYPES["EMAIL_OTP_VERIFICATION"], - ): - self._log("登录入口探测命中：该邮箱大概率已是 OpenAI 账号", "warning") - self._mark_email_as_registered() - self._last_register_password_error = ( - "该邮箱已存在 OpenAI 账号。" - "若是刚刚注册中断，请优先使用上一轮任务日志里的“生成密码”走登录续跑；" - "拿不到旧密码再更换邮箱。" - ) - except Exception as probe_error: - self._log(f"登录入口探测失败: {probe_error}", "warning") - else: - self._last_register_password_error = ( - f"注册密码接口返回异常: {normalized_error_msg or f'HTTP {response.status_code}'}" - ) - except Exception: - self._last_register_password_error = f"注册密码接口返回异常: HTTP {response.status_code}" - - return False, None - - return True, password - - except Exception as e: - self._log(f"密码注册失败: {e}", "error") - self._last_register_password_error = str(e) - return False, None - - def _register_password_with_retry( - self, - did: Optional[str] = None, - sen_token: Optional[str] = None, - ) -> Tuple[bool, Optional[str]]: - """Retry password registration when OpenAI returns a generic recoverable 400.""" - max_attempts = 3 - retryable_markers = ( - "failed to create account", - "create account", - "invalid_request_error", - "http 400", - ) - - for attempt in range(1, max_attempts + 1): - success, password = self._register_password(did, sen_token) - if success: - return True, password - - error_text = str(self._last_register_password_error or "").strip().lower() - if attempt >= max_attempts: - break - if not any(marker in error_text for marker in retryable_markers): - break - - self._log( - f"密码注册命中可重试 400，准备重新生成密码后重试 ({attempt}/{max_attempts})...", - "warning", - ) - time.sleep(min(2 * attempt, 4)) - - return False, None - - def _mark_email_as_registered(self): - """标记邮箱为已注册状态（用于防止重复尝试）""" - try: - with get_db() as db: - # 检查是否已存在该邮箱的记录 - existing = crud.get_account_by_email(db, self.email) - if not existing: - # 创建一个失败记录，标记该邮箱已注册过 - crud.create_account( - db, - email=self.email, - password="", # 空密码表示未成功注册 - email_service=self.email_service.service_type.value, - email_service_id=self.email_info.get("service_id") if self.email_info else None, - status="failed", - extra_data={"register_failed_reason": "email_already_registered_on_openai"} - ) - self._log(f"已在数据库中标记邮箱 {self.email} 为已注册状态") - except Exception as e: - logger.warning(f"标记邮箱状态失败: {e}") - - def _send_verification_code(self, referer: Optional[str] = None) -> bool: - """发送验证码""" - try: - # 记录发送时间戳 - self._otp_sent_at = time.time() - send_referer = str(referer or "https://auth.openai.com/create-account/password").strip() - - response = self.session.get( - OPENAI_API_ENDPOINTS["send_otp"], - headers={ - "referer": send_referer, - "accept": "application/json", - }, - ) - - self._log(f"验证码发送状态: {response.status_code}") - return response.status_code == 200 - - except Exception as e: - self._log(f"发送验证码失败: {e}", "error") - return False - - def _get_verification_code(self, timeout: Optional[int] = None) -> Optional[str]: - """获取验证码""" - try: - mailbox_email = str(self.inbox_email or self.email or "").strip() - self._log(f"正在等待邮箱 {mailbox_email} 的验证码...") - - email_id = self.email_info.get("service_id") if self.email_info else None - fetch_timeout = int(timeout) if timeout and int(timeout) > 0 else 120 - code = self.email_service.get_verification_code( - email=mailbox_email, - email_id=email_id, - timeout=fetch_timeout, - pattern=OTP_CODE_PATTERN, - otp_sent_at=self._otp_sent_at, - ) - - if code: - self._log(f"成功获取验证码: {code}") - return code - else: - self._log("等待验证码超时", "error") - return None - - except Exception as e: - self._log(f"获取验证码失败: {e}", "error") - return None - - def _validate_verification_code(self, code: str) -> bool: - """验证验证码""" - try: - self._last_otp_validation_code = str(code or "").strip() - self._last_otp_validation_status_code = None - self._last_otp_validation_outcome = "" - code_body = f'{{"code":"{code}"}}' - - did = str(self.device_id or self.session.cookies.get("oai-did") or "").strip() - otp_token = self._check_sentinel(did, flow="email_otp_validate") if did else None - headers = self._build_json_headers( - referer="https://auth.openai.com/email-verification", - include_device_id=True, - did=did, - ) - if otp_token: - headers["openai-sentinel-token"] = otp_token - response = self.session.post( - OPENAI_API_ENDPOINTS["validate_otp"], - headers=headers, - data=code_body, - ) - - self._log(f"验证码校验状态: {response.status_code}") - self._last_otp_validation_status_code = int(response.status_code) - self._last_otp_validation_outcome = "success" if response.status_code == 200 else "http_non_200" - if response.status_code == 200: - # 记录 OTP 校验返回中的 continue/workspace 提示，供 native 收尾兜底 - try: - import urllib.parse as urlparse - payload = response.json() or {} - candidates: List[Dict[str, Any]] = [] - if isinstance(payload, dict): - candidates.append(payload) - for key in ("data", "result", "next", "payload"): - value = payload.get(key) - if isinstance(value, dict): - candidates.append(value) - - found_continue = "" - found_workspace = "" - for item in candidates: - if not isinstance(item, dict): - continue - if not found_workspace: - found_workspace = str( - item.get("workspace_id") - or item.get("workspaceId") - or item.get("default_workspace_id") - or ((item.get("workspace") or {}).get("id") if isinstance(item.get("workspace"), dict) else "") - or "" - ).strip() - if not found_continue: - for key in ("continue_url", "continueUrl", "next_url", "nextUrl", "redirect_url", "redirectUrl", "url"): - candidate = str(item.get(key) or "").strip() - if not candidate: - continue - if candidate.startswith("/"): - candidate = urlparse.urljoin(OPENAI_API_ENDPOINTS["validate_otp"], candidate) - found_continue = candidate - break - if found_workspace and found_continue: - break - - if found_workspace: - self._last_validate_otp_workspace_id = found_workspace - self._log(f"OTP 校验返回 Workspace ID: {found_workspace}") - if found_continue: - self._last_validate_otp_continue_url = found_continue - self._log(f"OTP 校验返回 continue_url: {found_continue[:100]}...") - except Exception as parse_err: - self._log(f"解析 OTP 校验返回信息失败: {parse_err}", "warning") - - return response.status_code == 200 - - except Exception as e: - err_text = str(e or "").lower() - if ( - "timed out" in err_text - or "timeout" in err_text - or "curl: (28)" in err_text - or "operation timed out" in err_text - ): - self._last_otp_validation_outcome = "network_timeout" - else: - self._last_otp_validation_outcome = "network_error" - self._log(f"验证验证码失败: {e}", "error") - return False - - def _verify_email_otp_with_retry( - self, - stage_label: str = "验证码", - max_attempts: int = 3, - fetch_timeout: Optional[int] = None, - attempted_codes: Optional[set[str]] = None, - ) -> bool: - """ - 获取并校验验证码（带重试）。 - 用于规避邮箱里历史验证码导致的 400（第一次取到旧码，第二次取新码）。 - """ - # 每轮验证码阶段开始前，清理上轮 OTP 校验缓存，避免 continue_url/workspace 被旧阶段污染。 - self._last_validate_otp_continue_url = None - self._last_validate_otp_workspace_id = None - if attempted_codes is None: - attempted_codes = set() - for attempt in range(1, max_attempts + 1): - code = ( - self._get_verification_code(timeout=fetch_timeout) - if fetch_timeout - else self._get_verification_code() - ) - if not code: - if attempt < max_attempts: - self._log( - f"{stage_label}第 {attempt}/{max_attempts} 次未取到验证码，稍后重试...", - "warning", - ) - time.sleep(2) - continue - return False - - if code in attempted_codes: - allow_same_code_retry = ( - self._last_otp_validation_code == code - and self._last_otp_validation_outcome in {"network_timeout", "network_error"} - ) - if allow_same_code_retry: - self._log( - f"{stage_label}第 {attempt}/{max_attempts} 次命中重复验证码 {code}，" - f"但上次校验为网络异常（{self._last_otp_validation_outcome}），重试同码...", - "warning", - ) - if self._validate_verification_code(code): - return True - if attempt < max_attempts: - time.sleep(2) - continue - return False - - if attempt < max_attempts: - self._log( - f"{stage_label}第 {attempt}/{max_attempts} 次命中重复验证码 {code}，等待新邮件...", - "warning", - ) - time.sleep(2) - continue - return False - - attempted_codes.add(code) - - if self._validate_verification_code(code): - return True - - if attempt < max_attempts: - self._log( - f"{stage_label}第 {attempt}/{max_attempts} 次校验未通过，疑似旧验证码，自动重试下一封...", - "warning", - ) - time.sleep(2) - - return False - - def _create_user_account(self) -> bool: - """创建用户账户""" - try: - user_info = generate_random_user_info() - self._log(f"生成用户信息: {user_info['name']}, 年龄: {user_info.get('age', '?')}岁, 出生日期: {user_info['birthdate']}") - # 必须包含 birthdate，OpenAI API 会校验该必填项 - api_payload = {"name": user_info["name"], "birthdate": user_info["birthdate"]} - create_account_body = json.dumps(api_payload) - - did = str(self.device_id or self.session.cookies.get("oai-did") or "").strip() - create_token = self._check_sentinel(did, flow="oauth_create_account") if did else None - headers = self._build_json_headers( - referer="https://auth.openai.com/about-you", - include_device_id=True, - did=did, - ) - if create_token: - headers["openai-sentinel-token"] = create_token - response = self.session.post( - OPENAI_API_ENDPOINTS["create_account"], - headers=headers, - data=create_account_body, - ) - - self._log(f"账户创建状态: {response.status_code}") - - if response.status_code != 200: - self._log(f"账户创建失败: {response.text[:200]}", "warning") - return False - - try: - data = response.json() or {} - continue_url = str(data.get("continue_url") or "").strip() - if continue_url: - self._create_account_continue_url = continue_url - self._log(f"create_account 返回 continue_url，已缓存: {continue_url[:100]}...") - account_id = str( - data.get("account_id") - or data.get("chatgpt_account_id") - or (data.get("account") or {}).get("id") - or "" - ).strip() - if account_id: - self._create_account_account_id = account_id - self._log(f"create_account ���回 account_id，已缓存: {account_id}") - workspace_id = str( - data.get("workspace_id") - or data.get("default_workspace_id") - or (data.get("workspace") or {}).get("id") - or "" - ).strip() - if (not workspace_id) and isinstance(data.get("workspaces"), list) and data.get("workspaces"): - workspace_id = str((data.get("workspaces")[0] or {}).get("id") or "").strip() - if workspace_id: - self._create_account_workspace_id = workspace_id - self._log(f"create_account 返回 workspace_id，已缓存: {workspace_id}") - refresh_token = str(data.get("refresh_token") or "").strip() - if refresh_token: - self._create_account_refresh_token = refresh_token - self._log("create_account 返回 refresh_token，已缓存") - except Exception: - pass - - return True - - except Exception as e: - self._log(f"创建账户失败: {e}", "error") - return False - - def _get_workspace_id(self) -> Optional[str]: - """获取 Workspace ID""" - try: - def _extract_workspace_id(payload: Any) -> str: - if not isinstance(payload, dict): - return "" - workspace_id = str( - payload.get("workspace_id") - or payload.get("default_workspace_id") - or ((payload.get("workspace") or {}).get("id") if isinstance(payload.get("workspace"), dict) else "") - or "" - ).strip() - if workspace_id: - return workspace_id - workspaces = payload.get("workspaces") or [] - if isinstance(workspaces, list) and workspaces: - return str((workspaces[0] or {}).get("id") or "").strip() - return "" - - auth_cookie = str(self.session.cookies.get("oai-client-auth-session") or "").strip() - if not auth_cookie: - self._log("未能获取到授权 Cookie，尝试从 auth-info 里取 workspace", "warning") - - # 解码 JWT - import base64 - import json as json_module - import urllib.parse as urlparse - - try: - candidate_payloads: List[str] = [] - if auth_cookie: - segments = auth_cookie.split(".") - # 对齐 ABCard：优先 JWT payload 段（第 2 段） - if len(segments) >= 2 and segments[1]: - candidate_payloads.append(segments[1]) - if segments and segments[0]: - candidate_payloads.append(segments[0]) - # 极端情况下 cookie 可能直接是 JSON 字符串 - candidate_payloads.append(auth_cookie) - - for payload in candidate_payloads: - raw = str(payload or "").strip() - if not raw: - continue - auth_json = None - try: - pad = "=" * ((4 - (len(raw) % 4)) % 4) - decoded = base64.urlsafe_b64decode((raw + pad).encode("ascii")) - auth_json = json_module.loads(decoded.decode("utf-8")) - except Exception: - try: - auth_json = json_module.loads(raw) - except Exception: - auth_json = None - - workspace_id = _extract_workspace_id(auth_json) - if workspace_id: - self._log(f"Workspace ID: {workspace_id}") - return workspace_id - - # 兜底：从 oai-client-auth-info（URL 编码 JSON）提取 workspace - auth_info_raw = str(self.session.cookies.get("oai-client-auth-info") or "").strip() - if auth_info_raw: - auth_info_text = auth_info_raw - for _ in range(2): - decoded = urlparse.unquote(auth_info_text) - if decoded == auth_info_text: - break - auth_info_text = decoded - try: - auth_info_json = json_module.loads(auth_info_text) - workspace_id = _extract_workspace_id(auth_info_json) - if workspace_id: - self._log(f"Workspace ID (auth-info): {workspace_id}") - return workspace_id - except Exception as auth_info_err: - self._log(f"解析 auth-info Cookie 失败: {auth_info_err}", "warning") - - # 兜底：复用 create_account 缓存 - cached_workspace = str(self._create_account_workspace_id or "").strip() - if cached_workspace: - self._log(f"Workspace ID (create_account缓存): {cached_workspace}") - return cached_workspace - - self._log("授权 Cookie 里没有 workspace 信息", "warning") - return None - - except Exception as e: - self._log(f"解析授权 Cookie 失败: {e}", "warning") - return None - - except Exception as e: - self._log(f"获取 Workspace ID 失败: {e}", "error") - return None - - def _select_organization( - self, - *, - org_id: str, - project_id: str = "", - referer: str, - did: str = "", - ) -> Optional[str]: - """提交 organization/select，返回 continue_url 或回调地址。""" - try: - payload = {"org_id": org_id} - if project_id: - payload["project_id"] = project_id - headers = self._build_json_headers( - referer=referer, - include_device_id=True, - did=did, - ) - response = self.session.post( - OPENAI_API_ENDPOINTS["select_organization"], - headers=headers, - data=json.dumps(payload), - allow_redirects=False, - ) - self._log(f"organization/select 状态: {response.status_code}") - - location = str(response.headers.get("Location") or "").strip() - if response.status_code in [301, 302, 303, 307, 308] and location: - import urllib.parse - continue_url = urllib.parse.urljoin(OPENAI_API_ENDPOINTS["select_organization"], location) - self._log(f"Organization Continue URL (Location): {continue_url[:100]}...") - return continue_url - - if response.status_code != 200: - self._log(f"organization/select 失败: {response.status_code}", "warning") - return None - - try: - payload = response.json() or {} - except Exception as json_err: - self._log(f"解析 organization/select 响应失败: {json_err}", "warning") - return None - - continue_url = str(payload.get("continue_url") or "").strip() - if continue_url: - continue_url = continue_url.replace("\\/", "/") - self._log(f"Organization Continue URL: {continue_url[:100]}...") - return continue_url - return None - except Exception as e: - self._log(f"选择 Organization 失败: {e}", "error") - return None - - def _select_workspace(self, workspace_id: str) -> Optional[str]: - """选择 Workspace，并在需要时提交 organization/select。""" - try: - select_body = json.dumps({"workspace_id": workspace_id}) - did = str(self.device_id or self.session.cookies.get("oai-did") or "").strip() - headers = self._build_json_headers( - referer="https://auth.openai.com/sign-in-with-chatgpt/codex/consent", - include_device_id=True, - did=did, - ) - - response = self.session.post( - OPENAI_API_ENDPOINTS["select_workspace"], - headers=headers, - data=select_body, - allow_redirects=False, - ) - - location = str(response.headers.get("Location") or "").strip() - if response.status_code in [301, 302, 303, 307, 308] and location: - import urllib.parse - continue_url = urllib.parse.urljoin(OPENAI_API_ENDPOINTS["select_workspace"], location) - self._log(f"Continue URL (Location): {continue_url[:100]}...") - return continue_url - - if response.status_code != 200: - self._log(f"选择 workspace 失败: {response.status_code}", "error") - self._log(f"响应: {response.text[:200]}", "warning") - return None - - payload: Dict[str, Any] = {} - continue_url = "" - try: - payload = response.json() or {} - continue_url = str(payload.get("continue_url") or "").strip() - except Exception as json_err: - body_text = str(response.text or "") - self._log(f"workspace/select 非 JSON 响应，尝试文本兜底解析: {json_err}", "warning") - m = re.search(r'"continue_url"\s*:\s*"([^"]+)"', body_text) - if m: - continue_url = str(m.group(1) or "").strip() - if not continue_url: - m2 = re.search(r"https://auth\.openai\.com/[^\s\"'<>]+", body_text) - if m2: - continue_url = str(m2.group(0) or "").strip() - - orgs = [] - if isinstance(payload, dict): - orgs = (((payload.get("data") or {}).get("orgs")) or []) if isinstance(payload.get("data"), dict) else [] - - if orgs: - first_org = orgs[0] or {} - org_id = str(first_org.get("id") or "").strip() - project_id = str((((first_org.get("projects") or [None])[0]) or {}).get("id") or "").strip() - if org_id: - org_continue = self._select_organization( - org_id=org_id, - project_id=project_id, - referer=continue_url or "https://auth.openai.com/sign-in-with-chatgpt/codex/consent", - did=did, - ) - if org_continue: - return org_continue - - if not continue_url: - if location: - import urllib.parse - continue_url = urllib.parse.urljoin(OPENAI_API_ENDPOINTS["select_workspace"], location) - else: - self._log("workspace/select 响应里缺少 continue_url", "error") - return None - - if continue_url: - continue_url = continue_url.replace("\\/", "/") - self._log(f"Continue URL: {continue_url[:100]}...") - return continue_url - return None - except Exception as e: - self._log(f"选择 Workspace 失败: {e}", "error") - return None - - def _follow_redirects(self, start_url: str) -> Tuple[Optional[str], str]: - """手动跟随重定向链，返回 (callback_url, final_url)。""" - try: - def _is_oauth_callback(url: str) -> bool: - try: - import urllib.parse as _urlparse - - parsed = _urlparse.urlparse(url) - path = (parsed.path or "").lower() - if ("/auth/callback" not in path) and ("/api/auth/callback/openai" not in path): - return False - query = _urlparse.parse_qs(parsed.query or "", keep_blank_values=True) - # 只要带 code 或 error，就认为已经进入回调阶段（避免被本地 503 干扰识别） - return bool(query.get("code") or query.get("error")) - except Exception: - return False - - current_url = start_url - callback_url: Optional[str] = None - max_redirects = 12 - - for i in range(max_redirects): - self._log(f"重定向 {i+1}/{max_redirects}: {current_url[:100]}...") - if _is_oauth_callback(current_url) and not callback_url: - callback_url = current_url - self._log(f"命中回调 URL: {current_url[:120]}...") - # 已拿到 callback，不再请求本地 callback 地址，避免 503 干扰后续判断 - break - - response = self.session.get( - current_url, - allow_redirects=False, - timeout=15 - ) - - location = response.headers.get("Location") or "" - - if "/api/auth/callback/openai" in current_url and not callback_url: - callback_url = current_url - - # 如果不是重定向状态码，停止 - if response.status_code not in [301, 302, 303, 307, 308]: - self._log(f"非重定向状态码: {response.status_code}") - break - - if not location: - self._log("重定向响应缺少 Location 头") - break - - # 构建下一个 URL - import urllib.parse - next_url = urllib.parse.urljoin(current_url, location) - - # 命中回调时仅记录，不提前返回；继续跟到底，让 next-auth 充分落 cookie。 - if _is_oauth_callback(next_url) and not callback_url: - callback_url = next_url - self._log(f"找到回调 URL: {next_url[:100]}...") - current_url = next_url - break - - current_url = next_url - - # 对齐 ABCard：补打一跳 chatgpt 首页，确保 next-auth cookie 完整落地。 - try: - if not current_url.rstrip("/").endswith("chatgpt.com"): - self.session.get( - "https://chatgpt.com/", - headers={ - "accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", - "referer": current_url, - "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" - ), - }, - timeout=20, - ) - except Exception as home_err: - self._log(f"重定向结束后首页补跳异常: {home_err}", "warning") - - if not callback_url: - self._log("未能在重定向链中找到回调 URL", "warning") - return callback_url, current_url - - except Exception as e: - self._log(f"跟随重定向失败: {e}", "error") - return None, start_url - - def _handle_oauth_callback(self, callback_url: str) -> Optional[Dict[str, Any]]: - """处理 OAuth 回调""" - try: - if not self.oauth_start: - self._log("OAuth 流程未初始化", "error") - return None - - self._log("处理 OAuth 回调，最后一哆嗦，稳住别抖...") - token_info = self.oauth_manager.handle_callback( - callback_url=callback_url, - expected_state=self.oauth_start.state, - code_verifier=self.oauth_start.code_verifier - ) - - self._log("OAuth 授权成功，通关文牒到手") - return token_info - - except Exception as e: - self._log(f"处理 OAuth 回调失败: {e}", "error") - return None - - def _run_primary_registration(self) -> RegistrationResult: - """ - 执行完整的注册流程 - - 支持已注册账号自动登录： - - 如果检测到邮箱已注册，自动切换到登录流程 - - 已注册账号跳过：设置密码、发送验证码、创建用户账户 - - 共用步骤：获取验证码、验证验证码、Workspace 和 OAuth 回调 - - Returns: - RegistrationResult: 注册结果 - """ - result = RegistrationResult(success=False, logs=self.logs) - - try: - self._is_existing_account = False - self._token_acquisition_requires_login = False - self._otp_sent_at = None - self._create_account_continue_url = None - self._create_account_workspace_id = None - self._create_account_account_id = None - self._create_account_refresh_token = None - self._last_validate_otp_continue_url = None - self._last_validate_otp_workspace_id = None - - self._log("=" * 60) - self._log("注册流程启动，开始替你敲门") - self._check_cancelled("启动") - self._log("=" * 60) - self._log(f"注册入口链路配置: {self.registration_entry_flow}") - configured_entry_flow = self.registration_entry_flow - service_type_raw = getattr(self.email_service, "service_type", "") - service_type_value = str(getattr(service_type_raw, "value", service_type_raw) or "").strip().lower() - effective_entry_flow = configured_entry_flow - if service_type_value == "outlook": - self._log("检测到 Outlook 邮箱，自动使用 Outlook 入口链路（无需在设置中选择）") - effective_entry_flow = "outlook" - - # 1. 检查 IP 地理位置 - self._log("1. 先看看这条网络从哪儿来，别一开局就站错片场...") - self._check_cancelled("IP 检查前") - ip_ok, location = self._check_ip_location() - if not ip_ok: - result.error_message = f"IP 地理位置不支持: {location}" - self._log(f"IP 检查失败: {location}", "error") - return result - - self._log(f"IP 位置: {location}") - - # 2. 创建邮箱 - self._log("2. 开个新邮箱，准备收信...") - self._check_cancelled("创建邮箱前") - if not self._create_email(): - result.error_message = "创建邮箱失败" - return result - - result.email = self.email - - # 3. 准备首轮授权流程 - self._check_cancelled("准备授权前") - did, sen_token = self._prepare_authorize_flow("首次授权") - if not did: - result.error_message = "获取 Device ID 失败" - return result - result.device_id = did - if not sen_token: - result.error_message = "Sentinel POW 验证失败" - return result - - # 4. 提交注册入口邮箱 - self._log("4. 递上邮箱，看看 OpenAI 这球怎么接...") - self._check_cancelled("提交注册表单前") - signup_result = self._submit_signup_form(did, sen_token) - if not signup_result.success: - result.error_message = f"提交注册表单失败: {signup_result.error_message}" - return result - - if self._is_existing_account: - self._log("检测到这是老朋友账号，直接切去登录拿 token，不走弯路") - else: - self._log("5. 设置密码，别让小偷偷笑...") - self._check_cancelled("设置密码前") - password_ok, _ = self._register_password_with_retry(did, sen_token) - if not password_ok: - result.error_message = self._last_register_password_error or "注册密码失败" - return result - - self._log("6. 催一下注册验证码出门，邮差该冲刺了...") - self._check_cancelled("发送验证码前") - if not self._send_verification_code(): - result.error_message = "发送验证码失败" - return result - - self._log("7. 等验证码飞来，邮箱请注意查收...") - self._log("8. 对一下验证码，看看是不是本人...") - self._check_cancelled("验证验证码前") - if not self._verify_email_otp_with_retry(stage_label="注册验证码", max_attempts=3): - result.error_message = "验证验证码失败" - return result - - self._log("9. 给账号办个正式户口，名字写档案里...") - self._check_cancelled("创建账号前") - if not self._create_user_account(): - result.error_message = "创建用户账户失败" - return result - - # 注册完成后，统一走二次 OAuth 登录流程获取 token - self._check_cancelled("重启登录流程前") - login_ready, login_error = self._restart_login_flow() - if not login_ready: - result.error_message = login_error - return result - - # 统一使用 native backup 收尾链路：登录验证码 -> Workspace -> redirect -> OAuth callback -> token - self._check_cancelled("Token 交换前") - if not self._complete_token_exchange_native_backup(result): - return result - - # 10. 完成 - self._log("=" * 60) - if self._is_existing_account: - self._log("登录成功，老朋友顺利回家") - else: - self._log("注册成功，账号已经稳稳落地，可以开香槟了") - self._log(f"邮箱: {result.email}") - self._log(f"Device ID: {result.device_id or '-'}") - self._log(f"Account ID: {result.account_id}") - self._log(f"Workspace ID: {result.workspace_id}") - self._log("=" * 60) - - result.success = True - settings = get_settings() - client_id = str(getattr(settings, "openai_client_id", "") or getattr(self.oauth_manager, "client_id", "") or "").strip() - token_metadata = build_token_completeness_metadata( - access_token=result.access_token, - refresh_token=result.refresh_token, - id_token=result.id_token, - account_id=result.account_id, - ) - result.metadata = { - "email_service": self.email_service.service_type.value, - "proxy_used": self.proxy_url, - "registered_at": datetime.now().isoformat(), - "is_existing_account": self._is_existing_account, - "token_acquired_via_relogin": self._token_acquisition_requires_login, - "client_id": client_id, - "device_id": result.device_id, - "has_session_token": bool(result.session_token), - "registration_entry_flow": configured_entry_flow, - "registration_entry_flow_effective": effective_entry_flow, - # 对齐 K:\1\2：原生入口允许无 session_token 成功，但会标记待补。 - "session_token_pending": (effective_entry_flow == "native") and (not bool(result.session_token)), - **token_metadata, - } - - return result - - except RegistrationCancelled as e: - result.error_message = str(e) - self._log("注册流程已按取消请求中止", "warning") - return result - - except Exception as e: - self._log(f"注册过程中发生未预期错误: {e}", "error") - result.error_message = str(e) - return result - - def _build_anyauto_fallback_result( - self, - flow_result: Optional[Dict[str, Any]], - primary_error: str = "", - ) -> RegistrationResult: - """Map PR60 AnyAuto V2 output into the current RegistrationResult structure.""" - result = RegistrationResult(success=False, logs=self.logs) - result.email = str(self.email or "") - result.password = str(self.password or "") - result.device_id = str(self.device_id or "") - - if not flow_result or not flow_result.get("success"): - fallback_error = str((flow_result or {}).get("error_message") or "注册失败").strip() - if primary_error and fallback_error and fallback_error != primary_error: - result.error_message = f"{primary_error} | anyauto fallback: {fallback_error}" - else: - result.error_message = fallback_error or primary_error or "注册失败" - result.metadata = { - "registration_flow": "any-auto-register-fallback", - "fallback_attempted": True, - "primary_error": primary_error, - "fallback_success": False, - } - return result - - result.success = True - result.access_token = str(flow_result.get("access_token") or "") - result.refresh_token = str(flow_result.get("refresh_token") or "") - result.id_token = str(flow_result.get("id_token") or "") - result.session_token = str(flow_result.get("session_token") or "") - result.account_id = str(flow_result.get("account_id") or "") - result.workspace_id = str(flow_result.get("workspace_id") or "") - result.source = "register" - - if not result.account_id: - token_payload = result.access_token or result.id_token - result.account_id = str(self._extract_account_id_from_access_token(token_payload) or "").strip() - if (not result.account_id) and result.id_token: - try: - account_info = self.oauth_manager.extract_account_info(result.id_token) - result.account_id = str(account_info.get("account_id") or "").strip() - except Exception: - pass - - settings = get_settings() - client_id = str( - getattr(settings, "openai_client_id", "") - or getattr(self.oauth_manager, "client_id", "") - or "" - ).strip() - metadata = dict(flow_result.get("metadata") or {}) - token_metadata = build_token_completeness_metadata( - access_token=result.access_token, - refresh_token=result.refresh_token, - id_token=result.id_token, - account_id=result.account_id, - ) - metadata.update( - { - "email_service": self.email_service.service_type.value, - "proxy_used": self.proxy_url, - "registered_at": datetime.now().isoformat(), - "registration_flow": "any-auto-register-fallback", - "fallback_attempted": True, - "primary_error": primary_error, - "client_id": client_id, - "device_id": result.device_id, - "has_session_token": bool(result.session_token), - **token_metadata, - } - ) - result.metadata = metadata - return result - - def _run_anyauto_fallback(self, primary_error: str = "") -> RegistrationResult: - """Run the PR60 AnyAuto V2 engine as a controlled fallback.""" - settings = get_settings() - max_retries = int(getattr(settings, "registration_max_retries", 3) or 3) - browser_mode = str( - getattr(settings, "registration_anyauto_browser_mode", "protocol") or "protocol" - ).strip() - - flow_engine = AnyAutoRegistrationEngine( - email_service=self.email_service, - proxy_url=self.proxy_url, - callback_logger=self._log, - max_retries=max_retries, - browser_mode=browser_mode or "protocol", - extra_config=None, - ) - flow_result = flow_engine.run() - - self.email_info = flow_engine.email_info - self.email = flow_engine.email - self.inbox_email = flow_engine.inbox_email - self.password = flow_engine.password - self.session = flow_engine.session - self.device_id = flow_engine.device_id - - fallback_result = self._build_anyauto_fallback_result(flow_result, primary_error=primary_error) - if fallback_result.session_token: - self.session_token = fallback_result.session_token - return fallback_result - - def _should_try_anyauto_fallback(self, result: RegistrationResult) -> bool: - settings = get_settings() - enabled = bool(getattr(settings, "registration_enable_anyauto_fallback", True)) - if not enabled or result.success: - return False - - error_text = str(result.error_message or "").strip().lower() - if not error_text: - return True - - non_retryable_markers = ( - "unsupported country", - "invalid email service", - "email service not found", - ) - if any(marker in error_text for marker in non_retryable_markers): - return False - - retryable_markers = ( - "access_token", - "refresh_token", - "session", - "oauth", - "callback", - "authorization code", - "workspace", - "consent", - "otp", - "verification code", - "phone", - "add_phone", - "add-phone", - "sentinel", - "failed to create account", - "create account", - "invalid_request_error", - "http 400", - "registration failed", - ) - return any(marker in error_text for marker in retryable_markers) - - def run(self) -> RegistrationResult: - """Run the current primary flow first, then selectively fall back to PR60 AnyAuto V2.""" - primary_result = self._run_primary_registration() - if primary_result.success: - return primary_result - - if not self._should_try_anyauto_fallback(primary_result): - return primary_result - - primary_error = str(primary_result.error_message or "").strip() - self._log("主注册链路未成功，开始尝试 PR60 anyauto V2 回退流程...", "warning") - fallback_result = self._run_anyauto_fallback(primary_error=primary_error) - if fallback_result.success: - self._log("PR60 anyauto V2 回退流程成功，已补上 V2 注册兜底能力") - return fallback_result - - self._log(f"PR60 anyauto V2 回退流程也失败了: {fallback_result.error_message}", "warning") - return fallback_result - - def save_to_database( - self, - result: RegistrationResult, - account_label: Optional[str] = None, - role_tag: Optional[str] = None, - ) -> bool: - """ - 保存注册结果到数据库 - - Args: - result: 注册结果 - - Returns: - 是否保存成功 - """ - if not result.success: - return False - - try: - # 获取默认 client_id - settings = get_settings() - - with get_db() as db: - # 保存账户信息 - account = crud.create_account( - db, - email=result.email, - password=result.password, - client_id=settings.openai_client_id, - session_token=result.session_token, - cookies=self._dump_session_cookies(), - email_service=self.email_service.service_type.value, - email_service_id=self.email_info.get("service_id") if self.email_info else None, - account_id=result.account_id, - workspace_id=result.workspace_id, - access_token=result.access_token, - refresh_token=result.refresh_token, - id_token=result.id_token, - proxy_used=self.proxy_url, - extra_data=result.metadata, - source=result.source, - account_label=account_label, - role_tag=role_tag, - ) - - self._log(f"账户已存进数据库，落袋为安，ID: {account.id}") - return True - - except Exception as e: - self._log(f"保存到数据库失败: {e}", "error") - return False