operations_project/apps/permissions/services/permission_service.py

# 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
        )