管理员直接修改用户权限

role_based_system/urls.py

@@ -34,8 +34,8 @@ urlpatterns = [
 ]
 
 # 添加调试工具栏（仅在DEBUG模式下）
-if settings.DEBUG:
+# if settings.DEBUG:
-    import debug_toolbar
+#     import debug_toolbar
-    urlpatterns = [
+#     urlpatterns = [
-        path('__debug__/', include(debug_toolbar.urls)),
+#         path('__debug__/', include(debug_toolbar.urls)),
-    ] + urlpatterns
+#     ] + urlpatterns

user_management/migrations/0001_initial.py

@@ -1,4 +1,4 @@
-# Generated by Django 5.1.5 on 2025-02-26 09:23
+# Generated by Django 5.1.5 on 2025-03-17 05:47
 
 import django.contrib.auth.models
 import django.contrib.auth.validators
@@ -73,79 +73,22 @@ class Migration(migrations.Migration):
             name='KnowledgeBase',
             fields=[
                 ('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
-                ('name', models.CharField(max_length=100, verbose_name='知识库名称')),
+                ('user_id', models.UUIDField(verbose_name='创建者ID')),
+                ('name', models.CharField(max_length=100, unique=True, verbose_name='知识库名称')),
                 ('desc', models.TextField(blank=True, null=True, verbose_name='知识库描述')),
                 ('type', models.CharField(choices=[('admin', '管理级知识库'), ('leader', '部门级知识库'), ('member', '成员级知识库'), ('private', '私有知识库'), ('secret', '公司级别私密知识库')], default='private', max_length=20, verbose_name='知识库类型')),
                 ('department', models.CharField(blank=True, max_length=50, null=True)),
                 ('group', models.CharField(blank=True, max_length=50, null=True)),
-                ('user_id', models.CharField(max_length=50, verbose_name='创建者ID')),
                 ('documents', models.JSONField(default=list)),
                 ('char_length', models.IntegerField(default=0)),
                 ('document_count', models.IntegerField(default=0)),
                 ('external_id', models.UUIDField(blank=True, null=True)),
                 ('create_time', models.DateTimeField(auto_now_add=True)),
                 ('update_time', models.DateTimeField(auto_now=True)),
-                ('owners', models.ManyToManyField(related_name='owned_knowledge_bases', to=settings.AUTH_USER_MODEL, verbose_name='所有者')),
             ],
             options={
                 'db_table': 'knowledge_bases',
-            },
+                'indexes': [models.Index(fields=['type'], name='knowledge_b_type_0439e7_idx'), models.Index(fields=['department'], name='knowledge_b_departm_e739fd_idx'), models.Index(fields=['group'], name='knowledge_b_group_3dcf34_idx')],
-        ),
-        migrations.CreateModel(
-            name='ChatHistory',
-            fields=[
-                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
-                ('conversation_id', models.CharField(db_index=True, max_length=100)),
-                ('parent_id', models.CharField(blank=True, max_length=100, null=True)),
-                ('role', models.CharField(choices=[('user', '用户'), ('assistant', 'AI助手'), ('system', '系统')], max_length=20)),
-                ('content', models.TextField()),
-                ('tokens', models.IntegerField(default=0, help_text='消息token数')),
-                ('metadata', models.JSONField(blank=True, default=dict)),
-                ('created_at', models.DateTimeField(auto_now_add=True)),
-                ('is_deleted', models.BooleanField(default=False)),
-                ('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL)),
-                ('knowledge_base', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='user_management.knowledgebase')),
-            ],
-            options={
-                'db_table': 'chat_history',
-                'ordering': ['created_at'],
-            },
-        ),
-        migrations.CreateModel(
-            name='KnowledgeBasePermission',
-            fields=[
-                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
-                ('can_read', models.BooleanField(default=False, verbose_name='查看权限')),
-                ('can_edit', models.BooleanField(default=False, verbose_name='修改权限')),
-                ('can_delete', models.BooleanField(default=False, verbose_name='删除权限')),
-                ('status', models.CharField(choices=[('active', '生效中'), ('expired', '已过期'), ('revoked', '已撤销')], default='active', max_length=10, verbose_name='状态')),
-                ('granted_at', models.DateTimeField(auto_now_add=True, verbose_name='授权时间')),
-                ('expires_at', models.DateTimeField(blank=True, null=True, verbose_name='过期时间')),
-                ('granted_by', models.ForeignKey(null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='granted_permissions', to=settings.AUTH_USER_MODEL, verbose_name='授权人')),
-                ('knowledge_base', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='user_permissions', to='user_management.knowledgebase', verbose_name='知识库')),
-                ('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='knowledge_base_permissions', to=settings.AUTH_USER_MODEL, verbose_name='用户')),
-            ],
-            options={
-                'verbose_name': '知识库权限',
-                'verbose_name_plural': '知识库权限',
-                'db_table': 'knowledge_base_permissions',
-            },
-        ),
-        migrations.CreateModel(
-            name='Notification',
-            fields=[
-                ('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
-                ('type', models.CharField(choices=[('permission_request', '权限申请'), ('permission_approved', '权限批准'), ('permission_rejected', '权限拒绝'), ('permission_expired', '权限过期'), ('system_notice', '系统通知')], max_length=20)),
-                ('title', models.CharField(max_length=100)),
-                ('content', models.TextField()),
-                ('is_read', models.BooleanField(default=False)),
-                ('related_resource', models.CharField(blank=True, max_length=100)),
-                ('created_at', models.DateTimeField(auto_now_add=True)),
-                ('receiver', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='received_notifications', to=settings.AUTH_USER_MODEL)),
-                ('sender', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='sent_notifications', to=settings.AUTH_USER_MODEL)),
-            ],
-            options={
-                'ordering': ['-created_at'],
             },
         ),
         migrations.CreateModel(
@@ -181,40 +124,65 @@ class Migration(migrations.Migration):
                 'db_table': 'user_profiles',
             },
         ),
