393 lines
14 KiB
Python
393 lines
14 KiB
Python
"""
|
||
OpenAI Stripe Payment Automation
|
||
SEPA支付方式自动化模块 - 使用 curl_cffi HTTPClient
|
||
"""
|
||
|
||
import uuid
|
||
import time
|
||
import urllib.parse
|
||
from typing import Dict, Optional
|
||
import logging
|
||
from .http_client import HTTPClient
|
||
from .fingerprint import BrowserFingerprint
|
||
|
||
logging.basicConfig(level=logging.INFO)
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class StripePaymentHandler:
|
||
"""Stripe支付处理器 - 适用于OpenAI的SEPA支付"""
|
||
|
||
def __init__(self, checkout_session_url: str, http_client: Optional[HTTPClient] = None):
|
||
"""
|
||
初始化支付处理器
|
||
|
||
Args:
|
||
checkout_session_url: Stripe checkout的完整URL
|
||
例如: https://pay.openai.com/c/pay/cs_live_xxx#xxx
|
||
http_client: 可选的HTTPClient实例（共享session和cookies）
|
||
"""
|
||
self.checkout_session_url = checkout_session_url
|
||
self.session_id = self._extract_session_id(checkout_session_url)
|
||
|
||
# Stripe配置
|
||
self.stripe_public_key = "pk_live_51Pj377KslHRdbaPgTJYjThzH3f5dt1N1vK7LUp0qh0yNSarhfZ6nfbG7FFlh8KLxVkvdMWN5o6Mc4Vda6NHaSnaV00C2Sbl8Zs"
|
||
self.stripe_api_base = "https://api.stripe.com"
|
||
self.stripe_version = "2020-08-27;custom_checkout_beta=v1"
|
||
|
||
# 会话指纹（每次运行生成新的）
|
||
self.guid = str(uuid.uuid4()) + str(uuid.uuid4())[:8]
|
||
self.muid = str(uuid.uuid4()) + str(uuid.uuid4())[:6]
|
||
self.sid = str(uuid.uuid4()) + str(uuid.uuid4())[:6]
|
||
|
||
# 归因元数据
|
||
self.client_session_id = str(uuid.uuid4())
|
||
self.checkout_config_id = "9e2d84a8-5eec-41bf-aae8-24d59824ec84"
|
||
|
||
# HTTP客户端（使用curl_cffi）
|
||
if http_client:
|
||
self.http_client = http_client
|
||
else:
|
||
# 创建新的HTTP客户端
|
||
fingerprint = BrowserFingerprint()
|
||
self.http_client = HTTPClient(fingerprint)
|
||
|
||
def _extract_session_id(self, url: str) -> str:
|
||
"""从URL中提取session ID"""
|
||
# cs_live_xxx 格式
|
||
if "cs_live_" in url:
|
||
start = url.find("cs_live_")
|
||
end = url.find("#", start) if "#" in url[start:] else url.find("?", start)
|
||
if end == -1:
|
||
end = len(url)
|
||
return url[start:end]
|
||
raise ValueError("无法从URL中提取session_id")
|
||
|
||
def create_payment_method(
|
||
self,
|
||
iban: str,
|
||
name: str,
|
||
email: str,
|
||
address_line1: str,
|
||
city: str,
|
||
postal_code: str,
|
||
state: str,
|
||
country: str = "US"
|
||
) -> Optional[str]:
|
||
"""
|
||
创建支付方式
|
||
|
||
Args:
|
||
iban: 德国IBAN账号（例如：DE89370400440532013000）
|
||
name: 账户持有人姓名
|
||
email: 邮箱地址
|
||
address_line1: 地址第一行
|
||
city: 城市
|
||
postal_code: 邮编
|
||
state: 州/省（美国地址需要）
|
||
country: 国家代码（默认US）
|
||
|
||
Returns:
|
||
payment_method_id (pm_xxx) 或 None
|
||
"""
|
||
url = f"{self.stripe_api_base}/v1/payment_methods"
|
||
|
||
data = {
|
||
"type": "sepa_debit",
|
||
"sepa_debit[iban]": iban,
|
||
"billing_details[name]": name,
|
||
"billing_details[email]": email,
|
||
"billing_details[address][country]": country,
|
||
"billing_details[address][line1]": address_line1,
|
||
"billing_details[address][city]": city,
|
||
"billing_details[address][postal_code]": postal_code,
|
||
"billing_details[address][state]": state,
|
||
|
||
# 指纹追踪
|
||
"guid": self.guid,
|
||
"muid": self.muid,
|
||
"sid": self.sid,
|
||
|
||
# Stripe配置
|
||
"_stripe_version": self.stripe_version,
|
||
"key": self.stripe_public_key,
|
||
"payment_user_agent": "stripe.js/f4aa9d6f0f; stripe-js-v3/f4aa9d6f0f; checkout",
|
||
|
||
# 归因元数据
|
||
"client_attribution_metadata[client_session_id]": self.client_session_id,
|
||
"client_attribution_metadata[checkout_session_id]": self.session_id,
|
||
"client_attribution_metadata[merchant_integration_source]": "checkout",
|
||
"client_attribution_metadata[merchant_integration_version]": "hosted_checkout",
|
||
"client_attribution_metadata[payment_method_selection_flow]": "automatic",
|
||
"client_attribution_metadata[checkout_config_id]": self.checkout_config_id,
|
||
}
|
||
|
||
try:
|
||
logger.info(f"Creating payment method with IBAN: {iban[:8]}****{iban[-4:]}")
|
||
|
||
headers = self.http_client.fingerprint.get_headers(host='api.stripe.com')
|
||
headers.update({
|
||
"Content-Type": "application/x-www-form-urlencoded",
|
||
"Accept": "application/json",
|
||
"Origin": "https://pay.openai.com",
|
||
"Referer": "https://pay.openai.com/",
|
||
"Sec-Fetch-Site": "cross-site",
|
||
"Sec-Fetch-Mode": "cors",
|
||
"Sec-Fetch-Dest": "empty",
|
||
"Sec-Ch-Ua-Platform": '"Linux"',
|
||
"Accept-Language": "zh-CN,zh;q=0.9",
|
||
"Priority": "u=1, i"
|
||
})
|
||
|
||
response = self.http_client.session.post(
|
||
url,
|
||
data=data,
|
||
headers=headers,
|
||
timeout=30
|
||
)
|
||
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
payment_method_id = result.get("id")
|
||
logger.info(f"✅ Payment method created: {payment_method_id}")
|
||
logger.info(f"Bank code: {result.get('sepa_debit', {}).get('bank_code')}")
|
||
logger.info(f"Last4: {result.get('sepa_debit', {}).get('last4')}")
|
||
return payment_method_id
|
||
else:
|
||
logger.error(f"❌ Failed to create payment method: {response.status_code}")
|
||
logger.error(response.text)
|
||
return None
|
||
|
||
except Exception as e:
|
||
logger.error(f"❌ Exception creating payment method: {e}")
|
||
return None
|
||
|
||
def confirm_payment(
|
||
self,
|
||
payment_method_id: str,
|
||
captcha_token: Optional[str] = None
|
||
) -> bool:
|
||
"""
|
||
确认支付
|
||
|
||
Args:
|
||
payment_method_id: 支付方式ID（pm_xxx）
|
||
captcha_token: hCaptcha token（可选，如果需要人机验证）
|
||
|
||
Returns:
|
||
是否成功
|
||
"""
|
||
url = f"{self.stripe_api_base}/v1/payment_pages/{self.session_id}/confirm"
|
||
|
||
data = {
|
||
"eid": "NA",
|
||
"payment_method": payment_method_id,
|
||
"expected_amount": "0", # OpenAI Team通常是0初始金额
|
||
"consent[terms_of_service]": "accepted",
|
||
"expected_payment_method_type": "sepa_debit",
|
||
|
||
# Stripe配置
|
||
"_stripe_version": self.stripe_version,
|
||
"guid": self.guid,
|
||
"muid": self.muid,
|
||
"sid": self.sid,
|
||
"key": self.stripe_public_key,
|
||
"version": "f4aa9d6f0f",
|
||
|
||
# 校验和（这些值可能需要从页面JS中动态获取）
|
||
"init_checksum": "1i2GM0P7eFI4XpRyWa9ffzqQE4sToFkA",
|
||
"js_checksum": urllib.parse.quote("qto~d^n0=QU>azbu]blvv<\\v@=l`<cdbovabU&ov;;mOP$dNo?U^`w"),
|
||
|
||
# 归因元数据
|
||
"client_attribution_metadata[client_session_id]": self.client_session_id,
|
||
"client_attribution_metadata[checkout_session_id]": self.session_id,
|
||
"client_attribution_metadata[merchant_integration_source]": "checkout",
|
||
"client_attribution_metadata[merchant_integration_version]": "hosted_checkout",
|
||
"client_attribution_metadata[payment_method_selection_flow]": "automatic",
|
||
"client_attribution_metadata[checkout_config_id]": self.checkout_config_id,
|
||
}
|
||
|
||
# 如果有验证码token
|
||
if captcha_token:
|
||
data["passive_captcha_token"] = captcha_token
|
||
data["passive_captcha_ekey"] = ""
|
||
data["rv_timestamp"] = urllib.parse.quote("qto>n<Q=U&CyY&`>X^r<YNr<YN`<Y_C<Y_C<Y^`zY_`<Y^n{U>o&U&Cyd&QveO$sX=X<d&Yv[bdD[_YrY&YyY&##Y_YrYxdDY&X#dbQv[OMrd%n{U>e&U&CyX_\\#YO\\>Y&L$[OP>Y&oue>OuYxP>e=d;Y=QsX&\\<eRnDd=X;YOMuXxQsX=n<d_`#X&dDY&L#XxordbYyeRYsY%o?U^`w")
|
||
|
||
try:
|
||
logger.info(f"Confirming payment with method: {payment_method_id}")
|
||
|
||
headers = self.http_client.fingerprint.get_headers(host='api.stripe.com')
|
||
headers.update({
|
||
"Content-Type": "application/x-www-form-urlencoded",
|
||
"Accept": "application/json",
|
||
"Origin": "https://pay.openai.com",
|
||
"Referer": "https://pay.openai.com/",
|
||
"Sec-Fetch-Site": "cross-site",
|
||
"Sec-Fetch-Mode": "cors",
|
||
"Sec-Fetch-Dest": "empty",
|
||
"Sec-Ch-Ua-Platform": '"Linux"',
|
||
"Accept-Language": "zh-CN,zh;q=0.9",
|
||
"Priority": "u=1, i"
|
||
})
|
||
|
||
response = self.http_client.session.post(
|
||
url,
|
||
data=data,
|
||
headers=headers,
|
||
timeout=30
|
||
)
|
||
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
state = result.get("state")
|
||
logger.info(f"✅ Payment confirmation response state: {state}")
|
||
|
||
# 检查setup_intent状态
|
||
setup_intent = result.get("setup_intent", {})
|
||
if setup_intent.get("status") == "succeeded":
|
||
logger.info("✅ Setup intent succeeded")
|
||
return True
|
||
|
||
# 检查客户ID
|
||
customer = result.get("customer", {})
|
||
if customer.get("id"):
|
||
logger.info(f"✅ Customer created: {customer.get('id')}")
|
||
return True
|
||
|
||
return state in ["processing_subscription", "succeeded"]
|
||
else:
|
||
logger.error(f"❌ Failed to confirm payment: {response.status_code}")
|
||
logger.error(response.text)
|
||
return False
|
||
|
||
except Exception as e:
|
||
logger.error(f"❌ Exception confirming payment: {e}")
|
||
return False
|
||
|
||
def poll_payment_status(self, max_attempts: int = 20, interval: int = 3) -> Dict:
|
||
"""
|
||
轮询支付状态直到完成
|
||
|
||
Args:
|
||
max_attempts: 最大轮询次数
|
||
interval: 轮询间隔（秒）
|
||
|
||
Returns:
|
||
最终状态字典
|
||
"""
|
||
url = f"{self.stripe_api_base}/v1/payment_pages/{self.session_id}/poll"
|
||
params = {"key": self.stripe_public_key}
|
||
|
||
for attempt in range(max_attempts):
|
||
try:
|
||
logger.info(f"Polling payment status (attempt {attempt + 1}/{max_attempts})...")
|
||
|
||
headers = self.http_client.fingerprint.get_headers(host='api.stripe.com')
|
||
headers.update({
|
||
"Accept": "application/json",
|
||
"Origin": "https://pay.openai.com",
|
||
"Referer": "https://pay.openai.com/",
|
||
"Sec-Fetch-Site": "cross-site",
|
||
"Sec-Fetch-Mode": "cors",
|
||
"Sec-Fetch-Dest": "empty",
|
||
"Sec-Ch-Ua-Platform": '"Linux"',
|
||
"Accept-Language": "zh-CN,zh;q=0.9",
|
||
"Priority": "u=1, i"
|
||
})
|
||
|
||
response = self.http_client.session.get(
|
||
url,
|
||
params=params,
|
||
headers=headers,
|
||
timeout=30
|
||
)
|
||
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
state = result.get("state")
|
||
logger.info(f"Current state: {state}")
|
||
|
||
if state == "succeeded":
|
||
logger.info("✅ PAYMENT SUCCEEDED!")
|
||
logger.info(f"Success URL: {result.get('success_url')}")
|
||
return result
|
||
elif state in ["failed", "canceled"]:
|
||
logger.error(f"❌ Payment {state}")
|
||
return result
|
||
|
||
# 继续轮询
|
||
time.sleep(interval)
|
||
else:
|
||
logger.warning(f"Poll returned status {response.status_code}")
|
||
time.sleep(interval)
|
||
|
||
except Exception as e:
|
||
logger.error(f"❌ Exception polling status: {e}")
|
||
time.sleep(interval)
|
||
|
||
logger.warning("⚠️ Max polling attempts reached")
|
||
return {"state": "timeout"}
|
||
|
||
def complete_payment(
|
||
self,
|
||
iban: str,
|
||
name: str,
|
||
email: str,
|
||
address_line1: str,
|
||
city: str,
|
||
postal_code: str,
|
||
state: str,
|
||
country: str = "US",
|
||
captcha_token: Optional[str] = None
|
||
) -> bool:
|
||
"""
|
||
完整的支付流程：创建支付方式 → 确认支付 → 轮询状态
|
||
|
||
Returns:
|
||
是否成功
|
||
"""
|
||
logger.info("=" * 60)
|
||
logger.info("Starting complete payment flow")
|
||
logger.info(f"Session ID: {self.session_id}")
|
||
logger.info("=" * 60)
|
||
|
||
# Step 1: 创建支付方式
|
||
payment_method_id = self.create_payment_method(
|
||
iban=iban,
|
||
name=name,
|
||
email=email,
|
||
address_line1=address_line1,
|
||
city=city,
|
||
postal_code=postal_code,
|
||
state=state,
|
||
country=country
|
||
)
|
||
|
||
if not payment_method_id:
|
||
logger.error("Failed at step 1: create payment method")
|
||
return False
|
||
|
||
# Step 2: 确认支付
|
||
confirmed = self.confirm_payment(
|
||
payment_method_id=payment_method_id,
|
||
captcha_token=captcha_token
|
||
)
|
||
|
||
if not confirmed:
|
||
logger.error("Failed at step 2: confirm payment")
|
||
return False
|
||
|
||
# Step 3: 轮询状态
|
||
final_status = self.poll_payment_status()
|
||
|
||
if final_status.get("state") == "succeeded":
|
||
logger.info("=" * 60)
|
||
logger.info("✅ PAYMENT COMPLETED SUCCESSFULLY")
|
||
logger.info("=" * 60)
|
||
return True
|
||
else:
|
||
logger.error("=" * 60)
|
||
logger.error(f"❌ Payment failed with state: {final_status.get('state')}")
|
||
logger.error("=" * 60)
|
||
return False
|