payment-platform/backend/app/plugin/module_payment/notification/service.py

from typing import Any

from app.core.logger import log
from app.core.alipay.config import get_alipay_config
from app.core.redis_crud import RedisCURD
from redis.asyncio.client import Redis

from app.api.v1.module_system.auth.schema import AuthSchema
from app.plugin.module_payment.notification.crud import AlipayNotifyLogCRUD
from .schemas import AlipayNotifyBase
from .enums import AlipayNotifyMethodEnum
from .handlers import (
    AccountHandler,
    BaseHandler,
    BillHandler,
    EmployeeHandler,
    EnterpriseHandler,
    InstitutionHandler,
    OrderHandler,
    VoucherHandler,
)


class NotificationService:
    """通知消息分发服务"""

    _handlers: dict[str, type[BaseHandler]] = {
        AlipayNotifyMethodEnum.ENTERPRISE_CHANGE.value: EnterpriseHandler,
        AlipayNotifyMethodEnum.EMPLOYEE_CHANGE.value: EmployeeHandler,
        AlipayNotifyMethodEnum.CONSUME_CHANGE.value: BillHandler,
        AlipayNotifyMethodEnum.VOUCHER_CHANGE.value: VoucherHandler,
        # AlipayNotifyMethodEnum.INVOICE_ORDER_CHANGE.value: OrderHandler,
        AlipayNotifyMethodEnum.TRANS_AUTHORIZE_NOTIFY.value: AccountHandler,
        AlipayNotifyMethodEnum.FUND_CHANGE_NOTIFY.value: AccountHandler,
        AlipayNotifyMethodEnum.INSTITUTION_OPERATION.value: InstitutionHandler,
    }


    # Redis 配置
    _notify_cache_prefix = "alipay:notify:"
    _notify_cache_ttl = 604800  # 7天

    @classmethod
    async def dispatch(cls, msg_method: str, content: dict[str, Any], auth: AuthSchema, redis: Redis) -> bool:
        """
        分发通知消息到对应的处理器

        Args:
            msg_method: 消息方法名
            content: 通知内容
            auth: 认证信息

        Returns:
            处理是否成功
        """
        handler_class = cls._handlers.get(msg_method)

        if not handler_class:
            log.warning(f"支付宝通知消息未注册的消息处理器: {msg_method}")
            return True

        handler = handler_class()

        try:
            result = await handler.handle(method=msg_method, content=content, auth=auth, redis=redis)
            log.info(f"支付宝通知消息 - 处理完成: {msg_method}, result={result}")
            return result
        except Exception as e:
            log.error(f"支付宝通知消息 - 处理异常: {msg_method}, error={e}")
            return False

    @classmethod
    def verify_sign(cls, notify_data: AlipayNotifyBase) -> bool:
        """
        验签支付宝通知

        Args:
            notify_data: 支付宝通知数据

        Returns:
            验签是否通过
        """
        if not notify_data.sign:
            log.error("支付宝通知消息 - 验签失败: 缺少 sign 参数")
            return False

        if not notify_data.biz_content:
            log.error("支付宝通知消息 - 验签失败: 缺少 biz_content 参数")
            return False

        try:
            alipay_public_key = get_alipay_config().alipay_public_key

            if not alipay_public_key:
                log.error("支付宝通知消息 - 验签失败: 缺少支付宝公钥配置")
                return False

            from alipay.aop.api.util.SignatureUtils import verify_with_rsa
            import urllib.parse

            # 1. 构建参数字典
            # 2. 移除 sign、sign_type 参数
            params = notify_data.model_dump(exclude={"sign", "sign_type"}, exclude_none=True)

            # 3. 对参数值进行 url_decode
            decoded_params = {}
            for key, value in params.items():
                if isinstance(value, str):
                    decoded_params[key] = urllib.parse.unquote(value)
                else:
                    decoded_params[key] = value
        
            # 4. 按字典序排序
            sorted_keys = sorted(decoded_params.keys())
        
            # 5. 构建待验签字符串
            sign_str = '&'.join([f"{key}={decoded_params[key]}" for key in sorted_keys])
        
            log.debug(f"支付宝通知消息 - 待验签字符串: {sign_str}")

            verify_result = verify_with_rsa(
                alipay_public_key,
                sign_str.encode(get_alipay_config().charset),
                notify_data.sign
            )

            if not verify_result:
                log.warning(f"支付宝通知消息 - 验签失败: sign={notify_data.sign[:20]}..., sign_type={notify_data.sign_type}")

            return verify_result

        except Exception as e:
            log.error(f"支付宝通知消息 - 验签异常: {e}")
            return False

    @classmethod
    async def verify_and_dispatch(cls, notify_data: AlipayNotifyBase, redis: Redis, auth: AuthSchema) -> bool:
        """
        验签并分发通知消息

        Args:
            notify_data: 支付宝通知数据
            redis: Redis客户端对象
            auth: 认证信息

        Returns:
            处理是否成功
        """
        # 1. 先验签
        if not cls.verify_sign(notify_data):
            return False

        # 2. 检查是否已处理过该通知（幂等处理）
        notify_id = notify_data.notify_id
        if await cls._is_notify_processed(redis, notify_id):
            log.warning(f"支付宝通知消息 - 通知已处理: notify_id={notify_id}")
            return True
        
        # 记录日志
        await AlipayNotifyLogService.log_notify_service(auth, notify_data.model_dump())

        # 3. 解析 biz_content
        biz_content = notify_data.parse_biz_content()
        log.debug(f"支付宝通知消息 - 消息内容: {biz_content}")

        # 4. 从 enterprise_id 获取正确的 tenant_id
        enterprise_id = biz_content.get("enterprise_id", "")
        if enterprise_id and auth.tenant_id == -1:
            try:
                from app.plugin.module_payment.enterprise.model import EnterpriseModel
                from sqlalchemy import select
                stmt = select(EnterpriseModel.tenant_id).where(EnterpriseModel.enterprise_id == enterprise_id)
                result = await auth.db.execute(stmt)
                tenant_id = result.scalar_one_or_none()
                if tenant_id:
                    auth.tenant_id = tenant_id
                    log.debug(f"支付宝通知消息 - 从 enterprise_id 解析 tenant_id={tenant_id}")
            except Exception:
                log.warning(f"支付宝通知消息 - 解析 tenant_id 失败: enterprise_id={enterprise_id}")

        # 5. 分发处理
        success = await cls.dispatch(notify_data.method, biz_content, auth, redis)

        # 5. 处理成功后记录
        if success:
            await cls._mark_notify_processed(redis, notify_id)

        return success

    @classmethod
    async def _is_notify_processed(cls, redis: Redis, notify_id: str) -> bool:
        """
        检查通知是否已处理（幂等处理）

        Args:
            redis: Redis客户端对象
            notify_id: 通知ID

        Returns:
            是否已处理
        """
        key = f"{cls._notify_cache_prefix}{notify_id}"
        return await RedisCURD(redis).exists(key)

    @classmethod
    async def _mark_notify_processed(cls, redis: Redis, notify_id: str) -> None:
        """
        标记通知已处理

        Args:
            redis: Redis客户端对象
            notify_id: 通知ID
        """
        key = f"{cls._notify_cache_prefix}{notify_id}"
        await RedisCURD(redis).set(key, "1", expire=cls._notify_cache_ttl)



class AlipayNotifyLogService:
    """支付宝通知日志服务"""

    @classmethod
    async def log_notify_service(
        cls, auth: AuthSchema, message: dict, verify_result: bool = True
    ) -> bool:
        """
        记录支付宝通知
        
        Args:
            auth: 认证信息
            message: 通知数据（字典格式）
            verify_result: 验签结果
            process_result: 处理结果
            error: 错误信息
            
        Returns:
            bool: 记录是否成功
        """
        try:
            # 构建日志数据
            log_data = {
                "notify_id": message.get('notify_id'),
                "msg_method": message.get('msg_method'),
                "notify_type": message.get('notify_type'),
                "message": message,
                "verify_result": verify_result,
                "process_result": None,
                "error": None,
            }
            
            # 保存到数据库
            crud = AlipayNotifyLogCRUD(auth)
            await crud.create(log_data)
            log.info(f"记录支付宝通知: {message.get('msg_method')} - {message.get('notify_id')}")
            return True
        except Exception as e:
            log.error(f"记录支付宝通知失败: {e}")
            return False