paypal-integration

支持材料的可用性

以下为外部先决条件，并非随附文件：<support-root>/references/express-checkout.md、<support-root>/references/ipn-handling.md、<support-root>/references/refund-workflows.md、<support-root>/references/billing-agreements.md、<support-root>/assets/paypal-client.py、<support-root>/assets/ipn-processor.py、<support-root>/assets/recurring-billing.py。请从原始来源或现有安装中获取，并在使用前检查。必需文件不存在时，不要执行依赖该文件的步骤，也不要声称已完成；请报告无法执行的步骤。完整记录的内联工作流仍可在授权范围内使用。

PayPal 集成

全面掌握 PayPal 支付集成，包括 Express Checkout、IPN 处理、定期结算和退款工作流。

以下情况请勿使用此技能

• 任务与 PayPal 集成无关
• 你需要使用此范围以外的其他领域或工具

说明

• 明确目标、约束和所需输入。
• 应用相关最佳实践并验证结果。
• 提供可执行的步骤和验证方法。

以下情况请使用此技能

• 将 PayPal 集成为一种支付方式
• 实现快捷结账流程
• 使用 PayPal 设置定期结算
• 处理退款和支付争议
• 处理 PayPal Webhook（IPN）
• 支持国际支付
• 实现 PayPal 订阅

核心概念

1. 支付产品

PayPal Checkout

• 一次性付款
• 快捷结账体验
• 访客付款和 PayPal 账户付款

PayPal Subscriptions

• 定期结算
• 订阅计划
• 自动续订

PayPal Payouts

• 向多个收款人汇款
• 市场平台和平台支付

2. 集成方式

客户端（JavaScript SDK）

• Smart Payment Buttons
• 托管支付流程
• 最少的后端代码

服务端（REST API）

• 完全控制支付流程
• 自定义结账界面
• 高级功能

3. IPN（即时付款通知）

• 类似 Webhook 的付款通知
• 异步付款更新
• 需要验证

快速入门

在经过身份验证的后端上，使用当前的 JavaScript SDK 和 Orders v2 端点。不要将浏览器端捕获与第二次后端捕获结合使用，也不要将 Orders v2 ID 传递给旧版 Payments v1 SDK。

// Browser fragment; include the SDK with the public client ID, never the client secret.
paypal.Buttons({
  async createOrder() {
    const response = await fetch('/api/paypal/orders', {method: 'POST'});
    if (!response.ok) throw new Error('Order creation failed');
    return (await response.json()).id;
  },
  async onApprove(data) {
    const response = await fetch('/api/paypal/capture', {
      method: 'POST', headers: {'Content-Type': 'application/json'},
      body: JSON.stringify({orderID: data.orderID})
    });
    if (!response.ok) throw new Error('Capture confirmation failed');
    // Backend owns reconciliation/fulfilment; the browser callback is not payment evidence.
  }
}).render('#paypal-button-container');

路由处理程序是应用适配器，并非随附内容。它们必须对用户进行身份验证，实施 CSRF／来源控制，将订单绑定到用户拥有的购物车，从可信数据中确定价格／货币／收款方，为每项操作持久化稳定的 PayPal-Request-Id，并在原子化履约前验证捕获状态／金额／货币／收款方。超时应视为结果未定；应对现有请求进行核对，而不是再创建一笔扣款。通过服务器环境／机密管理器存储机密。使用 Decimal／字符串金额，并采用货币特定的精度；切勿使用二进制浮点数处理金额。以下代码片段是沙盒示例，并非生产级支付系统。

Express Checkout 实现

服务端创建订单

import requests
import json

