Class FolderTreeSerializer

apps/folders/serializers/folder.py:305–377

class FolderTreeSerializer(serializers.Serializer):
    workspace_id = serializers.CharField(required=True, allow_null=True, allow_blank=True, label=_('workspace id'))
    source = serializers.CharField(required=True, label=_('source'))

    @staticmethod
    def _check_tree_integrity(queryset):
        """检查树结构完整性"""
        for folder in queryset:
            if folder.lft >= folder.rght:
                return True  # 需要重建
            if folder.is_leaf_node() and folder.get_children().exists():
                return True  # 需要重建
        return False

    @staticmethod
    def _having_read_permission_by_role(user_id: str, workspace_id: str, source: str):
        workspace_user_role_mapping_model = DatabaseModelManage.get_model("workspace_user_role_mapping")
        role_permission_mapping_model = DatabaseModelManage.get_model("role_permission_mapping_model")
        is_x_pack_ee = workspace_user_role_mapping_model is not None and role_permission_mapping_model is not None
        if is_x_pack_ee:
            return QuerySet(workspace_user_role_mapping_model).select_related('role', 'user').filter(
                Q(role__rolepermission__permission_id=f"{source}_FOLDER:READ") | Q(role__internal=True),
                workspace_id=workspace_id,
                user_id=user_id,
                role__type=RoleConstants.USER.value.__str__(),
            ).exists()

        return False

    def get_folder_tree(self,
                        current_user, name=None):
        self.is_valid(raise_exception=True)
        user_id = current_user.id
        workspace_id = self.data.get('workspace_id')
        source = self.data.get('source')

        Folder = get_folder_type(source)  # noqa

        # 检查特定工作空间的树结构完整性
        workspace_folders = Folder.objects.filter(workspace_id=workspace_id)
        # 如果发现数据不一致，重建整个表（这是 MPTT 的限制）
        if self._check_tree_integrity(workspace_folders):
            Folder.objects.rebuild()

        workspace_manage = is_workspace_manage(user_id, workspace_id)

        base_q = Q(workspace_id=workspace_id)

        if name is not None:
            base_q &= Q(name__contains=name)
        if not workspace_manage:
            having_read_permission_by_role = has_exact_permission_by_role(user_id, workspace_id,
                                                                          f"{source}_FOLDER:READ",
                                                                          RoleConstants.USER.value.__str__())
            permission_condition = ['VIEW']
            if having_read_permission_by_role:
                permission_condition = ['VIEW', 'ROLE']