PayPal与ChatGPT集成实战：从零搭建支付回调系统

1次阅读

没有评论

共计 2874 个字符，预计需要花费 8 分钟才能阅读完成。

ChatGPT 类应用与传统电商在支付场景有显著差异：

异步交互模式：用户可能在对话中途触发支付，需要保持上下文连贯
高并发回调：高峰时段可能同时收到大量 IPN（即时支付通知）
状态同步延迟：从支付成功到服务开通需要秒级响应

典型问题案例：某 AI 写作工具接入支付后，因未处理并发通知导致 30% 的订单重复开通会员。

REST API 特点
实时性高但需要主动轮询
适合前端直接调用的场景
需要处理 OAuth2.0 令牌过期问题
IPN 机制优势
服务端推送模式更节省资源
支持离线通知（最高保留 30 天）
天然防前端篡改

最终选择 IPN 的关键因素：ChatGPT 类应用更需要可靠的异步通知机制，而非实时支付确认。

from flask import Flask, request
import ssl

app = Flask(__name__)

@app.route('/ipn', methods=['POST'])
def ipn_listener():
    raw_data = request.get_data().decode('utf-8')
    # 验证阶段放这里
    return "OK", 200

if __name__ == '__main__':
    context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
    context.load_cert_chain('server.crt', 'server.key')
    app.run(ssl_context=context, port=443)

import hmac
from urllib.parse import parse_qs

def verify_ipn(raw_data):
    """
    :param raw_data: 原始 POST 数据
    :return: bool 验证结果
    """
    params = parse_qs(raw_data)
    original = params.pop('hmac')[0]

    # 1. 按字母顺序排序参数
    sorted_params = []
    for key in sorted(params.keys()):
        if key == 'hmac':
            continue
        sorted_params.append(f"{key}={params[key][0]}")

    # 2. 拼接字符串
    message = '&'.join(sorted_params)

    # 3. 获取 PayPal 密钥（需在后台配置）secret = os.getenv('PAYPAL_HMAC_KEY')

    # 4. HMAC-SHA256 验证
    signature = hmac.new(secret.encode(),
        message.encode(),
        hashlib.sha256
    ).hexdigest()

    return signature == original

用户支付成功后跳转回站内
通过 tx 参数向 PayPal 查询交易详情
与 IPN 通知做交叉验证

import requests

def fetch_pdt(tx_token):
    pdt_url = "https://www.paypal.com/cgi-bin/webscr"
    params = {
        "cmd": "_notify-synch",
        "tx": tx_token,
        "at": os.getenv('PDT_TOKEN')
    }

    response = requests.post(pdt_url, data=params)
    if response.text.startswith('SUCCESS'):
        return parse_pdt_response(response.text)
    return None

第一层：HMAC 签名验证（如前述代码）
第二层：向 PayPal 验证 IPN 真实性

def verify_with_paypal(raw_data):
    validation_url = "https://ipnpb.paypal.com/cgi-bin/webscr"
    headers = {
        'Content-Type': 'application/x-www-form-urlencoded',
        'Host': 'ipnpb.paypal.com'
    }
    resp = requests.post(
        validation_url,
        data=f"{raw_data}&cmd=_notify-validate",
        headers=headers
    )
    return resp.text == "VERIFIED"

建议数据库设计：

CREATE TABLE payments (txn_id VARCHAR(64) PRIMARY KEY,
    status ENUM('pending', 'completed', 'refunded'),
    processed_at TIMESTAMP,
    retry_count INT DEFAULT 0
);

处理逻辑：

# 在收到通知时先检查是否存在记录
with db.cursor() as cur:
    cur.execute("SELECT status FROM payments WHERE txn_id=%s", (txn_id,))
    if cur.fetchone():
        return "Duplicate"  # 已处理过

Sandbox：使用https://ipnpb.sandbox.paypal.com
生产环境：注意 HTTPS 证书必须由可信 CA 签发

检查请求头是否包含：

User-Agent: Python-Requests
Accept-Encoding: gzip

验证服务器时间是否同步（NTP 服务）
检查防火墙是否屏蔽 PayPal IP 段

# 启动隧道（注意必须指定域名）ngrok http --domain=your-domain.ngrok.io 5000

# 在 PayPal 后台配置：https://your-domain.ngrok.io/ipn

调试时建议：

使用 --log-level=debug 参数
配合 Wireshark 分析原始流量

import redis
r = redis.Redis(host='localhost', port=6379, db=0)

def cache_payment(txn_id, status):
    r.setex(f"payment:{txn_id}",
        3600,  # 1 小时过期
        json.dumps({"status": status})
    )

# 使用 Redis 管道加速
pipe = r.pipeline()
for notification in batch_notifications:
    pipe.get(f"payment:{notification['txn_id']}")
results = pipe.execute()

当遇到每秒 100+ 的 IPN 通知时，如何设计消息队列处理架构？可以考虑：