281 lines
8.2 KiB
Python
281 lines
8.2 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
ChatGPT Plus 支付账单获取模块
|
||
|
||
功能:
|
||
- 使用 access_token 自动获取 Stripe 支付会话
|
||
- 返回 checkout_session_id 和 client_secret 用于后续支付
|
||
|
||
使用方法:
|
||
from core.checkout import CheckoutFlow
|
||
|
||
session = OAISession(proxy=proxy)
|
||
await session.login(email, password)
|
||
|
||
checkout = CheckoutFlow(session)
|
||
result = await checkout.create_checkout_session()
|
||
"""
|
||
|
||
from typing import Dict, Any, Optional
|
||
from utils.logger import logger
|
||
|
||
|
||
class CheckoutFlow:
|
||
"""
|
||
ChatGPT Plus 支付结账流程
|
||
|
||
通过 /backend-api/payments/checkout 接口获取 Stripe 支付会话
|
||
"""
|
||
|
||
# API 端点
|
||
CHECKOUT_URL = "https://chatgpt.com/backend-api/payments/checkout"
|
||
|
||
# 支持的订阅计划
|
||
PLAN_PLUS = "chatgptplusplan"
|
||
PLAN_PRO = "chatgptproplan"
|
||
PLAN_TEAM = "chatgptteamplan"
|
||
|
||
def __init__(
|
||
self,
|
||
session,
|
||
plan_name: str = "chatgptplusplan",
|
||
country: str = "US",
|
||
currency: str = "USD"
|
||
):
|
||
"""
|
||
初始化结账流程
|
||
|
||
参数:
|
||
session: OAISession 实例（需已登录）
|
||
plan_name: 订阅计划名称
|
||
- chatgptplusplan: ChatGPT Plus ($20/月)
|
||
- chatgptproplan: ChatGPT Pro ($200/月)
|
||
country: 账单国家代码
|
||
currency: 货币代码
|
||
"""
|
||
self.session = session
|
||
self.plan_name = plan_name
|
||
self.country = country
|
||
self.currency = currency
|
||
|
||
# 验证登录状态
|
||
if not session.is_logged_in():
|
||
logger.warning("Session not logged in - checkout may fail")
|
||
|
||
async def create_checkout_session(
|
||
self,
|
||
checkout_ui_mode: str = "redirect"
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
创建 Stripe 结账会话
|
||
|
||
参数:
|
||
checkout_ui_mode: 结账UI模式
|
||
- "redirect": 重定向到 Stripe 托管页面
|
||
- "custom": 使用自定义嵌入式表单
|
||
|
||
返回:
|
||
成功时返回:
|
||
{
|
||
"status": "success",
|
||
"checkout_session_id": "cs_live_xxx",
|
||
"client_secret": "cs_live_xxx_secret_xxx",
|
||
"publishable_key": "pk_live_xxx",
|
||
"plan_name": "chatgptplusplan",
|
||
"payment_status": "unpaid",
|
||
"url": "https://checkout.stripe.com/..." (如果 ui_mode=redirect)
|
||
}
|
||
|
||
失败时返回:
|
||
{
|
||
"status": "failed",
|
||
"error": "错误信息",
|
||
"status_code": HTTP状态码
|
||
}
|
||
"""
|
||
logger.info(f"Creating checkout session for plan: {self.plan_name}")
|
||
|
||
# 构建请求体
|
||
payload = {
|
||
"plan_name": self.plan_name,
|
||
"billing_details": {
|
||
"country": self.country,
|
||
"currency": self.currency
|
||
},
|
||
"checkout_ui_mode": checkout_ui_mode
|
||
}
|
||
|
||
# 构建请求头
|
||
headers = self._build_headers()
|
||
|
||
try:
|
||
response = self.session.post(
|
||
self.CHECKOUT_URL,
|
||
json=payload,
|
||
headers=headers
|
||
)
|
||
|
||
return self._parse_response(response)
|
||
|
||
except Exception as e:
|
||
logger.error(f"Checkout request failed: {e}")
|
||
return {
|
||
"status": "failed",
|
||
"error": str(e),
|
||
"message": f"Request exception: {type(e).__name__}"
|
||
}
|
||
|
||
def _build_headers(self) -> Dict[str, str]:
|
||
"""
|
||
构建结账请求头
|
||
|
||
包含必要的 Authorization 和 OpenAI 特定头
|
||
"""
|
||
headers = {
|
||
"Content-Type": "application/json",
|
||
"Accept": "*/*",
|
||
"Origin": "https://chatgpt.com",
|
||
"Referer": "https://chatgpt.com/",
|
||
"Oai-Language": "en-US",
|
||
"Oai-Device-Id": self.session.oai_did,
|
||
}
|
||
|
||
# 添加 Authorization 头（Bearer token）
|
||
if self.session.access_token:
|
||
headers["Authorization"] = f"Bearer {self.session.access_token}"
|
||
else:
|
||
logger.warning("No access_token available for Authorization header")
|
||
|
||
return headers
|
||
|
||
def _parse_response(self, response) -> Dict[str, Any]:
|
||
"""
|
||
解析结账响应
|
||
|
||
参数:
|
||
response: HTTP 响应对象
|
||
|
||
返回:
|
||
解析后的结果字典
|
||
"""
|
||
status_code = response.status_code
|
||
|
||
if status_code == 200:
|
||
try:
|
||
data = response.json()
|
||
|
||
result = {
|
||
"status": "success",
|
||
"tag": data.get("tag"),
|
||
"checkout_session_id": data.get("checkout_session_id"),
|
||
"client_secret": data.get("client_secret"),
|
||
"publishable_key": data.get("publishable_key"),
|
||
"processor_entity": data.get("processor_entity"),
|
||
"checkout_ui_mode": data.get("checkout_ui_mode"),
|
||
"plan_name": data.get("plan_name"),
|
||
"payment_status": data.get("payment_status"),
|
||
"url": data.get("url"),
|
||
"automatic_tax_enabled": data.get("automatic_tax_enabled"),
|
||
"requires_manual_approval": data.get("requires_manual_approval"),
|
||
}
|
||
|
||
logger.success(f"Checkout session created: {result['checkout_session_id'][:30]}...")
|
||
return result
|
||
|
||
except Exception as e:
|
||
logger.error(f"Failed to parse checkout response: {e}")
|
||
return {
|
||
"status": "failed",
|
||
"error": f"JSON parse error: {e}",
|
||
"raw_response": response.text[:500]
|
||
}
|
||
|
||
elif status_code == 401:
|
||
logger.error("Unauthorized - invalid or expired token")
|
||
return {
|
||
"status": "failed",
|
||
"error": "Unauthorized - token invalid or expired",
|
||
"status_code": 401
|
||
}
|
||
|
||
elif status_code == 403:
|
||
logger.error("Forbidden - possible Cloudflare block or permission issue")
|
||
return {
|
||
"status": "failed",
|
||
"error": "Forbidden - access denied",
|
||
"status_code": 403,
|
||
"raw_response": response.text[:500]
|
||
}
|
||
|
||
elif status_code == 400:
|
||
logger.error(f"Bad request: {response.text[:200]}")
|
||
return {
|
||
"status": "failed",
|
||
"error": "Bad request - invalid parameters",
|
||
"status_code": 400,
|
||
"raw_response": response.text[:500]
|
||
}
|
||
|
||
else:
|
||
logger.error(f"Unexpected status {status_code}: {response.text[:200]}")
|
||
return {
|
||
"status": "failed",
|
||
"error": f"HTTP {status_code}",
|
||
"status_code": status_code,
|
||
"raw_response": response.text[:500]
|
||
}
|
||
|
||
|
||
async def get_checkout_for_account(
|
||
email: str,
|
||
password: str,
|
||
proxy: str = None,
|
||
plan_name: str = "chatgptplusplan"
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
便捷函数：为指定账号获取支付链接
|
||
|
||
参数:
|
||
email: 账号邮箱
|
||
password: 账号密码
|
||
proxy: 代理地址（可选）
|
||
plan_name: 订阅计划
|
||
|
||
返回:
|
||
包含 checkout 信息的字典
|
||
"""
|
||
from core.session import OAISession
|
||
|
||
session = None
|
||
try:
|
||
session = OAISession(proxy=proxy, impersonate="chrome124")
|
||
|
||
# 登录
|
||
logger.info(f"Logging in as {email}...")
|
||
login_result = await session.login(email, password)
|
||
|
||
if login_result.get("status") != "success":
|
||
return {
|
||
"status": "failed",
|
||
"error": f"Login failed: {login_result.get('error')}",
|
||
"email": email
|
||
}
|
||
|
||
# 获取 checkout
|
||
checkout = CheckoutFlow(session, plan_name=plan_name)
|
||
result = await checkout.create_checkout_session()
|
||
result["email"] = email
|
||
|
||
return result
|
||
|
||
finally:
|
||
if session:
|
||
session.close()
|
||
|
||
|
||
# 导出
|
||
__all__ = [
|
||
"CheckoutFlow",
|
||
"get_checkout_for_account",
|
||
]
|