-        migrations.AddIndex(
+        migrations.CreateModel(
-            model_name='knowledgebase',
+            name='ChatHistory',
-            index=models.Index(fields=['type'], name='knowledge_b_type_0439e7_idx'),
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('conversation_id', models.CharField(db_index=True, max_length=100)),
+                ('parent_id', models.CharField(blank=True, max_length=100, null=True)),
+                ('role', models.CharField(choices=[('user', '用户'), ('assistant', 'AI助手'), ('system', '系统')], max_length=20)),
+                ('content', models.TextField()),
+                ('tokens', models.IntegerField(default=0, help_text='消息token数')),
+                ('metadata', models.JSONField(blank=True, default=dict)),
+                ('created_at', models.DateTimeField(auto_now_add=True)),
+                ('is_deleted', models.BooleanField(default=False)),
+                ('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL)),
+                ('knowledge_base', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='user_management.knowledgebase')),
+            ],
+            options={
+                'db_table': 'chat_history',
+                'ordering': ['created_at'],
+                'indexes': [models.Index(fields=['conversation_id', 'created_at'], name='chat_histor_convers_33721a_idx'), models.Index(fields=['user', 'created_at'], name='chat_histor_user_id_aa050a_idx')],
+            },
         ),
-        migrations.AddIndex(
+        migrations.CreateModel(
-            model_name='knowledgebase',
+            name='KnowledgeBasePermission',
-            index=models.Index(fields=['department'], name='knowledge_b_departm_e739fd_idx'),
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('can_read', models.BooleanField(default=False, verbose_name='查看权限')),
+                ('can_edit', models.BooleanField(default=False, verbose_name='修改权限')),
+                ('can_delete', models.BooleanField(default=False, verbose_name='删除权限')),
+                ('status', models.CharField(choices=[('active', '生效中'), ('expired', '已过期'), ('revoked', '已撤销')], default='active', max_length=10, verbose_name='状态')),
+                ('granted_at', models.DateTimeField(auto_now_add=True, verbose_name='授权时间')),
+                ('expires_at', models.DateTimeField(blank=True, null=True, verbose_name='过期时间')),
+                ('granted_by', models.ForeignKey(null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='granted_permissions', to=settings.AUTH_USER_MODEL, verbose_name='授权人')),
+                ('knowledge_base', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='user_permissions', to='user_management.knowledgebase', verbose_name='知识库')),
+                ('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='knowledge_base_permissions', to=settings.AUTH_USER_MODEL, verbose_name='用户')),
+            ],
+            options={
+                'verbose_name': '知识库权限',
+                'verbose_name_plural': '知识库权限',
+                'db_table': 'knowledge_base_permissions',
+                'indexes': [models.Index(fields=['knowledge_base', 'user', 'status'], name='knowledge_b_knowled_88e81e_idx')],
+                'unique_together': {('knowledge_base', 'user')},
+            },
         ),
