ZhAgent/apps/system_manage/views/user_resource_permission.py

# coding=utf-8
"""
    @project: MaxKB
    @Author：虎虎
    @file： workspace_user_resource_permission.py
    @date：2025/4/28 16:38
    @desc:
"""
from django.db.models import QuerySet
from django.utils.translation import gettext_lazy as _
from drf_spectacular.utils import extend_schema
from rest_framework.request import Request
from rest_framework.views import APIView

from common import result
from common.auth import TokenAuth
from common.auth.authentication import has_permissions
from common.constants.permission_constants import RoleConstants, Permission, Group, Operate, ViewPermission, \
    CompareConstants
from common.log.log import log
from system_manage.api.user_resource_permission import UserResourcePermissionAPI, EditUserResourcePermissionAPI, \
    ResourceUserPermissionAPI, ResourceUserPermissionPageAPI, ResourceUserPermissionEditAPI, \
    UserResourcePermissionPageAPI
from system_manage.serializers.user_resource_permission import UserResourcePermissionSerializer, \
    ResourceUserPermissionSerializer
from users.models import User


def get_user_operation_object(user_id):
    user_model = QuerySet(model=User).filter(id=user_id).first()
    if user_model is not None:
        return {
            "name": user_model.username
        }
    return {}


class WorkSpaceUserResourcePermissionView(APIView):
    authentication_classes = [TokenAuth]

    @extend_schema(
        methods=['GET'],
        description=_('Obtain resource authorization list'),
        operation_id=_('Obtain resource authorization list'),  # type: ignore
        parameters=UserResourcePermissionAPI.get_parameters(),
        responses=UserResourcePermissionAPI.get_response(),
        tags=[_('Resources authorization')]  # type: ignore
    )
    @has_permissions(
        lambda r, kwargs: Permission(group=Group(kwargs.get('resource') + '_WORKSPACE_USER_RESOURCE_PERMISSION'),
                                     operate=Operate.READ),
        RoleConstants.ADMIN, RoleConstants.WORKSPACE_MANAGE.get_workspace_role())
    def get(self, request: Request, workspace_id: str, user_id: str, resource: str):
        return result.success(UserResourcePermissionSerializer(
            data={'workspace_id': workspace_id, 'user_id': user_id, 'auth_target_type': resource}
        ).list({'name': request.query_params.get('name'),
                'permission': request.query_params.getlist('permission[]')}, request.user))

    @extend_schema(
        methods=['PUT'],
        description=_('Modify the resource authorization list'),
        operation_id=_('Modify the resource authorization list'),  # type: ignore
        parameters=EditUserResourcePermissionAPI.get_parameters(),
        request=EditUserResourcePermissionAPI.get_request(),
        responses=EditUserResourcePermissionAPI.get_response(),
        tags=[_('Resources authorization')]  # type: ignore
    )
    @log(menu='System', operate='Modify the resource authorization list',
         get_operation_object=lambda r, k: get_user_operation_object(k.get('user_id'))
         )
    @has_permissions(
        lambda r, kwargs: Permission(group=Group(kwargs.get('resource') + '_WORKSPACE_USER_RESOURCE_PERMISSION'),
                                     operate=Operate.EDIT),
        RoleConstants.ADMIN, RoleConstants.WORKSPACE_MANAGE.get_workspace_role())
    def put(self, request: Request, workspace_id: str, user_id: str, resource: str):
        return result.success(UserResourcePermissionSerializer(
            data={'workspace_id': workspace_id, 'user_id': user_id, 'auth_target_type': resource}
        ).edit(request.data, request.user))

    class Page(APIView):
        authentication_classes = [TokenAuth]

        @extend_schema(
            methods=['GET'],
            description=_('Obtain resource authorization list by page'),
            summary=_('Obtain resource authorization list by page'),
            operation_id=_('Obtain resource authorization list by page'),  # type: ignore
            request=None,
            parameters=UserResourcePermissionPageAPI.get_parameters(),
            responses=UserResourcePermissionPageAPI.get_response(),
            tags=[_('Resources authorization')]  # type: ignore
        )
        @has_permissions(
            lambda r, kwargs: Permission(group=Group(kwargs.get('resource') + '_WORKSPACE_USER_RESOURCE_PERMISSION'),
                                         operate=Operate.READ),
            RoleConstants.ADMIN, RoleConstants.WORKSPACE_MANAGE.get_workspace_role())
        def get(self, request: Request, workspace_id: str, user_id: str, resource: str, current_page: str,
                page_size: str):
            return result.success(UserResourcePermissionSerializer(
                data={'workspace_id': workspace_id, 'user_id': user_id, 'auth_target_type': resource}
                ).page({'name': request.query_params.get('name'),
                        'permission': request.query_params.getlist('permission[]')}, current_page, page_size, request.user))


