name: stripe-integration description: 实现Stripe支付处理，用于健壮、PCI合规的支付流程，包括结账、订阅和Webhooks。在集成Stripe支付、构建订阅系统或实施安全结账流程时使用。
Stripe集成
掌握Stripe支付处理集成，用于健壮、PCI合规的支付流程，包括结账、订阅、Webhooks和退款。
何时使用此技能
- 在Web/移动应用中实现支付处理
- 设置订阅计费系统
- 处理一次性支付和定期收费
- 处理退款和争议
- 管理客户支付方式
- 为欧洲支付实施SCA（强客户认证）
- 使用Stripe Connect构建市场支付流程
核心概念
1. 支付流程
结账会话（托管）
- Stripe托管的支付页面
- 最小化PCI合规负担
- 最快实现
- 支持一次性和定期支付
支付意图（自定义UI）
- 完全控制支付UI
- 需要使用Stripe.js以确保PCI合规
- 更复杂的实现
- 更好的定制选项
设置意图（保存支付方式）
- 在不收费的情况下收集支付方式
- 用于订阅和未来支付
- 需要客户确认
2. Webhooks
关键事件：
payment_intent.succeeded：支付完成payment_intent.payment_failed：支付失败customer.subscription.updated：订阅更改customer.subscription.deleted：订阅取消charge.refunded：退款处理invoice.payment_succeeded：订阅支付成功
3. 订阅
组件：
- 产品：您销售的内容
- 价格：金额和频率
- 订阅：客户的定期支付
- 发票：为每个计费周期生成
4. 客户管理
- 创建和管理客户记录
- 存储多种支付方式
- 跟踪客户元数据
- 管理账单详情
快速开始
import stripe
stripe.api_key = "sk_test_..."
# 创建结账会话
session = stripe.checkout.Session.create(
payment_method_types=['card'],
line_items=[{
'price_data': {
'currency': 'usd',
'product_data': {
'name': '高级订阅',
},
'unit_amount': 2000, # $20.00
'recurring': {
'interval': 'month',
},
},
'quantity': 1,
}],
mode='subscription',
success_url='https://yourdomain.com/success?session_id={CHECKOUT_SESSION_ID}',
cancel_url='https://yourdomain.com/cancel',
)
# 重定向用户到session.url
print(session.url)
支付实现模式
模式1：一次性支付（托管结账）
def create_checkout_session(amount, currency='usd'):
"""创建一次性支付结账会话。"""
try:
session = stripe.checkout.Session.create(
payment_method_types=['card'],
line_items=[{
'price_data': {
'currency': currency,
'product_data': {
'name': '购买',
'images': ['https://example.com/product.jpg'],
},
'unit_amount': amount, # 以分为单位
},
'quantity': 1,
}],
mode='payment',
success_url='https://yourdomain.com/success?session_id={CHECKOUT_SESSION_ID}',
cancel_url='https://yourdomain.com/cancel',
metadata={
'order_id': 'order_123',
'user_id': 'user_456'
}
)
return session
except stripe.error.StripeError as e:
# 处理错误
print(f"Stripe错误: {e.user_message}")
raise
模式2：自定义支付意图流程
def create_payment_intent(amount, currency='usd', customer_id=None):
"""为自定义结账UI创建支付意图。"""
intent = stripe.PaymentIntent.create(
amount=amount,
currency=currency,
customer=customer_id,
automatic_payment_methods={
'enabled': True,
},
metadata={
'integration_check': 'accept_a_payment'
}
)
return intent.client_secret # 发送到前端
# 前端（JavaScript）
"""
const stripe = Stripe('pk_test_...');
const elements = stripe.elements();
const cardElement = elements.create('card');
cardElement.mount('#card-element');
const {error, paymentIntent} = await stripe.confirmCardPayment(
clientSecret,
{
payment_method: {
card: cardElement,
billing_details: {
name: '客户姓名'
}
}
}
);
if (error) {
// 处理错误
} else if (paymentIntent.status === 'succeeded') {
// 支付成功
}
"""
模式3：订阅创建
def create_subscription(customer_id, price_id):
"""为客户创建订阅。"""
try:
subscription = stripe.Subscription.create(
customer=customer_id,
items=[{'price': price_id}],
payment_behavior='default_incomplete',
payment_settings={'save_default_payment_method': 'on_subscription'},
expand=['latest_invoice.payment_intent'],
)
return {
'subscription_id': subscription.id,
'client_secret': subscription.latest_invoice.payment_intent.client_secret
}
except stripe.error.StripeError as e:
print(f"订阅创建失败: {e}")
raise
模式4：客户门户
def create_customer_portal_session(customer_id):
"""创建客户门户会话以管理订阅。"""
session = stripe.billing_portal.Session.create(
customer=customer_id,
return_url='https://yourdomain.com/account',
)
return session.url # 重定向客户到此
Webhook处理
安全Webhook端点
from flask import Flask, request
import stripe
app = Flask(__name__)
endpoint_secret = 'whsec_...'
@app.route('/webhook', methods=['POST'])
def webhook():
payload = request.data
sig_header = request.headers.get('Stripe-Signature')
try:
event = stripe.Webhook.construct_event(
payload, sig_header, endpoint_secret
)
except ValueError:
# 无效负载
return '无效负载', 400
except stripe.error.SignatureVerificationError:
# 无效签名
return '无效签名', 400
# 处理事件
if event['type'] == 'payment_intent.succeeded':
payment_intent = event['data']['object']
handle_successful_payment(payment_intent)
elif event['type'] == 'payment_intent.payment_failed':
payment_intent = event['data']['object']
handle_failed_payment(payment_intent)
elif event['type'] == 'customer.subscription.deleted':
subscription = event['data']['object']
handle_subscription_canceled(subscription)
return '成功', 200
def handle_successful_payment(payment_intent):
"""处理成功支付。"""
customer_id = payment_intent.get('customer')
amount = payment_intent['amount']
metadata = payment_intent.get('metadata', {})
# 更新您的数据库
# 发送确认邮件
# 履行订单
print(f"支付成功: {payment_intent['id']}")
def handle_failed_payment(payment_intent):
"""处理失败支付。"""
error = payment_intent.get('last_payment_error', {})
print(f"支付失败: {error.get('message')}")
# 通知客户
# 更新订单状态
def handle_subscription_canceled(subscription):
"""处理订阅取消。"""
customer_id = subscription['customer']
# 更新用户访问
# 发送取消邮件
print(f"订阅取消: {subscription['id']}")
Webhook最佳实践
import hashlib
import hmac
def verify_webhook_signature(payload, signature, secret):
"""手动验证Webhook签名。"""
expected_sig = hmac.new(
secret.encode('utf-8'),
payload,
hashlib.sha256
).hexdigest()
return hmac.compare_digest(signature, expected_sig)
def handle_webhook_idempotently(event_id, handler):
"""确保Webhook仅处理一次。"""
# 检查事件是否已处理
if is_event_processed(event_id):
return
# 处理事件
try:
handler()
mark_event_processed(event_id)
except Exception as e:
log_error(e)
# Stripe将重试失败的Webhooks
raise
客户管理
def create_customer(email, name, payment_method_id=None):
"""创建Stripe客户。"""
customer = stripe.Customer.create(
email=email,
name=name,
payment_method=payment_method_id,
invoice_settings={
'default_payment_method': payment_method_id
} if payment_method_id else None,
metadata={
'user_id': '12345'
}
)
return customer
def attach_payment_method(customer_id, payment_method_id):
"""将支付方式附加到客户。"""
stripe.PaymentMethod.attach(
payment_method_id,
customer=customer_id
)
# 设为默认
stripe.Customer.modify(
customer_id,
invoice_settings={
'default_payment_method': payment_method_id
}
)
def list_customer_payment_methods(customer_id):
"""列出客户的所有支付方式。"""
payment_methods = stripe.PaymentMethod.list(
customer=customer_id,
type='card'
)
return payment_methods.data
退款处理
def create_refund(payment_intent_id, amount=None, reason=None):
"""创建退款。"""
refund_params = {
'payment_intent': payment_intent_id
}
if amount:
refund_params['amount'] = amount # 部分退款
if reason:
refund_params['reason'] = reason # 'duplicate', 'fraudulent', 'requested_by_customer'
refund = stripe.Refund.create(**refund_params)
return refund
def handle_dispute(charge_id, evidence):
"""用证据更新争议。"""
stripe.Dispute.modify(
charge_id,
evidence={
'customer_name': evidence.get('customer_name'),
'customer_email_address': evidence.get('customer_email'),
'shipping_documentation': evidence.get('shipping_proof'),
'customer_communication': evidence.get('communication'),
}
)
测试
# 使用测试模式密钥
stripe.api_key = "sk_test_..."
# 测试卡号
TEST_CARDS = {
'success': '4242424242424242',
'declined': '4000000000000002',
'3d_secure': '4000002500003155',
'insufficient_funds': '4000000000009995'
}
def test_payment_flow():
"""测试完整支付流程。"""
# 创建测试客户
customer = stripe.Customer.create(
email="test@example.com"
)
# 创建支付意图
intent = stripe.PaymentIntent.create(
amount=1000,
currency='usd',
customer=customer.id,
payment_method_types=['card']
)
# 用测试卡确认
confirmed = stripe.PaymentIntent.confirm(
intent.id,
payment_method='pm_card_visa' # 测试支付方式
)
assert confirmed.status == 'succeeded'
资源
- references/checkout-flows.md：详细结账实现
- references/webhook-handling.md：Webhook安全和处理
- references/subscription-management.md：订阅生命周期
- references/customer-management.md：客户和支付方式处理
- references/invoice-generation.md：发票和计费
- assets/stripe-client.py：生产就绪的Stripe客户端包装器
- assets/webhook-handler.py：完整的Webhook处理器
- assets/checkout-config.json：结账配置模板
最佳实践
- 始终使用Webhooks：不要仅依赖客户端确认
- 幂等性：以幂等方式处理Webhook事件
- 错误处理：优雅处理所有Stripe错误
- 测试模式：在生产前用测试密钥全面测试
- 元数据：使用元数据链接Stripe对象到您的数据库
- 监控：跟踪支付成功率和错误
- PCI合规：绝不在服务器上处理原始卡数据
- SCA就绪：为欧洲支付实施3D Secure
常见陷阱
- 未验证Webhooks：始终验证Webhook签名
- 缺少Webhook事件：处理所有相关Webhook事件
- 硬编码金额：使用分/最小货币单位
- 无重试逻辑：为API调用实现重试
- 忽略测试模式：用测试卡测试所有边缘情况