419 lines
16 KiB
Python
419 lines
16 KiB
Python
"""
|
||
Stripe SEPA 支付 API 模块
|
||
- 使用纯 API 方式完成 Stripe SEPA 支付
|
||
- 参考 team-reg-go/stripe/stripe.go 实现
|
||
"""
|
||
|
||
import re
|
||
import time
|
||
import uuid
|
||
import random
|
||
|
||
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}")
|
||
|
||
|
||
# Stripe 配置常量 (与 team-reg-go 保持一致)
|
||
STRIPE_VERSION = "2020-08-27;custom_checkout_beta=v1"
|
||
STRIPE_JS_VERSION = "c8cd270e71"
|
||
OPENAI_STRIPE_PUBLIC_KEY = "pk_live_51Pj377KslHRdbaPgTJYjThzH3f5dt1N1vK7LUp0qh0yNSarhfZ6nfbG7FFlh8KLxVkvdMWN5o6Mc4Vda6NHaSnaV00C2Sbl8Zs"
|
||
|
||
|
||
def generate_stripe_fingerprint() -> str:
|
||
"""生成 Stripe 指纹 ID"""
|
||
return uuid.uuid4().hex
|
||
|
||
|
||
def extract_session_id(checkout_url: str) -> str:
|
||
"""从 checkout URL 提取 session_id"""
|
||
match = re.search(r'(cs_(?:live|test)_[a-zA-Z0-9]+)', checkout_url)
|
||
if match:
|
||
return match.group(1)
|
||
return ""
|
||
|
||
|
||
class StripePaymentAPI:
|
||
"""Stripe SEPA 支付 API 处理器"""
|
||
|
||
def __init__(self, checkout_url: str, session=None, proxy: str = None):
|
||
"""初始化
|
||
|
||
Args:
|
||
checkout_url: Stripe checkout URL
|
||
session: 可选的 curl_cffi session (复用已有会话)
|
||
proxy: 代理地址
|
||
"""
|
||
self.checkout_url = checkout_url
|
||
self.session_id = extract_session_id(checkout_url)
|
||
self.stripe_public_key = OPENAI_STRIPE_PUBLIC_KEY
|
||
self.init_checksum = ""
|
||
self.js_checksum = ""
|
||
self.guid = generate_stripe_fingerprint()
|
||
self.muid = generate_stripe_fingerprint()
|
||
self.sid = generate_stripe_fingerprint()
|
||
self.client_session_id = ""
|
||
self.checkout_config_id = ""
|
||
|
||
# 创建或复用 session
|
||
if session:
|
||
self.session = session
|
||
elif CURL_CFFI_AVAILABLE:
|
||
self.session = curl_requests.Session(
|
||
impersonate="chrome",
|
||
verify=False,
|
||
proxies={"http": proxy, "https": proxy} if proxy else {}
|
||
)
|
||
else:
|
||
self.session = requests.Session()
|
||
if proxy:
|
||
self.session.proxies = {"http": proxy, "https": proxy}
|
||
|
||
def _get_stripe_headers(self) -> dict:
|
||
"""获取 Stripe API 请求头"""
|
||
return {
|
||
"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": '"Windows"',
|
||
"Accept-Language": "en-US,en;q=0.9",
|
||
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36"
|
||
}
|
||
|
||
def fetch_checkout_page(self) -> bool:
|
||
"""获取 checkout 页面参数"""
|
||
try:
|
||
headers = {
|
||
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
|
||
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36"
|
||
}
|
||
resp = self.session.get(self.checkout_url, headers=headers, timeout=30)
|
||
|
||
if resp.status_code == 200:
|
||
html = resp.text
|
||
|
||
# 提取 initChecksum
|
||
init_match = re.search(r'"initChecksum":\s*"([^"]+)"', html)
|
||
if init_match:
|
||
self.init_checksum = init_match.group(1)
|
||
|
||
# 提取 jsChecksum
|
||
js_match = re.search(r'"jsChecksum":\s*"([^"]+)"', html)
|
||
if js_match:
|
||
self.js_checksum = js_match.group(1)
|
||
|
||
return True
|
||
return False
|
||
except Exception as e:
|
||
log_progress(f"[!] 获取 checkout 页面失败: {e}")
|
||
return False
|
||
|
||
def create_payment_method(self, iban: str, name: str, email: str,
|
||
address: str, city: str, postal_code: str,
|
||
country: str = "DE") -> str:
|
||
"""Step 1: 创建支付方式
|
||
|
||
Returns:
|
||
str: payment_method_id，失败返回空字符串
|
||
"""
|
||
api_url = "https://api.stripe.com/v1/payment_methods"
|
||
|
||
self.client_session_id = str(uuid.uuid4())
|
||
self.checkout_config_id = str(uuid.uuid4())
|
||
|
||
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,
|
||
"billing_details[address][city]": city,
|
||
"billing_details[address][postal_code]": postal_code,
|
||
"guid": self.guid,
|
||
"muid": self.muid,
|
||
"sid": self.sid,
|
||
"_stripe_version": STRIPE_VERSION,
|
||
"key": self.stripe_public_key,
|
||
"payment_user_agent": f"stripe.js/{STRIPE_JS_VERSION}; stripe-js-v3/{STRIPE_JS_VERSION}; 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:
|
||
resp = self.session.post(api_url, data=data, headers=self._get_stripe_headers(), timeout=30)
|
||
|
||
if resp.status_code == 200:
|
||
result = resp.json()
|
||
payment_method_id = result.get("id", "")
|
||
if payment_method_id:
|
||
return payment_method_id
|
||
|
||
error_text = resp.text[:200] if resp.text else "无响应"
|
||
log_progress(f"[X] 创建支付方式失败: {resp.status_code} - {error_text}")
|
||
return ""
|
||
except Exception as e:
|
||
log_progress(f"[X] 创建支付方式异常: {e}")
|
||
return ""
|
||
|
||
def confirm_payment(self, payment_method_id: str, captcha_token: str = "",
|
||
rv_timestamp: str = "") -> tuple:
|
||
"""Step 2: 确认支付
|
||
|
||
Returns:
|
||
tuple: (success: bool, error_message: str)
|
||
"""
|
||
api_url = f"https://api.stripe.com/v1/payment_pages/{self.session_id}/confirm"
|
||
|
||
data = {
|
||
"eid": "NA",
|
||
"payment_method": payment_method_id,
|
||
"expected_amount": "0",
|
||
"consent[terms_of_service]": "accepted",
|
||
"tax_id_collection[purchasing_as_business]": "false",
|
||
"expected_payment_method_type": "sepa_debit",
|
||
"_stripe_version": STRIPE_VERSION,
|
||
"guid": self.guid,
|
||
"muid": self.muid,
|
||
"sid": self.sid,
|
||
"key": self.stripe_public_key,
|
||
"version": STRIPE_JS_VERSION,
|
||
"referrer": "https://chatgpt.com",
|
||
"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,
|
||
}
|
||
|
||
if self.init_checksum:
|
||
data["init_checksum"] = self.init_checksum
|
||
if self.js_checksum:
|
||
data["js_checksum"] = self.js_checksum
|
||
if captcha_token:
|
||
data["passive_captcha_token"] = captcha_token
|
||
data["passive_captcha_ekey"] = ""
|
||
if rv_timestamp:
|
||
data["rv_timestamp"] = rv_timestamp
|
||
|
||
try:
|
||
resp = self.session.post(api_url, data=data, headers=self._get_stripe_headers(), timeout=30)
|
||
|
||
if resp.status_code == 200:
|
||
result = resp.json()
|
||
state = result.get("state", "")
|
||
|
||
if state in ["succeeded", "processing", "processing_subscription"]:
|
||
return True, ""
|
||
elif state == "failed":
|
||
error = result.get("error", {})
|
||
error_msg = error.get("message", str(error)) if isinstance(error, dict) else str(error)
|
||
return False, f"支付失败: {error_msg}"
|
||
else:
|
||
# 其他状态继续轮询
|
||
return True, ""
|
||
|
||
error_text = resp.text[:200] if resp.text else "无响应"
|
||
return False, f"确认失败: {resp.status_code} - {error_text}"
|
||
except Exception as e:
|
||
return False, f"确认异常: {e}"
|
||
|
||
def poll_payment_status(self, max_attempts: int = 20) -> tuple:
|
||
"""Step 3: 轮询支付状态
|
||
|
||
Returns:
|
||
tuple: (state: str, error_message: str)
|
||
"""
|
||
api_url = f"https://api.stripe.com/v1/payment_pages/{self.session_id}/poll?key={self.stripe_public_key}"
|
||
|
||
for attempt in range(max_attempts):
|
||
try:
|
||
resp = self.session.get(api_url, headers=self._get_stripe_headers(), timeout=30)
|
||
|
||
if resp.status_code == 200:
|
||
result = resp.json()
|
||
state = result.get("state", "")
|
||
|
||
if state == "succeeded":
|
||
return "succeeded", ""
|
||
elif state in ["failed", "canceled"]:
|
||
return state, f"支付 {state}"
|
||
|
||
time.sleep(2)
|
||
except Exception as e:
|
||
log_progress(f"[!] 轮询异常: {e}")
|
||
time.sleep(2)
|
||
|
||
return "timeout", "轮询超时"
|
||
|
||
def complete_payment(self, iban: str, name: str, email: str,
|
||
address: str, city: str, postal_code: str,
|
||
country: str = "DE") -> tuple:
|
||
"""执行完整支付流程
|
||
|
||
Returns:
|
||
tuple: (success: bool, error_message: str)
|
||
"""
|
||
# 获取页面参数
|
||
self.fetch_checkout_page()
|
||
|
||
# Step 1: 创建支付方式
|
||
payment_method_id = self.create_payment_method(
|
||
iban, name, email, address, city, postal_code, country
|
||
)
|
||
if not payment_method_id:
|
||
return False, "创建支付方式失败"
|
||
|
||
# Step 2: 确认支付
|
||
success, error = self.confirm_payment(payment_method_id)
|
||
if not success:
|
||
return False, error
|
||
|
||
# Step 3: 轮询状态
|
||
state, error = self.poll_payment_status(15)
|
||
if state == "succeeded":
|
||
return True, ""
|
||
|
||
return False, error or f"支付失败: {state}"
|
||
|
||
|
||
def api_payment_with_retry(checkout_url: str, email: str, session=None, proxy: str = None,
|
||
max_retries: int = 3, get_iban_func=None,
|
||
get_address_func=None, get_name_func=None,
|
||
progress_callback=None) -> tuple:
|
||
"""使用 API 完成支付流程（带 IBAN 重试）
|
||
|
||
Args:
|
||
checkout_url: Stripe checkout URL
|
||
email: 邮箱地址
|
||
session: 可选的 curl_cffi session
|
||
proxy: 代理地址
|
||
max_retries: 最大重试次数
|
||
get_iban_func: 获取 IBAN 的函数，签名: func() -> str
|
||
get_address_func: 获取地址的函数，签名: func() -> tuple(street, postal_code, city)
|
||
get_name_func: 获取姓名的函数，签名: func() -> str
|
||
progress_callback: 进度回调函数
|
||
|
||
Returns:
|
||
tuple: (success: bool, error_message: str)
|
||
"""
|
||
def log_cb(msg):
|
||
if progress_callback:
|
||
progress_callback(msg)
|
||
else:
|
||
log_progress(msg)
|
||
|
||
# 默认的 IBAN/地址/姓名生成函数
|
||
if get_iban_func is None:
|
||
def get_iban_func():
|
||
# 从 auto_gpt_team 导入
|
||
try:
|
||
from auto_gpt_team import get_sepa_ibans
|
||
ibans = get_sepa_ibans()
|
||
return random.choice(ibans) if ibans else ""
|
||
except:
|
||
return ""
|
||
|
||
if get_address_func is None:
|
||
def get_address_func():
|
||
try:
|
||
from auto_gpt_team import SEPA_ADDRESSES
|
||
return random.choice(SEPA_ADDRESSES)
|
||
except:
|
||
return ("Alexanderplatz 1", "10178", "Berlin")
|
||
|
||
if get_name_func is None:
|
||
def get_name_func():
|
||
try:
|
||
from auto_gpt_team import FIRST_NAMES, LAST_NAMES
|
||
return f"{random.choice(FIRST_NAMES)} {random.choice(LAST_NAMES)}"
|
||
except:
|
||
return "Max Mustermann"
|
||
|
||
# 初始化 Stripe 支付处理器
|
||
stripe_handler = StripePaymentAPI(checkout_url, session=session, proxy=proxy)
|
||
|
||
last_error = ""
|
||
|
||
for retry in range(max_retries):
|
||
# 生成支付信息
|
||
iban = get_iban_func()
|
||
if not iban:
|
||
return False, "没有可用的 IBAN"
|
||
|
||
street, postal_code, city = get_address_func()
|
||
account_name = get_name_func()
|
||
|
||
if retry == 0:
|
||
log_status("API支付", "SEPA 支付处理中...")
|
||
log_cb(f"IBAN: {iban[:8]}...")
|
||
log_cb(f"地址: {street}, {postal_code} {city}")
|
||
log_cb(f"姓名: {account_name}")
|
||
else:
|
||
log_cb(f"[!] 重试 {retry}/{max_retries-1}，更换 IBAN...")
|
||
log_cb(f"新 IBAN: {iban[:8]}...")
|
||
|
||
success, error = stripe_handler.complete_payment(
|
||
iban, account_name, email, street, city, postal_code, "DE"
|
||
)
|
||
|
||
if success:
|
||
log_status("API支付", "[OK] 支付成功")
|
||
return True, ""
|
||
|
||
last_error = error
|
||
|
||
# 分析错误类型，决定是否重试
|
||
error_lower = error.lower() if error else ""
|
||
|
||
# IBAN/BIC 相关错误 - 换 IBAN 重试
|
||
if "bank_account_unusable" in error_lower or "bic" in error_lower or "iban" in error_lower:
|
||
log_cb(f"[!] IBAN 无效: {error}")
|
||
if retry < max_retries - 1:
|
||
continue
|
||
|
||
# 可恢复错误 - 重试
|
||
retryable_errors = ["400", "500", "timeout", "eof", "connection", "确认失败"]
|
||
is_retryable = any(e in error_lower for e in retryable_errors)
|
||
|
||
if is_retryable and retry < max_retries - 1:
|
||
log_cb(f"[!] 支付错误: {error}，重试中...")
|
||
time.sleep(1)
|
||
continue
|
||
|
||
# 不可恢复错误或重试用尽
|
||
break
|
||
|
||
log_status("API支付", f"[X] 支付失败: {last_error}")
|
||
return False, last_error
|
||
|
||
|
||
def is_stripe_api_available() -> bool:
|
||
"""检查 Stripe API 模式是否可用"""
|
||
return CURL_CFFI_AVAILABLE
|