class WorkspaceResourceUserPermissionView(APIView):
    authentication_classes = [TokenAuth]

    @extend_schema(
        methods=['GET'],
        description=_('Get user authorization status of resource'),
        summary=_('Get user authorization status of resource'),
        operation_id=_('Get user authorization status of resource'),  # type: ignore
        parameters=ResourceUserPermissionAPI.get_parameters(),
        responses=ResourceUserPermissionAPI.get_response(),
        tags=[_('Resources authorization')]  # type: ignore
    )
    @has_permissions(
        lambda r, kwargs: Permission(group=Group(kwargs.get('resource')),
                                     operate=Operate.AUTH,
                                     resource_path=f"/WORKSPACE/{kwargs.get('workspace_id')}:ROLE/WORKSPACE_MANAGE"),
        lambda r, kwargs: Permission(group=Group(kwargs.get('resource')),
                                     operate=Operate.AUTH,
                                     resource_path=f"/WORKSPACE/{kwargs.get('workspace_id')}/{kwargs.get('resource').replace('_FOLDER','')}/{kwargs.get('target')}"),
        ViewPermission([RoleConstants.USER.get_workspace_role()],
                       [lambda r, kwargs: Permission(group=Group(kwargs.get('resource').replace('_FOLDER','')),
                                                     operate=Operate.SELF,
                                                     resource_path=f"/WORKSPACE/{kwargs.get('workspace_id')}/{kwargs.get('resource').replace('_FOLDER','')}/{kwargs.get('target')}")],
                       CompareConstants.AND),
        RoleConstants.WORKSPACE_MANAGE.get_workspace_role())
    def get(self, request: Request, workspace_id: str, target: str, resource: str):
        return result.success(ResourceUserPermissionSerializer(
            data={'workspace_id': workspace_id, "target": target, 'auth_target_type': resource.replace('_FOLDER',''),
                  }).list(
            {'username': request.query_params.get("username"),
             'role': request.query_params.get("role"),
             'nick_name': request.query_params.get("nick_name"),
             'permission': request.query_params.getlist("permission[]")
             }))

    @extend_schema(
        methods=['PUT'],
        description=_('Edit user authorization status of resource'),
        summary=_('Edit user authorization status of resource'),
        operation_id=_('Edit user authorization status of resource'),  # type: ignore
        parameters=ResourceUserPermissionEditAPI.get_parameters(),
        request=ResourceUserPermissionEditAPI.get_request(),
        responses=ResourceUserPermissionEditAPI.get_response(),
        tags=[_('Resources authorization')]  # type: ignore
    )
    @log(menu='System', operate='Edit user authorization status of resource',
         get_operation_object=lambda r, k: get_user_operation_object(k.get('user_id'))
         )
    @has_permissions(
        lambda r, kwargs: Permission(group=Group(kwargs.get('resource')),
                                     operate=Operate.AUTH,
                                     resource_path=f"/WORKSPACE/{kwargs.get('workspace_id')}:ROLE/WORKSPACE_MANAGE"),
        lambda r, kwargs: Permission(group=Group(kwargs.get('resource')),
                                     operate=Operate.AUTH,
                                     resource_path=f"/WORKSPACE/{kwargs.get('workspace_id')}/{kwargs.get('resource').replace('_FOLDER','')}/{kwargs.get('target')}"),
        ViewPermission([RoleConstants.USER.get_workspace_role()],
                       [lambda r, kwargs: Permission(group=Group(kwargs.get('resource').replace('_FOLDER','')),
                                                     operate=Operate.SELF,
                                                     resource_path=f"/WORKSPACE/{kwargs.get('workspace_id')}/{kwargs.get('resource').replace('_FOLDER','')}/{kwargs.get('target')}")],
                       CompareConstants.AND),
        RoleConstants.WORKSPACE_MANAGE.get_workspace_role())
    def put(self, request: Request, workspace_id: str, target: str, resource: str):
        return result.success(ResourceUserPermissionSerializer(
            data={'workspace_id': workspace_id, "target": target, 'auth_target_type': resource.replace('_FOLDER',''), })
                              .edit(instance=request.data, current_user_id=request.user.id))

    class Page(APIView):
        authentication_classes = [TokenAuth]

        @extend_schema(
            methods=['GET'],
            description=_('Get user authorization status of resource by page'),
            summary=_('Get user authorization status of resource by page'),
            operation_id=_('Get user authorization status of resource by page'),  # type: ignore
            parameters=ResourceUserPermissionPageAPI.get_parameters(),
            responses=ResourceUserPermissionPageAPI.get_response(),
            tags=[_('Resources authorization')]  # type: ignore
        )
        @has_permissions(
            lambda r, kwargs: Permission(group=Group(kwargs.get('resource')),
                                         operate=Operate.AUTH,
                                         resource_path=f"/WORKSPACE/{kwargs.get('workspace_id')}:ROLE/WORKSPACE_MANAGE"),
        lambda r, kwargs: Permission(group=Group(kwargs.get('resource')),
                                     operate=Operate.AUTH,
                                     resource_path=f"/WORKSPACE/{kwargs.get('workspace_id')}/{kwargs.get('resource').replace('_FOLDER','')}/{kwargs.get('target')}"),
             ViewPermission([RoleConstants.USER.get_workspace_role()],
                           [lambda r, kwargs: Permission(group=Group(kwargs.get('resource').replace('_FOLDER','')),
                                     operate=Operate.SELF,
                                     resource_path=f"/WORKSPACE/{kwargs.get('workspace_id')}/{kwargs.get('resource').replace('_FOLDER','')}/{kwargs.get('target')}")],
                           CompareConstants.AND),
            RoleConstants.WORKSPACE_MANAGE.get_workspace_role())
        def get(self, request: Request, workspace_id: str, target: str, resource: str, current_page: int,
                page_size: int):
            return result.success(ResourceUserPermissionSerializer(
                data={'workspace_id': workspace_id, "target": target, 'auth_target_type': resource.replace('_FOLDER',''), }
            ).page({'username': request.query_params.get("username"),
                    'role': request.query_params.get("role"),
                    'nick_name': request.query_params.get("nick_name"),
                    'permission': request.query_params.getlist("permission[]")}, current_page, page_size,
                   ))