----/后端源码/yifan.action-ai.cn/app/api/v1/module_system/notice/service.py

# -*- coding: utf-8 -*-

from app.core.base_schema import BatchSetAvailable
from app.core.exceptions import CustomException
from app.utils.excel_util import ExcelUtil

from ..auth.schema import AuthSchema
from .schema import NoticeCreateSchema, NoticeUpdateSchema, NoticeOutSchema, NoticeQueryParam
from .crud import NoticeCRUD


class NoticeService:
    """
    公告管理模块服务层
    """
    
    @classmethod
    async def get_notice_detail_service(cls, auth: AuthSchema, id: int) -> dict:
        """
        获取公告详情。
        
        参数:
        - auth (AuthSchema): 认证信息模型。
        - id (int): 公告ID。
        
        返回:
        - Dict: 公告详情字典。
        """
        notice_obj = await NoticeCRUD(auth).get_by_id_crud(id=id)
        return NoticeOutSchema.model_validate(notice_obj).model_dump()
        
    @classmethod
    async def get_notice_list_available_service(cls, auth: AuthSchema) -> list[dict]:
        """
        获取可用的公告列表。
        
        参数:
        - auth (AuthSchema): 认证信息模型。
        
        返回:
        - list[dict]: 可用公告详情字典列表。
        """
        notice_obj_list = await NoticeCRUD(auth).get_list_crud(search={'status': '0',})
        return [NoticeOutSchema.model_validate(notice_obj).model_dump() for notice_obj in notice_obj_list]

    @classmethod
    async def get_notice_list_service(cls, auth: AuthSchema, search: NoticeQueryParam | None = None, order_by: list[dict] | None = None) -> list[dict]:
        """
        获取公告列表。
        
        参数:
        - auth (AuthSchema): 认证信息模型。
        - search (NoticeQueryParam | None): 查询参数模型。
        - order_by (list[dict] | None): 排序参数列表。
        
        返回:
        - list[dict]: 公告详情字典列表。
        """
        notice_obj_list = await NoticeCRUD(auth).get_list_crud(search=search.__dict__, order_by=order_by)
        return [NoticeOutSchema.model_validate(notice_obj).model_dump() for notice_obj in notice_obj_list]
    
    @classmethod
    async def create_notice_service(cls, auth: AuthSchema, data: NoticeCreateSchema) -> dict:
        """
        创建公告。
        
        参数:
        - auth (AuthSchema): 认证信息模型。
        - data (NoticeCreateSchema): 创建公告负载模型。
        
        返回:
        - dict: 创建的公告详情字典。
        
        异常:
        - CustomException: 创建失败，该公告通知已存在。
        """
        notice = await NoticeCRUD(auth).get(notice_title=data.notice_title)
        if notice:
            raise CustomException(msg='创建失败，该公告通知已存在')
        notice_obj = await NoticeCRUD(auth).create_crud(data=data)
        return NoticeOutSchema.model_validate(notice_obj).model_dump()
    
    @classmethod
    async def update_notice_service(cls, auth: AuthSchema, id: int, data: NoticeUpdateSchema) -> dict:
        """
        更新公告。
        
        参数:
        - auth (AuthSchema): 认证信息模型。
        - id (int): 公告ID。
        - data (NoticeUpdateSchema): 更新公告负载模型。
        
        返回:
        - dict: 更新的公告详情字典。
        
        异常:
        - CustomException: 更新失败，该公告通知不存在或公告通知标题重复。
        """
        notice = await NoticeCRUD(auth).get_by_id_crud(id=id)
        if not notice:
            raise CustomException(msg='更新失败，该公告通知不存在')
        exist_notice = await NoticeCRUD(auth).get(notice_title=data.notice_title)
        if exist_notice and exist_notice.id != id:
            raise CustomException(msg='更新失败，公告通知标题重复')
        notice_obj = await NoticeCRUD(auth).update_crud(id=id, data=data)
        return NoticeOutSchema.model_validate(notice_obj).model_dump()
    
    @classmethod
    async def delete_notice_service(cls, auth: AuthSchema, ids: list[int]) -> None:
        """
        删除公告。
        
        参数:
        - auth (AuthSchema): 认证信息模型。
        - ids (list[int]): 删除的ID列表。
        
        异常:
        - CustomException: 删除失败，删除对象不能为空或该公告通知不存在。
        """ 
        if len(ids) < 1:
            raise CustomException(msg='删除失败，删除对象不能为空')
        for id in ids:
            notice = await NoticeCRUD(auth).get_by_id_crud(id=id)
            if not notice:
                raise CustomException(msg='删除失败，该公告通知不存在')
        await NoticeCRUD(auth).delete_crud(ids=ids)
    
    @classmethod
    async def set_notice_available_service(cls, auth: AuthSchema, data: BatchSetAvailable) -> None:
        """
        批量设置公告状态。
        
        参数:
        - auth (AuthSchema): 认证信息模型。
        - data (BatchSetAvailable): 批量设置可用负载模型。
        
        异常:
        - CustomException: 批量设置失败，该公告通知不存在。
        """ 
        await NoticeCRUD(auth).set_available_crud(ids=data.ids, status=data.status)
    
    @classmethod
    async def export_notice_service(cls, notice_list: list[dict]) -> bytes:
        """
        导出公告列表。
        
        参数:
        - notice_list (list[dict]): 公告详情字典列表。
        
        返回:
        - bytes: Excel 文件的字节流。
        """
        mapping_dict = {
            'id': '编号',
            'notice_title': '公告标题', 
            'notice_type': '公告类型（1通知 2公告）',
            'notice_content': '公告内容',
            'status': '状态',
            'description': '备注',
            'created_time': '创建时间',
            'updated_time': '更新时间',
            'created_id': '创建者ID',
            'updated_id': '更新者ID',
        }

        # 复制数据并转换状态
        data = notice_list.copy()
        for item in data:
            # 处理状态
            item['status'] = '启用' if item.get('status') == '0' else '停用'
            # 处理公告类型
            item['notice_type'] = '通知' if item.get('notice_type') == '1' else '公告'
            item['creator'] = item.get('creator', {}).get('name', '未知') if isinstance(item.get('creator'), dict) else '未知'

        return ExcelUtil.export_list2excel(list_data=data, mapping_dict=mapping_dict)