338 lines
14 KiB
Python
338 lines
14 KiB
Python
# # apps/common/services/permission_service.py
|
||
# import logging
|
||
# from django.db import transaction
|
||
# from django.db.models import Q
|
||
# from django.utils import timezone
|
||
# from rest_framework.exceptions import ValidationError
|
||
# from apps.user.models import User
|
||
# from apps.knowledge_base.models import KnowledgeBase
|
||
# from apps.permissions.models import Permission, KnowledgeBasePermission as KBPermissionModel
|
||
|
||
# logger = logging.getLogger(__name__)
|
||
|
||
# class PermissionService:
|
||
# def can_manage_knowledge_base(self, user, knowledge_base):
|
||
# """检查用户是否是知识库的创建者"""
|
||
# return str(knowledge_base.user_id) == str(user.id)
|
||
|
||
# def check_extend_permission(self, permission, user):
|
||
# """检查是否有权限延长权限有效期"""
|
||
# knowledge_base = permission.knowledge_base
|
||
# if knowledge_base.type == 'private':
|
||
# return knowledge_base.user_id == user.id
|
||
# if knowledge_base.type == 'leader':
|
||
# return user.role == 'admin'
|
||
# if knowledge_base.type == 'member':
|
||
# return user.role == 'admin' or (
|
||
# user.role == 'leader' and user.department == knowledge_base.department
|
||
# )
|
||
# return False
|
||
|
||
# def create_permission_request(self, user, validated_data, notification_service):
|
||
# """创建权限申请并发送通知"""
|
||
# knowledge_base = validated_data['knowledge_base']
|
||
# if str(knowledge_base.user_id) == str(user.id):
|
||
# raise ValidationError({
|
||
# "code": 400,
|
||
# "message": "您是此知识库的创建者，无需申请权限",
|
||
# "data": None
|
||
# })
|
||
|
||
# approver = User.objects.get(id=knowledge_base.user_id)
|
||
# requested_permissions = validated_data.get('permissions', {})
|
||
# expires_at = validated_data.get('expires_at')
|
||
|
||
# if not any([requested_permissions.get('can_read'),
|
||
# requested_permissions.get('can_edit'),
|
||
# requested_permissions.get('can_delete')]):
|
||
# raise ValidationError("至少需要申请一种权限（读/改/删）")
|
||
|
||
# if not expires_at:
|
||
# raise ValidationError("请指定权限到期时间")
|
||
|
||
# existing_request = Permission.objects.filter(
|
||
# knowledge_base=knowledge_base,
|
||
# applicant=user,
|
||
# status='pending'
|
||
# ).first()
|
||
# if existing_request:
|
||
# raise ValidationError("您已有一个待处理的权限申请")
|
||
|
||
# existing_permission = Permission.objects.filter(
|
||
# knowledge_base=knowledge_base,
|
||
# applicant=user,
|
||
# status='approved',
|
||
# expires_at__gt=timezone.now()
|
||
# ).first()
|
||
# if existing_permission:
|
||
# raise ValidationError("您已有此知识库的访问权限")
|
||
|
||
# with transaction.atomic():
|
||
# permission = Permission.objects.create(
|
||
# knowledge_base=knowledge_base,
|
||
# applicant=user,
|
||
# approver=approver,
|
||
# permissions=requested_permissions,
|
||
# expires_at=expires_at,
|
||
# status='pending'
|
||
# )
|
||
|
||
# permission_types = []
|
||
# if requested_permissions.get('can_read'):
|
||
# permission_types.append('读取')
|
||
# if requested_permissions.get('can_edit'):
|
||
# permission_types.append('编辑')
|
||
# if requested_permissions.get('can_delete'):
|
||
# permission_types.append('删除')
|
||
# permission_str = '、'.join(permission_types)
|
||
|
||
# notification_service.send_notification(
|
||
# user=approver,
|
||
# title="新的权限申请",
|
||
# content=f"用户 {user.name} 申请了知识库 '{knowledge_base.name}' 的{permission_str}权限",
|
||
# notification_type="permission_request",
|
||
# related_object_id=str(permission.id)
|
||
# )
|
||
|
||
# return permission
|
||
|
||
# def approve_permission(self, user, permission, response_message, notification_service):
|
||
# """审批权限申请"""
|
||
# if not self.can_manage_knowledge_base(user, permission.knowledge_base):
|
||
# raise ValidationError({
|
||
# 'code': 403,
|
||
# 'message': '只有知识库创建者可以审批此申请',
|
||
# 'data': None
|
||
# })
|
||
|
||
# with transaction.atomic():
|
||
# permission.status = 'approved'
|
||
# permission.approver = user
|
||
# permission.response_message = response_message
|
||
# permission.save()
|
||
|
||
# kb_permission = KBPermissionModel.objects.filter(
|
||
# knowledge_base=permission.knowledge_base,
|
||
# user=permission.applicant
|
||
# ).first()
|
||
|
||
# if kb_permission:
|
||
# kb_permission.can_read = permission.permissions.get('can_read', False)
|
||
# kb_permission.can_edit = permission.permissions.get('can_edit', False)
|
||
# kb_permission.can_delete = permission.permissions.get('can_delete', False)
|
||
# kb_permission.granted_by = user
|
||
# kb_permission.status = 'active'
|
||
# kb_permission.expires_at = permission.expires_at
|
||
# kb_permission.save()
|
||
# logger.info(f"更新知识库权限记录: {kb_permission.id}")
|
||
# else:
|
||
# kb_permission = KBPermissionModel.objects.create(
|
||
# knowledge_base=permission.knowledge_base,
|
||
# user=permission.applicant,
|
||
# can_read=permission.permissions.get('can_read', False),
|
||
# can_edit=permission.permissions.get('can_edit', False),
|
||
# can_delete=permission.permissions.get('can_delete', False),
|
||
# granted_by=user,
|
||
# status='active',
|
||
# expires_at=permission.expires_at
|
||
# )
|
||
# logger.info(f"创建新的知识库权限记录: {kb_permission.id}")
|
||
|
||
# notification_service.send_notification(
|
||
# user=permission.applicant,
|
||
# title="权限申请已通过",
|
||
# content=f"您对知识库 '{permission.knowledge_base.name}' 的权限申请已通过",
|
||
# notification_type="permission_approved",
|
||
# related_object_id=str(permission.id)
|
||
# )
|
||
|
||
# return permission
|
||
|
||
# def reject_permission(self, user, permission, response_message, notification_service):
|
||
# """拒绝权限申请"""
|
||
# if not self.can_manage_knowledge_base(user, permission.knowledge_base):
|
||
# raise ValidationError({
|
||
# 'code': 403,
|
||
# 'message': '只有知识库创建者可以审批此申请',
|
||
# 'data': None
|
||
# })
|
||
|
||
# if permission.status != 'pending':
|
||
# raise ValidationError({
|
||
# 'code': 400,
|
||
# 'message': '该申请已被处理',
|
||
# 'data': None
|
||
# })
|
||
|
||
# if not response_message:
|
||
# raise ValidationError({
|
||
# 'code': 400,
|
||
# 'message': '请填写拒绝原因',
|
||
# 'data': None
|
||
# })
|
||
|
||
# with transaction.atomic():
|
||
# permission.status = 'rejected'
|
||
# permission.approver = user
|
||
# permission.response_message = response_message
|
||
# permission.save()
|
||
|
||
# notification_service.send_notification(
|
||
# user=permission.applicant,
|
||
# title="权限申请已拒绝",
|
||
# content=f"您对知识库 '{permission.knowledge_base.name}' 的权限申请已被拒绝\n拒绝原因：{response_message}",
|
||
# notification_type="permission_rejected",
|
||
# related_object_id=str(permission.id)
|
||
# )
|
||
|
||
# return permission
|
||
|
||
# def extend_permission(self, user, permission, new_expires_at, notification_service):
|
||
# """延长权限有效期"""
|
||
# if not self.check_extend_permission(permission, user):
|
||
# raise ValidationError({
|
||
# "code": 403,
|
||
# "message": "您没有权限延长此权限",
|
||
# "data": None
|
||
# })
|
||
|
||
# if not new_expires_at:
|
||
# raise ValidationError({
|
||
# "code": 400,
|
||
# "message": "请设置新的过期时间",
|
||
# "data": None
|
||
# })
|
||
|
||
# try:
|
||
# new_expires_at = timezone.datetime.strptime(new_expires_at, '%Y-%m-%dT%H:%M:%SZ')
|
||
# new_expires_at = timezone.make_aware(new_expires_at)
|
||
# if new_expires_at <= timezone.now():
|
||
# raise ValidationError({
|
||
# "code": 400,
|
||
# "message": "过期时间不能早于或等于当前时间",
|
||
# "data": None
|
||
# })
|
||
# except ValueError:
|
||
# raise ValidationError({
|
||
# "code": 400,
|
||
# "message": "过期时间格式错误，应为 ISO 格式 (YYYY-MM-DDThh:mm:ssZ)",
|
||
# "data": None
|
||
# })
|
||
|
||
# with transaction.atomic():
|
||
# permission.expires_at = new_expires_at
|
||
# permission.save()
|
||
|
||
# kb_permission = KBPermissionModel.objects.get(
|
||
# knowledge_base=permission.knowledge_base,
|
||
# user=permission.applicant
|
||
# )
|
||
# kb_permission.expires_at = new_expires_at
|
||
# kb_permission.save()
|
||
|
||
# notification_service.send_notification(
|
||
# user=permission.applicant,
|
||
# title="权限有效期延长",
|
||
# content=f"您对知识库 '{permission.knowledge_base.name}' 的权限有效期已延长至 {new_expires_at.strftime('%Y-%m-%d %H:%M:%S')}",
|
||
# notification_type="permission_extended",
|
||
# related_object_id=str(permission.id)
|
||
# )
|
||
|
||
# return permission
|
||
|
||
# def update_user_permission(self, admin_user, user_id, knowledge_base_id, permissions, expires_at_str, notification_service):
|
||
# """管理员更新用户权限"""
|
||
# if admin_user.role != 'admin':
|
||
# raise ValidationError({
|
||
# 'code': 403,
|
||
# 'message': '只有管理员可以直接修改权限',
|
||
# 'data': None
|
||
# })
|
||
|
||
# if not all([user_id, knowledge_base_id, permissions]):
|
||
# raise ValidationError({
|
||
# 'code': 400,
|
||
# 'message': '缺少必要参数',
|
||
# 'data': None
|
||
# })
|
||
|
||
# required_permission_fields = ['can_read', 'can_edit', 'can_delete']
|
||
# if not all(field in permissions for field in required_permission_fields):
|
||
# raise ValidationError({
|
||
# 'code': 400,
|
||
# 'message': '权限参数格式错误，必须包含 can_read、can_edit、can_delete',
|
||
# 'data': None
|
||
# })
|
||
|
||
# try:
|
||
# user = User.objects.get(id=user_id)
|
||
# knowledge_base = KnowledgeBase.objects.get(id=knowledge_base_id)
|
||
# except User.DoesNotExist:
|
||
# raise ValidationError({
|
||
# 'code': 404,
|
||
# 'message': f'用户ID {user_id} 不存在',
|
||
# 'data': None
|
||
# })
|
||
# except KnowledgeBase.DoesNotExist:
|
||
# raise ValidationError({
|
||
# 'code': 404,
|
||
# 'message': f'知识库ID {knowledge_base_id} 不存在',
|
||
# 'data': None
|
||
# })
|
||
|
||
# if knowledge_base.type == 'private' and str(knowledge_base.user_id) != str(user.id):
|
||
# raise ValidationError({
|
||
# 'code': 403,
|
||
# 'message': '不能修改其他用户的私有知识库权限',
|
||
# 'data': None
|
||
# })
|
||
|
||
# if user.role == 'member' and permissions.get('can_delete'):
|
||
# raise ValidationError({
|
||
# 'code': 400,
|
||
# 'message': '普通成员不能获得删除权限',
|
||
# 'data': None
|
||
# })
|
||
|
||
# expires_at = None
|
||
# if expires_at_str:
|
||
# try:
|
||
# expires_at = timezone.datetime.strptime(expires_at_str, '%Y-%m-%dT%H:%M:%SZ')
|
||
# expires_at = timezone.make_aware(expires_at)
|
||
# if expires_at <= timezone.now():
|
||
# raise ValidationError({
|
||
# 'code': 400,
|
||
# 'message': '过期时间不能早于或等于当前时间',
|
||
# 'data': None
|
||
# })
|
||
# except ValueError:
|
||
# raise ValidationError({
|
||
# 'code': 400,
|
||
# 'message': '过期时间格式错误，应为 ISO 格式 (YYYY-MM-DDThh:mm:ssZ)',
|
||
# 'data': None
|
||
# })
|
||
|
||
# with transaction.atomic():
|
||
# permission, created = KBPermissionModel.objects.update_or_create(
|
||
# user=user,
|
||
# knowledge_base=knowledge_base,
|
||
# defaults={
|
||
# 'can_read': permissions.get('can_read', False),
|
||
# 'can_edit': permissions.get('can_edit', False),
|
||
# 'can_delete': permissions.get('can_delete', False),
|
||
# 'granted_by': admin_user,
|
||
# 'status': 'active',
|
||
# 'expires_at': expires_at
|
||
# }
|
||
# )
|
||
|
||
# notification_service.send_notification(
|
||
# user=user,
|
||
# title="知识库权限更新",
|
||
# content=f"管理员已{created and '授予' or '更新'}您对知识库 '{knowledge_base.name}' 的权限",
|
||
# notification_type="permission_updated",
|
||
# related_object_id=str(permission.id)
|
||
# )
|
||
|
||
# return permission, created
|
||
|