taijibaga/taiji-AI-PAD
9
0
Fork 0
forked from xiaohei/taiji-AI-PAD
Code Pull Requests Activity
Files
feature/chenchen
taiji-AI-PAD/plans/PayPal支付集成-后端实施文档.md
T
zhanggangyong a3d7ef6a6e 备份
2026-03-15 15:30:24 +00:00

39 KiB
Raw Permalink Blame History

PayPal 支付集成 - 后端实施文档

1. 概述

1.1 需求背景

用户在使用平台 Agent 时，当 EU 余额不足时可以通过 PayPal 在线充值。

1.2 核心概念

  • EU（执行单元）：系统计费单位，1 EU = 1 USD
  • 充值流程：用户支付 USD → 系统增加等额 EU 余额

1.3 PayPal 凭证

  • 环境：Sandbox（测试环境）
  • Client ID：AWJcBVeccSgDDhcZcYEbf4SJKxq9Uk_qVNlvk9mCewzl9o1Cp0onPzOD-v26-Mye9F1cKF6SzipuTtQZ
  • Secret：EJVeShfyCCTLvNejkhs6F943gYfNyNFkbmw-CFOA0VEGLsiqic0GPthYzVQLBajzH-v8PVpJ0SM51ccL

⚠️ 重要提醒：这是客户支付的真金白银，代码必须严谨，所有金额操作必须有事务保护和审计日志。

2. API 接口设计

2.1 接口列表

接口 方法 描述
/api/user/billing/paypal/create-order POST 创建 PayPal 订单
/api/user/billing/paypal/capture-order POST 捕获支付（验证并完成充值）
/api/user/billing/paypal/order/{order_id} GET 查询订单状态
/whitelist/payment/callback/paypal POST PayPal Webhook 回调

2.2 创建订单

请求

POST /api/user/billing/paypal/create-order
Authorization: Bearer <token>
Content-Type: application/json

{
    "amount": 10.00,
    "currency": "USD"
}

响应

{
    "success": true,
    "data": {
        "orderId": "5O190127TN364715T",
        "status": "CREATED",
        "amount": 10.00,
        "currency": "USD",
        "euAmount": 10.00
    }
}

2.3 捕获支付

请求

POST /api/user/billing/paypal/capture-order
Authorization: Bearer <token>
Content-Type: application/json

{
    "orderId": "5O190127TN364715T"
}

响应

{
    "success": true,
    "data": {
        "orderId": "5O190127TN364715T",
        "status": "COMPLETED",
        "amount": 10.00,
        "euAmount": 10.00,
        "newBalance": 110.00,
        "captureId": "3C679366HH908993F",
        "payerEmail": "buyer@example.com"
    },
    "message": "充值成功，已增加 10.00 EU"
}

2.4 查询订单状态

请求

GET /api/user/billing/paypal/order/5O190127TN364715T
Authorization: Bearer <token>

响应

{
    "success": true,
    "data": {
        "orderId": "5O190127TN364715T",
        "status": "captured",
        "paypalStatus": "COMPLETED",
        "amount": 10.00,
        "euAmount": 10.00,
        "createdAt": "2026-03-11T07:00:00Z",
        "capturedAt": "2026-03-11T07:01:00Z"
    }
}

2.5 Webhook 回调

请求

POST /whitelist/payment/callback/paypal
Content-Type: application/json
PayPal-Transmission-Id: <transmission_id>
PayPal-Transmission-Time: <timestamp>
PayPal-Transmission-Sig: <signature>
PayPal-Cert-Url: <cert_url>
PayPal-Auth-Algo: <algorithm>

{
    "id": "WH-XXX",
    "event_type": "PAYMENT.CAPTURE.COMPLETED",
    "resource": {
        "id": "3C679366HH908993F",
        "status": "COMPLETED",
        "amount": {
            "currency_code": "USD",
            "value": "10.00"
        }
    }
}

响应

{
    "success": true
}

3. 支付流程

