vue-fastapi-admin/ruoyi-fastapi-backend/module_admin/service/log_service.py

from module_admin.dao.log_dao import *
from module_admin.service.dict_service import Request, DictDataService
from module_admin.entity.vo.common_vo import CrudResponseModel
from utils.common_util import export_list2excel, CamelCaseUtil


class OperationLogService:
    """
    操作日志管理模块服务层
    """

    @classmethod
    def get_operation_log_list_services(cls, query_db: Session, query_object: OperLogQueryModel):
        """
        获取操作日志列表信息service
        :param query_db: orm对象
        :param query_object: 查询参数对象
        :return: 操作日志列表信息对象
        """
        operation_log_list_result = OperationLogDao.get_operation_log_list(query_db, query_object)

        return CamelCaseUtil.transform_result(operation_log_list_result)

    @classmethod
    def add_operation_log_services(cls, query_db: Session, page_object: OperLogModel):
        """
        新增操作日志service
        :param query_db: orm对象
        :param page_object: 新增操作日志对象
        :return: 新增操作日志校验结果
        """
        try:
            OperationLogDao.add_operation_log_dao(query_db, page_object)
            query_db.commit()
            result = dict(is_success=True, message='新增成功')
        except Exception as e:
            query_db.rollback()
            result = dict(is_success=False, message=str(e))

        return CrudResponseModel(**result)

    @classmethod
    def delete_operation_log_services(cls, query_db: Session, page_object: DeleteOperLogModel):
        """
        删除操作日志信息service
        :param query_db: orm对象
        :param page_object: 删除操作日志对象
        :return: 删除操作日志校验结果
        """
        if page_object.oper_ids.split(','):
            oper_id_list = page_object.oper_ids.split(',')
            try:
                for oper_id in oper_id_list:
                    OperationLogDao.delete_operation_log_dao(query_db, OperLogModel(operId=oper_id))
                query_db.commit()
                result = dict(is_success=True, message='删除成功')
            except Exception as e:
                query_db.rollback()
                raise e
        else:
            result = dict(is_success=False, message='传入操作日志id为空')
        return CrudResponseModel(**result)

    @classmethod
    def clear_operation_log_services(cls, query_db: Session):
        """
        清除操作日志信息service
        :param query_db: orm对象
        :return: 清除操作日志校验结果
        """
        try:
            OperationLogDao.clear_operation_log_dao(query_db)
            query_db.commit()
            result = dict(is_success=True, message='清除成功')
        except Exception as e:
            query_db.rollback()
            raise e

        return CrudResponseModel(**result)

    @classmethod
    async def export_operation_log_list_services(cls, request: Request, operation_log_list: List):
        """
        导出操作日志信息service
        :param request: Request对象
        :param operation_log_list: 操作日志信息列表
        :return: 操作日志信息对应excel的二进制数据
        """
        # 创建一个映射字典，将英文键映射到中文键
        mapping_dict = {
            "operId": "日志编号",
            "title": "系统模块",
            "businessType": "操作类型",
            "method": "方法名称",
            "requestMethod": "请求方式",
            "operName": "操作人员",
            "deptName": "部门名称",
            "operUrl": "请求URL",
            "operIp": "操作地址",
            "operLocation": "操作地点",
            "operParam": "请求参数",
            "jsonResult": "返回参数",
            "status": "操作状态",
            "error_msg": "错误消息",
            "operTime": "操作日期",
            "costTime": "消耗时间（毫秒）"
        }

        data = operation_log_list
        operation_type_list = await DictDataService.query_dict_data_list_from_cache_services(request.app.state.redis, dict_type='sys_oper_type')
        operation_type_option = [dict(label=item.get('dictLabel'), value=item.get('dictValue')) for item in operation_type_list]
        operation_type_option_dict = {item.get('value'): item for item in operation_type_option}

        for item in data:
            if item.get('status') == 0:
                item['status'] = '成功'
            else:
                item['status'] = '失败'
            if str(item.get('businessType')) in operation_type_option_dict.keys():
                item['businessType'] = operation_type_option_dict.get(str(item.get('businessType'))).get('label')

        new_data = [{mapping_dict.get(key): value for key, value in item.items() if mapping_dict.get(key)} for item in data]
        binary_data = export_list2excel(new_data)

        return binary_data


