VberAdminPy/VapAdmin/BACKEND/domain/services/admin/sys_oper_log_service.py

from typing import Optional, List, Dict, Any

from sqlalchemy import select, func, or_

from core.exceptions import ServiceWarning
from domain.dtos import SysOperLogDto, PageDto, PageResultDto
from domain.models import SysOperLogModel
from domain.services.base_services import ServiceBase


class SysOperLogService(ServiceBase):
    def __init__(self, db_name: Optional[str] = None):
        super().__init__(db_name)

    async def get_page_list(
        self,
        page_dto: PageDto,
    ) -> PageResultDto[SysOperLogDto]:
        """
        获取操作日志列表
        :param page_dto: 分页查询DTO
        :return: 操作日志列表
        """
        try:
            async with await self._get_async_db() as db:
                query = select(SysOperLogModel)
                query = self._apply_search(query, page_dto.search)
                query = self._apply_filter(
                    query, page_dto.filters, page_dto.filter_conditions
                )
                total = await db.scalar(select(func.count()).select_from(query))
                query = self._apply_order(query, page_dto.order)
                query = query.offset(page_dto.offset).limit(page_dto.limit)
                result = await db.execute(query)
                models = result.scalars().all()
                return PageResultDto(
                    total=total,
                    rows=self._apply_map_get_dto_list(models),
                    page_size=page_dto.page_size,
                )
        except Exception as e:
            raise ServiceWarning(f"获取操作日志列表失败: {str(e)}")

    @staticmethod
    def _apply_search(query, search: str):
        """
        根据关键字构建查询对象
        :param query: 查询对象
        :param search: 关键字
        :return: 查询对象
        """
        if search:
            query = query.where(
                or_(
                    SysOperLogModel.oper_name.like(f"%{search}%"),
                    SysOperLogModel.oper_location.like(f"%{search}%"),
                    SysOperLogModel.oper_ip.like(f"%{search}%"),
                    SysOperLogModel.oper_url.like(f"%{search}%"),
                )
            )
        return query

    def _apply_filter(
        self,
        query,
        filters: Optional[Dict[str, Any]] = None,
        filter_conditions: Optional[List[Dict[str, Any]]] = None,
    ):
        """
        根据过滤条件构建查询对象
        :param query: 查询对象
        :param filters: 过滤条件
        :return: 查询对象
        """

        return self._apply_filter_base(
            SysOperLogModel, query, filters, filter_conditions
        )

    def _apply_order(self, query, order: str):
        """
        根据排序字段构建查询对象
        :param query: 查询对象
        :param order: 排序字段
        :return: 查询对象
        """
        if not order:
            order = "oper_time desc"
        return self._apply_order_base(SysOperLogModel, query, order)

    def _apply_map_get_dto_list(self, models) -> List[SysOperLogDto]:
        """
        将模型对象转换为DTO对象
        :param models: 模型对象列表
        :return: DTO对象
        """
        return [self._apply_map_get_dto(model) for model in models]

    @staticmethod
    def _apply_map_get_dto(model) -> SysOperLogDto:
        """
        将模型对象转换为DTO对象
        :param model: 模型对象
        :return: DTO对象
        """
        return SysOperLogDto.from_model(model)