Stripe支付集成Skill stripe-integration

---

name: stripe-integration description: 实现Stripe支付处理，用于稳健、PCI合规的支付流程，包括结账、订阅和webhooks。适用于集成Stripe支付、构建订阅系统或实施安全结账流程时使用。

Stripe集成

掌握Stripe支付处理集成，用于稳健、PCI合规的支付流程，包括结账、订阅、webhooks和退款。

何时使用此技能

• 在Web/移动应用程序中实现支付处理
• 设置订阅计费系统
• 处理一次性支付和定期扣款
• 处理退款和争议
• 管理客户支付方式
• 为欧洲支付实施SCA（强客户认证）
• 使用Stripe Connect构建市场支付流程

核心概念

1. 支付流程

结账会话

• 推荐用于大多数集成
• 支持所有UI路径：
  • Stripe托管的结账页面
  • 嵌入式结账表单
  • 使用ui_mode='custom'的自定义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(
    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(
            line_items=[{
                'price_data': {
                    'currency': currency,
                    'product_data': {
                        'name': '蓝色T恤',
                        '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_checkout_session_for_elements(amount, currency='usd'):
    """为支付元素创建配置的结账会话。"""
    session = stripe.checkout.Session.create(
        mode='payment',
        ui_mode='custom',
        line_items=[{
            'price_data': {
                'currency': currency,
                'product_data': {'name': '蓝色T恤'},
                'unit_amount': amount,
            },
            'quantity': 1,
        }],
        return_url='https://yourdomain.com/complete?session_id={CHECKOUT_SESSION_ID}'
    )
    return session.client_secret  # 发送到前端

const stripe = Stripe("pk_test_...");
const appearance = { theme: "stripe" };

const checkout = stripe.initCheckout({
  clientSecret,
  elementsOptions: { appearance },
});
const loadActionsResult = await checkout.loadActions();

if (loadActionsResult.type === "success") {
  const { actions } = loadActionsResult;
  const session = actions.getSession();

  const button = document.getElementById("pay-button");
  const checkoutContainer = document.getElementById("checkout-container");
  const emailInput = document.getElementById("email");
  const emailErrors = document.getElementById("email-errors");
  const errors = document.getElementById("confirm-errors");

  // 显示表示总金额的格式化字符串
  checkoutContainer.append(`总计: ${session.total.total.amount}`);

  // 挂载支付元素
  const paymentElement = checkout.createPaymentElement();
  paymentElement.mount("#payment-element");

  // 存储电子邮件以提交
  emailInput.addEventListener("blur", () => {
    actions.updateEmail(emailInput.value).then((result) => {
      if (result.error) emailErrors.textContent = result.error.message;
    });
  });

  // 处理表单提交
  button.addEventListener("click", () => {
    actions.confirm().then((result) => {
      if (result.type === "error") errors.textContent = result.error.message;
    });
  });
}

模式3: 元素与支付意图

模式2（元素与结账会话）是Stripe推荐的方法，但您也可以使用支付意图作为替代。

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  # 发送到前端

// 挂载支付元素并通过支付意图确认
const stripe = Stripe("pk_test_...");
const appearance = { theme: "stripe" };
const elements = stripe.elements({ appearance, clientSecret });

const paymentElement = elements.create("payment");
paymentElement.mount("#payment-element");

document.getElementById("pay-button").addEventListener("click", async () => {
  const { error } = await stripe.confirmPayment({
    elements,
    confirmParams: {
      return_url: "https://yourdomain.com/complete",
    },
  });

  if (error) {
    document.getElementById("errors").textContent = error.message;
  }
});

模式4: 订阅创建

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

模式5: 客户门户

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,
        automatic_payment_methods={
            'enabled': True
        },
        currency='usd',
        customer=customer.id
    )

    # 使用测试卡确认
    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: 结账配置模板

最佳实践

1. 始终使用Webhooks: 不要仅依赖客户端确认
2. 幂等性: 以幂等方式处理webhook事件
3. 错误处理: 优雅处理所有Stripe错误
4. 测试模式: 在生产前用测试密钥彻底测试
5. 元数据: 使用元数据链接Stripe对象到数据库
6. 监控: 跟踪支付成功率和错误
7. PCI合规: 切勿在服务器上处理原始卡数据
8. SCA就绪: 为欧洲支付实施3D安全

常见陷阱

• 未验证Webhooks: 始终验证webhook签名
• 遗漏Webhook事件: 处理所有相关webhook事件
• 硬编码金额: 使用分/最小货币单位
• 无重试逻辑: 为API调用实现重试
• 忽略测试模式: 用测试卡测试所有边缘情况