class PayPalClient:
    def __init__(self, client_id, client_secret, mode='sandbox'):
        if mode not in ('sandbox', 'live'):
            raise ValueError('Explicit sandbox or live mode required')
        self.client_id = client_id
        self.client_secret = client_secret
        self.base_url = 'https://api-m.sandbox.paypal.com' if mode == 'sandbox' else 'https://api-m.paypal.com'
        self.access_token = self.get_access_token()

    def get_access_token(self):
        """Get OAuth access token."""
        url = f"{self.base_url}/v1/oauth2/token"
        headers = {"Accept": "application/json", "Accept-Language": "en_US"}

        response = requests.post(
            url,
            headers=headers,
            data={"grant_type": "client_credentials"},
            auth=(self.client_id, self.client_secret), timeout=15, allow_redirects=False
        )

        response.raise_for_status()
        return response.json()['access_token']

    def create_order(self, amount, request_id, currency='USD'):
        """Create a PayPal order."""
        url = f"{self.base_url}/v2/checkout/orders"
        headers = {
            "Content-Type": "application/json",
            "PayPal-Request-Id": request_id, # Persisted per operation, unchanged on retry.
            "Authorization": f"Bearer {self.access_token}"
        }

        payload = {
            "intent": "CAPTURE",
            "purchase_units": [{
                "amount": {
                    "currency_code": currency,
                    "value": str(amount)
                }
            }]
        }

        response = requests.post(url, headers=headers, json=payload, timeout=15, allow_redirects=False)
        response.raise_for_status()
        return response.json()

    def capture_order(self, order_id, request_id):
        """Capture payment for an order."""
        url = f"{self.base_url}/v2/checkout/orders/{order_id}/capture"
        headers = {
            "Content-Type": "application/json",
            "PayPal-Request-Id": request_id, # Persisted per operation, unchanged on retry.
            "Authorization": f"Bearer {self.access_token}"
        }

        response = requests.post(url, headers=headers, timeout=15, allow_redirects=False)
        response.raise_for_status()
        return response.json()

    def get_order_details(self, order_id):
        """Get order details."""
        url = f"{self.base_url}/v2/checkout/orders/{order_id}"
        headers = {
            "Authorization": f"Bearer {self.access_token}"
        }

        response = requests.get(url, headers=headers, timeout=15, allow_redirects=False)
        response.raise_for_status()
        return response.json()

IPN（即时付款通知）处理

IPN 是一种旧版通知协议，并非 REST Webhooks API。对于当前的 REST 集成，请使用文档中说明的 Webhook ID 和环境验证 Webhook 签名。限制入站请求大小。VERIFIED 只能证明来源：还需验证收款方、货币、金额、用户拥有的发票／订单以及交易状态，然后通过唯一事件／交易记录进行原子化去重。单独使用先检查后执行的辅助函数会引发竞态。应根据原始记录处理撤销／退款，并且只有在持久化记录后才确认。

IPN 验证与处理

from flask import Flask, request
import requests
from urllib.parse import parse_qs

app = Flask(__name__)

@app.route('/ipn', methods=['POST'])
def handle_ipn():
    """Handle PayPal IPN notifications."""
    # Get IPN message
    raw_ipn = request.get_data(cache=True)
    ipn_data = request.form.to_dict()

    # Verify IPN with PayPal
    if not verify_ipn(raw_ipn):
        return 'IPN verification failed', 400

    # Process IPN based on transaction type
    payment_status = ipn_data.get('payment_status')
    txn_type = ipn_data.get('txn_type')

    if payment_status == 'Completed':
        handle_payment_completed(ipn_data)
    elif payment_status == 'Refunded':
        handle_refund(ipn_data)
    elif payment_status == 'Reversed':
        handle_chargeback(ipn_data)

    return 'IPN processed', 200

def verify_ipn(raw_ipn):
    """Echo original encoded body in order; do not reconstruct it from a dict."""
    paypal_url = 'https://ipnpb.sandbox.paypal.com/cgi-bin/webscr'
    response = requests.post(paypal_url, data=b'cmd=_notify-validate&' + raw_ipn,
                             headers={'Content-Type': 'application/x-www-form-urlencoded'},
                             timeout=15, allow_redirects=False)
    response.raise_for_status()
    return response.text.strip() == 'VERIFIED'

