接入支付之后我睡不着：异步回调、幂等、掉单与对账的完整方案

1. 用户点「开通会员」
      ↓
2. 服务端创建订单（状态 PENDING，生成 out_trade_no）
      ↓
3. 调用支付平台「下单」接口，拿到支付链接/二维码
      ↓
4. 用户完成支付（在支付平台的页面/App 里）
      ↓
5a. 同步返回：支付平台把浏览器跳回你的 return_url（带参数）
5b. 异步通知：支付平台服务器 POST 你的 notify_url（可能多次）
      ↓
6. 服务端确认收款 → 发货（开通会员）
      ↓
7. 用户看到"开通成功"

def pay_return(request):
    """同步返回：只做展示，不改订单状态！"""
    out_trade_no = request.GET.get('out_trade_no')
    # 查一下本地状态，告诉用户"正在确认支付结果"
    order = Order.objects.filter(out_trade_no=out_trade_no).first()
    return render(request, 'pay_return.html', {'order': order})

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


def verify_sign(params: dict, sign: str, alipay_public_key: str) -> bool:
    """验证支付宝回调签名。
    params: 回调的所有参数（dict），sign: 签名值（base64）
    """
    # 1. 剔除 sign 和 sign_type
    data = {k: v for k, v in params.items() if k not in ('sign', 'sign_type')}
    # 2. 按 key 的 ASCII 升序排序
    items = sorted(data.items())
    # 3. 拼成 key=value&key=value（值要 URL 解码后的原值）
    content = '&'.join(f'{k}={urllib.parse.unquote(str(v))}' for k, v in items)

    # 4. 用支付宝公钥验证
    pub_key = serialization.load_pem_public_key(
        f'-----BEGIN PUBLIC KEY-----\n{alipay_public_key}\n-----END PUBLIC KEY-----'.encode()
    )
    try:
        pub_key.verify(
            base64.b64decode(sign),
            content.encode('utf-8'),
            padding.PKCS1v15(),
            hashes.SHA256(),
        )
        return True
    except Exception:
        return False

from django.db import transaction
from django.db.models import F


@csrf_exempt
@require_POST
def alipay_notify(request):
    params = request.POST.dict()
    sign = params.get('sign', '')

    # 1. 验签
    if not verify_sign(params, sign, settings.ALIPAY_PUBLIC_KEY):
        logger.warning('alipay notify sign verify failed: %s', params.get('out_trade_no'))
        return HttpResponse('fail')

    out_trade_no = params.get('out_trade_no')
    trade_status = params.get('trade_status')
    trade_no = params.get('trade_no')            # 平台流水号
    total_amount = params.get('total_amount')

    # 2. 只处理"成功"状态（还有 TRADE_CLOSED / WAIT_BUYER_PAY 等）
    if trade_status not in ('TRADE_SUCCESS', 'TRADE_FINISHED'):
        return HttpResponse('success')           # 非成功状态也要返回 success，否则平台一直重试

    try:
        with transaction.atomic():
            # 3. 加锁取订单
            order = Order.objects.select_for_update().get(out_trade_no=out_trade_no)

            # 4. 金额校验（关键！）
            if Decimal(total_amount) != order.amount:
                logger.error('amount mismatch: %s vs %s', total_amount, order.amount)
                return HttpResponse('fail')

            # 5. 幂等：已经处理过就直接返回
            if order.status == 'PAID':
                return HttpResponse('success')

            # 6. 状态推进 + 发货
            order.status = 'PAID'
            order.trade_no = trade_no
            order.paid_at = timezone.now()
            order.save(update_fields=['status', 'trade_no', 'paid_at', 'updated_at'])
            fulfill_order(order)                  # 开通会员等

    except Order.DoesNotExist:
        logger.error('order not found: %s', out_trade_no)
        return HttpResponse('fail')               # 订单不存在，让平台重试（可能是我们的主从延迟）

    return HttpResponse('success')

# 用条件更新代替"查询 + 判断 + 更新"，避免任何并发窗口
updated = Order.objects.filter(
    out_trade_no=out_trade_no,
    status='PENDING',                 # ← 只有待支付才能改成已支付
).update(status='PAID', trade_no=trade_no, paid_at=timezone.now())

if updated == 0:
    # 没更新到 → 要么订单不存在，要么已经处理过了
    logger.info('order %s already processed or not found', out_trade_no)
    return HttpResponse('success')

# 更新成功了才发货
order = Order.objects.get(out_trade_no=out_trade_no)
fulfill_order(order)

