#!/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", ]