sequenceDiagram
    participant U as 用户
    participant F as 前端
    participant B as 后端 MCP-Server
    participant P as PayPal API

    Note over U,P: 阶段1: 创建订单
    U->>F: 1. 输入充值金额，点击 PayPal 按钮
    F->>B: 2. POST /api/user/billing/paypal/create-order
    Note right of B: 验证用户身份<br/>验证金额有效性
    B->>P: 3. POST /v2/checkout/orders
    P-->>B: 4. 返回 order_id, status=CREATED
    B->>B: 5. 创建 RechargeRecord 记录
    B-->>F: 6. 返回 order_id

    Note over U,P: 阶段2: 用户支付
    F->>P: 7. PayPal JS SDK 弹出支付窗口
    U->>P: 8. 用户登录 PayPal 并确认支付
    P-->>F: 9. 支付成功，返回 order_id

    Note over U,P: 阶段3: 捕获支付
    F->>B: 10. POST /api/user/billing/paypal/capture-order
    B->>P: 11. POST /v2/checkout/orders/{id}/capture
    P-->>B: 12. 返回 status=COMPLETED, 支付详情
    B->>B: 13. 验证支付金额
    B->>B: 14. 更新 RechargeRecord 状态
    B->>B: 15. 增加用户 EU 余额
    B-->>F: 16. 返回充值成功

    Note over U,P: 阶段4: 完成
    F-->>U: 17. 显示充值成功，更新余额显示

    Note over U,P: 可选: Webhook 异步通知
    P->>B: POST /whitelist/payment/callback/paypal
    B->>B: 验证签名，更新订单状态
    B-->>P: 200 OK

4. 数据库说明

4.1 使用现有表

本方案使用现有的 recharge_records 表存储 PayPal 订单，无需新建表。

RechargeRecord 表字段映射：

字段 用途 PayPal 数据
user_id 用户 ID 当前登录用户
amount 充值金额 用户输入的金额
payment_method 支付方式 固定值 "paypal"
order_id 订单号 PayPal 返回的 order_id
status 订单状态 pending → success / failed / amount_mismatch
completed_at 完成时间 支付成功时的时间戳

4.2 订单状态流转

pending (创建订单)
    ↓
    ├── success (支付成功，余额已增加)
    ├── failed (PayPal 返回非 COMPLETED 状态)
    └── amount_mismatch (金额不匹配，拒绝充值)

4.3 可选：扩展字段（如需存储更多 PayPal 信息）

如果需要存储更多 PayPal 信息（如 payer_email、capture_id），可以添加迁移：

-- 可选迁移：扩展 recharge_records 表
ALTER TABLE recharge_records ADD COLUMN paypal_capture_id VARCHAR(100);
ALTER TABLE recharge_records ADD COLUMN paypal_payer_email VARCHAR(255);
ALTER TABLE recharge_records ADD COLUMN paypal_payer_id VARCHAR(100);
ALTER TABLE recharge_records ADD COLUMN paypal_raw_response JSONB;

注意：当前方案不需要这些扩展字段，PayPal 信息已通过日志记录用于审计。

5. 实施步骤

5.1 安装依赖

在 requirements.txt 中添加：

paypal-server-sdk>=0.5.0

5.2 添加配置

在 config.py 中添加：

# PayPal 配置
paypal_client_id: str = os.getenv("PAYPAL_CLIENT_ID", "")
paypal_client_secret: str = os.getenv("PAYPAL_CLIENT_SECRET", "")
paypal_environment: str = os.getenv("PAYPAL_ENVIRONMENT", "sandbox")  # sandbox 或 production
paypal_webhook_id: str = os.getenv("PAYPAL_WEBHOOK_ID", "")  # 用于验证 Webhook 签名

@property
def paypal_api_base(self) -> str:
    if self.paypal_environment == "production":
        return "https://api-m.paypal.com"
    return "https://api-m.sandbox.paypal.com"

5.3 创建 PayPal 客户端

创建文件 app/paypal_client.py：

"""
PayPal 客户端封装
"""

import logging
from paypalserversdk.http.auth.o_auth_2 import ClientCredentialsAuthCredentials
from paypalserversdk.logging.configuration.api_logging_configuration import (
    LoggingConfiguration,
    RequestLoggingConfiguration,
    ResponseLoggingConfiguration,
)
from paypalserversdk.paypal_serversdk_client import PaypalServersdkClient
from paypalserversdk.controllers.orders_controller import OrdersController
from paypalserversdk.models.order_request import OrderRequest
from paypalserversdk.models.checkout_payment_intent import CheckoutPaymentIntent
from paypalserversdk.models.purchase_unit_request import PurchaseUnitRequest
from paypalserversdk.models.amount_with_breakdown import AmountWithBreakdown
from paypalserversdk.api_helper import ApiHelper

from config import settings
import structlog

logger = structlog.get_logger(__name__)


