kangda_robot_backend/ruoyi-fastapi-backend/module_admin/service/message_service.py

from sqlalchemy.ext.asyncio import AsyncSession
from typing import List
from config.constant import CommonConstant
from exceptions.exception import ServiceException
from module_admin.entity.vo.common_vo import CrudResponseModel
from module_admin.dao.message_dao import MessageDao
from module_admin.entity.vo.message_vo import DeleteMessageModel, MessageModel, MessagePageQueryModel, EditMessageModel
from utils.common_util import CamelCaseUtil
from utils.excel_util import ExcelUtil


class MessageService:
    """
    系统消息模块服务层
    """
    # 获取告警消息列表
    @classmethod
    async def get_alert_message_services(cls, query_db: AsyncSession, pageNum, pageSize):
        t = MessagePageQueryModel()
        t.type = '0'
        t.page_num = pageNum
        t.page_size = pageSize
        return await MessageDao.get_message_list(query_db, t, True)
    
    
    # 获取告警消息分布
    @classmethod
    async def get_alarm_type_distribution_services(cls, query_db: AsyncSession):
        result = await MessageDao.get_alarm_type_distribution(query_db)
        # d = dict()
        # t = 0
        # for k, v in result.items():
        #     d[k] = v
        #     t += v
        # for k, v in d.items():
        #     result[k] = round(v/t, 2)
        return result

    @classmethod
    async def handle_all_message_services(cls,query_db: AsyncSession):

        try:
            await MessageDao.handle_all_message(query_db)
            await query_db.commit()
            return CrudResponseModel(is_success=True, message='处理成功')
        except Exception as e:
            await query_db.rollback()
            raise e



    @classmethod
    async def get_message_list_services(
        cls, query_db: AsyncSession, query_object: MessagePageQueryModel, is_page: bool = False
    ):
        """
        获取系统消息列表信息service

        :param query_db: orm对象
        :param query_object: 查询参数对象
        :param is_page: 是否开启分页
        :return: 系统消息列表信息对象
        """
        message_list_result = await MessageDao.get_message_list(query_db, query_object, is_page)

        return message_list_result

    @classmethod
    async def count_message(cls, query_db: AsyncSession,):
        """
        获取系统消息数量service

        :param query_db: orm对象
        :return:
            total 总消息数
            unhandled 未处理消息数
            handled 已处理消息数
        """
        total, unhandled, handled = await MessageDao.count_message(query_db)
        return {'total': total, 'unhandled': unhandled, 'handled': handled}

    @classmethod
    async def add_message_services(cls, query_db: AsyncSession, page_object: MessageModel):
        """
        新增系统消息信息service

        :param query_db: orm对象
        :param page_object: 新增系统消息对象
        :return: 新增系统消息校验结果
        """
        try:
            await MessageDao.add_message_dao(query_db, page_object)
            await query_db.commit()
            return CrudResponseModel(is_success=True, message='新增成功')
        except Exception as e:
            await query_db.rollback()
            raise e

    @classmethod
    async def edit_message_services(cls, query_db: AsyncSession, page_object: EditMessageModel):
        """
        编辑系统消息信息service

        :param query_db: orm对象
        :param page_object: 编辑系统消息对象
        :return: 编辑系统消息校验结果
        """
        edit_message = page_object.model_dump(exclude_unset=True, exclude={'create_time', 'create_by', })
        message_info = await cls.message_detail_services(query_db, page_object.message_id)
        if message_info.message_id:
            try:
                await MessageDao.edit_message_dao(query_db, edit_message)
                await query_db.commit()
                return CrudResponseModel(is_success=True, message='更新成功')
            except Exception as e:
                await query_db.rollback()
                raise e
        else:
            raise ServiceException(message='系统消息不存在')

    @classmethod
    async def delete_message_services(cls, query_db: AsyncSession, page_object: DeleteMessageModel):
        """
        删除系统消息信息service

        :param query_db: orm对象
        :param page_object: 删除系统消息对象
        :return: 删除系统消息校验结果
        """
        if page_object.message_ids:
            message_id_list = page_object.message_ids.split(',')
            try:
                for message_id in message_id_list:
                    await MessageDao.delete_message_dao(query_db, MessageModel(messageId=message_id))
                await query_db.commit()
                return CrudResponseModel(is_success=True, message='删除成功')
            except Exception as e:
                await query_db.rollback()
                raise e
        else:
            raise ServiceException(message='传入消息主键为空')

    @classmethod
    async def message_detail_services(cls, query_db: AsyncSession, message_id: int):
        """
        获取系统消息详细信息service

        :param query_db: orm对象
        :param message_id: 消息主键
        :return: 消息主键对应的信息
        """
        message = await MessageDao.get_message_detail_by_id(query_db, message_id=message_id)
        if message:
            result = MessageModel(**CamelCaseUtil.transform_result(message))
        else:
            result = MessageModel(**dict())

        return result

    @staticmethod
    async def export_message_list_services(message_list: List):
        """
        导出系统消息信息service

        :param message_list: 系统消息信息列表
        :return: 系统消息信息对应excel的二进制数据
        """
        # 创建一个映射字典，将英文键映射到中文键
        mapping_dict = {
            'messageId': '消息主键',
            'type': '类型',
            'status': '状态',
            'detail': '消息描述',
            'createTime': '创建时间',
            'createBy': '创建者',
            'updateTime': '更新时间',
            'updateBy': '更新者',
        }
        binary_data = ExcelUtil.export_list2excel(message_list, mapping_dict)

        return binary_data