payment-platform/backend/app/plugin/module_payment/notification/handlers/enterprise_handler.py

from redis.asyncio import Redis

from app.api.v1.module_system.auth.schema import AuthSchema
from app.core.logger import log
from app.plugin.module_payment.enterprise.enums import EnterpriseStatusEnum
from app.plugin.module_payment.enterprise.schema import EnterpriseCreateOrUpdateSchema
from app.plugin.module_payment.enterprise.service import EnterpriseService
from app.utils.snowflake import extract_tenant_id_from_id_str

from ..schemas import EnterpriseChangeContent
from .base_handler import BaseHandler


class EnterpriseHandler(BaseHandler[dict]):
    """企业变更通知处理器"""

    async def _update_enterprise_status(
        self,
        data: EnterpriseChangeContent,
        auth: AuthSchema,
    ) -> bool:
        """更新企业状态

        Args:
            data: 通知数据
            auth: 认证信息

        Returns:
            更新是否成功
        """
        # 创建更新数据
        update_data = EnterpriseCreateOrUpdateSchema(
            enterprise_id=data.enterprise_id,
            out_biz_no=data.out_biz_no,
            account_id=data.account_id,
            status=data.action,
            remark=data.remark,
        )

        try:
            auth.tenant_id = extract_tenant_id_from_id_str(data.out_biz_no)
            await EnterpriseService.update_service_from_alipay(auth, update_data)
            log.info(f"企业状态更新成功: {data.enterprise_id} -> {data.action}")
            return True
        except Exception as e:
            log.error(f"更新企业状态失败: {e}")
            return False

    async def handle(self, method: str, content: dict, auth: AuthSchema, redis: Redis) -> bool:
        """
        处理企业变更通知

        动作类型:
        - ENTERPRISE_CREATE: 企业入驻
        - ENTERPRISE_ACTIVATED: 企业签约成功
        - ENTERPRISE_UNSIGN: 企业解约
        """
        try:
            notify_data = EnterpriseChangeContent(**content)
        except Exception as e:
            log.error(f"支付宝通知消息 - 解析企业变更通知消息内容失败: {e}")
            return False

        action = notify_data.action
        enterprise_id = notify_data.enterprise_id

        log.info(f"支付宝通知消息 - 处理企业变更通知: action={action}, enterprise_id={enterprise_id}")

        try:
            if action == EnterpriseStatusEnum.ENTERPRISE_CREATE.value:
                return await self._handle_create(notify_data, auth)
            elif action == EnterpriseStatusEnum.ENTERPRISE_ACTIVATED.value:
                return await self._handle_activated(notify_data, auth)
            elif action == EnterpriseStatusEnum.ENTERPRISE_UNSIGN.value:
                return await self._handle_unsign(notify_data, auth)
            elif action == EnterpriseStatusEnum.ENTERPRISE_WITHDRAW.value:
                return await self._handle_withdraw(notify_data, auth)
            elif action == EnterpriseStatusEnum.ENTERPRISE_AUTH.value:
                return await self._handle_auth(notify_data, auth)
            elif action == EnterpriseStatusEnum.ENTERPRISE_AUTH_REJECTED.value:
                return await self._handle_auth_rejected(notify_data, auth)
            else:
                log.warning(f"支付宝通知消息 - 未知的企业变更动作: {action}")
                return True
        except Exception as e:
            log.error(f"支付宝通知消息 - 处理企业变更通知异常: {e}")
            return False

    async def _handle_create(self, data: EnterpriseChangeContent, auth: AuthSchema) -> bool:
        """处理企业入驻"""
        return await self._update_enterprise_status(data=data, auth=auth)

    async def _handle_activated(self, data: EnterpriseChangeContent, auth: AuthSchema) -> bool:
        """处理企业签约成功"""

        return await self._update_enterprise_status(data=data, auth=auth)

    async def _handle_unsign(self, data: EnterpriseChangeContent, auth: AuthSchema) -> bool:
        """处理企业解约"""

        return await self._update_enterprise_status(data=data, auth=auth)

    async def _handle_withdraw(self, data: EnterpriseChangeContent, auth: AuthSchema) -> bool:
        """处理企业注销"""

        return await self._update_enterprise_status(data=data, auth=auth)

    async def _handle_auth(self, data: EnterpriseChangeContent, auth: AuthSchema) -> bool:
        """处理企业认证"""

        return await self._update_enterprise_status(data=data, auth=auth)

    async def _handle_auth_rejected(self, data: EnterpriseChangeContent, auth: AuthSchema) -> bool:
        """处理企业认证拒绝"""

        return await self._update_enterprise_status(data=data, auth=auth)