class PayPalClient:
    """PayPal 客户端封装"""
    
    def __init__(self):
        self._client = None
        self._orders_controller = None
    
    def _get_client(self) -> PaypalServersdkClient:
        """获取 PayPal SDK 客户端（懒加载）"""
        if self._client is None:
            self._client = PaypalServersdkClient(
                client_credentials_auth_credentials=ClientCredentialsAuthCredentials(
                    o_auth_client_id=settings.paypal_client_id,
                    o_auth_client_secret=settings.paypal_client_secret,
                ),
                environment=settings.paypal_environment,
                logging_configuration=LoggingConfiguration(
                    log_level=logging.INFO,
                    mask_sensitive_headers=True,
                    request_logging_config=RequestLoggingConfiguration(
                        log_body=True
                    ),
                    response_logging_config=ResponseLoggingConfiguration(
                        log_body=True
                    ),
                ),
            )
        return self._client
    
    @property
    def orders_controller(self) -> OrdersController:
        """获取订单控制器"""
        if self._orders_controller is None:
            self._orders_controller = self._get_client().orders
        return self._orders_controller
    
    async def create_order(self, amount: float, currency: str = "USD") -> dict:
        """
        创建 PayPal 订单
        
        Args:
            amount: 支付金额
            currency: 货币类型，默认 USD
            
        Returns:
            包含 order_id 和状态的字典
        """
        order_request = OrderRequest(
            intent=CheckoutPaymentIntent.CAPTURE,
            purchase_units=[
                PurchaseUnitRequest(
                    amount=AmountWithBreakdown(
                        currency_code=currency,
                        value=f"{amount:.2f}",
                    ),
                    description="Taiji AI Platform - EU Balance Recharge",
                )
            ],
        )
        
        try:
            response = self.orders_controller.orders_create({"body": order_request})
            
            if response.body:
                order = response.body
                logger.info(
                    "PayPal 订单创建成功",
                    order_id=order.id,
                    status=order.status,
                    amount=amount
                )
                return {
                    "order_id": order.id,
                    "status": order.status,
                    "links": [{"rel": link.rel, "href": link.href} for link in order.links] if order.links else []
                }
            else:
                raise Exception("PayPal 返回空响应")
                
        except Exception as e:
            logger.error("PayPal 订单创建失败", error=str(e))
            raise
    
    async def capture_order(self, order_id: str) -> dict:
        """
        捕获 PayPal 订单（完成支付）
        
        Args:
            order_id: PayPal 订单 ID
            
        Returns:
            包含支付详情的字典
        """
        try:
            response = self.orders_controller.orders_capture({"id": order_id})
            
            if response.body:
                order = response.body
                
                # 提取支付信息
                capture_info = {}
                if order.purchase_units and len(order.purchase_units) > 0:
                    pu = order.purchase_units[0]
                    if pu.payments and pu.payments.captures and len(pu.payments.captures) > 0:
                        capture = pu.payments.captures[0]
                        capture_info = {
                            "capture_id": capture.id,
                            "amount": float(capture.amount.value) if capture.amount else 0,
                            "currency": capture.amount.currency_code if capture.amount else "USD",
                        }
                
                # 提取付款人信息
                payer_info = {}
                if order.payer:
                    payer_info = {
                        "payer_id": order.payer.payer_id,
                        "email": order.payer.email_address,
                        "name": f"{order.payer.name.given_name} {order.payer.name.surname}" if order.payer.name else None,
                    }
                
                logger.info(
                    "PayPal 订单捕获成功",
                    order_id=order.id,
                    status=order.status,
                    capture_id=capture_info.get("capture_id")
                )
                
                return {
                    "order_id": order.id,
                    "status": order.status,
                    **capture_info,
                    **payer_info,
                    "raw_response": ApiHelper.json_serialize(order),
                }
            else:
                raise Exception("PayPal 返回空响应")
                
        except Exception as e:
            logger.error("PayPal 订单捕获失败", order_id=order_id, error=str(e))
            raise
    
    async def get_order(self, order_id: str) -> dict:
        """
        查询 PayPal 订单状态
        
        Args:
            order_id: PayPal 订单 ID
            
        Returns:
            订单详情
        """
        try:
            response = self.orders_controller.orders_get({"id": order_id})
            
            if response.body:
                order = response.body
                return {
                    "order_id": order.id,
                    "status": order.status,
                    "create_time": order.create_time,
                    "update_time": order.update_time,
                }
            else:
                raise Exception("PayPal 返回空响应")
                
        except Exception as e:
            logger.error("PayPal 订单查询失败", order_id=order_id, error=str(e))
            raise


# 单例
_paypal_client = None

def get_paypal_client() -> PayPalClient:
    """获取 PayPal 客户端单例"""
    global _paypal_client
    if _paypal_client is None:
        _paypal_client = PayPalClient()
    return _paypal_client

5.4 创建路由

创建文件 app/routes/paypal.py：

"""
PayPal 支付路由
用户在线充值功能

重要：这是客户支付的真金白银，所有操作必须：
1. 有完整的审计日志
2. 有事务保护
3. 有金额验证
"""

