从零到一开发支付库：电商系统对接微信支付支付宝全链路实战案例与避坑指南

写这篇文章的动机，源于我踩过的每一个坑，以及深夜帮同学改支付对接时的那句”这参数怎么和文档说得不一样”。

---

支付对接这件事，到底在对接什么

很多人第一次做支付对接，以为就是调个接口、传个参数，钱就到账了。

实际上，支付是一个闭环系统：

用户下单 → 创建支付请求 → 唤起支付 → 用户付款 → 支付结果回调 → 
通知商户 → 更新订单 → 退款/售后 → 对账

每一步都可能出问题，而你的支付库要处理的，正是这一整条链路。

我见过太多项目只做了”创建订单”和”接收回调”，结果退款时对账对不上，半夜被运维电话叫醒，查出来的问题是一个签名验证漏了——这种坑，后面会详细讲。

---

设计一个支付库，先想清楚这五件事

1. 统一接口，屏蔽差异

微信和支付宝的接口风格完全不同：

• 微信用的是 HTTP POST + XML/JSON + 签名
• 支付宝用的是 HTTP GET/POST + 签名 + 回调

但你的上层业务代码不应该关心这些差异。所以我的做法是定义一个统一的接口：

from abc import ABC, abstractmethod
from dataclasses import dataclass
from enum import Enum
from typing import Optional
import uuid
from datetime import datetime


class PaymentChannel(Enum):
    WECHAT = "wechat"
    ALIPAY = "alipay"


class PaymentStatus(Enum):
    PENDING = "pending"
    SUCCESS = "success"
    FAILED = "failed"
    CLOSED = "closed"
    REFUNDING = "refunding"
    REFUNDED = "refunded"


@dataclass
class PaymentOrder:
    """统一的支付订单模型"""
    order_id: str                    # 业务订单号
    channel: PaymentChannel          # 支付渠道
    amount: int                      # 金额（分）
    currency: str                    # 货币单位，默认 CNY
    description: str                 # 商品描述
    notify_url: str                  # 回调地址
    return_url: str                  # 跳转地址
    external_trace_id: str           # 第三方流水号
    status: PaymentStatus = PaymentStatus.PENDING
    created_at: datetime = None
    updated_at: datetime = None
    metadata: dict = None            # 扩展字段

    def __post_init__(self):
        if self.created_at is None:
            self.created_at = datetime.now()
        if self.updated_at is None:
            self.updated_at = self.created_at
        if self.metadata is None:
            self.metadata = {}

上层调用方只需要这样用：

# 业务代码完全不用关心是微信还是支付宝
order = PaymentOrder(
    order_id="20240101ABC123",
    channel=PaymentChannel.WECHAT,
    amount=9900,          # 99元，单位分
    description="VIP会员订阅",
    notify_url="https://your-domain.com/pay/callback"
)

result = payment_gateway.create_payment(order)
# result 里有一个 prepay_id 或 alipay_trade_no，用于唤起支付

2. 策略模式 + 工厂模式，别用 if-else

一开始很多人会这么写：

# 千万别这么写
if channel == "wechat":
    # 微信逻辑
    ...
elif channel == "alipay":
    # 支付宝逻辑
    ...

这段代码每加一个渠道就要改一遍，而且逻辑越来越乱。我用的是策略模式：

from abc import ABC, abstractmethod
from typing import Dict, Type

class PaymentChannelStrategy(ABC):
    """支付渠道策略接口"""

    @abstractmethod
    def create_payment(self, order: PaymentOrder) -> dict:
        """创建支付订单，返回唤起支付所需参数"""
        pass

    @abstractmethod
    def handle_callback(self, raw_data: dict) -> dict:
        """处理支付回调，返回解析后的支付结果"""
        pass

    @abstractmethod
    def query_payment(self, order_id: str) -> dict:
        """查询支付状态"""
        pass

    @abstractmethod
    def refund(self, order_id: str, amount: int, reason: str = "") -> dict:
        """发起退款"""
        pass

    @abstractmethod
    def verify_signature(self, data: dict, signature: str) -> bool:
        """验证回调签名"""
        pass


