checkout
This commit is contained in:
dela
280
core/checkout.py
Normal file
280
core/checkout.py
Normal file
@@ -0,0 +1,280 @@
|
||||
#!/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",
|
||||
]
|
||||
Reference in New Issue
Block a user