import structlog
from datetime import datetime
from decimal import Decimal
from fastapi import APIRouter, Depends, HTTPException, status, Request
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from pydantic import BaseModel, Field

from database import get_db
from models import User, Balance, RechargeRecord
from app.auth import require_auth
from app.schemas import SuccessResponse
from app.paypal_client import get_paypal_client
from app.billing import add_balance
from app.audit import log_audit_event  # 审计日志

logger = structlog.get_logger(__name__)
router = APIRouter(prefix="/api/user/billing/paypal", tags=["PayPal支付"])
webhook_router = APIRouter(tags=["PayPal Webhook"])


# ============= 请求/响应模型 =============

class CreatePayPalOrderRequest(BaseModel):
    """创建 PayPal 订单请求"""
    amount: float = Field(..., gt=0, le=10000, description="充值金额 (USD)")
    currency: str = Field("USD", pattern="^USD$", description="货币类型，目前仅支持 USD")


class CapturePayPalOrderRequest(BaseModel):
    """捕获 PayPal 订单请求"""
    orderId: str = Field(..., min_length=1, description="PayPal 订单 ID")


# ============= 用户路由 =============

@router.post("/create-order", response_model=SuccessResponse)
async def create_paypal_order(
    req: CreatePayPalOrderRequest,
    request: Request,  # 用于审计日志获取 IP
    principal: dict = Depends(require_auth),
    db: AsyncSession = Depends(get_db)
):
    """
    创建 PayPal 订单
    
    用户发起充值请求，后端创建 PayPal 订单并返回订单 ID。
    前端使用订单 ID 调用 PayPal JS SDK 弹出支付窗口。
    
    金额说明：
    - 1 USD = 1 EU
    - 最小充值金额：1 USD
    - 最大充值金额：10000 USD
    """
    user_id = principal.get("user_id")
    order_id = None  # 用于审计日志
    
    # 验证用户存在
    result = await db.execute(select(User).where(User.id == user_id))
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="用户不存在"
        )
    
    try:
        # 创建 PayPal 订单
        paypal_client = get_paypal_client()
        order_result = await paypal_client.create_order(
            amount=req.amount,
            currency=req.currency
        )
        order_id = order_result["order_id"]
        
        # 创建本地订单记录
        recharge_record = RechargeRecord(
            user_id=user_id,
            amount=Decimal(str(req.amount)),
            payment_method="paypal",
            order_id=order_id,
            status="pending",
        )
        db.add(recharge_record)
        await db.commit()
        
        logger.info(
            "PayPal 订单创建成功",
            user_id=user_id,
            order_id=order_id,
            amount=req.amount
        )
        
        # ========== 审计日志：订单创建成功 ==========
        await log_audit_event(
            action="payment.paypal.order_created",
            resource_type="paypal_order",
            resource_id=order_id,
            user_id=str(user_id),
            success=True,
            details={
                "amount": req.amount,
                "currency": req.currency,
                "eu_amount": req.amount,
                "paypal_status": order_result["status"],
            },
            request=request,
            db=db
        )
        
        return SuccessResponse(
            data={
                "orderId": order_id,
                "status": order_result["status"],
                "amount": req.amount,
                "currency": req.currency,
                "euAmount": req.amount,  # 1:1 转换
            },
            message="订单创建成功，请完成支付"
        )
        
    except Exception as e:
        logger.error("创建 PayPal 订单失败", user_id=user_id, error=str(e))
        
        # ========== 审计日志：订单创建失败 ==========
        await log_audit_event(
            action="payment.paypal.order_created",
            resource_type="paypal_order",
            resource_id=order_id,
            user_id=str(user_id),
            success=False,
            details={"amount": req.amount, "currency": req.currency},
            error_message=str(e),
            request=request,
            db=db
        )
        
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"创建订单失败: {str(e)}"
        )