class WechatPaymentStrategy(PaymentChannelStrategy):
    def create_payment(self, order: PaymentOrder) -> dict:
        # 微信JSAPI/APP/小程序/H5各有不同的接口
        # 这里以JSAPI为例
        pass

    def handle_callback(self, raw_data: dict) -> dict:
        # 微信回调数据是XML，需要解析
        pass

    def verify_signature(self, data: dict, signature: str) -> bool:
        # 微信V3用的是HMAC-SHA256，V2用的是MD5
        pass


class AlipayPaymentStrategy(PaymentChannelStrategy):
    def create_payment(self, order: PaymentOrder) -> dict:
        # 支付宝用的是 alipay.trade.precreate（扫码）或
        # alipay.trade.wap.create（手机网站）
        pass

    def handle_callback(self, raw_data: dict) -> dict:
        # 支付宝回调是POST，参数在表单里
        pass

    def verify_signature(self, data: dict, signature: str) -> bool:
        # 支付宝用的是RSA2签名
        pass


class PaymentGateway:
    """支付网关，负责路由到具体策略"""

    def __init__(self):
        self._strategies: Dict[PaymentChannel, PaymentChannelStrategy] = {}

    def register(self, channel: PaymentChannel, strategy: PaymentChannelStrategy):
        self._strategies[channel] = strategy

    def create_payment(self, order: PaymentOrder) -> dict:
        strategy = self._get_strategy(order.channel)
        return strategy.create_payment(order)

    def handle_callback(self, channel: PaymentChannel, raw_data: dict) -> dict:
        strategy = self._get_strategy(channel)
        return strategy.handle_callback(raw_data)

    def _get_strategy(self, channel: PaymentChannel) -> PaymentChannelStrategy:
        strategy = self._strategies.get(channel)
        if not strategy:
            raise ValueError(f"不支持的支付渠道: {channel}")
        return strategy

这样加一个新渠道，只需要：

1. 实现 PaymentChannelStrategy 接口
2. 在 PaymentGateway 里注册

业务代码一行都不用改。

---

微信支付的坑，我踩了一遍又一遍

坑一：签名算法变了，文档还没更新

微信支付有 V2 和 V3 两个版本，差异非常大：

特性	微信V2	微信V3
签名算法	MD5 / HMAC-SHA256	HMAC-SHA256-256
请求格式	XML	JSON
回调通知	XML	JSON
证书	不需要API证书	需要APIv3密钥+证书
商户证书	只需要商户号	需要商户私钥证书

很多人直接从网上抄了V2的代码，结果用V3的接口，签名永远验证不过。

正确做法：确认你申请的是V2还是V3接口，然后严格按照对应版本的文档来。

V3的签名生成逻辑：

import hashlib
import hmac
import base64
from datetime import datetime
from typing import Optional


class WechatV3Signer:
    """微信支付V3签名工具"""

    def __init__(self, mch_id: str, api_v3_key: str, private_key: str):
        self.mch_id = mch_id
        self.api_v3_key = api_v3_key.encode()
        self.private_key = private_key.encode()

    def sign(self, method: str, url: str, body: str, timestamp: int, nonce: str) -> str:
        """
        生成Authorization头中的签名
        """
        message = f"{method}\n{url}\n{timestamp}\n{nonce}\n{body}\n"
        signing = self._rsa_sign(message.encode())
        schema = "WECHATPAY2-SHA256-RSA2048"
        token = f'serial_no="YOUR_CERT_SERIAL",{schema},timestamp="{timestamp}",nonce="{nonce}",signature="{signing}"'
        return token

    def _rsa_sign(self, message: bytes) -> str:
        from cryptography.hazmat.primitives import hashes, serialization
        from cryptography.hazmat.primitives.asymmetric import padding
        from cryptography.hazmat.backends import default_backend

        private_key = serialization.load_pem_private_key(
            self.private_key, password=None, backend=default_backend()
        )
        signature = private_key.sign(
            message,
            padding.PKCS1v15(),
            hashes.SHA256()
        )
        return base64.b64encode(signature).decode()

    def verify_response(self, timestamp: str, nonce: str, body: str, signature: str) -> bool:
        """验证微信回调响应签名"""
        message = f"{timestamp}\n{nonce}\n{body}\n"
        # 用微信平台证书公钥验证
        ...

