139 lines
4.9 KiB
Python
139 lines
4.9 KiB
Python
# apps/permissions/services/permission_service.py
|
|
import logging
|
|
from apps.accounts.models import User
|
|
from apps.knowledge_base.models import KnowledgeBase
|
|
from apps.permissions.models import KnowledgeBasePermission as KBPermissionModel
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class KnowledgeBasePermissionMixin:
|
|
"""知识库权限管理混入类"""
|
|
|
|
def _can_read(self, type, user, department=None, group=None, creator_id=None, knowledge_base_id=None):
|
|
"""检查读取权限"""
|
|
try:
|
|
# 1. 检查显式权限表
|
|
if knowledge_base_id:
|
|
permission = KBPermissionModel.objects.filter(
|
|
knowledge_base_id=knowledge_base_id,
|
|
user=user,
|
|
can_read=True,
|
|
status='active'
|
|
).first()
|
|
if permission:
|
|
return True
|
|
|
|
# 2. 检查角色权限
|
|
if type == 'private':
|
|
return str(user.id) == str(creator_id)
|
|
if type == 'member':
|
|
return user.department == department
|
|
if type == 'leader':
|
|
return user.department == department and user.role in ['leader', 'admin']
|
|
if type == 'admin':
|
|
return True
|
|
return False
|
|
|
|
except Exception as e:
|
|
logger.error(f"检查读取权限时出错: {str(e)}")
|
|
return False
|
|
|
|
def _can_edit(self, type, user, department=None, group=None, creator_id=None, knowledge_base_id=None):
|
|
"""检查编辑权限"""
|
|
try:
|
|
# 1. 检查显式权限表
|
|
if knowledge_base_id:
|
|
permission = KBPermissionModel.objects.filter(
|
|
knowledge_base_id=knowledge_base_id,
|
|
user=user,
|
|
can_edit=True,
|
|
status='active'
|
|
).first()
|
|
if permission:
|
|
return True
|
|
|
|
# 2. 检查角色权限
|
|
if type == 'private':
|
|
return str(user.id) == str(creator_id)
|
|
if type == 'member':
|
|
return user.department == department and user.role in ['leader', 'admin']
|
|
if type == 'leader':
|
|
return user.department == department and user.role in ['leader', 'admin']
|
|
if type == 'admin':
|
|
return True
|
|
return False
|
|
|
|
except Exception as e:
|
|
logger.error(f"检查编辑权限时出错: {str(e)}")
|
|
return False
|
|
|
|
def _can_delete(self, type, user, department=None, group=None, creator_id=None, knowledge_base_id=None):
|
|
"""检查删除权限"""
|
|
try:
|
|
# 1. 检查显式权限表
|
|
if knowledge_base_id:
|
|
permission = KBPermissionModel.objects.filter(
|
|
knowledge_base_id=knowledge_base_id,
|
|
user=user,
|
|
can_delete=True,
|
|
status='active'
|
|
).first()
|
|
if permission:
|
|
return True
|
|
|
|
# 2. 检查角色权限
|
|
if type == 'private':
|
|
return str(user.id) == str(creator_id)
|
|
if type == 'member':
|
|
return user.department == department and user.role == 'admin'
|
|
if type == 'leader':
|
|
return user.department == department and user.role == 'admin'
|
|
if type == 'admin':
|
|
return True
|
|
return False
|
|
|
|
except Exception as e:
|
|
logger.error(f"检查删除权限时出错: {str(e)}")
|
|
return False
|
|
|
|
def check_knowledge_base_permission(self, knowledge_base, user, required_permission='read'):
|
|
"""统一的知识库权限检查方法"""
|
|
if not knowledge_base:
|
|
return False
|
|
|
|
try:
|
|
permission = KBPermissionModel.objects.filter(
|
|
knowledge_base_id=knowledge_base.id,
|
|
user=user,
|
|
status='active'
|
|
).first()
|
|
|
|
if permission:
|
|
if required_permission == 'read':
|
|
return permission.can_read
|
|
elif required_permission == 'edit':
|
|
return permission.can_edit
|
|
elif required_permission == 'delete':
|
|
return permission.can_delete
|
|
|
|
except Exception as e:
|
|
logger.error(f"检查显式权限时出错: {str(e)}")
|
|
|
|
permission_method = {
|
|
'read': self._can_read,
|
|
'edit': self._can_edit,
|
|
'delete': self._can_delete
|
|
}.get(required_permission)
|
|
|
|
if not permission_method:
|
|
return False
|
|
|
|
return permission_method(
|
|
type=knowledge_base.type,
|
|
user=user,
|
|
department=knowledge_base.department,
|
|
group=knowledge_base.group,
|
|
creator_id=knowledge_base.user_id,
|
|
knowledge_base_id=knowledge_base.id
|
|
)
|
|
|