def handle_payment_completed(ipn_data):
    """Process completed payment."""
    txn_id = ipn_data.get('txn_id')
    payer_email = ipn_data.get('payer_email')
    mc_gross = ipn_data.get('mc_gross')
    item_name = ipn_data.get('item_name')

    # Check if already processed (prevent duplicates)
    if is_transaction_processed(txn_id):
        return

    # Update database
    # Send confirmation email
    # Fulfill order
    print(f"Payment completed: {txn_id}, Amount: ${mc_gross}")

def handle_refund(ipn_data):
    """Handle refund."""
    parent_txn_id = ipn_data.get('parent_txn_id')
    mc_gross = ipn_data.get('mc_gross')

    # Process refund in your system
    print(f"Refund processed: {parent_txn_id}, Amount: ${mc_gross}")

def handle_chargeback(ipn_data):
    """Handle payment reversal/chargeback."""
    txn_id = ipn_data.get('txn_id')
    reason_code = ipn_data.get('reason_code')

    # Handle chargeback
    print(f"Chargeback: {txn_id}, Reason: {reason_code}")

在支持的情况下，变更操作示例要求使用已持久化且针对特定操作的 PayPal-Request-Id、由服务器拥有且不可变的金额，以及经过验证的不透明资源 ID。令牌过期与刷新、经过验证的商户身份以及持久化事件处理，均属于必需的集成工作；任何片段都无法独立实现订单履约。退款和定期扣款都需要针对该确切操作的授权。

订阅/定期计费

创建订阅计划

def create_subscription_plan(name, amount, request_id, interval='MONTH'):
    """Create a subscription plan."""
    client = PayPalClient(CLIENT_ID, CLIENT_SECRET)

    url = f"{client.base_url}/v1/billing/plans"
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {client.access_token}",
        "PayPal-Request-Id": request_id
    }

    payload = {
        "product_id": "PRODUCT_ID",  # Create product first
        "name": name,
        "billing_cycles": [{
            "frequency": {
                "interval_unit": interval,
                "interval_count": 1
            },
            "tenure_type": "REGULAR",
            "sequence": 1,
            "total_cycles": 0,  # Infinite
            "pricing_scheme": {
                "fixed_price": {
                    "value": str(amount),
                    "currency_code": "USD"
                }
            }
        }],
        "payment_preferences": {
            "auto_bill_outstanding": True,
            "setup_fee": {
                "value": "0",
                "currency_code": "USD"
            },
            "setup_fee_failure_action": "CONTINUE",
            "payment_failure_threshold": 3
        }
    }

    response = requests.post(url, headers=headers, json=payload, timeout=15, allow_redirects=False)
    response.raise_for_status()
    return response.json()

def create_subscription(plan_id, subscriber_email, request_id):
    """Create a subscription for a customer."""
    client = PayPalClient(CLIENT_ID, CLIENT_SECRET)

    url = f"{client.base_url}/v1/billing/subscriptions"
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {client.access_token}",
        "PayPal-Request-Id": request_id
    }

    payload = {
        "plan_id": plan_id,
        "subscriber": {
            "email_address": subscriber_email
        },
        "application_context": {
            "return_url": "https://yourdomain.com/subscription/success",
            "cancel_url": "https://yourdomain.com/subscription/cancel"
        }
    }

    response = requests.post(url, headers=headers, json=payload, timeout=15, allow_redirects=False)
    response.raise_for_status()
    subscription = response.json()

    # Get approval URL
    for link in subscription.get('links', []):
        if link['rel'] == 'approve':
            return {
                'subscription_id': subscription['id'],
                'approval_url': link['href']
            }

退款工作流