@router.post("/capture-order", response_model=SuccessResponse)
async def capture_paypal_order(
    req: CapturePayPalOrderRequest,
    request: Request,  # 用于审计日志获取 IP
    principal: dict = Depends(require_auth),
    db: AsyncSession = Depends(get_db)
):
    """
    捕获 PayPal 订单（完成支付）
    
    用户在 PayPal 完成支付后，前端调用此接口验证支付并完成充值。
    
    流程：
    1. 验证订单属于当前用户
    2. 调用 PayPal API 捕获支付
    3. 验证支付金额
    4. 增加用户 EU 余额
    5. 更新订单状态
    """
    user_id = principal.get("user_id")
    
    # 查找本地订单记录
    result = await db.execute(
        select(RechargeRecord).where(
            RechargeRecord.order_id == req.orderId,
            RechargeRecord.user_id == user_id,
            RechargeRecord.payment_method == "paypal"
        )
    )
    recharge_record = result.scalar_one_or_none()
    
    if not recharge_record:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="订单不存在或不属于当前用户"
        )
    
    if recharge_record.status == "success":
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="订单已完成，请勿重复操作"
        )
    
    try:
        # 捕获 PayPal 订单
        paypal_client = get_paypal_client()
        capture_result = await paypal_client.capture_order(req.orderId)
        
        if capture_result["status"] != "COMPLETED":
            recharge_record.status = "failed"
            await db.commit()
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail=f"支付未完成，状态: {capture_result['status']}"
            )
        
        # 验证支付金额（严格模式：金额必须完全匹配）
        captured_amount = Decimal(str(capture_result.get("amount", 0)))
        expected_amount = recharge_record.amount
        
        if captured_amount != expected_amount:
            logger.error(
                "支付金额不匹配，拒绝充值",
                order_id=req.orderId,
                expected=float(expected_amount),
                captured=float(captured_amount),
                user_id=user_id
            )
            recharge_record.status = "amount_mismatch"
            await db.commit()
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail=f"支付金额不匹配: 期望 ${expected_amount}, 实际 ${captured_amount}"
            )
        
        # 增加用户余额
        eu_amount = captured_amount  # 1 USD = 1 EU
        success, message = await add_balance(
            user_id=user_id,
            amount=eu_amount,
            db=db,
            description=f"PayPal 充值 (订单: {req.orderId})"
        )
        
        if not success:
            raise HTTPException(
                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail=f"余额更新失败: {message}"
            )
        
        # 更新订单状态
        recharge_record.status = "success"
        recharge_record.completed_at = datetime.utcnow()
        
        await db.commit()
        
        # 获取新余额
        balance_result = await db.execute(
            select(Balance).where(Balance.user_id == user_id)
        )
        balance = balance_result.scalar_one_or_none()
        new_balance = float(balance.eu_balance) if balance else 0
        
        logger.info(
            "PayPal 充值成功",
            user_id=user_id,
            order_id=req.orderId,
            amount=float(captured_amount),
            new_balance=new_balance
        )
        
        # ========== 审计日志：支付成功 ==========
        await log_audit_event(
            action="payment.paypal.order_captured",
            resource_type="paypal_order",
            resource_id=req.orderId,
            user_id=str(user_id),
            success=True,
            details={
                "amount": float(captured_amount),
                "eu_amount": float(eu_amount),
                "new_balance": new_balance,
                "capture_id": capture_result.get("capture_id"),
                "payer_email": capture_result.get("email"),
                "payer_id": capture_result.get("payer_id"),
            },
            request=request,
            db=db
        )
        
        return SuccessResponse(
            data={
                "orderId": req.orderId,
                "status": "COMPLETED",
                "amount": float(captured_amount),
                "euAmount": float(eu_amount),
                "newBalance": new_balance,
                "captureId": capture_result.get("capture_id"),
                "payerEmail": capture_result.get("email"),
            },
            message=f"充值成功，已增加 {eu_amount} EU"
        )
        
    except HTTPException as e:
        await db.rollback()  # 确保回滚
        
        # ========== 审计日志：业务失败 ==========
        action = "payment.paypal.amount_mismatch" if "金额不匹配" in str(e.detail) else "payment.paypal.order_failed"
        await log_audit_event(
            action=action,
            resource_type="paypal_order",
            resource_id=req.orderId,
            user_id=str(user_id),
            success=False,
            details={
                "expected_amount": float(recharge_record.amount) if recharge_record else None,
                "status_code": e.status_code,
            },
            error_message=str(e.detail),
            request=request,
            db=db
        )
        raise
        
    except Exception as e:
        await db.rollback()  # 确保回滚
        logger.error(
            "捕获 PayPal 订单失败",
            order_id=req.orderId,
            user_id=user_id,
            error=str(e),
            error_type=type(e).__name__
        )
        
        # ========== 审计日志：系统异常 ==========
        await log_audit_event(
            action="payment.paypal.order_failed",
            resource_type="paypal_order",
            resource_id=req.orderId,
            user_id=str(user_id),
            success=False,
            details={"error_type": type(e).__name__},
            error_message=str(e),
            request=request,
            db=db
        )
        
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"支付验证失败: {str(e)}"
        )