-        migrations.AddIndex(
+        migrations.CreateModel(
-            model_name='knowledgebase',
+            name='Notification',
-            index=models.Index(fields=['group'], name='knowledge_b_group_3dcf34_idx'),
+            fields=[
-        ),
+                ('id', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
-        migrations.AddIndex(
+                ('type', models.CharField(choices=[('permission_request', '权限申请'), ('permission_approved', '权限批准'), ('permission_rejected', '权限拒绝'), ('permission_expired', '权限过期'), ('system_notice', '系统通知')], max_length=20)),
-            model_name='chathistory',
+                ('title', models.CharField(max_length=100)),
-            index=models.Index(fields=['conversation_id', 'created_at'], name='chat_histor_convers_33721a_idx'),
+                ('content', models.TextField()),
-        ),
+                ('is_read', models.BooleanField(default=False)),
-        migrations.AddIndex(
+                ('related_resource', models.CharField(blank=True, max_length=100)),
-            model_name='chathistory',
+                ('created_at', models.DateTimeField(auto_now_add=True)),
-            index=models.Index(fields=['user', 'created_at'], name='chat_histor_user_id_aa050a_idx'),
+                ('receiver', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='received_notifications', to=settings.AUTH_USER_MODEL)),
-        ),
+                ('sender', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='sent_notifications', to=settings.AUTH_USER_MODEL)),
-        migrations.AddIndex(
+            ],
-            model_name='knowledgebasepermission',
+            options={
-            index=models.Index(fields=['knowledge_base', 'user', 'status'], name='knowledge_b_knowled_88e81e_idx'),
+                'ordering': ['-created_at'],
-        ),
+                'indexes': [models.Index(fields=['receiver', '-created_at'], name='user_manage_receive_fcb0eb_idx'), models.Index(fields=['type', 'is_read'], name='user_manage_type_362052_idx')],
-        migrations.AlterUniqueTogether(
+            },
-            name='knowledgebasepermission',
-            unique_together={('knowledge_base', 'user')},
-        ),
-        migrations.AddIndex(
-            model_name='notification',
-            index=models.Index(fields=['receiver', '-created_at'], name='user_manage_receive_fcb0eb_idx'),
-        ),
-        migrations.AddIndex(
-            model_name='notification',
-            index=models.Index(fields=['type', 'is_read'], name='user_manage_type_362052_idx'),
         ),
     ]

user_management/views.py

@@ -42,6 +42,7 @@ from rest_framework import serializers
 import traceback
 
 
+
 # 添加模型导入
 from .models import (
     User,
@@ -1665,7 +1666,6 @@ class KnowledgeBaseViewSet(viewsets.ModelViewSet):
                 "data": None
             }, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
         
-
 class PermissionViewSet(viewsets.ModelViewSet):
     serializer_class = PermissionSerializer
     permission_classes = [IsAuthenticated]
@@ -1693,10 +1693,70 @@ class PermissionViewSet(viewsets.ModelViewSet):
         query = Q(applicant=user)  # 自己发出的申请
         query |= Q(knowledge_base_id__in=managed_kb_ids)  # 有管理权限的知识库的申请
         
+        # 使用 select_related 优化查询，预加载关联的对象
         return Permission.objects.filter(query).distinct().select_related(
-            'knowledge_base', 'applicant', 'approver'
+            'knowledge_base',  # 预加载知识库信息
+            'applicant',      # 预加载申请人信息
+            'approver'        # 预加载审批人信息
         )
 
+    def list(self, request, *args, **kwargs):
+        """获取权限申请列表，包含详细信息"""
+        queryset = self.get_queryset()
+        page = self.paginate_queryset(queryset)
+        
+        if page is not None:
+            data = []
+            for permission in page:
+                permission_data = {
+                    'id': str(permission.id),
+                    'knowledge_base': {
+                        'id': str(permission.knowledge_base.id),
+                        'name': permission.knowledge_base.name,
+                        'type': permission.knowledge_base.type,
+                    },
+                    'applicant': {
+                        'id': str(permission.applicant.id),
+                        'username': permission.applicant.username,
+                        'name': permission.applicant.name,
+                        'department': permission.applicant.department,
+                    },
+                    'permissions': permission.permissions,
+                    'status': permission.status,
+                    'created_at': permission.created_at.strftime('%Y-%m-%d %H:%M:%S'),
+                    'expires_at': permission.expires_at.strftime('%Y-%m-%d %H:%M:%S') if permission.expires_at else None,
+                }
+                
+                # 添加审批人信息（如果已审批）
+                if permission.approver:
+                    permission_data['approver'] = {
+                        'id': str(permission.approver.id),
+                        'username': permission.approver.username,
+                        'name': permission.approver.name,
+                        'department': permission.approver.department,
+                    }
+                    permission_data['response_message'] = permission.response_message
+                
+                data.append(permission_data)
+            
+            return Response({
+                'code': 200,
+                'message': '获取权限申请列表成功',
+                'data': {
+                    'total': self.paginator.page.paginator.count,
+                    'results': data
+                }
+            })
+        
+        return Response({
+            'code': 200,
+            'message': '获取权限申请列表成功',
+            'data': {
+                'total': queryset.count(),
+                'results': []
+            }
+        })
+
     def perform_create(self, serializer):
         """创建权限申请并发送通知给知识库创建者"""
         # 获取知识库