坑二：回调地址必须公网可访问

微信回调会发一个 HTTP POST 到你的 notify_url。这个地址必须：

• 公网可访问（不能是内网地址）
• 支持 HTTPS（微信支付强制要求）
• 24小时在线

很多项目在本地开发时用的是 http://localhost:8080/notify，上线后才发现收不到回调。

解决方案：开发阶段用 ngrok 或类似工具暴露本地服务：

# 用ngrok暴露本地8080端口
ngrok http 8080
# 得到类似 https://abc123.ngrok.io 的地址
# 把这个地址配置到微信商户平台

坑三：回调里一定要验签，不能相信任何数据

这是最重要的安全原则。微信回调会带一个签名，你必须验证签名才能处理业务。

def verify_wechat_callback(self, request) -> dict:
    """
    验证微信支付V3回调
    """
    # 1. 获取请求头
    timestamp = request.headers.get("Wechatpay-Timestamp")
    nonce = request.headers.get("Wechatpay-Nonce")
    signature = request.headers.get("Wechatpay-Signature")
    serial = request.headers.get("Wechatpay-Serial")

    # 2. 获取请求体
    body = request.get_data().decode("utf-8")

    # 3. 验证签名（关键！）
    if not self._verify_v3_signature(timestamp, nonce, body, signature):
        logger.warning("微信回调签名验证失败")
        return {"code": "SIGN_FAILED", "message": "签名验证失败"}

    # 4. 解密回调内容（V3加密了敏感信息）
    result = self._decrypt_callback(body)

    # 5. 检查业务状态
    if result.get("status") != "SUCCESS":
        logger.warning(f"支付失败: {result}")
        return {"code": "PAYMENT_FAILED", "message": "支付失败"}

    return result

坑四：退款要用商户证书，不能用API密钥

很多人退款时直接用API密钥签名，结果报 签名错误。

微信退款接口要求使用商户私钥证书签名，不是API密钥。证书在商户平台下载，通常是 .pem 格式。

def refund(self, order: PaymentOrder, refund_amount: int, reason: str = "") -> dict:
    """
    微信退款（需要使用商户证书）
    """
    url = f"https://api.mch.weixin.qq.com/v3/refund/domestic/refunds"

    # 退款请求体
    body = {
        "out_trade_no": order.order_id,
        "out_refund_no": f"REFUND_{uuid.uuid4().hex[:16]}",
        "amount": {
            "refund": refund_amount,
            "total": order.amount,
            "currency": order.currency
        },
        "reason": reason
    }

    # 用商户私钥签名
    timestamp = str(int(datetime.now().timestamp()))
    nonce = uuid.uuid4().hex[:32]
    sign = self.signer.sign("POST", "/v3/refund/domestic/refunds", json.dumps(body), timestamp, nonce)

    headers = {
        "Authorization": sign,
        "Content-Type": "application/json",
        "Accept": "application/json",
        "Wechatpay-Serial": self.cert_serial  # 证书序列号
    }

    response = requests.post(url, json=body, headers=headers)
    return response.json()

坑五：重复回调处理

微信支付可能会重复发送同一个回调。如果你的业务逻辑是”收到回调就更新订单状态”，重复处理会导致：

• 重复发货
• 重复退款
• 订单金额对不上

所以一定要做幂等性处理：