@router.get("/order/{order_id}", response_model=SuccessResponse)
async def get_paypal_order(
    order_id: str,
    principal: dict = Depends(require_auth),
    db: AsyncSession = Depends(get_db)
):
    """
    查询 PayPal 订单状态
    """
    user_id = principal.get("user_id")
    
    # 查找本地订单记录
    result = await db.execute(
        select(RechargeRecord).where(
            RechargeRecord.order_id == order_id,
            RechargeRecord.user_id == user_id,
            RechargeRecord.payment_method == "paypal"
        )
    )
    recharge_record = result.scalar_one_or_none()
    
    if not recharge_record:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="订单不存在"
        )
    
    return SuccessResponse(
        data={
            "orderId": order_id,
            "status": recharge_record.status,
            "amount": float(recharge_record.amount),
            "euAmount": float(recharge_record.amount),
            "createdAt": recharge_record.created_at.isoformat() if recharge_record.created_at else None,
            "completedAt": recharge_record.completed_at.isoformat() if recharge_record.completed_at else None,
        }
    )


# ============= Webhook 路由 =============

@webhook_router.post("/whitelist/payment/callback/paypal")
async def paypal_webhook(
    request: Request,
    db: AsyncSession = Depends(get_db)
):
    """
    PayPal Webhook 回调
    
    用于接收 PayPal 异步通知，处理边缘情况（如用户关闭页面但支付已完成）。
    
    支持的事件类型：
    - PAYMENT.CAPTURE.COMPLETED: 支付捕获完成
    - PAYMENT.CAPTURE.DENIED: 支付被拒绝
    - PAYMENT.CAPTURE.REFUNDED: 支付已退款
    """
    try:
        # 获取请求体
        body = await request.json()
        
        # 获取 PayPal 签名头
        headers = {
            "PayPal-Transmission-Id": request.headers.get("PayPal-Transmission-Id"),
            "PayPal-Transmission-Time": request.headers.get("PayPal-Transmission-Time"),
            "PayPal-Transmission-Sig": request.headers.get("PayPal-Transmission-Sig"),
            "PayPal-Cert-Url": request.headers.get("PayPal-Cert-Url"),
            "PayPal-Auth-Algo": request.headers.get("PayPal-Auth-Algo"),
        }
        
        # TODO: 验证 Webhook 签名
        # 生产环境必须验证签名，防止伪造请求
        # 使用 PayPal Webhook Signature Verification API
        
        event_type = body.get("event_type")
        resource = body.get("resource", {})
        
        logger.info(
            "收到 PayPal Webhook",
            event_type=event_type,
            resource_id=resource.get("id")
        )
        
        if event_type == "PAYMENT.CAPTURE.COMPLETED":
            # 支付完成
            capture_id = resource.get("id")
            # 根据 capture_id 查找并更新订单
            # 这里作为备份机制，主要逻辑在 capture-order 接口
            pass
            
        elif event_type == "PAYMENT.CAPTURE.DENIED":
            # 支付被拒绝
            capture_id = resource.get("id")
            logger.warning("PayPal 支付被拒绝", capture_id=capture_id)
            
        elif event_type == "PAYMENT.CAPTURE.REFUNDED":
            # 支付已退款
            capture_id = resource.get("id")
            logger.info("PayPal 支付已退款", capture_id=capture_id)
            # TODO: 处理退款逻辑，扣减用户余额
        
        return {"success": True}
        
    except Exception as e:
        logger.error("处理 PayPal Webhook 失败", error=str(e))
        # 返回 200 避免 PayPal 重试
        return {"success": False, "error": str(e)}

5.5 注册路由

在 main.py 中注册路由：

from app.routes.paypal import router as paypal_router, webhook_router as paypal_webhook_router

# 用户路由（需要认证）
app.include_router(paypal_router)

# Webhook 路由（无需认证，在 whitelist 路径下）
app.include_router(paypal_webhook_router)

5.6 环境变量配置

在 .env 文件中添加：

# PayPal Sandbox 配置（测试环境）
PAYPAL_CLIENT_ID=AWJcBVeccSgDDhcZcYEbf4SJKxq9Uk_qVNlvk9ewzl9o1Cp0onPzOD-v26-Mye9F1cKF6SzipuTtQZ
PAYPAL_CLIENT_SECRET=EJVeShfyCCTLvNejkhs6F943gYfNyNFkbmw-CFOA0VEGLsiqic0GPthYzVQLBajzH-v8PVpJ0SM51ccL
PAYPAL_ENVIRONMENT=sandbox
PAYPAL_WEBHOOK_ID=<在 PayPal Developer 配置 Webhook 后获取>

6. 生产环境部署

6.1 获取 Live 凭证

  1. 登录 https://developer.paypal.com
  2. 进入 Dashboard → My Apps & Credentials
  3. 切换到 "Live" 标签
  4. 创建新应用或查看现有应用的 Live 凭证

