""" 知识库管理器 - 权限管理 Mixin 提供基于角色和部门的向量库访问控制功能。 权限模型: - admin: 可访问所有向量库,拥有所有操作权限 - manager: 可访问公开库和本部门库,拥有读写权限 - user: 可访问公开库和本部门库,仅有只读权限 向量库命名规则: - public_kb: 公开知识库,所有人可读 - dept_{部门名}: 部门知识库,仅对应部门可访问 主要方法: - get_accessible_collections: 获取用户可访问的向量库列表 - check_permission: 检查用户对指定向量库的操作权限 """ import logging from typing import List from .base import PUBLIC_KB_NAME, normalize_department_name logger = logging.getLogger(__name__) class PermissionMixin: """ 权限管理 Mixin 提供向量库级别的访问控制,基于用户角色和部门判断权限。 依赖属性(需由主类提供): - self._metadata: 元数据字典(包含 collections 信息) """ def get_accessible_collections( self, role: str, department: str, operation: str = "read" ) -> List[str]: """ 获取用户可访问的向量库列表 根据用户角色和部门,返回其有权限访问的向量库名称列表。 权限规则: - admin: 可访问所有向量库 - manager: 可读公开库 + 本部门库,可写本部门库 - user: 仅可读公开库 + 本部门库 Args: role: 用户角色,可选值: 'admin' | 'manager' | 'user' department: 用户部门(支持中文名如"财务部"或英文标识如"finance") operation: 操作类型,可选值: 'read' | 'write' | 'delete' | 'sync' Returns: 可访问的向量库名称列表 Example: >>> kb_manager.get_accessible_collections("admin", "finance") ['public_kb', 'dept_finance', 'dept_hr', ...] >>> kb_manager.get_accessible_collections("user", "技术部", "read") ['public_kb', 'dept_tech'] """ result = [] if role == "admin": for info in self.list_collections(): result.append(info.name) return result if PUBLIC_KB_NAME in self._metadata.get("collections", {}): result.append(PUBLIC_KB_NAME) if department: normalized_dept = normalize_department_name(department) if normalized_dept: dept_kb = f"dept_{normalized_dept}" if dept_kb in self._metadata.get("collections", {}): if operation == "read": result.append(dept_kb) elif operation in ("write", "delete", "sync"): if role == "manager": result.append(dept_kb) else: logger.warning(f"部门名称无法标准化: {department}") return result def check_permission( self, role: str, department: str, kb_name: str, operation: str = "read" ) -> bool: """ 检查用户对指定向量库的操作权限 Args: role: 用户角色,可选值: 'admin' | 'manager' | 'user' department: 用户部门 kb_name: 目标向量库名称 operation: 操作类型,可选值: 'read' | 'write' | 'delete' | 'sync' Returns: 有权限返回 True,无权限返回 False Example: >>> kb_manager.check_permission("user", "finance", "dept_finance", "read") True >>> kb_manager.check_permission("user", "finance", "dept_finance", "write") False """ accessible = self.get_accessible_collections(role, department, operation) return kb_name in accessible