Spaces:
Sleeping
Sleeping
| """ | |
| 注册流程引擎 | |
| 从 main.py 中提取并重构的注册流程 | |
| """ | |
| import re | |
| import json | |
| import time | |
| import logging | |
| import secrets | |
| import string | |
| from typing import Optional, Dict, Any, Tuple, Callable | |
| from dataclasses import dataclass | |
| from datetime import datetime | |
| from curl_cffi import requests as cffi_requests | |
| 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_CHARSET, | |
| AccountStatus, | |
| TaskStatus, | |
| ) | |
| from ..config.settings import get_settings | |
| logger = logging.getLogger(__name__) | |
| 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 = "" # 会话令牌 | |
| 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 "", | |
| "error_message": self.error_message, | |
| "logs": self.logs or [], | |
| "metadata": self.metadata or {}, | |
| "source": self.source, | |
| } | |
| 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 | |
| ): | |
| """ | |
| 初始化注册引擎 | |
| 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 | |
| # 创建 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 # 传递代理配置 | |
| ) | |
| # 状态变量 | |
| self.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.logs: list = [] | |
| self._otp_sent_at: Optional[float] = None # OTP 发送时间戳 | |
| self._is_existing_account: bool = False # 是否为已注册账号（用于自动登录） | |
| self._token_acquisition_requires_login: bool = False # 新注册账号需要二次登录拿 token | |
| 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 _generate_password(self, length: int = DEFAULT_PASSWORD_LENGTH) -> str: | |
| """生成随机密码""" | |
| return ''.join(secrets.choice(PASSWORD_CHARSET) for _ in range(length)) | |
| 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 | |
| self.email = self.email_info["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 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 | |
| return None | |
| def _check_sentinel(self, did: str) -> Optional[str]: | |
| """检查 Sentinel 拦截""" | |
| try: | |
| sen_token = self.http_client.check_sentinel(did) | |
| if sen_token: | |
| self._log(f"Sentinel token 获取成功") | |
| return sen_token | |
| self._log("Sentinel 检查失败: 未获取到 token", "warning") | |
| return None | |
| except Exception as e: | |
| self._log(f"Sentinel 检查异常: {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: 提交结果，包含账号状态判断 | |
| """ | |
| try: | |
| request_body = json.dumps({ | |
| "username": { | |
| "value": self.email, | |
| "kind": "email", | |
| }, | |
| "screen_hint": screen_hint, | |
| }) | |
| headers = { | |
| "referer": referer, | |
| "accept": "application/json", | |
| "content-type": "application/json", | |
| } | |
| if sen_token: | |
| sentinel = json.dumps({ | |
| "p": "", | |
| "t": "", | |
| "c": sen_token, | |
| "id": did, | |
| "flow": "authorize_continue", | |
| }) | |
| headers["openai-sentinel-token"] = sentinel | |
| 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 != 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: | |
| self._log(f"{log_label}失败: {e}", "error") | |
| return SignupFormResult(success=False, error_message=str(e)) | |
| 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: | |
| """提交登录密码，进入邮箱验证码页面。""" | |
| try: | |
| response = self.session.post( | |
| OPENAI_API_ENDPOINTS["password_verify"], | |
| headers={ | |
| "referer": "https://auth.openai.com/log-in/password", | |
| "accept": "application/json", | |
| "content-type": "application/json", | |
| }, | |
| data=json.dumps({"password": self.password}), | |
| ) | |
| self._log(f"提交登录密码状态: {response.status_code}") | |
| 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: | |
| self._log(f"提交登录密码失败: {e}", "error") | |
| return SignupFormResult(success=False, error_message=str(e)) | |
| 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 = self._get_device_id() | |
| if not did: | |
| return None, None | |
| 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 | |
| def _complete_token_exchange(self, result: RegistrationResult) -> bool: | |
| """在登录态已建立后，继续完成 workspace 和 OAuth token 获取。""" | |
| self._log("等待登录验证码到场，最后这位嘉宾还在路上...") | |
| code = self._get_verification_code() | |
| if not code: | |
| result.error_message = "获取验证码失败" | |
| return False | |
| self._log("核对登录验证码，验明正身一下...") | |
| if not self._validate_verification_code(code): | |
| result.error_message = "验证码校验失败" | |
| return False | |
| self._log("摸一下 Workspace ID，看看该坐哪桌...") | |
| workspace_id = self._get_workspace_id() | |
| if not workspace_id: | |
| result.error_message = "获取 Workspace ID 失败" | |
| return False | |
| result.workspace_id = workspace_id | |
| self._log("选择 Workspace，安排个靠谱座位...") | |
| continue_url = self._select_workspace(workspace_id) | |
| if not continue_url: | |
| result.error_message = "选择 Workspace 失败" | |
| return False | |
| self._log("顺着重定向面包屑往前走，别跟丢了...") | |
| callback_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 = 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" | |
| 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 _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 _register_password(self) -> Tuple[bool, Optional[str]]: | |
| """注册密码""" | |
| try: | |
| # 生成密码 | |
| password = self._generate_password() | |
| self.password = password # 保存密码到实例变量 | |
| self._log(f"生成密码: {password}") | |
| # 提交密码注册 | |
| register_body = json.dumps({ | |
| "password": password, | |
| "username": self.email | |
| }) | |
| response = self.session.post( | |
| OPENAI_API_ENDPOINTS["register"], | |
| headers={ | |
| "referer": "https://auth.openai.com/create-account/password", | |
| "accept": "application/json", | |
| "content-type": "application/json", | |
| }, | |
| 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", "") | |
| # 检测邮箱已注册的情况 | |
| if "already" in error_msg.lower() or "exists" in error_msg.lower() or error_code == "user_exists": | |
| self._log(f"邮箱 {self.email} 可能已在 OpenAI 注册过", "error") | |
| # 标记此邮箱为已注册状态 | |
| self._mark_email_as_registered() | |
| except Exception: | |
| pass | |
| return False, None | |
| return True, password | |
| except Exception as e: | |
| self._log(f"密码注册失败: {e}", "error") | |
| 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) -> bool: | |
| """发送验证码""" | |
| try: | |
| # 记录发送时间戳 | |
| self._otp_sent_at = time.time() | |
| response = self.session.get( | |
| OPENAI_API_ENDPOINTS["send_otp"], | |
| headers={ | |
| "referer": "https://auth.openai.com/create-account/password", | |
| "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) -> Optional[str]: | |
| """获取验证码""" | |
| try: | |
| self._log(f"正在等待邮箱 {self.email} 的验证码...") | |
| email_id = self.email_info.get("service_id") if self.email_info else None | |
| code = self.email_service.get_verification_code( | |
| email=self.email, | |
| email_id=email_id, | |
| timeout=30, | |
| 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: | |
| code_body = f'{{"code":"{code}"}}' | |
| response = self.session.post( | |
| OPENAI_API_ENDPOINTS["validate_otp"], | |
| headers={ | |
| "referer": "https://auth.openai.com/email-verification", | |
| "accept": "application/json", | |
| "content-type": "application/json", | |
| }, | |
| data=code_body, | |
| ) | |
| self._log(f"验证码校验状态: {response.status_code}") | |
| return response.status_code == 200 | |
| except Exception as e: | |
| self._log(f"验证验证码失败: {e}", "error") | |
| return False | |
| def _create_user_account(self) -> bool: | |
| """创建用户账户""" | |
| try: | |
| user_info = generate_random_user_info() | |
| self._log(f"生成用户信息: {user_info['name']}, 生日: {user_info['birthdate']}") | |
| create_account_body = json.dumps(user_info) | |
| response = self.session.post( | |
| OPENAI_API_ENDPOINTS["create_account"], | |
| headers={ | |
| "referer": "https://auth.openai.com/about-you", | |
| "accept": "application/json", | |
| "content-type": "application/json", | |
| }, | |
| data=create_account_body, | |
| ) | |
| self._log(f"账户创建状态: {response.status_code}") | |
| if response.status_code != 200: | |
| self._log(f"账户创建失败: {response.text[:200]}", "warning") | |
| return False | |
| return True | |
| except Exception as e: | |
| self._log(f"创建账户失败: {e}", "error") | |
| return False | |
| def _get_workspace_id(self) -> Optional[str]: | |
| """获取 Workspace ID""" | |
| try: | |
| auth_cookie = self.session.cookies.get("oai-client-auth-session") | |
| if not auth_cookie: | |
| self._log("未能获取到授权 Cookie", "error") | |
| return None | |
| # 解码 JWT | |
| import base64 | |
| import json as json_module | |
| try: | |
| segments = auth_cookie.split(".") | |
| if len(segments) < 1: | |
| self._log("授权 Cookie 格式错误", "error") | |
| return None | |
| # 解码第一个 segment | |
| payload = segments[0] | |
| pad = "=" * ((4 - (len(payload) % 4)) % 4) | |
| decoded = base64.urlsafe_b64decode((payload + pad).encode("ascii")) | |
| auth_json = json_module.loads(decoded.decode("utf-8")) | |
| workspaces = auth_json.get("workspaces") or [] | |
| if not workspaces: | |
| self._log("授权 Cookie 里没有 workspace 信息", "error") | |
| return None | |
| workspace_id = str((workspaces[0] or {}).get("id") or "").strip() | |
| if not workspace_id: | |
| self._log("无法解析 workspace_id", "error") | |
| return None | |
| self._log(f"Workspace ID: {workspace_id}") | |
| return workspace_id | |
| except Exception as e: | |
| self._log(f"解析授权 Cookie 失败: {e}", "error") | |
| return None | |
| except Exception as e: | |
| self._log(f"获取 Workspace ID 失败: {e}", "error") | |
| return None | |
| def _select_workspace(self, workspace_id: str) -> Optional[str]: | |
| """选择 Workspace""" | |
| try: | |
| select_body = f'{{"workspace_id":"{workspace_id}"}}' | |
| response = self.session.post( | |
| OPENAI_API_ENDPOINTS["select_workspace"], | |
| headers={ | |
| "referer": "https://auth.openai.com/sign-in-with-chatgpt/codex/consent", | |
| "content-type": "application/json", | |
| }, | |
| data=select_body, | |
| ) | |
| if response.status_code != 200: | |
| self._log(f"选择 workspace 失败: {response.status_code}", "error") | |
| self._log(f"响应: {response.text[:200]}", "warning") | |
| return None | |
| continue_url = str((response.json() or {}).get("continue_url") or "").strip() | |
| if not continue_url: | |
| self._log("workspace/select 响应里缺少 continue_url", "error") | |
| return None | |
| self._log(f"Continue URL: {continue_url[:100]}...") | |
| return continue_url | |
| except Exception as e: | |
| self._log(f"选择 Workspace 失败: {e}", "error") | |
| return None | |
| def _follow_redirects(self, start_url: str) -> Optional[str]: | |
| """跟随重定向链，寻找回调 URL""" | |
| try: | |
| current_url = start_url | |
| max_redirects = 6 | |
| for i in range(max_redirects): | |
| self._log(f"重定向 {i+1}/{max_redirects}: {current_url[:100]}...") | |
| response = self.session.get( | |
| current_url, | |
| allow_redirects=False, | |
| timeout=15 | |
| ) | |
| location = response.headers.get("Location") or "" | |
| # 如果不是重定向状态码，停止 | |
| 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) | |
| # 检查是否包含回调参数 | |
| if "code=" in next_url and "state=" in next_url: | |
| self._log(f"找到回调 URL: {next_url[:100]}...") | |
| return next_url | |
| current_url = next_url | |
| self._log("未能在重定向链中找到回调 URL", "error") | |
| return None | |
| except Exception as e: | |
| self._log(f"跟随重定向失败: {e}", "error") | |
| return None | |
| 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(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._log("=" * 60) | |
| self._log("注册流程启动，开始替你敲门") | |
| self._log("=" * 60) | |
| # 1. 检查 IP 地理位置 | |
| self._log("1. 先看看这条网络从哪儿来，别一开局就站错片场...") | |
| 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. 开个新邮箱，准备收信...") | |
| if not self._create_email(): | |
| result.error_message = "创建邮箱失败" | |
| return result | |
| result.email = self.email | |
| # 3. 准备首轮授权流程 | |
| did, sen_token = self._prepare_authorize_flow("首次授权") | |
| if not did: | |
| result.error_message = "获取 Device ID 失败" | |
| return result | |
| if not sen_token: | |
| result.error_message = "Sentinel POW 验证失败" | |
| return result | |
| # 4. 提交注册入口邮箱 | |
| self._log("4. 递上邮箱，看看 OpenAI 这球怎么接...") | |
| 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. 设置密码，别让小偷偷笑...") | |
| password_ok, _ = self._register_password() | |
| if not password_ok: | |
| result.error_message = "注册密码失败" | |
| return result | |
| self._log("6. 催一下注册验证码出门，邮差该冲刺了...") | |
| if not self._send_verification_code(): | |
| result.error_message = "发送验证码失败" | |
| return result | |
| self._log("7. 等验证码飞来，邮箱请注意查收...") | |
| code = self._get_verification_code() | |
| if not code: | |
| result.error_message = "获取验证码失败" | |
| return result | |
| self._log("8. 对一下验证码，看看是不是本人...") | |
| if not self._validate_verification_code(code): | |
| result.error_message = "验证验证码失败" | |
| return result | |
| self._log("9. 给账号办个正式户口，名字写档案里...") | |
| if not self._create_user_account(): | |
| result.error_message = "创建用户账户失败" | |
| return result | |
| login_ready, login_error = self._restart_login_flow() | |
| if not login_ready: | |
| result.error_message = login_error | |
| return result | |
| if not self._complete_token_exchange(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"Account ID: {result.account_id}") | |
| self._log(f"Workspace ID: {result.workspace_id}") | |
| self._log("=" * 60) | |
| result.success = True | |
| 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, | |
| } | |
| return result | |
| except Exception as e: | |
| self._log(f"注册过程中发生未预期错误: {e}", "error") | |
| result.error_message = str(e) | |
| return result | |
| def save_to_database(self, result: RegistrationResult) -> 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, | |
| 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 | |
| ) | |
| self._log(f"账户已存进数据库，落袋为安，ID: {account.id}") | |
| return True | |
| except Exception as e: | |
| self._log(f"保存到数据库失败: {e}", "error") | |
| return False | |