6.2 配置 Webhook

  1. 在 PayPal Developer Dashboard 中配置 Webhook
  2. Webhook URL: https://your-domain.com/whitelist/payment/callback/paypal
  3. 选择事件类型：
    • PAYMENT.CAPTURE.COMPLETED
    • PAYMENT.CAPTURE.DENIED
    • PAYMENT.CAPTURE.REFUNDED
  4. 保存后获取 Webhook ID

6.3 更新生产环境变量

PAYPAL_CLIENT_ID=<Live Client ID>
PAYPAL_CLIENT_SECRET=<Live Client Secret>
PAYPAL_ENVIRONMENT=production
PAYPAL_WEBHOOK_ID=<Live Webhook ID>

7. 安全考虑

7.1 关键安全措施（必须实现）

安全措施 说明 实现位置
金额验证 捕获支付时验证 PayPal 返回的实际金额与订单金额一致 capture_paypal_order()
幂等性保护 检查订单状态，防止重复充值（status == "success" 时拒绝） capture_paypal_order()
用户归属验证 确保订单属于当前登录用户（user_id 匹配） capture_paypal_order()
数据库事务 余额更新和订单状态更新在同一事务中，失败时回滚 capture_paypal_order()
行锁保护 add_balance() 使用 FOR UPDATE 锁防止并发超充 billing.py
日志审计 记录所有支付操作，包括 user_id、order_id、amount 全部接口

7.2 金额验证逻辑（重要）

# 在 capture_paypal_order 中必须验证金额
captured_amount = Decimal(str(capture_result.get("amount", 0)))
expected_amount = recharge_record.amount

# 严格模式：金额必须完全匹配
if captured_amount != expected_amount:
    logger.error(
        "支付金额不匹配，拒绝充值",
        order_id=req.orderId,
        expected=float(expected_amount),
        captured=float(captured_amount)
    )
    recharge_record.status = "amount_mismatch"
    await db.commit()
    raise HTTPException(
        status_code=status.HTTP_400_BAD_REQUEST,
        detail=f"支付金额不匹配: 期望 {expected_amount}, 实际 {captured_amount}"
    )

7.3 事务安全

# 确保余额更新和订单状态更新在同一事务中
try:
    # 1. 增加用户余额（使用行锁）
    success, message = await add_balance(
        user_id=user_id,
        amount=eu_amount,
        db=db,
        description=f"PayPal 充值 (订单: {req.orderId})"
    )
    
    if not success:
        raise HTTPException(...)
    
    # 2. 更新订单状态
    recharge_record.status = "success"
    recharge_record.completed_at = datetime.utcnow()
    
    # 3. 统一提交
    await db.commit()
    
except Exception as e:
    # 4. 失败时回滚
    await db.rollback()
    raise

7.4 Webhook 安全

  1. 签名验证：生产环境必须验证 PayPal Webhook 签名
  2. 幂等处理：同一事件可能多次推送，需要幂等处理
  3. HTTPS：Webhook URL 必须使用 HTTPS
  4. IP 白名单：可选，限制只接受 PayPal IP 的请求

7.5 防欺诈检查

# 可选：添加防欺诈检查
async def check_fraud_risk(user_id: str, amount: Decimal, db: AsyncSession) -> bool:
    """
    检查欺诈风险
    - 短时间内多次大额充值
    - 新用户首次大额充值
    - 异常 IP 地址
    """
    # 检查最近 1 小时内的充值次数
    recent_count = await db.execute(
        select(func.count(RechargeRecord.id))
        .where(RechargeRecord.user_id == user_id)
        .where(RechargeRecord.created_at > datetime.utcnow() - timedelta(hours=1))
    )
    if recent_count.scalar() > 5:
        return False  # 风险过高
    
    return True

8. 审计日志与问题追溯

8.1 审计日志集成

项目已有完善的审计日志系统 app/audit.py，PayPal 支付需要集成以下审计事件：

新增审计操作类型（在 AUDIT_ACTIONS 中添加）：

# 在 app/audit.py 的 AUDIT_ACTIONS 中添加
"payment.paypal.order_created": "PayPal订单创建",
"payment.paypal.order_captured": "PayPal支付完成",
"payment.paypal.order_failed": "PayPal支付失败",
"payment.paypal.amount_mismatch": "PayPal金额不匹配",
"payment.paypal.webhook_received": "PayPal Webhook接收",

8.2 关键节点日志记录

在路由代码中添加审计日志调用：

from app.audit import log_audit_event