class LoginLogService:
    """
    登录日志管理模块服务层
    """

    @classmethod
    def get_login_log_list_services(cls, query_db: Session, query_object: LoginLogQueryModel):
        """
        获取登录日志列表信息service
        :param query_db: orm对象
        :param query_object: 查询参数对象
        :return: 登录日志列表信息对象
        """
        operation_log_list_result = LoginLogDao.get_login_log_list(query_db, query_object)

        return CamelCaseUtil.transform_result(operation_log_list_result)

    @classmethod
    def add_login_log_services(cls, query_db: Session, page_object: LogininforModel):
        """
        新增登录日志service
        :param query_db: orm对象
        :param page_object: 新增登录日志对象
        :return: 新增登录日志校验结果
        """
        try:
            LoginLogDao.add_login_log_dao(query_db, page_object)
            query_db.commit()
            result = dict(is_success=True, message='新增成功')
        except Exception as e:
            query_db.rollback()
            result = dict(is_success=False, message=str(e))

        return CrudResponseModel(**result)

    @classmethod
    def delete_login_log_services(cls, query_db: Session, page_object: DeleteLoginLogModel):
        """
        删除操作日志信息service
        :param query_db: orm对象
        :param page_object: 删除操作日志对象
        :return: 删除操作日志校验结果
        """
        if page_object.info_ids.split(','):
            info_id_list = page_object.info_ids.split(',')
            try:
                for info_id in info_id_list:
                    LoginLogDao.delete_login_log_dao(query_db, LogininforModel(infoId=info_id))
                query_db.commit()
                result = dict(is_success=True, message='删除成功')
            except Exception as e:
                query_db.rollback()
                raise e
        else:
            result = dict(is_success=False, message='传入登录日志id为空')
        return CrudResponseModel(**result)

    @classmethod
    def clear_login_log_services(cls, query_db: Session):
        """
        清除操作日志信息service
        :param query_db: orm对象
        :return: 清除操作日志校验结果
        """
        try:
            LoginLogDao.clear_login_log_dao(query_db)
            query_db.commit()
            result = dict(is_success=True, message='清除成功')
        except Exception as e:
            query_db.rollback()
            raise e

        return CrudResponseModel(**result)

    @classmethod
    async def unlock_user_services(cls, request: Request, unlock_user: UnlockUser):
        locked_user = await request.app.state.redis.get(f"account_lock:{unlock_user.user_name}")
        if locked_user:
            await request.app.state.redis.delete(f"account_lock:{unlock_user.user_name}")
            result = dict(is_success=True, message='解锁成功')
        else:
            result = dict(is_success=False, message='该用户未锁定')
        return CrudResponseModel(**result)

    @staticmethod
    def export_login_log_list_services(login_log_list: List):
        """
        导出登录日志信息service
        :param login_log_list: 登录日志信息列表
        :return: 登录日志信息对应excel的二进制数据
        """
        # 创建一个映射字典，将英文键映射到中文键
        mapping_dict = {
            "infoId": "访问编号",
            "userName": "用户名称",
            "ipaddr": "登录地址",
            "loginLocation": "登录地点",
            "browser": "浏览器",
            "os": "操作系统",
            "status": "登录状态",
            "msg": "操作信息",
            "loginTime": "登录日期"
        }

        data = login_log_list

        for item in data:
            if item.get('status') == '0':
                item['status'] = '成功'
            else:
                item['status'] = '失败'
        new_data = [{mapping_dict.get(key): value for key, value in item.items() if mapping_dict.get(key)} for item in data]
        binary_data = export_list2excel(new_data)

        return binary_data