kms-backend/endpoints/web/role.py

# -*- coding:utf-8 -*-
"""
@Created on : 2023/6/30 15:17 
@Author: hxl
@Des: 
"""
from fastapi import APIRouter, Request, Depends
from pydantic import BaseModel
from helper import respond_to, login_required
from models import Role, Archive, ModuleMenu, User

router = APIRouter(prefix='/role',dependencies=[Depends(login_required)]) # dependencies=[Depends(login_required)]


class UpdateRoleAuth(BaseModel):
    name:str
    grade:int = 0
    portion:dict
    masks:dict
    comment:str | None


class RoleDeleteRequest(BaseModel):
    role_ids: list

    

@router.get('', summary='角色列表')
async def index(request:Request, page_no: int = 1, page_size: int = 20):
    """
    获取角色列表
    :return:
    """
    role_id = request.state.current_user.role_id
    current_role = await Role.get(id=role_id)
    offset = (page_no - 1) * page_size
    query = Role.filter(grade__lte=current_role.grade)
    count = await query.count()
    roles = await query.offset(offset).limit(page_size).all()

    return respond_to(data=dict(count=count, roles= roles))

@router.get('/get_role_list_all',summary='获取角色名称列表')
async def get_role_list_all(id:str):
    """
    获取角色名称列表,按当前登录角色的等级，返回小于该等级的角色列表
    :return:
    """
    user_obj = await User.get_or_none(id = id)
    role_user_obj = await Role.get_or_none(id=user_obj.role_id)
    roles = await Role.filter(grade__lte=role_user_obj.grade).values()
    # 等级名称
    grade_name ={
        0:'普通用户',
        50:'管理员',
        75:'平台管理员',
        100:'超级管理员'
    }
    roles_list = []
    for item in roles :
        item['grade_name'] = grade_name[item.get("grade")]
        roles_list.append(item)
    return respond_to(data=roles_list)

@router.get('/get_module_menu_all',summary='获取模块菜单列表')
async def get_module_menu_all():
    """
    获取模块菜单列表
    :return:
    """
    archives = await Archive.all()
    modules = await ModuleMenu.all()
    # 中控权限
    portion = list(filter( lambda item : item.id>2000,modules))
    # 柜体权限
    masks = list(filter(lambda item : item.id<2000,modules))
    data={
        "archives":archives,
        "portion":portion,
        'masks':masks
    }
    return respond_to(data=data)

@router.get('/get_role_module_menu',summary='获取当前角色的权限模块')
async def get_role_module_menu(role_id:str):
    """
    获取当前角色的权限模块
    :param id: str role_id
    :return:
    """
    print('ddddddddddddddddddddddddd::',id)
    archives = await Archive.all()
    modules = await ModuleMenu.all()
    # user_obj = await User.get_or_none(id=id)
    # role_obj = await Role.get_or_none(id = user_obj.role_id)
    role_obj = await Role.get_or_none(id=role_id)
    # 中控所有权限
    portion_all = list(filter( lambda item : item.id>2000,modules))
    # 柜体所有权限
    masks_all = list(filter(lambda item : item.id<2000,modules))
    # # 角色的中控权限
    protion =  role_obj.portion
    # # 角色的柜体权限
    masks =  role_obj.masks
    # protion_keys = [list(d.keys()) for d in protion]
    # masks_keys = [list(d.keys()) for d in masks]
    # 构造返回给前端"portion": {"<试剂类型ID>": {"<模块ID>": "<模块值>", ...}, "<试剂类型ID>": {"<模块ID>": "<模块值>", ...}},
    # print('ddddddddddddddddddddddddddddddddd::',protion)
    # portion_list = []
    # for archive in archives:
    #     portion_name = {archive.id:list(filter(lambda item: str(item.id) in protion.get(archive.id).keys(), modules)) if protion.get(archive.id) else {}}
    #     portion_list.append(portion_name)
    # # portion_name = list(filter(lambda item: str(item.id) in protion.keys(), modules)) if protion else {}
    # masks_name = list(filter(lambda item: str(item.id) in masks.keys(), modules)) if masks else {}


    data = {
        "name":role_obj.name,
        "id":role_obj.id,
        "protion":portion_all,
        "masks":masks_all,
        "comment":role_obj.comment,
        'grade':role_obj.grade,
        "params":{
            "archives":archives,
            "protion":protion,
            "masks":masks
        }
    }
    return respond_to(data=data)

@router.post('/create_role_module_menu',summary='新增角色权限')
async def create_role_module_menu(port:UpdateRoleAuth):
    """
    新增角色权限
    :param post: 新增角色信息的请求体
    :return:
    """
    archives = await Archive.all()
    role_obj = await Role.create(**port.dict())
    # # 角色的中控权限
    protion = role_obj.portion
    # # 角色的柜体权限
    masks = role_obj.masks
    data = {
        "name": role_obj.name,
        "id": role_obj.id,
        "archives": archives,
        "protion": protion,
        "masks": masks,
    }
    return respond_to(data=data)


@router.put('/update_role_module_menu/{id}',summary='编辑角色权限')
async def update_role_module_menu(id:str,port:UpdateRoleAuth):
    """
    :param id: str role_id
    :param post: 编辑角色信息的请求体
    :return:
    """
    archives = await Archive.all()
    role_origin = await Role.get_or_none(id=id)
    role_obj = await role_origin.update_from_dict(port.dict())
    await role_obj.save()
    # # 角色的中控权限
    protion = role_obj.portion
    # # 角色的柜体权限
    masks = role_obj.masks
    data = {
        "name": role_obj.name,
        "id": role_obj.id,
        "archives": archives,
        "protion": protion,
        "masks": masks,
    }
    return respond_to(data=data)

@router.delete('/delete_role',summary='删除角色')
async def delete_role(body: RoleDeleteRequest):
    """
    删除角色，如果该角色有绑定用户，则不允许删除
    :param id
    :return
    """
    for role_id in body.role_ids:
        user_count = await User.filter(role_id=role_id).count()
        role_obj = await Role.get_or_none(id=role_id)
        if user_count != 0 :
            return respond_to(code=400,desc=f'删除失败，{role_obj.name} 角色已被其他用户绑定')

    await role_obj.filter(id__in=body.role_ids).delete()
    return respond_to(desc='删除角色成功')