# 异步通知处理 - Python 示例

适用产品：AI 网页应用收款、AI 移动应用收款。AI 按量付费不使用异步通知。

```python
import re
from flask import Blueprint, Response, request


def create_alipay_notify_blueprint(deps):
    """deps 接入项目真实 SDK 验签适配器、支付配置、订单仓储、通知事件仓储和脱敏日志。"""
    blueprint = Blueprint("alipay_notify", __name__)

    @blueprint.post("/alipay/notify")
    def alipay_notify():
        params = {}
        try:
            params = request.form.to_dict(flat=True)
            config = deps["payment_config_repository"].find_by_app_id(params.get("app_id"))
            if config is None:
                return fail(deps, "alipay notify app config not found", params)

            if not verify_notify_with_alipay_sdk(params, deps["alipay_notify_verifier"], config):
                return fail(deps, "alipay notify signature check failed", params)

            order = deps["order_repository"].find_by_out_trade_no(params.get("out_trade_no"))
            if order is None or not business_fields_match(order, params, config):
                return fail(deps, "alipay notify business check failed", params)

            inserted = deps["notify_event_repository"].insert_once(
                notify_id=params.get("notify_id"),
                trade_no=params.get("trade_no"),
                out_trade_no=params.get("out_trade_no"),
                trade_status=params.get("trade_status"),
                raw_params=params,
            )
            if not inserted:
                return Response("success", mimetype="text/plain")

            if is_paid_trade_notification(params):
                deps["order_repository"].mark_paid_idempotently(order["out_trade_no"], params.get("trade_no"), params)
            else:
                deps["order_repository"].record_non_paid_trade_event(order["out_trade_no"], params.get("trade_status"), params)

            return Response("success", mimetype="text/plain")
        except Exception:
            safe_log_notify_error(deps, "alipay notify unexpected exception", params)
            return Response("fail", mimetype="text/plain")

    return blueprint


def business_fields_match(order, params, config):
    return (
        params.get("app_id") == config["app_id"]
        and params.get("out_trade_no") == order["out_trade_no"]
        and amount_equals(order["total_amount"], params.get("total_amount"))
        and expected_seller_matches(config, params.get("seller_id"), params.get("seller_email"))
    )


def is_paid_trade_notification(params):
    paid_status = params.get("trade_status") in ("TRADE_SUCCESS", "TRADE_FINISHED")
    return paid_status and not params.get("out_biz_no") and not params.get("gmt_refund") and not params.get("refund_fee")


def amount_equals(expected, actual):
    expected_amount = normalize_amount_text(expected)
    actual_amount = normalize_amount_text(actual)
    return expected_amount is not None and expected_amount == actual_amount


def normalize_amount_text(value):
    if value is None:
        return None
    match = re.fullmatch(r"(\d+)(?:\.(\d{1,2}))?", str(value).strip())
    if not match:
        return None
    return f"{int(match.group(1))}.{(match.group(2) or '').ljust(2, '0')}"


def expected_seller_matches(config, seller_id, seller_email):
    expected_seller_id = config.get("seller_id")
    expected_seller_email = config.get("seller_email")
    return (
        bool(expected_seller_id and seller_id == expected_seller_id)
        or bool(expected_seller_email and seller_email == expected_seller_email)
    )


def verify_notify_with_alipay_sdk(params, verifier, config):
    """
    使用当前项目已安装的支付宝 SDK 验签能力校验 params["sign"]。
    verifier.verify(params, config) 必须封装目标项目当前 SDK 实际通知验签 API；落地前先查已安装包文档或类型定义，不能手写临时 RSA 验签。
    """
    return verifier.verify(params, config)


def fail(deps, message, params):
    safe_log_notify_error(deps, message, params)
    return Response("fail", mimetype="text/plain")


def safe_log_notify_error(deps, message, params):
    try:
        logger = deps.get("logger")
        if logger is not None:
            logger.warning(message, extra={"alipay_notify": sanitize_notify_params(params)})
    except Exception:
        # 日志失败不能影响支付宝收到纯文本 fail。
        pass


def sanitize_notify_params(params):
    safe_params = dict(params)
    safe_params.pop("sign", None)
    return safe_params
```

要点：

- 支付宝通知是 POST 表单，不能用 JSON body 解析。
- 验签必须调用当前项目支付宝 SDK 的实际验签 API，验签前不得丢字段或改值。
- `app_id`、`seller_id` / `seller_email` 从运行时支付配置校验，不从订单对象默认取值。
- 金额校验使用字符串规范化、`Decimal` 或项目定点数工具，不要用浮点数直接比较。
- 只有付款成功状态且不是退款、关单、分账等事件时，才认定付款成功。
- 内存订单或内存幂等只能用于非生产 demo，生产必须使用持久化订单表和通知事件表。
- 处理成功后返回纯文本 `success`；异常、验签失败或业务校验失败返回 `fail`。