def handle_wechat_callback(self, callback_data: dict) -> dict:
    """
    处理微信回调，确保幂等
    """
    out_trade_no = callback_data["out_trade_no"]
    transaction_id = callback_data.get("transaction_id", "")

    # 1. 查询本地订单状态
    order = self.order_repo.find_by_order_id(out_trade_no)
    if order.status == PaymentStatus.SUCCESS:
        # 已经处理过了，直接返回成功
        return {"code": "SUCCESS", "message": "已处理"}

    if order.status in [PaymentStatus.REFUNDED, PaymentStatus.CLOSED]:
        return {"code": "SUCCESS", "message": "订单已关闭"}

    # 2. 更新订单状态
    order.status = PaymentStatus.SUCCESS
    order.external_trace_id = transaction_id
    order.updated_at = datetime.now()
    self.order_repo.save(order)

    # 3. 触发后续业务逻辑（发货、通知等）
    self.on_payment_success(order)

    return {"code": "SUCCESS", "message": "成功"}

---

支付宝的坑，画风不太一样

坑一：RSA签名和V3完全不同

支付宝用的是 RSA2 签名（SHA256 with RSA），和微信的 HMAC-SHA256 完全不一样。

支付宝签名需要准备两个东西：

• 应用私钥：你自己生成的，用于签名请求
• 支付宝公钥：从支付宝平台获取，用于验证回调签名

from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding
import base64


class AlipaySigner:
    """支付宝RSA2签名工具"""

    def __init__(self, app_id: str, private_key: str, alipay_public_key: str):
        self.app_id = app_id
        self.private_key = private_key
        self.alipay_public_key = alipay_public_key

    def sign(self, params: dict) -> str:
        """
        生成支付宝签名
        params: 所有需要签名的参数（不包括sign本身）
        """
        # 1. 按key排序
        sorted_params = sorted(params.items())

        # 2. 拼接成字符串
        sign_string = "&".join(f"{k}={v}" for k, v in sorted_params)

        # 3. RSA2签名
        private_key = serialization.load_pem_private_key(
            self.private_key.encode(),
            password=None
        )
        signature = private_key.sign(
            sign_string.encode(),
            padding.PKCS1v15(),
            hashes.SHA256()
        )
        return base64.b64encode(signature).decode()

    def verify(self, params: dict, sign: str) -> bool:
        """
        验证支付宝回调签名
        """
        # 从params中移除sign字段
        verify_params = {k: v for k, v in params.items() if k != "sign"}
        sorted_params = sorted(verify_params.items())
        sign_string = "&".join(f"{k}={v}" for k, v in sorted_params)

        public_key = serialization.load_pem_public_key(
            self.alipay_public_key.encode()
        )
        try:
            signature = base64.b64decode(sign.encode())
            public_key.verify(
                signature,
                sign_string.encode(),
                padding.PKCS1v15(),
                hashes.SHA256()
            )
            return True
        except Exception:
            return False

坑二：charset 和 sign_type 必须传

很多新手调支付宝接口报 ILLEGAL_SIGN 或 SIGN_ERROR，原因多半是漏传了这两个参数：

# 支付宝要求的公共参数
common_params = {
    "app_id": self.app_id,
    "method": method,           # 接口方法名
    "charset": "utf-8",         # 字符集，必须传
    "sign_type": "RSA2",        # 签名类型，必须传
    "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
    "version": "1.0",
    "notify_url": notify_url,
    "return_url": return_url,
}

坑三：异步通知和同步通知是两回事

支付宝有两个通知：

• 同步通知（return_url）：用户付款后跳转回的页面，不可靠，不能用于确认订单
• 异步通知（notify_url）：支付宝服务器主动推给你的，这个才是可靠的