# ========== 创建订单时记录 ==========
@router.post("/create-order")
async def create_paypal_order(...):
    try:
        # ... 创建订单逻辑 ...
        
        # 记录成功
        await log_audit_event(
            action="payment.paypal.order_created",
            resource_type="paypal_order",
            resource_id=order_result["order_id"],
            user_id=str(user_id),
            success=True,
            details={
                "amount": req.amount,
                "currency": req.currency,
                "eu_amount": req.amount,
                "paypal_status": order_result["status"],
            },
            request=request,
            db=db
        )
        
    except Exception as e:
        # 记录失败
        await log_audit_event(
            action="payment.paypal.order_created",
            resource_type="paypal_order",
            resource_id=None,
            user_id=str(user_id),
            success=False,
            details={"amount": req.amount, "currency": req.currency},
            error_message=str(e),
            request=request,
            db=db
        )
        raise


# ========== 捕获支付时记录 ==========
@router.post("/capture-order")
async def capture_paypal_order(...):
    try:
        # ... 捕获支付逻辑 ...
        
        # 记录成功
        await log_audit_event(
            action="payment.paypal.order_captured",
            resource_type="paypal_order",
            resource_id=req.orderId,
            user_id=str(user_id),
            success=True,
            details={
                "amount": float(captured_amount),
                "eu_amount": float(eu_amount),
                "new_balance": new_balance,
                "capture_id": capture_result.get("capture_id"),
                "payer_email": capture_result.get("email"),
            },
            request=request,
            db=db
        )
        
    except HTTPException as e:
        # 记录业务失败（金额不匹配等）
        action = "payment.paypal.amount_mismatch" if "金额不匹配" in str(e.detail) else "payment.paypal.order_failed"
        await log_audit_event(
            action=action,
            resource_type="paypal_order",
            resource_id=req.orderId,
            user_id=str(user_id),
            success=False,
            details={"expected_amount": float(recharge_record.amount) if recharge_record else None},
            error_message=str(e.detail),
            request=request,
            db=db
        )
        raise

8.3 日志查询与问题追溯

按订单号查询：

# 查询某个 PayPal 订单的所有日志
logs = await query_audit_logs(
    resource_type="paypal_order",
    resource_id="5O190127TN364715T",
    db=db
)

按用户查询充值历史：

# 查询用户所有 PayPal 相关操作
logs = await query_audit_logs(
    user_id=user_id,
    action="payment.paypal.order_captured",  # 或不传 action 查询所有
    start_date=datetime(2026, 3, 1),
    end_date=datetime(2026, 3, 31),
    db=db
)

查询失败的支付：

# 查询所有失败的 PayPal 支付
logs = await query_audit_logs(
    resource_type="paypal_order",
    success=False,
    db=db
)

8.4 日志字段说明

字段 说明 示例
action 操作类型 payment.paypal.order_captured
resource_type 资源类型 paypal_order
resource_id PayPal 订单号 5O190127TN364715T
user_id 用户 ID uuid
success 是否成功 true/false
details.amount 支付金额 10.00
details.eu_amount EU 数量 10.00
details.new_balance 充值后余额 110.00
details.capture_id PayPal 捕获 ID 3C679366HH908993F
details.payer_email 付款人邮箱 buyer@example.com
error_message 错误信息 支付金额不匹配: 期望 $10.00, 实际 $5.00
ip_address 客户端 IP 192.168.1.1
created_at 操作时间 2026-03-11T08:00:00Z

8.5 问题追溯流程

场景：用户反馈充值未到账

  1. 获取订单号：从用户处获取 PayPal 订单号或充值时间

  2. 查询审计日志：

    SELECT * FROM audit_logs
WHERE resource_type = 'paypal_order'
AND resource_id = '5O190127TN364715T'
ORDER BY created_at;

  3. 分析日志：

    • 如果只有 order_created 没有 order_captured → 用户未完成支付
    • 如果有 order_failed → 查看 error_message 了解失败原因
    • 如果有 amount_mismatch → 金额被篡改，需要人工处理
    • 如果有 order_captured 且 success=true → 充值成功，检查余额表

  4. 检查数据库：

    -- 检查充值记录
SELECT * FROM recharge_records WHERE order_id = '5O190127TN364715T';

-- 检查用户余额
SELECT * FROM balances WHERE user_id = '<user_id>';

  5. 检查 PayPal 后台：

    • 登录 PayPal Developer Dashboard
    • 查看订单状态和交易详情

9. 任务清单

  • 安装 paypal-server-sdk 依赖
  • 添加 PayPal 配置到 config.py
  • 在 app/audit.py 添加 PayPal 审计操作类型
  • 创建 app/paypal_client.py 客户端封装
  • 创建 app/routes/paypal.py 路由（含审计日志）
  • 在 app/routes/__init__.py 注册路由
  • 添加环境变量配置
  • 编写单元测试
  • 配置 PayPal Webhook
View Git Blame Copy Permalink