628 lines
21 KiB
Python
628 lines
21 KiB
Python
"""
|
||
ChatGPT API 注册模块 (协议模式)
|
||
- 使用 curl_cffi 通过 API 快速完成注册
|
||
- 支持 Cookie 注入到浏览器完成支付
|
||
"""
|
||
|
||
import time
|
||
import random
|
||
import re
|
||
import uuid
|
||
from urllib.parse import unquote
|
||
|
||
try:
|
||
from curl_cffi import requests as curl_requests
|
||
CURL_CFFI_AVAILABLE = True
|
||
except ImportError:
|
||
CURL_CFFI_AVAILABLE = False
|
||
curl_requests = None
|
||
|
||
import requests
|
||
|
||
|
||
def log_status(step, message):
|
||
"""日志输出"""
|
||
timestamp = time.strftime("%H:%M:%S")
|
||
print(f"[{timestamp}] [{step}] {message}")
|
||
|
||
|
||
def log_progress(message):
|
||
"""进度输出"""
|
||
print(f" {message}")
|
||
|
||
|
||
def request_with_retry(func, *args, max_retries=3, **kwargs):
|
||
"""带重试的请求"""
|
||
for i in range(max_retries):
|
||
try:
|
||
return func(*args, **kwargs)
|
||
except Exception as e:
|
||
if i == max_retries - 1:
|
||
raise e
|
||
time.sleep(1)
|
||
|
||
|
||
class ChatGPTAPIRegister:
|
||
"""ChatGPT API 注册类 (协议模式)"""
|
||
|
||
def __init__(self, proxy=None):
|
||
"""初始化
|
||
|
||
Args:
|
||
proxy: 代理地址，如 "http://127.0.0.1:7890"
|
||
"""
|
||
if not CURL_CFFI_AVAILABLE:
|
||
raise ImportError("协议模式需要安装 curl_cffi: pip install curl_cffi")
|
||
|
||
self.session = curl_requests.Session(
|
||
impersonate="edge",
|
||
verify=False,
|
||
proxies={"http": proxy, "https": proxy} if proxy else {}
|
||
)
|
||
self.auth_session_logging_id = str(uuid.uuid4())
|
||
self.oai_did = ""
|
||
self.csrf_token = ""
|
||
self.authorize_url = ""
|
||
self.access_token = ""
|
||
|
||
def init_session(self) -> bool:
|
||
"""初始化会话，获取必要的 cookies 和 tokens"""
|
||
try:
|
||
resp = request_with_retry(self.session.get, "https://chatgpt.com")
|
||
if resp.status_code != 200:
|
||
log_progress(f"[X] 初始化失败: HTTP {resp.status_code}")
|
||
return False
|
||
|
||
self.oai_did = self.session.cookies.get("oai-did")
|
||
csrf_cookie = self.session.cookies.get("__Host-next-auth.csrf-token")
|
||
|
||
if csrf_cookie:
|
||
self.csrf_token = unquote(csrf_cookie).split("|")[0]
|
||
else:
|
||
log_progress("[X] 未获取到 CSRF token")
|
||
return False
|
||
|
||
# 访问登录页面
|
||
request_with_retry(
|
||
self.session.get,
|
||
f"https://chatgpt.com/auth/login?openaicom-did={self.oai_did}"
|
||
)
|
||
return True
|
||
except Exception as e:
|
||
log_progress(f"[X] 初始化异常: {e}")
|
||
return False
|
||
|
||
def get_authorize_url(self, email: str) -> bool:
|
||
"""获取授权 URL"""
|
||
try:
|
||
url = f"https://chatgpt.com/api/auth/signin/openai?prompt=login&ext-oai-did={self.oai_did}&auth_session_logging_id={self.auth_session_logging_id}&screen_hint=login_or_signup&login_hint={email}"
|
||
payload = {
|
||
"callbackUrl": "https://chatgpt.com/",
|
||
"csrfToken": self.csrf_token,
|
||
"json": "true"
|
||
}
|
||
resp = request_with_retry(
|
||
self.session.post, url, data=payload,
|
||
headers={"Origin": "https://chatgpt.com"}
|
||
)
|
||
data = resp.json()
|
||
|
||
if data.get("url") and "auth.openai.com" in data["url"]:
|
||
self.authorize_url = data["url"]
|
||
return True
|
||
|
||
log_progress(f"[X] 授权 URL 无效: {data}")
|
||
return False
|
||
except Exception as e:
|
||
log_progress(f"[X] 获取授权 URL 异常: {e}")
|
||
return False
|
||
|
||
def start_authorize(self) -> bool:
|
||
"""启动授权流程"""
|
||
try:
|
||
resp = request_with_retry(
|
||
self.session.get, self.authorize_url, allow_redirects=True
|
||
)
|
||
return "create-account" in resp.url or "log-in" in resp.url
|
||
except Exception as e:
|
||
log_progress(f"[X] 启动授权异常: {e}")
|
||
return False
|
||
|
||
def register(self, email: str, password: str) -> bool:
|
||
"""注册账户"""
|
||
try:
|
||
resp = request_with_retry(
|
||
self.session.post,
|
||
"https://auth.openai.com/api/accounts/user/register",
|
||
json={"password": password, "username": email},
|
||
headers={
|
||
"Content-Type": "application/json",
|
||
"Origin": "https://auth.openai.com"
|
||
}
|
||
)
|
||
if resp.status_code == 200:
|
||
return True
|
||
log_progress(f"[X] 注册失败: {resp.status_code} - {resp.text[:200]}")
|
||
return False
|
||
except Exception as e:
|
||
log_progress(f"[X] 注册异常: {e}")
|
||
return False
|
||
|
||
def send_verification_email(self) -> bool:
|
||
"""发送验证邮件"""
|
||
try:
|
||
resp = request_with_retry(
|
||
self.session.get,
|
||
"https://auth.openai.com/api/accounts/email-otp/send",
|
||
allow_redirects=True
|
||
)
|
||
return resp.status_code == 200
|
||
except Exception as e:
|
||
log_progress(f"[X] 发送验证邮件异常: {e}")
|
||
return False
|
||
|
||
def validate_otp(self, otp_code: str) -> bool:
|
||
"""验证 OTP 码"""
|
||
try:
|
||
resp = request_with_retry(
|
||
self.session.post,
|
||
"https://auth.openai.com/api/accounts/email-otp/validate",
|
||
json={"code": otp_code},
|
||
headers={
|
||
"Content-Type": "application/json",
|
||
"Origin": "https://auth.openai.com"
|
||
}
|
||
)
|
||
if resp.status_code == 200:
|
||
return True
|
||
log_progress(f"[X] OTP 验证失败: {resp.status_code} - {resp.text[:200]}")
|
||
return False
|
||
except Exception as e:
|
||
log_progress(f"[X] OTP 验证异常: {e}")
|
||
return False
|
||
|
||
def create_account(self, name: str, birthdate: str) -> bool:
|
||
"""创建账户 (填写姓名和生日)
|
||
|
||
Args:
|
||
name: 姓名
|
||
birthdate: 生日，格式 "YYYY-MM-DD"
|
||
"""
|
||
try:
|
||
resp = request_with_retry(
|
||
self.session.post,
|
||
"https://auth.openai.com/api/accounts/create_account",
|
||
json={"name": name, "birthdate": birthdate},
|
||
headers={
|
||
"Content-Type": "application/json",
|
||
"Origin": "https://auth.openai.com"
|
||
}
|
||
)
|
||
if resp.status_code != 200:
|
||
log_progress(f"[X] 创建账户失败: {resp.status_code} - {resp.text[:200]}")
|
||
return False
|
||
|
||
data = resp.json()
|
||
continue_url = data.get("continue_url")
|
||
if continue_url:
|
||
request_with_retry(self.session.get, continue_url, allow_redirects=True)
|
||
return True
|
||
except Exception as e:
|
||
log_progress(f"[X] 创建账户异常: {e}")
|
||
return False
|
||
|
||
def login(self, email: str, password: str) -> bool:
|
||
"""使用密码登录 (复用注册时建立的会话)"""
|
||
try:
|
||
resp = request_with_retry(
|
||
self.session.post,
|
||
"https://auth.openai.com/api/accounts/password/verify",
|
||
json={"username": email, "password": password},
|
||
headers={
|
||
"Content-Type": "application/json",
|
||
"Origin": "https://auth.openai.com"
|
||
}
|
||
)
|
||
|
||
if resp.status_code == 200:
|
||
data = resp.json()
|
||
continue_url = data.get("continue_url")
|
||
if continue_url:
|
||
request_with_retry(self.session.get, continue_url, allow_redirects=True)
|
||
return True
|
||
|
||
log_progress(f"[X] 登录失败: {resp.status_code} - {resp.text[:200]}")
|
||
return False
|
||
except Exception as e:
|
||
log_progress(f"[X] 登录异常: {e}")
|
||
return False
|
||
|
||
def get_session_token(self) -> str:
|
||
"""获取 access token"""
|
||
try:
|
||
resp = request_with_retry(self.session.get, "https://chatgpt.com/api/auth/session")
|
||
if resp.status_code == 200:
|
||
data = resp.json()
|
||
token = data.get("accessToken")
|
||
if token:
|
||
self.access_token = token
|
||
return token
|
||
log_progress(f"[X] Session 响应无 token: {str(data)[:200]}")
|
||
else:
|
||
log_progress(f"[X] Session 请求失败: {resp.status_code}")
|
||
return ""
|
||
except Exception as e:
|
||
log_progress(f"[X] 获取 token 异常: {e}")
|
||
return ""
|
||
|
||
def get_checkout_url(self) -> str:
|
||
"""通过 API 获取支付页 URL"""
|
||
try:
|
||
token = self.access_token or self.get_session_token()
|
||
if not token:
|
||
log_progress("[X] 无法获取 access token")
|
||
return ""
|
||
|
||
payload = {
|
||
"plan_name": "chatgptteamplan",
|
||
"team_plan_data": {
|
||
"workspace_name": "Sepa",
|
||
"price_interval": "month",
|
||
"seat_quantity": 5
|
||
},
|
||
"billing_details": {
|
||
"country": "DE",
|
||
"currency": "EUR"
|
||
},
|
||
"promo_campaign": {
|
||
"promo_campaign_id": "team-1-month-free",
|
||
"is_coupon_from_query_param": True
|
||
},
|
||
"checkout_ui_mode": "redirect"
|
||
}
|
||
|
||
resp = request_with_retry(
|
||
self.session.post,
|
||
"https://chatgpt.com/backend-api/payments/checkout",
|
||
json=payload,
|
||
headers={
|
||
"Authorization": f"Bearer {token}",
|
||
"Content-Type": "application/json",
|
||
"Origin": "https://chatgpt.com"
|
||
}
|
||
)
|
||
|
||
if resp.status_code == 200:
|
||
data = resp.json()
|
||
checkout_url = data.get("url")
|
||
if checkout_url:
|
||
return checkout_url
|
||
log_progress(f"[X] 响应无 URL: {resp.text[:200]}")
|
||
else:
|
||
log_progress(f"[X] 获取支付页失败: {resp.status_code} - {resp.text[:200]}")
|
||
return ""
|
||
except Exception as e:
|
||
log_progress(f"[X] 获取支付页异常: {e}")
|
||
return ""
|
||
|
||
def get_cookies(self) -> list:
|
||
"""获取所有 cookies 用于注入浏览器"""
|
||
cookies = []
|
||
for cookie in self.session.cookies.jar:
|
||
cookies.append({
|
||
"name": cookie.name,
|
||
"value": cookie.value,
|
||
"domain": cookie.domain,
|
||
"path": cookie.path or "/",
|
||
"secure": cookie.secure,
|
||
})
|
||
return cookies
|
||
|
||
|
||
def get_verification_code_api(target_email: str, mail_api_base: str, mail_api_token: str, max_retries: int = 90) -> str:
|
||
"""通过 API 获取验证码
|
||
|
||
Args:
|
||
target_email: 目标邮箱
|
||
mail_api_base: 邮件 API 地址
|
||
mail_api_token: 邮件 API Token
|
||
max_retries: 最大重试次数
|
||
|
||
Returns:
|
||
str: 验证码，失败返回空字符串
|
||
"""
|
||
log_status("API监听", "正在监听邮件...")
|
||
headers = {"Authorization": mail_api_token, "Content-Type": "application/json"}
|
||
start_time = time.time()
|
||
|
||
for i in range(max_retries):
|
||
elapsed = int(time.time() - start_time)
|
||
try:
|
||
url = f"{mail_api_base}/api/public/emailList"
|
||
payload = {"toEmail": target_email, "timeSort": "desc", "size": 20}
|
||
resp = requests.post(url, headers=headers, json=payload, timeout=10)
|
||
|
||
if resp.status_code == 200:
|
||
data = resp.json()
|
||
if data.get('code') == 200:
|
||
mails = data.get('data', [])
|
||
if mails:
|
||
for mail in mails:
|
||
html_body = mail.get('content') or mail.get('text') or str(mail)
|
||
code_match = re.search(r'\b(\d{6})\b', html_body)
|
||
if code_match:
|
||
code = code_match.group(1)
|
||
log_status("捕获", f"[OK] 提取到验证码: {code}")
|
||
return code
|
||
except Exception:
|
||
pass
|
||
|
||
if i % 5 == 0:
|
||
print(f" [监听中] 已耗时 {elapsed}秒...")
|
||
time.sleep(2)
|
||
|
||
log_status("超时", "[X] 未能获取验证码")
|
||
return ""
|
||
|
||
|
||
def api_register_flow(
|
||
email: str,
|
||
password: str,
|
||
real_name: str,
|
||
birthdate: str,
|
||
mail_api_base: str,
|
||
mail_api_token: str,
|
||
proxy: str = None,
|
||
progress_callback=None
|
||
) -> ChatGPTAPIRegister:
|
||
"""执行 API 注册流程
|
||
|
||
Args:
|
||
email: 邮箱
|
||
password: 密码
|
||
real_name: 姓名
|
||
birthdate: 生日 (YYYY-MM-DD)
|
||
mail_api_base: 邮件 API 地址
|
||
mail_api_token: 邮件 API Token
|
||
proxy: 代理地址
|
||
progress_callback: 进度回调
|
||
|
||
Returns:
|
||
ChatGPTAPIRegister: 成功返回 reg 对象，失败返回 None
|
||
"""
|
||
def log_cb(msg):
|
||
if progress_callback:
|
||
progress_callback(msg)
|
||
else:
|
||
log_progress(msg)
|
||
|
||
reg = ChatGPTAPIRegister(proxy=proxy)
|
||
|
||
try:
|
||
log_status("API注册", "初始化会话...")
|
||
if not reg.init_session():
|
||
log_cb("[X] 初始化失败")
|
||
return None
|
||
log_cb("[OK] 会话初始化成功")
|
||
|
||
log_status("API注册", "获取授权 URL...")
|
||
if not reg.get_authorize_url(email):
|
||
log_cb("[X] 获取授权 URL 失败")
|
||
return None
|
||
log_cb("[OK] 授权 URL 获取成功")
|
||
|
||
log_status("API注册", "开始授权流程...")
|
||
if not reg.start_authorize():
|
||
log_cb("[X] 授权流程启动失败")
|
||
return None
|
||
log_cb("[OK] 授权流程已启动")
|
||
|
||
log_status("API注册", "注册账户...")
|
||
if not reg.register(email, password):
|
||
log_cb("[X] 注册失败")
|
||
return None
|
||
log_cb("[OK] 账户注册成功")
|
||
|
||
log_status("API注册", "发送验证邮件...")
|
||
if not reg.send_verification_email():
|
||
log_cb("[X] 发送验证邮件失败")
|
||
return None
|
||
log_cb("[OK] 验证邮件已发送")
|
||
|
||
# 获取验证码
|
||
otp_code = get_verification_code_api(email, mail_api_base, mail_api_token)
|
||
if not otp_code:
|
||
log_cb("[X] 未能获取验证码")
|
||
return None
|
||
|
||
log_status("API注册", f"验证 OTP: {otp_code}")
|
||
if not reg.validate_otp(otp_code):
|
||
log_cb("[X] OTP 验证失败")
|
||
return None
|
||
log_cb("[OK] OTP 验证成功")
|
||
|
||
log_status("API注册", "创建账户...")
|
||
if not reg.create_account(real_name, birthdate):
|
||
log_cb("[X] 创建账户失败")
|
||
return None
|
||
log_cb("[OK] 账户创建成功")
|
||
|
||
# 验证 session 是否有效
|
||
token = reg.get_session_token()
|
||
if token:
|
||
log_cb(f"[OK] Session 有效，Token: {token[:30]}...")
|
||
else:
|
||
log_cb("[!] 注册完成但 session 可能未完全建立")
|
||
|
||
return reg
|
||
|
||
except Exception as e:
|
||
log_status("错误", f"注册异常: {e}")
|
||
return None
|
||
|
||
|
||
def api_login_flow(
|
||
email: str,
|
||
password: str,
|
||
proxy: str = None,
|
||
progress_callback=None
|
||
) -> ChatGPTAPIRegister:
|
||
"""执行 API 登录流程
|
||
|
||
Args:
|
||
email: 邮箱
|
||
password: 密码
|
||
proxy: 代理地址
|
||
progress_callback: 进度回调
|
||
|
||
Returns:
|
||
ChatGPTAPIRegister: 成功返回 reg 对象，失败返回 None
|
||
"""
|
||
def log_cb(msg):
|
||
if progress_callback:
|
||
progress_callback(msg)
|
||
else:
|
||
log_progress(msg)
|
||
|
||
reg = ChatGPTAPIRegister(proxy=proxy)
|
||
|
||
try:
|
||
log_status("API登录", "初始化会话...")
|
||
if not reg.init_session():
|
||
log_cb("[X] 初始化失败")
|
||
return None
|
||
log_cb("[OK] 初始化成功")
|
||
|
||
log_status("API登录", "获取授权 URL...")
|
||
if not reg.get_authorize_url(email):
|
||
log_cb("[X] 获取授权 URL 失败")
|
||
return None
|
||
log_cb("[OK] 获取授权 URL 成功")
|
||
|
||
log_status("API登录", "开始授权流程...")
|
||
if not reg.start_authorize():
|
||
log_cb("[X] 授权流程失败")
|
||
return None
|
||
log_cb("[OK] 授权流程成功")
|
||
|
||
log_status("API登录", "密码验证...")
|
||
if not reg.login(email, password):
|
||
log_cb("[X] 登录失败")
|
||
return None
|
||
log_cb("[OK] 登录成功")
|
||
|
||
# 获取 token
|
||
token = reg.get_session_token()
|
||
if token:
|
||
log_status("API登录", f"Token: {token[:50]}...")
|
||
|
||
return reg
|
||
|
||
except Exception as e:
|
||
log_status("错误", f"登录异常: {e}")
|
||
return None
|
||
|
||
|
||
def is_api_mode_available() -> bool:
|
||
"""检查协议模式是否可用"""
|
||
return CURL_CFFI_AVAILABLE
|
||
|
||
|
||
def api_register_account_only(
|
||
email: str,
|
||
password: str,
|
||
real_name: str,
|
||
birthdate: str,
|
||
get_verification_code_func,
|
||
proxy: str = None,
|
||
progress_callback=None
|
||
) -> bool:
|
||
"""仅执行 API 注册流程（不含支付，用于邀请邮箱注册）
|
||
|
||
Args:
|
||
email: 邮箱
|
||
password: 密码
|
||
real_name: 姓名
|
||
birthdate: 生日 (YYYY-MM-DD)
|
||
get_verification_code_func: 获取验证码的函数，签名: func(email) -> str
|
||
proxy: 代理地址
|
||
progress_callback: 进度回调
|
||
|
||
Returns:
|
||
bool: 是否注册成功
|
||
"""
|
||
def log_cb(msg):
|
||
if progress_callback:
|
||
progress_callback(msg)
|
||
else:
|
||
log_progress(msg)
|
||
|
||
if not CURL_CFFI_AVAILABLE:
|
||
log_status("错误", "协议模式不可用，请安装 curl_cffi")
|
||
return False
|
||
|
||
reg = ChatGPTAPIRegister(proxy=proxy)
|
||
|
||
try:
|
||
log_status("API注册", "初始化会话...")
|
||
if not reg.init_session():
|
||
log_cb("[X] 初始化失败")
|
||
return False
|
||
log_cb("[OK] 会话初始化成功")
|
||
|
||
log_status("API注册", "获取授权 URL...")
|
||
if not reg.get_authorize_url(email):
|
||
log_cb("[X] 获取授权 URL 失败")
|
||
return False
|
||
log_cb("[OK] 授权 URL 获取成功")
|
||
|
||
log_status("API注册", "开始授权流程...")
|
||
if not reg.start_authorize():
|
||
log_cb("[X] 授权流程启动失败")
|
||
return False
|
||
log_cb("[OK] 授权流程已启动")
|
||
|
||
log_status("API注册", "注册账户...")
|
||
if not reg.register(email, password):
|
||
log_cb("[X] 注册失败")
|
||
return False
|
||
log_cb("[OK] 账户注册成功")
|
||
|
||
log_status("API注册", "发送验证邮件...")
|
||
if not reg.send_verification_email():
|
||
log_cb("[X] 发送验证邮件失败")
|
||
return False
|
||
log_cb("[OK] 验证邮件已发送")
|
||
|
||
# 使用传入的函数获取验证码
|
||
log_status("API注册", "等待验证码...")
|
||
otp_code = get_verification_code_func(email)
|
||
if not otp_code:
|
||
log_cb("[X] 未能获取验证码")
|
||
return False
|
||
|
||
log_status("API注册", f"验证 OTP: {otp_code}")
|
||
if not reg.validate_otp(otp_code):
|
||
log_cb("[X] OTP 验证失败")
|
||
return False
|
||
log_cb("[OK] OTP 验证成功")
|
||
|
||
log_status("API注册", "创建账户...")
|
||
if not reg.create_account(real_name, birthdate):
|
||
log_cb("[X] 创建账户失败")
|
||
return False
|
||
log_cb("[OK] 账户创建成功")
|
||
|
||
# 验证 session 是否有效
|
||
token = reg.get_session_token()
|
||
if token:
|
||
log_cb(f"[OK] 注册完成，Token: {token[:30]}...")
|
||
return True
|
||
else:
|
||
log_cb("[!] 注册完成但 session 可能未完全建立")
|
||
return True # 仍然返回成功，因为注册流程已完成
|
||
|
||
except Exception as e:
|
||
log_status("错误", f"注册异常: {e}")
|
||
return False
|