def handle_alipay_callback(self, request) -> dict:
    """
    处理支付宝异步通知
    注意：必须验证签名，必须返回 SUCCESS
    """
    # 支付宝会带一堆参数
    params = request.form.to_dict()

    # 1. 验证签名
    if not self.signer.verify(params, params.get("sign", "")):
        logger.warning("支付宝回调签名验证失败")
        return "failure"  # 支付宝会重试

    # 2. 验证app_id（防止其他应用的回调打进来）
    if params.get("app_id") != self.app_id:
        logger.warning(f"app_id不匹配: {params.get('app_id')}")
        return "failure"

    # 3. 检查交易状态
    trade_status = params.get("trade_status")
    if trade_status not in ["TRADE_SUCCESS", "TRADE_FINISHED"]:
        return "failure"

    # 4. 幂等处理
    out_trade_no = params.get("out_trade_no")
    trade_no = params.get("trade_no")

    order = self.order_repo.find_by_order_id(out_trade_no)
    if order.status == PaymentStatus.SUCCESS:
        return "success"  # 已处理，直接返回

    # 5. 更新订单
    order.status = PaymentStatus.SUCCESS
    order.external_trace_id = trade_no
    self.order_repo.save(order)

    self.on_payment_success(order)

    return "success"  # 必须返回success，否则支付宝会一直重试

坑四：支付宝的”交易查询”接口要注意时机

用户付款后，不要立刻调查询接口。支付宝的订单状态同步有延迟，建议：

1. 收到异步通知后直接更新
2. 如果5秒内没收到通知，再调查询接口
3. 查询接口最多重试3次，间隔递增

import time

async def confirm_payment_with_retry(self, order_id: str, max_retries: int = 3):
    """
    订单支付确认，带重试逻辑
    """
    for i in range(max_retries):
        result = await self.alipay.query_trade(order_id)
        if result.get("trade_status") in ["TRADE_SUCCESS", "TRADE_FINISHED"]:
            return result
        # 指数退避
        time.sleep(2 ** i)

    # 超过重试次数，标记需要人工介入
    return None

---

对账，这是最容易忽略的一步

很多项目做到”支付成功就完事了”，结果月底对账发现差了十几万。

对账的核心逻辑：把你自己的订单数据和支付渠道的账单做比对，找出差异。

import pandas as pd
from datetime import datetime, timedelta


class ReconciliationService:
    """对账服务"""

    def __init__(self, order_repo, wechat_client, alipay_client):
        self.order_repo = order_repo
        self.wechat_client = wechat_client
        self.alipay_client = alipay_client

    def reconcile(self, date: datetime) -> dict:
        """
        对某一天的账单进行对账
        """
        date_str = date.strftime("%Y%m%d")

        # 1. 获取本系统订单数据
        system_orders = self.order_repo.find_by_date(date)

        # 2. 获取微信支付账单
        wechat_bill = self.wechat_client.download_bill(date_str)

        # 3. 获取支付宝账单
        alipay_bill = self.alipay_client.download_bill(date_str)

        # 4. 比对
        results = {
            "date": date_str,
            "wechat": self._compare_orders(system_orders, wechat_bill, "wechat"),
            "alipay": self._compare_orders(system_orders, alipay_bill, "alipay"),
        }

        return results

    def _compare_orders(self, system_orders: list, third_party_bill: list, channel: str) -> dict:
        """
        比对系统订单和第三方账单
        """
        # 构建索引
        system_map = {order.order_id: order for order in system_orders}
        bill_map = {item["out_trade_no"]: item for item in third_party_bill}

        # 找差异
        discrepancies = []

        # 1. 系统有但第三方没有（可能退款了或者还没对账）
        for order_id, order in system_map.items():
            if order_id not in bill_map:
                discrepancies.append({
                    "type": "missing_in_bill",
                    "order_id": order_id,
                    "amount": order.amount,
                    "channel": channel
                })

        # 2. 第三方有但系统没有（可能是其他系统的订单）
        for item in third_party_bill:
            if item["out_trade_no"] not in system_map:
                discrepancies.append({
                    "type": "missing_in_system",
                    "order_id": item["out_trade_no"],
                    "amount": item["amount"],
                    "channel": channel
                })

        # 3. 金额不一致
        for order_id in system_map:
            if order_id in bill_map:
                sys_amount = system_map[order_id].amount
                bill_amount = int(bill_map[order_id]["amount"])
                if sys_amount != bill_amount:
                    discrepancies.append({
                        "type": "amount_mismatch",
                        "order_id": order_id,
                        "system_amount": sys_amount,
                        "bill_amount": bill_amount,
                        "channel": channel
                    })

        return {
            "total_system": len(system_map),
            "total_bill": len(bill_map),
            "discrepancies": discrepancies
        }