@@ -2001,6 +2061,375 @@ class PermissionViewSet(viewsets.ModelViewSet):
         
         return False
 
+    @action(detail=False, methods=['get'])
+    def user_permissions(self, request):
+        """获取指定用户的所有知识库权限"""
+        try:
+            # 获取用户名参数
+            username = request.query_params.get('username')
+            if not username:
+                return Response({
+                    'code': 400,
+                    'message': '请提供用户名',
+                    'data': None
+                }, status=status.HTTP_400_BAD_REQUEST)
+            
+            # 获取用户
+            try:
+                target_user = User.objects.get(username=username)
+            except User.DoesNotExist:
+                return Response({
+                    'code': 404,
+                    'message': f'用户 {username} 不存在',
+                    'data': None
+                }, status=status.HTTP_404_NOT_FOUND)
+            
+            # 获取该用户的所有权限记录
+            permissions = KBPermissionModel.objects.filter(
+                user=target_user,
+                status='active'
+            ).select_related('knowledge_base', 'granted_by')
+            
+            # 构建响应数据
+            permissions_data = []
+            for perm in permissions:
+                perm_data = {
+                    'id': str(perm.id),
+                    'knowledge_base': {
+                        'id': str(perm.knowledge_base.id),
+                        'name': perm.knowledge_base.name,
+                        'type': perm.knowledge_base.type,
+                        'department': perm.knowledge_base.department,
+                        'group': perm.knowledge_base.group
+                    },
+                    'permissions': {
+                        'can_read': perm.can_read,
+                        'can_edit': perm.can_edit,
+                        'can_delete': perm.can_delete
+                    },
+                    'granted_by': {
+                        'id': str(perm.granted_by.id) if perm.granted_by else None,
+                        'username': perm.granted_by.username if perm.granted_by else None,
+                        'name': perm.granted_by.name if perm.granted_by else None
+                    },
+                    'created_at': perm.created_at.strftime('%Y-%m-%d %H:%M:%S'),
+                    'expires_at': perm.expires_at.strftime('%Y-%m-%d %H:%M:%S') if perm.expires_at else None,
+                    'status': perm.status
+                }
+                permissions_data.append(perm_data)
+            
+            return Response({
+                'code': 200,
+                'message': '获取用户权限成功',
+                'data': {
+                    'user': {
+                        'id': str(target_user.id),
+                        'username': target_user.username,
+                        'name': target_user.name,
+                        'department': target_user.department,
+                        'role': target_user.role
+                    },
+                    'permissions': permissions_data
+                }
+            })
+            
+        except Exception as e:
+            logger.error(f"获取用户权限失败: {str(e)}")
+            logger.error(traceback.format_exc())
+            return Response({
+                'code': 500,
+                'message': f'获取用户权限失败: {str(e)}',
+                'data': None
+            }, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+    @action(detail=False, methods=['get'])
+    def all_permissions(self, request):
+        """管理员获取所有用户的知识库权限（不包括私有知识库）"""
+        try:
+            # 检查是否是管理员
+            if request.user.role != 'admin':
+                return Response({
+                    'code': 403,
+                    'message': '只有管理员可以查看所有权限',
+                    'data': None
+                }, status=status.HTTP_403_FORBIDDEN)
+            
+            # 获取查询参数
+            page = int(request.query_params.get('page', 1))
+            page_size = int(request.query_params.get('page_size', 10))
+            status_filter = request.query_params.get('status')  # active/expired
+            department = request.query_params.get('department')
+            kb_type = request.query_params.get('kb_type')  # 知识库类型筛选
+            
+            # 构建基础查询
+            queryset = KBPermissionModel.objects.filter(
+                ~Q(knowledge_base__type='private')  # 排除私有知识库
+            ).select_related(
+                'user',
+                'knowledge_base',
+                'granted_by'
+            )
+            
+            # 应用过滤条件
+            if status_filter == 'active':
+                queryset = queryset.filter(
+                    Q(expires_at__gt=timezone.now()) | Q(expires_at__isnull=True),
+                    status='active'
+                )
+            elif status_filter == 'expired':
+                queryset = queryset.filter(
+                    Q(expires_at__lte=timezone.now()) | Q(status='inactive')
+                )
+            
+            if department:
+                queryset = queryset.filter(user__department=department)
+            
+            if kb_type:
+                queryset = queryset.filter(knowledge_base__type=kb_type)
+            
+            # 计算总数
+            total = queryset.count()
+            
+            # 分页
+            start = (page - 1) * page_size
+            end = start + page_size
+            permissions = queryset.order_by('-granted_at')[start:end]
+            
+            # 获取所有相关的创建者ID
+            creator_ids = set(perm.knowledge_base.user_id for perm in permissions)
+            creators = {
+                str(user.id): user 
+                for user in User.objects.filter(id__in=creator_ids)
+            }
+            
+            # 构建响应数据
+            permissions_data = []
+            for perm in permissions:
+                creator = creators.get(str(perm.knowledge_base.user_id))
+                perm_data = {
+                    'id': str(perm.id),
+                    'user': {
+                        'id': str(perm.user.id),
+                        'username': perm.user.username,
+                        'name': getattr(perm.user, 'name', perm.user.username),
+                        'department': getattr(perm.user, 'department', None),
+                        'role': getattr(perm.user, 'role', None)
+                    },
+                    'knowledge_base': {
+                        'id': str(perm.knowledge_base.id),
+                        'name': perm.knowledge_base.name,
+                        'type': perm.knowledge_base.type,
+                        'department': perm.knowledge_base.department,
+                        'group': perm.knowledge_base.group,
+                        'creator': {
+                            'id': str(perm.knowledge_base.user_id),
+                            'name': creator.name if creator else None,
+                            'username': creator.username if creator else None
+                        }
+                    },
+                    'permissions': {
+                        'can_read': perm.can_read,
+                        'can_edit': perm.can_edit,
+                        'can_delete': perm.can_delete
+                    },
+                    'granted_by': {
+                        'id': str(perm.granted_by.id) if perm.granted_by else None,
+                        'username': perm.granted_by.username if perm.granted_by else None,
+                        'name': getattr(perm.granted_by, 'name', perm.granted_by.username) if perm.granted_by else None
+                    },
+                    'granted_at': perm.granted_at.strftime('%Y-%m-%d %H:%M:%S'),
+                    'expires_at': perm.expires_at.strftime('%Y-%m-%d %H:%M:%S') if perm.expires_at else None,
+                    'status': perm.status
+                }
+                permissions_data.append(perm_data)
+            
+            return Response({
+                'code': 200,
+                'message': '获取权限列表成功',
+                'data': {
+                    'total': total,
+                    'page': page,
+                    'page_size': page_size,
+                    'results': permissions_data
+                }
+            })
+            
+        except Exception as e:
+            logger.error(f"获取所有权限失败: {str(e)}")
+            logger.error(traceback.format_exc())
+            return Response({
+                'code': 500,
+                'message': f'获取所有权限失败: {str(e)}',
+                'data': None
+            }, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+    @action(detail=False, methods=['post'])
+    def update_permission(self, request):
+        """管理员更新用户的知识库权限"""
+        try:
+            # 检查是否是管理员
+            if request.user.role != 'admin':
+                return Response({
+                    'code': 403,
+                    'message': '只有管理员可以直接修改权限',
+                    'data': None
+                }, status=status.HTTP_403_FORBIDDEN)
+            
+            # 验证必要参数
+            user_id = request.data.get('user_id')
+            knowledge_base_id = request.data.get('knowledge_base_id')
+            permissions = request.data.get('permissions')
+            expires_at_str = request.data.get('expires_at')
+            
+            if not all([user_id, knowledge_base_id, permissions]):
+                return Response({
+                    'code': 400,
+                    'message': '缺少必要参数',
+                    'data': None
+                }, status=status.HTTP_400_BAD_REQUEST)
+            
+            # 验证权限参数格式
+            required_permission_fields = ['can_read', 'can_edit', 'can_delete']
+            if not all(field in permissions for field in required_permission_fields):
+                return Response({
+                    'code': 400,
+                    'message': '权限参数格式错误，必须包含 can_read、can_edit、can_delete',
+                    'data': None
+                }, status=status.HTTP_400_BAD_REQUEST)
+            
+            # 获取用户和知识库
+            try:
+                user = User.objects.get(id=user_id)
+                knowledge_base = KnowledgeBase.objects.get(id=knowledge_base_id)
+            except User.DoesNotExist:
+                return Response({
+                    'code': 404,
+                    'message': f'用户ID {user_id} 不存在',
+                    'data': None
+                }, status=status.HTTP_404_NOT_FOUND)
+            except KnowledgeBase.DoesNotExist:
+                return Response({
+                    'code': 404,
+                    'message': f'知识库ID {knowledge_base_id} 不存在',
+                    'data': None
+                }, status=status.HTTP_404_NOT_FOUND)
+            
+            # 检查知识库类型和用户角色的匹配
+            if knowledge_base.type == 'private' and str(knowledge_base.user_id) != str(user.id):
+                return Response({
+                    'code': 403,
+                    'message': '不能修改其他用户的私有知识库权限',
+                    'data': None
+                }, status=status.HTTP_403_FORBIDDEN)
+            
+            # 处理过期时间
+            expires_at = None
+            if expires_at_str:
+                try:
+                    # 将字符串转换为datetime对象
+                    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():
+                        return Response({
+                            'code': 400,
+                            'message': '过期时间不能早于或等于当前时间',
+                            'data': None
+                        }, status=status.HTTP_400_BAD_REQUEST)
+                except ValueError:
+                    return Response({
+                        'code': 400,
+                        'message': '过期时间格式错误，应为 ISO 格式 (YYYY-MM-DDThh:mm:ssZ)',
+                        'data': None
+                    }, status=status.HTTP_400_BAD_REQUEST)
+            
+            # 根据用户角色限制权限
+            if user.role == 'member' and permissions.get('can_delete'):
+                return Response({
+                    'code': 400,
+                    'message': '普通成员不能获得删除权限',
+                    'data': None
+                }, status=status.HTTP_400_BAD_REQUEST)
+            
+            # 更新或创建权限记录
+            try:
+                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': request.user,
+                            'status': 'active',
+                            'expires_at': expires_at
+                        }
+                    )
+                    
+                    # 发送通知给用户
+                    self.send_notification(
+                        user=user,
+                        title="知识库权限更新",
+                        content=f"管理员已{created and '授予' or '更新'}您对知识库 '{knowledge_base.name}' 的权限",
+                        notification_type="permission_updated",
+                        related_object_id=permission.id
+                    )
+            except IntegrityError as e:
+                return Response({
+                    'code': 500,
+                    'message': f'数据库操作失败: {str(e)}',
+                    'data': None
+                }, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
+            
+            return Response({
+                'code': 200,
+                'message': f"{'创建' if created else '更新'}权限成功",
+                'data': {
+                    'id': str(permission.id),
+                    'user': {
+                        'id': str(user.id),
+                        'username': user.username,
+                        'name': user.name,
+                        'department': user.department,
+                        'role': user.role
+                    },
+                    'knowledge_base': {
+                        'id': str(knowledge_base.id),
+                        'name': knowledge_base.name,
+                        'type': knowledge_base.type,
+                        'department': knowledge_base.department,
+                        'group': knowledge_base.group
+                    },
+                    'permissions': {
+                        'can_read': permission.can_read,
+                        'can_edit': permission.can_edit,
+                        'can_delete': permission.can_delete
+                    },
+                    'granted_by': {
+                        'id': str(request.user.id),
+                        'username': request.user.username,
+                        'name': request.user.name
+                    },
+                    'expires_at': permission.expires_at.strftime('%Y-%m-%d %H:%M:%S') if permission.expires_at else None,
+                    'created': created
+                }
+            })
+            
+        except Exception as e:
+            logger.error(f"更新权限失败: {str(e)}")
+            logger.error(traceback.format_exc())
+            return Response({
+                'code': 500,
+                'message': f'更新权限失败: {str(e)}',
+                'data': None
+            }, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
+
 class NotificationViewSet(viewsets.ModelViewSet):
     """通知视图集"""
     queryset = Notification.objects.all()