def create_refund(capture_id, request_id, amount=None, note=None):
    """Create a refund for a captured payment."""
    client = PayPalClient(CLIENT_ID, CLIENT_SECRET)

    url = f"{client.base_url}/v2/payments/captures/{capture_id}/refund"
    headers = {
        "Content-Type": "application/json",
        "PayPal-Request-Id": request_id,
        "Authorization": f"Bearer {client.access_token}"
    }

    payload = {}
    if amount is not None:
        payload["amount"] = {
            "value": str(amount),
            "currency_code": "USD"
        }

    # Validate positive Decimal refund <= remaining captured balance and capture currency first.
    if note:
        payload["note_to_payer"] = note

    response = requests.post(url, headers=headers, json=payload, timeout=15, allow_redirects=False)
    response.raise_for_status()
    return response.json()

def get_refund_details(refund_id):
    """Get refund details."""
    client = PayPalClient(CLIENT_ID, CLIENT_SECRET)

    url = f"{client.base_url}/v2/payments/refunds/{refund_id}"
    headers = {
        "Authorization": f"Bearer {client.access_token}"
    }

    response = requests.get(url, headers=headers, timeout=15, allow_redirects=False)
    response.raise_for_status()
    return response.json()

错误处理

class PayPalError(Exception):
    """Custom PayPal error."""
    pass

def handle_paypal_api_call(api_function):
    """Wrapper for PayPal API calls with error handling."""
    try:
        result = api_function()
        return result
    except requests.exceptions.RequestException as e:
        # Network error
        raise PayPalError(f"Network error: {str(e)}")
    except Exception as e:
        # Other errors
        raise PayPalError(f"PayPal API error: {str(e)}")

# Usage
try:
    order = handle_paypal_api_call(lambda: client.create_order("25.00", persisted_request_id))
except PayPalError as e:
    # Handle error appropriately
    log_error(e)

测试

# Use sandbox credentials
import os
SANDBOX_CLIENT_ID = os.environ["PAYPAL_SANDBOX_CLIENT_ID"]
SANDBOX_SECRET = os.environ["PAYPAL_SANDBOX_SECRET"]

# Test accounts
# Create test buyer and seller accounts at developer.paypal.com

def test_payment_flow():
    """Test complete payment flow."""
    client = PayPalClient(SANDBOX_CLIENT_ID, SANDBOX_SECRET, mode='sandbox')

    # Create order
    order = client.create_order("10.00", "sandbox-test-order-001")
    assert 'id' in order

    # Get approval URL
    approval_url = next((link['href'] for link in order['links'] if link['rel'] == 'approve'), None)
    assert approval_url is not None

    # After approval (manual step with test account)
    # Capture order
    # captured = client.capture_order(order['id'], 'sandbox-test-capture-001')
    # assert captured['status'] == 'COMPLETED'

资源

• references/express-checkout.md：Express Checkout 实现指南
• references/ipn-handling.md：IPN 验证与处理
• references/refund-workflows.md：退款处理模式
• references/billing-agreements.md：定期计费设置
• assets/paypal-client.py：生产环境 PayPal 客户端
• assets/ipn-processor.py：IPN webhook 处理器
• assets/recurring-billing.py：订阅管理

最佳实践

1. 始终验证 IPN：未经验证，绝不信任 IPN
2. 幂等处理：处理重复的 IPN 通知
3. 错误处理：实现健壮的错误处理
4. 日志记录：仅记录最少的事件 ID/状态；对令牌、付款人标识符和原始服务提供商错误进行脱敏
5. 充分测试：广泛使用沙盒
6. Webhook 备用机制：不要只依赖客户端回调
7. 货币处理：始终明确指定货币

常见问题

• 未验证 IPN：未经验证便接受 IPN
• 重复处理：未检查重复交易
• 环境错误：混用沙盒和生产环境的 URL/凭据
• Webhook 缺失：未处理所有支付状态
• 硬编码值：未将其设为可配置项以适配不同环境

限制

• 仅当任务明确符合上述范围时，才使用此技能。
• 不要将输出视为环境专属验证、测试或专家审查的替代品。
• 如果缺少必需的输入、权限、安全边界或成功标准，请停止并要求澄清。