feat: system knowledge base management with admin/user role separation
- Auto-create system KBs from data/knowledge_base/ subdirectories on startup - Upgrade existing user KBs to system KBs when matching directory found - Admin can upload to system KBs (saved to knowledge_base dir) - Regular users see only their own KBs on knowledge page - Fix API permission checks for system KBs (list, detail, documents, update, delete) - Add is_superuser to UserResponse and frontend User type - Fix sidebar session menu visibility for long titles Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
@@ -10,6 +10,8 @@ from sqlalchemy.orm import Session
|
||||
from sqlalchemy import and_
|
||||
|
||||
from ..models.document import Document, DocumentChunk
|
||||
from ..models.knowledge_base import KnowledgeBase
|
||||
from ..models.user import User
|
||||
from ..core.config import get_settings
|
||||
from .document_service import DocumentService
|
||||
|
||||
@@ -30,10 +32,13 @@ class KnowledgeBaseService:
|
||||
directory = self.knowledge_base_dir
|
||||
else:
|
||||
directory = Path(directory)
|
||||
|
||||
|
||||
if not directory.exists():
|
||||
return {"success": False, "message": f"目录不存在: {directory}"}
|
||||
|
||||
|
||||
# 先确保每个子目录都有对应的系统知识库
|
||||
self.ensure_system_knowledge_bases()
|
||||
|
||||
results = {
|
||||
"scanned_files": 0,
|
||||
"new_files": 0,
|
||||
@@ -41,14 +46,14 @@ class KnowledgeBaseService:
|
||||
"skipped_files": 0,
|
||||
"errors": []
|
||||
}
|
||||
|
||||
|
||||
# 递归扫描目录
|
||||
for file_path in directory.rglob("*"):
|
||||
if file_path.is_file() and self._is_supported_file(file_path):
|
||||
try:
|
||||
result = self.process_file(str(file_path))
|
||||
results["scanned_files"] += 1
|
||||
|
||||
|
||||
if result["status"] == "new":
|
||||
results["new_files"] += 1
|
||||
elif result["status"] == "updated":
|
||||
@@ -60,16 +65,82 @@ class KnowledgeBaseService:
|
||||
"file": str(file_path),
|
||||
"error": result["error"]
|
||||
})
|
||||
|
||||
|
||||
except Exception as e:
|
||||
results["errors"].append({
|
||||
"file": str(file_path),
|
||||
"error": str(e)
|
||||
})
|
||||
|
||||
|
||||
results["success"] = len(results["errors"]) == 0
|
||||
return results
|
||||
|
||||
|
||||
def ensure_system_knowledge_bases(self):
|
||||
"""确保 knowledge_base_dir 下每个子目录都有对应的系统知识库"""
|
||||
if not self.knowledge_base_dir.exists():
|
||||
return
|
||||
|
||||
for subdir in self.knowledge_base_dir.iterdir():
|
||||
if subdir.is_dir():
|
||||
self._get_or_create_system_kb(subdir.name)
|
||||
|
||||
def _get_or_create_system_kb(self, name: str) -> int:
|
||||
"""按名称查找或创建系统知识库，返回 knowledge_base_id"""
|
||||
kb = self.db.query(KnowledgeBase).filter(
|
||||
KnowledgeBase.name == name,
|
||||
KnowledgeBase.is_system == True
|
||||
).first()
|
||||
|
||||
if kb:
|
||||
return kb.id
|
||||
|
||||
# 检查是否已有同名非系统KB，如有则升级为系统KB
|
||||
existing = self.db.query(KnowledgeBase).filter(
|
||||
KnowledgeBase.name == name,
|
||||
KnowledgeBase.is_system == False
|
||||
).first()
|
||||
if existing:
|
||||
existing.is_system = True
|
||||
admin = self.db.query(User).filter(User.is_superuser == True).first()
|
||||
if admin:
|
||||
existing.user_id = admin.id
|
||||
self.db.commit()
|
||||
self.db.refresh(existing)
|
||||
print(f"[KB] 升级为系统知识库: {name} (id={existing.id})")
|
||||
return existing.id
|
||||
|
||||
# 找到 admin 用户（或任意 superuser）作为 owner
|
||||
admin = self.db.query(User).filter(User.is_superuser == True).first()
|
||||
if not admin:
|
||||
admin = self.db.query(User).first()
|
||||
|
||||
kb = KnowledgeBase(
|
||||
name=name,
|
||||
description=f"系统知识库：{name}",
|
||||
user_id=admin.id if admin else 1,
|
||||
is_system=True
|
||||
)
|
||||
self.db.add(kb)
|
||||
self.db.commit()
|
||||
self.db.refresh(kb)
|
||||
print(f"[KB] 自动创建系统知识库: {name} (id={kb.id})")
|
||||
return kb.id
|
||||
|
||||
def _resolve_knowledge_base(self, file_path: Path) -> Optional[int]:
|
||||
"""从文件路径解析对应的系统知识库 ID
|
||||
|
||||
data/knowledge_base/国土空间规划文献/paper.pdf → 知识库 "国土空间规划文献"
|
||||
"""
|
||||
try:
|
||||
relative = file_path.relative_to(self.knowledge_base_dir)
|
||||
parts = relative.parts
|
||||
if len(parts) >= 2:
|
||||
subdir_name = parts[0]
|
||||
return self._get_or_create_system_kb(subdir_name)
|
||||
except ValueError:
|
||||
pass
|
||||
return None
|
||||
|
||||
def process_file(self, file_path: str) -> Dict[str, Any]:
|
||||
"""处理单个文件（检查、提取、入库）"""
|
||||
try:
|
||||
@@ -101,16 +172,25 @@ class KnowledgeBaseService:
|
||||
|
||||
if existing_doc:
|
||||
# 检查是否需要更新
|
||||
if (existing_doc.last_modified and
|
||||
if (existing_doc.last_modified and
|
||||
existing_doc.last_modified >= last_modified and
|
||||
existing_doc.file_hash == file_hash):
|
||||
return {"status": "skipped", "message": "文件未修改"}
|
||||
|
||||
|
||||
# 如果文档没有关联知识库，尝试关联
|
||||
if existing_doc.knowledge_base_id is None:
|
||||
kb_id = self._resolve_knowledge_base(file_path)
|
||||
if kb_id:
|
||||
existing_doc.knowledge_base_id = kb_id
|
||||
self.db.commit()
|
||||
|
||||
# 更新现有文档
|
||||
return self._update_document(existing_doc, file_path, file_size, last_modified, file_hash)
|
||||
else:
|
||||
# 解析知识库 ID
|
||||
kb_id = self._resolve_knowledge_base(file_path)
|
||||
# 创建新文档
|
||||
return self._create_document(file_path, file_size, last_modified, file_hash)
|
||||
return self._create_document(file_path, file_size, last_modified, file_hash, knowledge_base_id=kb_id)
|
||||
|
||||
except Exception as e:
|
||||
return {"status": "error", "error": str(e)}
|
||||
|
||||
Reference in New Issue
Block a user