100d1a06eb
- 后端 API（Flask + Gunicorn） - RAG 引擎（混合检索 + 云端 Reranker + 引用溯源） - 文档解析（MinerU + 多格式支持） - Docker 生产部署配置 - 排除前端项目、敏感配置、模型文件
120 lines
3.8 KiB
Python
120 lines
3.8 KiB
Python
"""
|
||
知识库管理器 - 权限管理 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
|