对账建议每天自动执行，差异数据发送到钉钉/飞书/企业微信，有人工处理工单跟进。

---

退款链路，比想象中复杂

退款不只是”调个接口”那么简单。真实场景要考虑：

1. 部分退款

用户买了100块的东西，退了30块，还能再退70块。需要记录已退款金额。

@dataclass
class RefundRecord:
    """退款记录"""
    refund_id: str
    order_id: str
    refund_amount: int       # 本次退款金额
    total_refunded: int      # 累计退款金额
    reason: str
    status: str              # pending/success/failed
    external_refund_no: str  # 第三方退款单号
    created_at: datetime

2. 退款回调

微信和支付宝退款后也会发回调，同样需要验签和幂等处理。

3. 退款查询

退款不是实时到账的，需要轮询查询退款状态。

async def refund_and_monitor(self, order: PaymentOrder, refund_amount: int) -> dict:
    """
    发起退款并监控状态
    """
    # 1. 发起退款
    result = await self.gateway.refund(order, refund_amount)
    refund_id = result["refund_id"]

    # 2. 轮询退款状态
    for i in range(10):
        await asyncio.sleep(5)
        status = await self.gateway.query_refund(refund_id)
        if status in ["SUCCESS", "CLOSED"]:
            return {"refund_id": refund_id, "status": status}

    # 超时，标记为需要人工处理
    return {"refund_id": refund_id, "status": "PENDING_MANUAL"}

---

一些真实的线上事故

事故一：退款接口被刷

有个项目没有对退款接口做频率限制，被黑产批量调用，短时间内退了十几万。

教训：

• 退款接口必须做频率限制（同一个订单1小时内只能退一次）
• 大额退款需要人工审核
• 退款金额不能超过订单金额

事故二：回调地址配置错误

有人把测试环境的回调地址配到了生产环境，导致测试数据污染了生产账单。

教训：

• 测试和生产环境严格隔离
• 回调地址用环境变量区分，不要硬编码

事故三：时间戳不一致

订单创建时间和支付超时时间用的是不同时钟，导致支付链接提前失效。

教训：

• 所有时间操作统一用 UTC
• 支付超时时间建议用相对时间（如15分钟），不要用绝对时间

---

一个完整的支付库模块结构

payment-sdk/
├── src/
│   ├── __init__.py
│   ├── gateway.py              # 支付网关（统一入口）
│   ├── strategies/
│   │   ├── __init__.py
│   │   ├── base.py             # 策略基类
│   │   ├── wechat.py           # 微信支付策略
│   │   └── alipay.py           # 支付宝策略
│   ├── models/
│   │   ├── __init__.py
│   │   ├── order.py            # 订单模型
│   │   └── refund.py           # 退款模型
│   ├── utils/
│   │   ├── __init__.py
│   │   ├── signers.py          # 签名工具
│   │   ├── validators.py       # 验签工具
│   │   └── encryption.py       # 加密工具（V3敏感信息）
│   └── exceptions.py           # 自定义异常
├── tests/
│   ├── test_wechat.py
│   ├── test_alipay.py
│   └── test_gateway.py
├── examples/
│   └── usage_example.py
├── pyproject.toml
└── README.md

---

最后说几句

做支付对接，细节决定生死。

一个签名算法搞错，可能让黑客伪造回调；一个幂等性没做，可能让财务损失真金白银；一个对账没做，月底查账查到你怀疑人生。

建议的做法：

1. 先跑通流程：用沙箱环境把整个链路跑通
2. 再抠细节：签名、验签、幂等、对账
3. 最后压测：高并发下回调会不会丢、订单会不会重复
4. 上线后监控：支付成功率、回调延迟、对账差异

支付系统是电商的命脉，宁可多花一周做测试，也不要上线后半夜接电话。

希望这篇文章能帮你少踩几个坑。如果还有具体问题，欢迎留言讨论。