life-echo/api/payment/alipay_pay.py

"""
支付宝 OpenAPI 封装
使用 python-alipay-sdk 实现 APP 支付
"""
import logging
from typing import Dict, Optional

from .config import AlipayConfig
from .schemas import PaymentResult, NotifyResult, PaymentStatus
from .exceptions import PaymentConfigError, PaymentCreateError, PaymentNotifyError, PaymentQueryError

logger = logging.getLogger(__name__)


class AlipayClient:
    """支付宝客户端"""

    def __init__(self, config: AlipayConfig):
        self._config = config
        self._client = None

    def _ensure_client(self):
        """确保支付宝客户端已初始化"""
        if not self._config.is_configured:
            raise PaymentConfigError("支付宝配置不完整，请检查环境变量")

        if self._client is None:
            try:
                from alipay import AliPay

                self._client = AliPay(
                    appid=self._config.app_id,
                    app_notify_url=self._config.notify_url,
                    app_private_key_string=self._config.private_key,
                    alipay_public_key_string=self._config.alipay_public_key,
                    sign_type=self._config.sign_type,
                    debug=False,  # 生产环境
                )
                logger.info("支付宝客户端初始化成功")
            except Exception as e:
                raise PaymentConfigError(f"支付宝客户端初始化失败: {e}")

    def create_app_order(
        self,
        out_trade_no: str,
        total_amount: int,
        subject: str,
    ) -> PaymentResult:
        """
        创建 APP 支付订单

        Args:
            out_trade_no: 商户订单号
            total_amount: 金额（单位：分）
            subject: 商品标题

        Returns:
            PaymentResult 包含支付宝 order_string
        """
        self._ensure_client()

        try:
            # 支付宝金额单位是元，需要从分转换
            amount_yuan = f"{total_amount / 100:.2f}"

            order_string = self._client.api_alipay_trade_app_pay(
                out_trade_no=out_trade_no,
                total_amount=amount_yuan,
                subject=subject,
                notify_url=self._config.notify_url,
            )

            if order_string:
                logger.info(f"支付宝订单创建成功: {out_trade_no}")
                return PaymentResult(
                    success=True,
                    payment_method="alipay",
                    out_trade_no=out_trade_no,
                    alipay_order_string=order_string,
                )
            else:
                raise PaymentCreateError("支付宝下单失败: 返回空订单字符串")

        except PaymentCreateError:
            raise
        except Exception as e:
            logger.error(f"支付宝订单创建异常: {e}")
            raise PaymentCreateError(f"支付宝下单异常: {e}")

    def verify_notify(self, params: Dict[str, str]) -> NotifyResult:
        """
        验证支付宝异步通知签名

        Args:
            params: 通知参数字典（从 POST form 解析）

        Returns:
            NotifyResult 包含交易信息
        """
        self._ensure_client()

        try:
            # 提取签名和签名类型
            sign = params.pop("sign", None)
            sign_type = params.pop("sign_type", None)

            if not sign:
                raise PaymentNotifyError("支付宝回调缺少签名参数")

            # 验证签名
            success = self._client.verify(params, sign)

            if success:
                trade_status = params.get("trade_status", "")
                out_trade_no = params.get("out_trade_no", "")
                trade_no = params.get("trade_no", "")
                total_amount_str = params.get("total_amount", "0")

                # 将元转换为分
                total_amount = int(float(total_amount_str) * 100)

                logger.info(
                    f"支付宝回调验证成功: out_trade_no={out_trade_no}, "
                    f"trade_status={trade_status}, amount={total_amount}"
                )

                return NotifyResult(
                    success=True,
                    out_trade_no=out_trade_no,
                    trade_no=trade_no,
                    total_amount=total_amount,
                    trade_status=trade_status,
                )
            else:
                raise PaymentNotifyError("支付宝回调验签失败")

        except PaymentNotifyError:
            raise
        except Exception as e:
            logger.error(f"支付宝回调处理异常: {e}")
            raise PaymentNotifyError(f"支付宝回调处理失败: {e}")

    def query_order(self, out_trade_no: str) -> PaymentStatus:
        """
        主动查询支付宝订单状态

        Args:
            out_trade_no: 商户订单号

        Returns:
            PaymentStatus 订单状态
        """
        self._ensure_client()

        try:
            result = self._client.api_alipay_trade_query(out_trade_no=out_trade_no)

            if result:
                code = result.get("code", "")
                if code == "10000":
                    trade_status = result.get("trade_status", "")
                    trade_no = result.get("trade_no", "")
                    total_amount_str = result.get("total_amount", "0")
                    total_amount = int(float(total_amount_str) * 100)

                    # 将支付宝状态映射到统一状态
                    unified_status = self._map_trade_status(trade_status)

                    return PaymentStatus(
                        success=True,
                        out_trade_no=out_trade_no,
                        trade_no=trade_no,
                        trade_status=unified_status,
                        total_amount=total_amount,
                    )
                else:
                    error_msg = result.get("sub_msg", result.get("msg", "未知错误"))
                    raise PaymentQueryError(f"查询支付宝订单失败: {error_msg}")
            else:
                raise PaymentQueryError("查询支付宝订单返回空结果")

        except PaymentQueryError:
            raise
        except Exception as e:
            logger.error(f"查询支付宝订单异常: {e}")
            raise PaymentQueryError(f"查询支付宝订单异常: {e}")

    def close_order(self, out_trade_no: str) -> bool:
        """
        关闭支付宝订单

        Args:
            out_trade_no: 商户订单号

        Returns:
            是否关闭成功
        """
        self._ensure_client()

        try:
            result = self._client.api_alipay_trade_close(out_trade_no=out_trade_no)
            if result and result.get("code") == "10000":
                logger.info(f"支付宝订单已关闭: {out_trade_no}")
                return True
            else:
                error_msg = result.get("sub_msg", "未知错误") if result else "空结果"
                logger.warning(f"关闭支付宝订单失败: {error_msg}")
                return False
        except Exception as e:
            logger.error(f"关闭支付宝订单异常: {e}")
            return False

    @staticmethod
    def _map_trade_status(alipay_status: str) -> str:
        """将支付宝交易状态映射到统一状态"""
        status_map = {
            "WAIT_BUYER_PAY": "NOTPAY",
            "TRADE_CLOSED": "CLOSED",
            "TRADE_SUCCESS": "SUCCESS",
            "TRADE_FINISHED": "SUCCESS",
        }
        return status_map.get(alipay_status, alipay_status)