@shared_task
def reconcile_pending_orders():
    """扫描超过 N 分钟仍未支付的订单，主动向支付平台查询真实状态。"""
    cutoff = timezone.now() - timedelta(minutes=5)
    # 只处理"创建超过 5 分钟、还处于 PENDING"的订单
    qs = Order.objects.filter(
        status='PENDING',
        created_at__lt=cutoff,
        created_at__gt=timezone.now() - timedelta(days=1),   # 只补最近一天
    ).values_list('id', 'out_trade_no')[:200]

    for order_id, out_trade_no in qs:
        try:
            result = alipay_query(out_trade_no)      # 调用 alipay.trade.query
        except Exception as e:
            logger.warning('query failed %s: %s', out_trade_no, e)
            continue

        if result.get('trade_status') in ('TRADE_SUCCESS', 'TRADE_FINISHED'):
            # 走和回调一样的处理逻辑（同一个函数！）
            handle_paid(out_trade_no, result.get('trade_no'), result.get('total_amount'))
        elif result.get('trade_status') == 'TRADE_CLOSED':
            Order.objects.filter(id=order_id, status='PENDING').update(status='CLOSED')
        # WAIT_BUYER_PAY：用户还没付款，跳过

if Decimal(params['total_amount']) != order.amount:
    logger.error('amount mismatch')
    return HttpResponse('fail')

if params.get('seller_id') and params['seller_id'] != settings.ALIPAY_SELLER_ID:
    logger.error('seller mismatch')
    return HttpResponse('fail')

def fulfill_order(order):
    """发货：开通会员。也要幂等。"""
    user = order.user

    # 用 get_or_create + 幂等键（订单号）保证不重复发放
    obj, created = Membership.objects.get_or_create(
        user=user,
        source_order=order,
        defaults={
            'level': order.product.level,
            'expires_at': timezone.now() + timedelta(days=order.product.days),
        },
    )
    if not created:
        logger.info('membership already granted for order %s', order.out_trade_no)
        return
    # 续期逻辑（如果已有会员，延长而不是覆盖）
    extend_membership(user, order.product)

from django.db.models import F
from django.utils import timezone

# 安全：数据库层累加
Membership.objects.filter(user=user).update(
    expires_at=F('expires_at') + timedelta(days=365)     # 注意：Django 对 timedelta 的支持
)

# 或者更明确地用 SQL 函数

1. 每天凌晨下载前一天的平台账单（对账单 API）
2. 逐条比对：平台有的 / 本地有的 / 金额是否一致 / 状态是否一致
3. 差异分类处理：
   - 平台有、本地没有 → 漏单（补发货）
   - 本地已支付、平台没有 → 可能的假支付（重点查！）
   - 金额不一致 → 人工处理
4. 输出对账报告（邮件/群消息）

@shared_task
def daily_reconcile(date_str=None):
    """每日对账。date_str: '2026-03-11'，默认昨天。"""
    date_str = date_str or (timezone.localdate() - timedelta(days=1)).isoformat()

    # 1. 拿账单下载链接
    url = alipay_bill_download_url(date_str)
    # 2. 下载并解析（支付宝账单是 csv，gbk 编码，最后一行有汇总）
    rows = download_and_parse_bill(url)

    # 3. 本地订单
    local = {o.out_trade_no: o for o in Order.objects.filter(
        created_at__date=date_str, status='PAID')}

    diff_missing = []      # 平台有、本地没有（漏单）
    diff_extra = []        # 本地有、平台没有（可疑）
    diff_amount = []       # 金额不一致

    for row in rows:
        if row['trade_status'] not in ('TRADE_SUCCESS', 'TRADE_FINISHED'):
            continue
        no = row['out_trade_no']
        if no not in local:
            diff_missing.append(row)
        elif Decimal(row['total_amount']) != local[no].amount:
            diff_amount.append((row, local[no].amount))

    for no, order in local.items():
        if no not in {r['out_trade_no'] for r in rows}:
            diff_extra.append(order)

    # 4. 报告
    report = {
        'date': date_str,
        'platform_count': len(rows),
        'local_count': len(local),
        'missing': diff_missing,
        'extra': diff_extra,
        'amount_mismatch': diff_amount,
    }
    send_report(report)      # 发邮件/群

    if diff_missing:
        logger.error('reconcile: %d orders missing locally!', len(diff_missing))
        # 自动补发货（谨慎：先人工确认，或者只对明确的漏单自动补）
    return report

def request_refund(order, amount):
    with transaction.atomic():
        locked = Order.objects.select_for_update().get(id=order.id)
        refunded = locked.refunded_amount or Decimal('0')
        refundable = locked.amount - refunded
        if amount > refundable:
            raise ValueError(f'可退金额不足：{refundable}')
        # 记录退款单（幂等键：out_request_no）
        Refund.objects.create(order=locked, amount=amount, out_request_no=uuid4().hex)
        locked.refunded_amount = refunded + amount
        locked.save(update_fields=['refunded_amount'])
    # 调用退款 API（在事务外，避免长事务）
    alipay_refund(order.out_trade_no, amount, out_request_no)

logger.info('pay notify received', extra={
    'out_trade_no': out_trade_no,
    'trade_status': trade_status,
    'amount': total_amount,
    'sign_ok': sign_ok,
})