927 lines
33 KiB
Python
927 lines
33 KiB
Python
"""
|
||
知识库同步服务 - 自动检测文档变更并触发增量更新
|
||
|
||
功能：
|
||
1. 文件变更监控 - 使用 watchdog 监控 documents 目录
|
||
2. 哈希比对 - 识别文件具体变更类型（新增/修改/删除）
|
||
3. 增量向量化 - 仅处理变更文件
|
||
4. 变更日志 - 记录变更历史
|
||
|
||
使用方式：
|
||
from knowledge.sync import KnowledgeSyncService
|
||
|
||
# 启动同步服务
|
||
sync_service = KnowledgeSyncService()
|
||
sync_service.start() # 启动后台监控
|
||
|
||
# 手动触发同步
|
||
result = sync_service.sync_now()
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import json
|
||
import hashlib
|
||
import threading
|
||
import time
|
||
from datetime import datetime
|
||
from typing import Dict, List, Optional, Callable
|
||
from dataclasses import dataclass, asdict
|
||
from enum import Enum
|
||
import logging
|
||
|
||
from data.db import get_connection, init_databases
|
||
|
||
# 缓存支持
|
||
try:
|
||
from core.cache import get_cache_manager
|
||
CACHE_AVAILABLE = True
|
||
except ImportError:
|
||
CACHE_AVAILABLE = False
|
||
|
||
# 设置日志
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
||
)
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 尝试导入 watchdog
|
||
try:
|
||
from watchdog.observers import Observer
|
||
from watchdog.events import FileSystemEventHandler, FileCreatedEvent, FileModifiedEvent, FileDeletedEvent
|
||
HAS_WATCHDOG = True
|
||
except ImportError:
|
||
HAS_WATCHDOG = False
|
||
logger.warning("watchdog 未安装，文件监控功能不可用。请运行: pip install watchdog")
|
||
|
||
|
||
class ChangeType(Enum):
|
||
"""变更类型"""
|
||
ADDED = "added" # 新增
|
||
MODIFIED = "modified" # 修改
|
||
DELETED = "deleted" # 删除
|
||
|
||
|
||
class SyncStatus(Enum):
|
||
"""同步状态"""
|
||
IDLE = "idle" # 空闲
|
||
RUNNING = "running" # 运行中
|
||
COMPLETED = "completed" # 已完成
|
||
FAILED = "failed" # 失败
|
||
|
||
|
||
@dataclass
|
||
class DocumentChange:
|
||
"""文档变更记录"""
|
||
document_id: str # 文档ID（相对路径）
|
||
document_name: str # 文件名
|
||
change_type: ChangeType # 变更类型
|
||
old_hash: Optional[str] # 旧哈希
|
||
new_hash: Optional[str] # 新哈希
|
||
change_time: datetime # 变更时间
|
||
processed: bool = False # 是否已处理
|
||
error_message: Optional[str] = None
|
||
|
||
def to_dict(self):
|
||
return {
|
||
"document_id": self.document_id,
|
||
"document_name": self.document_name,
|
||
"change_type": self.change_type.value,
|
||
"old_hash": self.old_hash,
|
||
"new_hash": self.new_hash,
|
||
"change_time": self.change_time.isoformat(),
|
||
"processed": self.processed,
|
||
"error_message": self.error_message
|
||
}
|
||
|
||
|
||
@dataclass
|
||
class SyncResult:
|
||
"""同步结果"""
|
||
status: SyncStatus
|
||
start_time: datetime
|
||
end_time: Optional[datetime]
|
||
documents_processed: int
|
||
documents_added: int
|
||
documents_modified: int
|
||
documents_deleted: int
|
||
errors: List[str]
|
||
|
||
def to_dict(self):
|
||
return {
|
||
"status": self.status.value,
|
||
"start_time": self.start_time.isoformat(),
|
||
"end_time": self.end_time.isoformat() if self.end_time else None,
|
||
"documents_processed": self.documents_processed,
|
||
"documents_added": self.documents_added,
|
||
"documents_modified": self.documents_modified,
|
||
"documents_deleted": self.documents_deleted,
|
||
"errors": self.errors
|
||
}
|
||
|
||
|
||
class SyncDatabase:
|
||
"""同步数据库管理"""
|
||
|
||
def __init__(self):
|
||
"""初始化数据库"""
|
||
init_databases()
|
||
|
||
def get_document_hash(self, document_id: str) -> Optional[Dict]:
|
||
"""获取文档的当前哈希"""
|
||
with get_connection("knowledge") as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
SELECT document_id, document_name, content_hash, file_size, last_modified
|
||
FROM document_hashes WHERE document_id = ?
|
||
''', (document_id,))
|
||
row = cursor.fetchone()
|
||
|
||
if row:
|
||
return {
|
||
"document_id": row[0],
|
||
"document_name": row[1],
|
||
"content_hash": row[2],
|
||
"file_size": row[3],
|
||
"last_modified": row[4]
|
||
}
|
||
return None
|
||
|
||
def set_document_hash(self, document_id: str, document_name: str,
|
||
content_hash: str, file_size: int, last_modified: datetime):
|
||
"""设置文档哈希"""
|
||
with get_connection("knowledge") as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
INSERT OR REPLACE INTO document_hashes
|
||
(document_id, document_name, content_hash, file_size, last_modified, updated_at)
|
||
VALUES (?, ?, ?, ?, ?, CURRENT_TIMESTAMP)
|
||
''', (document_id, document_name, content_hash, file_size, last_modified))
|
||
|
||
def delete_document_hash(self, document_id: str):
|
||
"""删除文档哈希记录"""
|
||
with get_connection("knowledge") as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('DELETE FROM document_hashes WHERE document_id = ?', (document_id,))
|
||
|
||
def get_all_document_hashes(self) -> Dict[str, Dict]:
|
||
"""获取所有文档哈希"""
|
||
with get_connection("knowledge") as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
SELECT document_id, document_name, content_hash, file_size, last_modified
|
||
FROM document_hashes
|
||
''')
|
||
rows = cursor.fetchall()
|
||
|
||
return {
|
||
row[0]: {
|
||
"document_id": row[0],
|
||
"document_name": row[1],
|
||
"content_hash": row[2],
|
||
"file_size": row[3],
|
||
"last_modified": row[4]
|
||
}
|
||
for row in rows
|
||
}
|
||
|
||
def log_change(self, change: DocumentChange) -> int:
|
||
"""记录变更"""
|
||
with get_connection("knowledge") as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
INSERT INTO change_logs
|
||
(document_id, document_name, change_type, old_hash, new_hash, change_time, processed, error_message)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
||
''', (
|
||
change.document_id,
|
||
change.document_name,
|
||
change.change_type.value,
|
||
change.old_hash,
|
||
change.new_hash,
|
||
change.change_time,
|
||
change.processed,
|
||
change.error_message
|
||
))
|
||
return cursor.lastrowid
|
||
|
||
def get_change_logs(self, limit: int = 100, processed: Optional[bool] = None,
|
||
days: int = 30) -> List[Dict]:
|
||
"""获取变更日志"""
|
||
with get_connection("knowledge") as conn:
|
||
cursor = conn.cursor()
|
||
|
||
sql = '''
|
||
SELECT id, document_id, document_name, change_type, old_hash, new_hash,
|
||
change_time, processed, error_message
|
||
FROM change_logs
|
||
WHERE change_time >= datetime('now', ?)
|
||
'''
|
||
params = [f'-{days} days']
|
||
|
||
if processed is not None:
|
||
sql += ' AND processed = ?'
|
||
params.append(1 if processed else 0)
|
||
|
||
sql += ' ORDER BY change_time DESC LIMIT ?'
|
||
params.append(limit)
|
||
|
||
cursor.execute(sql, params)
|
||
rows = cursor.fetchall()
|
||
|
||
return [
|
||
{
|
||
"id": row[0],
|
||
"document_id": row[1],
|
||
"document_name": row[2],
|
||
"change_type": row[3],
|
||
"old_hash": row[4],
|
||
"new_hash": row[5],
|
||
"change_time": row[6],
|
||
"processed": bool(row[7]),
|
||
"error_message": row[8]
|
||
}
|
||
for row in rows
|
||
]
|
||
|
||
def mark_change_processed(self, change_id: int, error_message: str = None):
|
||
"""标记变更已处理"""
|
||
with get_connection("knowledge") as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
UPDATE change_logs
|
||
SET processed = 1, error_message = ?
|
||
WHERE id = ?
|
||
''', (error_message, change_id))
|
||
|
||
def log_sync_status(self, result: SyncResult) -> int:
|
||
"""记录同步状态"""
|
||
with get_connection("knowledge") as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
INSERT INTO sync_status
|
||
(sync_type, status, start_time, end_time, documents_processed,
|
||
documents_added, documents_modified, documents_deleted, error_message)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
''', (
|
||
"incremental",
|
||
result.status.value,
|
||
result.start_time,
|
||
result.end_time,
|
||
result.documents_processed,
|
||
result.documents_added,
|
||
result.documents_modified,
|
||
result.documents_deleted,
|
||
"; ".join(result.errors) if result.errors else None
|
||
))
|
||
return cursor.lastrowid
|
||
|
||
def get_sync_history(self, limit: int = 20) -> List[Dict]:
|
||
"""获取同步历史"""
|
||
with get_connection("knowledge") as conn:
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
SELECT id, sync_type, status, start_time, end_time,
|
||
documents_processed, documents_added, documents_modified, documents_deleted, error_message
|
||
FROM sync_status
|
||
ORDER BY start_time DESC
|
||
LIMIT ?
|
||
''', (limit,))
|
||
rows = cursor.fetchall()
|
||
|
||
return [
|
||
{
|
||
"id": row[0],
|
||
"sync_type": row[1],
|
||
"status": row[2],
|
||
"start_time": row[3],
|
||
"end_time": row[4],
|
||
"documents_processed": row[5],
|
||
"documents_added": row[6],
|
||
"documents_modified": row[7],
|
||
"documents_deleted": row[8],
|
||
"error_message": row[9]
|
||
}
|
||
for row in rows
|
||
]
|
||
|
||
|
||
class FileChangeHandler(FileSystemEventHandler if HAS_WATCHDOG else object):
|
||
"""文件变更处理器"""
|
||
|
||
def __init__(self, sync_service: 'KnowledgeSyncService'):
|
||
if HAS_WATCHDOG:
|
||
super().__init__()
|
||
self.sync_service = sync_service
|
||
# v5 统一解析支持的所有格式
|
||
self.supported_extensions = {'.pdf', '.docx', '.doc', '.xlsx', '.xls', '.pptx', '.txt', '.png', '.jpg', '.jpeg', '.bmp', '.tiff'}
|
||
self._pending_changes = {} # 防抖：短时间内多次修改只记录一次
|
||
self._debounce_seconds = 2
|
||
|
||
def _is_supported_file(self, file_path: str) -> bool:
|
||
"""检查是否为支持的文件类型"""
|
||
ext = os.path.splitext(file_path)[1].lower()
|
||
return ext in self.supported_extensions
|
||
|
||
def _debounce_change(self, file_path: str, change_type: ChangeType):
|
||
"""防抖处理：短时间内多次修改合并为一次"""
|
||
current_time = time.time()
|
||
|
||
if file_path in self._pending_changes:
|
||
last_time, last_type = self._pending_changes[file_path]
|
||
# 如果是修改事件且距离上次事件很近，忽略
|
||
if current_time - last_time < self._debounce_seconds:
|
||
return
|
||
|
||
self._pending_changes[file_path] = (current_time, change_type)
|
||
|
||
# 延迟处理
|
||
threading.Timer(self._debounce_seconds, self._process_change, args=[file_path, change_type]).start()
|
||
|
||
def _process_change(self, file_path: str, change_type: ChangeType):
|
||
"""处理文件变更"""
|
||
try:
|
||
# 计算相对路径
|
||
rel_path = os.path.relpath(file_path, self.sync_service.documents_path).replace(chr(92), "/")
|
||
document_name = os.path.basename(file_path)
|
||
|
||
logger.info(f"检测到文件变更: {rel_path} ({change_type.value})")
|
||
|
||
# 创建变更记录
|
||
change = DocumentChange(
|
||
document_id=rel_path,
|
||
document_name=document_name,
|
||
change_type=change_type,
|
||
old_hash=None,
|
||
new_hash=None,
|
||
change_time=datetime.now()
|
||
)
|
||
|
||
# 获取旧哈希
|
||
old_doc = self.sync_service.db.get_document_hash(rel_path)
|
||
if old_doc:
|
||
change.old_hash = old_doc['content_hash']
|
||
|
||
# 计算新哈希（如果不是删除）
|
||
if change_type != ChangeType.DELETED and os.path.exists(file_path):
|
||
change.new_hash = self.sync_service.calculate_file_hash(file_path)
|
||
|
||
# 记录变更
|
||
self.sync_service.db.log_change(change)
|
||
|
||
# 触发回调
|
||
if self.sync_service.on_change_callback:
|
||
self.sync_service.on_change_callback(change)
|
||
|
||
except Exception as e:
|
||
logger.error(f"处理文件变更失败: {file_path}, 错误: {e}")
|
||
|
||
def on_created(self, event):
|
||
"""文件创建事件"""
|
||
if event.is_directory:
|
||
return
|
||
if not self._is_supported_file(event.src_path):
|
||
return
|
||
self._debounce_change(event.src_path, ChangeType.ADDED)
|
||
|
||
def on_modified(self, event):
|
||
"""文件修改事件"""
|
||
if event.is_directory:
|
||
return
|
||
if not self._is_supported_file(event.src_path):
|
||
return
|
||
self._debounce_change(event.src_path, ChangeType.MODIFIED)
|
||
|
||
def on_deleted(self, event):
|
||
"""文件删除事件"""
|
||
if event.is_directory:
|
||
return
|
||
if not self._is_supported_file(event.src_path):
|
||
return
|
||
self._debounce_change(event.src_path, ChangeType.DELETED)
|
||
|
||
def on_moved(self, event):
|
||
"""文件移动事件"""
|
||
if event.is_directory:
|
||
return
|
||
# 移动视为删除旧文件 + 创建新文件
|
||
if self._is_supported_file(event.src_path):
|
||
self._debounce_change(event.src_path, ChangeType.DELETED)
|
||
if self._is_supported_file(event.dest_path):
|
||
self._debounce_change(event.dest_path, ChangeType.ADDED)
|
||
|
||
|
||
class KnowledgeSyncService:
|
||
"""知识库同步服务"""
|
||
|
||
def __init__(self, documents_path: str = None):
|
||
"""
|
||
初始化同步服务
|
||
|
||
Args:
|
||
documents_path: 文档目录路径，默认为 ./documents
|
||
"""
|
||
self.documents_path = documents_path or os.path.join(
|
||
os.path.dirname(os.path.abspath(__file__)), "documents"
|
||
)
|
||
self.db = SyncDatabase()
|
||
|
||
self._observer = None
|
||
self._running = False
|
||
self.on_change_callback: Optional[Callable] = None
|
||
self.on_sync_callback: Optional[Callable] = None
|
||
|
||
|
||
|
||
@staticmethod
|
||
def calculate_file_hash(file_path: str) -> str:
|
||
"""计算文件哈希"""
|
||
hasher = hashlib.md5()
|
||
try:
|
||
with open(file_path, 'rb') as f:
|
||
for chunk in iter(lambda: f.read(8192), b''):
|
||
hasher.update(chunk)
|
||
return hasher.hexdigest()
|
||
except Exception as e:
|
||
logger.error(f"计算文件哈希失败: {file_path}, 错误: {e}")
|
||
return ""
|
||
|
||
def scan_documents(self) -> Dict[str, Dict]:
|
||
"""扫描文档目录，返回所有文档信息"""
|
||
documents = {}
|
||
# v5 统一解析支持的所有格式
|
||
supported_extensions = {'.pdf', '.docx', '.doc', '.xlsx', '.xls', '.pptx', '.txt', '.png', '.jpg', '.jpeg', '.bmp', '.tiff'}
|
||
|
||
for root, dirs, files in os.walk(self.documents_path):
|
||
for filename in files:
|
||
ext = os.path.splitext(filename)[1].lower()
|
||
if ext not in supported_extensions:
|
||
continue
|
||
|
||
file_path = os.path.join(root, filename)
|
||
rel_path = os.path.relpath(file_path, self.documents_path).replace(chr(92), "/")
|
||
|
||
try:
|
||
file_stat = os.stat(file_path)
|
||
documents[rel_path] = {
|
||
"document_id": rel_path,
|
||
"document_name": filename,
|
||
"file_path": file_path,
|
||
"file_size": file_stat.st_size,
|
||
"last_modified": datetime.fromtimestamp(file_stat.st_mtime),
|
||
"content_hash": self.calculate_file_hash(file_path)
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"扫描文档失败: {rel_path}, 错误: {e}")
|
||
|
||
return documents
|
||
|
||
def detect_changes(self) -> List[DocumentChange]:
|
||
"""检测文档变更"""
|
||
changes = []
|
||
current_docs = self.scan_documents()
|
||
stored_docs = self.db.get_all_document_hashes()
|
||
|
||
current_ids = set(current_docs.keys())
|
||
stored_ids = set(stored_docs.keys())
|
||
|
||
# 新增的文档
|
||
for doc_id in current_ids - stored_ids:
|
||
doc = current_docs[doc_id]
|
||
changes.append(DocumentChange(
|
||
document_id=doc_id,
|
||
document_name=doc["document_name"],
|
||
change_type=ChangeType.ADDED,
|
||
old_hash=None,
|
||
new_hash=doc["content_hash"],
|
||
change_time=datetime.now()
|
||
))
|
||
|
||
# 删除的文档
|
||
for doc_id in stored_ids - current_ids:
|
||
doc = stored_docs[doc_id]
|
||
changes.append(DocumentChange(
|
||
document_id=doc_id,
|
||
document_name=doc["document_name"],
|
||
change_type=ChangeType.DELETED,
|
||
old_hash=doc["content_hash"],
|
||
new_hash=None,
|
||
change_time=datetime.now()
|
||
))
|
||
|
||
# 修改的文档
|
||
for doc_id in current_ids & stored_ids:
|
||
current_doc = current_docs[doc_id]
|
||
stored_doc = stored_docs[doc_id]
|
||
|
||
if current_doc["content_hash"] != stored_doc["content_hash"]:
|
||
changes.append(DocumentChange(
|
||
document_id=doc_id,
|
||
document_name=current_doc["document_name"],
|
||
change_type=ChangeType.MODIFIED,
|
||
old_hash=stored_doc["content_hash"],
|
||
new_hash=current_doc["content_hash"],
|
||
change_time=datetime.now()
|
||
))
|
||
|
||
return changes
|
||
|
||
def process_change(self, change: DocumentChange) -> bool:
|
||
"""处理单个变更"""
|
||
try:
|
||
file_path = os.path.join(self.documents_path, change.document_id)
|
||
|
||
# 从 document_id 中解析目标向量库
|
||
# document_id 格式: "public/filename.pdf" 或 "finance/filename.pdf"
|
||
kb_name = self._get_kb_name_from_path(change.document_id)
|
||
|
||
# 导入知识库管理器
|
||
from knowledge.manager import get_kb_manager
|
||
kb_manager = get_kb_manager()
|
||
|
||
if change.change_type == ChangeType.ADDED:
|
||
# 新增文档 - 使用多向量库方法
|
||
new_version = self._generate_version_id(kb_name, change.document_name)
|
||
chunks_added = kb_manager.add_file_to_kb(
|
||
kb_name=kb_name,
|
||
filepath=file_path,
|
||
extra_metadata={
|
||
'status': 'active',
|
||
'version': new_version,
|
||
'change_time': datetime.now().isoformat()
|
||
}
|
||
)
|
||
# 更新哈希记录
|
||
self.db.set_document_hash(
|
||
change.document_id,
|
||
change.document_name,
|
||
change.new_hash,
|
||
os.path.getsize(file_path) if os.path.exists(file_path) else 0,
|
||
datetime.now()
|
||
)
|
||
|
||
# 创建版本记录
|
||
try:
|
||
from knowledge.document_versions import get_version_query
|
||
version_query = get_version_query()
|
||
version_query.create_version_record(
|
||
collection=kb_name,
|
||
document_id=change.document_name,
|
||
version=new_version,
|
||
status="active",
|
||
change_summary="新增文档",
|
||
created_by="sync_service",
|
||
chunk_count=chunks_added
|
||
)
|
||
except Exception as e:
|
||
logger.warning(f"创建版本记录失败: {e}")
|
||
|
||
logger.info(f"已添加文档到 {kb_name}: {change.document_id}, 片段数: {chunks_added}")
|
||
|
||
elif change.change_type == ChangeType.MODIFIED:
|
||
# 修改文档：版本管理策略
|
||
# 执行顺序：SQLite 标记旧版本 → ChromaDB 替换切片 → SQLite 创建新版本
|
||
|
||
# 1. 获取当前版本号 & 生成新版本号
|
||
old_version = self._get_current_version(kb_name, change.document_name)
|
||
new_version = self._generate_version_id(kb_name, change.document_name)
|
||
|
||
# 2. 在 SQLite 中标记旧版本为 superseded（ChromaDB 切片由 Phase 3 去重自动清理）
|
||
if old_version:
|
||
try:
|
||
kb_manager.mark_document_as_superseded(
|
||
kb_name,
|
||
change.document_name,
|
||
new_version=new_version,
|
||
reason="文档更新"
|
||
)
|
||
logger.info(
|
||
f"标记旧版本为 superseded: "
|
||
f"{change.document_name} {old_version} -> {new_version}"
|
||
)
|
||
except Exception as e:
|
||
logger.warning(f"标记旧版本失败: {e}")
|
||
|
||
# 3. 添加新版本（Phase 3 去重会自动清理同名旧切片）
|
||
chunks_added = kb_manager.add_file_to_kb(
|
||
kb_name=kb_name,
|
||
filepath=file_path,
|
||
extra_metadata={
|
||
'status': 'active',
|
||
'version': new_version,
|
||
'previous_version': old_version or '',
|
||
'change_time': datetime.now().isoformat()
|
||
}
|
||
)
|
||
|
||
# 4. 更新哈希记录
|
||
self.db.set_document_hash(
|
||
change.document_id,
|
||
change.document_name,
|
||
change.new_hash,
|
||
os.path.getsize(file_path) if os.path.exists(file_path) else 0,
|
||
datetime.now()
|
||
)
|
||
|
||
# 5. 在 SQLite 中创建新版本记录
|
||
try:
|
||
from knowledge.document_versions import get_version_query
|
||
version_query = get_version_query()
|
||
version_query.create_version_record(
|
||
collection=kb_name,
|
||
document_id=change.document_name,
|
||
version=new_version,
|
||
status="active",
|
||
change_summary=f"从 {old_version or '(无)'} 更新",
|
||
supersedes=old_version,
|
||
created_by="sync_service",
|
||
chunk_count=chunks_added
|
||
)
|
||
except Exception as e:
|
||
logger.warning(f"创建版本记录失败: {e}")
|
||
|
||
logger.info(
|
||
f"已更新文档: {change.document_id}, "
|
||
f"版本: {old_version or '(无)'} -> {new_version}, "
|
||
f"添加 {chunks_added} 片段"
|
||
)
|
||
|
||
elif change.change_type == ChangeType.DELETED:
|
||
# 删除文档
|
||
deleted = kb_manager.delete_document(kb_name, change.document_name)
|
||
# 删除哈希记录
|
||
self.db.delete_document_hash(change.document_id)
|
||
logger.info(f"已删除文档: {change.document_id}, 删除 {deleted} 片段")
|
||
|
||
# ==================== 缓存失效 ====================
|
||
# 文档变更后递增知识库版本号，使旧缓存自动失效
|
||
if CACHE_AVAILABLE:
|
||
try:
|
||
cache = get_cache_manager()
|
||
cache.increment_kb_version(kb_name)
|
||
logger.debug(f"已递增知识库版本号: {kb_name}")
|
||
except Exception as e:
|
||
logger.warning(f"递增缓存版本号失败: {e}")
|
||
|
||
# 语义缓存没有知识库版本字段，文档增删改后必须清空；否则回答缓存
|
||
# 可能继续返回已过时的来源、引用或图片路径。
|
||
try:
|
||
from core.semantic_cache import get_semantic_cache
|
||
semantic_cache = get_semantic_cache()
|
||
if semantic_cache:
|
||
semantic_cache.clear()
|
||
logger.debug(f"已清空语义缓存（文档变更触发）: {kb_name}")
|
||
except Exception as e:
|
||
logger.warning(f"清空语义缓存失败: {e}")
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"处理变更失败: {change.document_id}, 错误: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
def _get_kb_name_from_path(self, document_id: str) -> str:
|
||
"""
|
||
从文档ID中解析目标向量库名称
|
||
|
||
Args:
|
||
document_id: 文档ID，格式如 "public_kb/filename.pdf" 或 "dept_hr/filename.pdf"
|
||
|
||
Returns:
|
||
向量库名称（目录名 = 向量库名）
|
||
"""
|
||
# 统一路径分隔符（兼容 Windows 和 Linux）
|
||
normalized = document_id.replace('\\', '/')
|
||
# 获取第一级目录名（即向量库名）
|
||
parts = normalized.split('/')
|
||
if len(parts) > 1:
|
||
return parts[0] # 目录名即向量库名
|
||
else:
|
||
return 'public_kb' # 默认公开库
|
||
|
||
def _get_current_version(self, kb_name: str, filename: str) -> str:
|
||
"""
|
||
获取文档当前版本号
|
||
|
||
Args:
|
||
kb_name: 知识库名称
|
||
filename: 文件名
|
||
|
||
Returns:
|
||
当前版本号，如 "v1", "v2"，不存在则返回 None
|
||
"""
|
||
try:
|
||
from knowledge.document_versions import get_version_query
|
||
version_query = get_version_query()
|
||
active_version = version_query.get_active_version(kb_name, filename)
|
||
return active_version.version if active_version else None
|
||
except Exception as e:
|
||
logger.warning(f"获取当前版本失败: {e}")
|
||
return None
|
||
|
||
def _generate_version_id(self, kb_name: str, filename: str) -> str:
|
||
"""
|
||
生成新版本号
|
||
|
||
基于所有版本记录（不限状态）中的最高版本号递增，
|
||
避免已替代版本被覆盖后版本号回退到 v1。
|
||
|
||
Args:
|
||
kb_name: 知识库名称
|
||
filename: 文件名
|
||
|
||
Returns:
|
||
新版本号，如 "v1", "v2", "v3"
|
||
"""
|
||
try:
|
||
from knowledge.document_versions import get_version_query
|
||
version_query = get_version_query()
|
||
return version_query.get_next_version(kb_name, filename)
|
||
except Exception as e:
|
||
logger.warning(f"生成版本号失败: {e}")
|
||
# 回退：基于当前 active 版本递增
|
||
current_version = self._get_current_version(kb_name, filename)
|
||
if not current_version:
|
||
return "v1"
|
||
try:
|
||
version_num = int(current_version.replace('v', ''))
|
||
return f"v{version_num + 1}"
|
||
except (ValueError, AttributeError):
|
||
return "v1"
|
||
|
||
def _record_version_change(
|
||
self,
|
||
kb_name: str,
|
||
filename: str,
|
||
old_version: str,
|
||
new_version: str,
|
||
reason: str
|
||
):
|
||
"""
|
||
记录版本变更到数据库
|
||
|
||
Args:
|
||
kb_name: 知识库名称
|
||
filename: 文件名
|
||
old_version: 旧版本号
|
||
new_version: 新版本号
|
||
reason: 变更原因
|
||
"""
|
||
try:
|
||
from knowledge.document_versions import get_version_query
|
||
version_query = get_version_query()
|
||
version_query.log_version_change(
|
||
collection=kb_name,
|
||
document_id=filename,
|
||
change_type="update",
|
||
old_version=old_version,
|
||
new_version=new_version,
|
||
old_status="active",
|
||
new_status="active",
|
||
reason=reason,
|
||
changed_by="sync_service"
|
||
)
|
||
except Exception as e:
|
||
logger.warning(f"记录版本变更失败: {e}")
|
||
|
||
def sync_now(self) -> SyncResult:
|
||
"""立即执行同步"""
|
||
logger.info("开始同步...")
|
||
|
||
result = SyncResult(
|
||
status=SyncStatus.RUNNING,
|
||
start_time=datetime.now(),
|
||
end_time=None,
|
||
documents_processed=0,
|
||
documents_added=0,
|
||
documents_modified=0,
|
||
documents_deleted=0,
|
||
errors=[]
|
||
)
|
||
|
||
try:
|
||
# 检测变更
|
||
changes = self.detect_changes()
|
||
|
||
# 处理变更
|
||
for change in changes:
|
||
success = self.process_change(change)
|
||
result.documents_processed += 1
|
||
|
||
if success:
|
||
if change.change_type == ChangeType.ADDED:
|
||
result.documents_added += 1
|
||
elif change.change_type == ChangeType.MODIFIED:
|
||
result.documents_modified += 1
|
||
elif change.change_type == ChangeType.DELETED:
|
||
result.documents_deleted += 1
|
||
else:
|
||
result.errors.append(f"处理失败: {change.document_id}")
|
||
|
||
# 记录变更
|
||
self.db.log_change(change)
|
||
|
||
result.status = SyncStatus.COMPLETED
|
||
|
||
except Exception as e:
|
||
result.status = SyncStatus.FAILED
|
||
result.errors.append(str(e))
|
||
logger.error(f"同步失败: {e}")
|
||
|
||
result.end_time = datetime.now()
|
||
|
||
# 记录同步状态
|
||
self.db.log_sync_status(result)
|
||
|
||
# 触发回调
|
||
if self.on_sync_callback:
|
||
self.on_sync_callback(result)
|
||
|
||
logger.info(f"同步完成: 处理 {result.documents_processed} 个文档, "
|
||
f"新增 {result.documents_added}, "
|
||
f"修改 {result.documents_modified}, "
|
||
f"删除 {result.documents_deleted}")
|
||
|
||
return result
|
||
|
||
def start(self):
|
||
"""启动文件监控"""
|
||
if not HAS_WATCHDOG:
|
||
logger.error("watchdog 未安装，无法启动文件监控")
|
||
return False
|
||
|
||
if self._running:
|
||
logger.warning("文件监控已在运行")
|
||
return True
|
||
|
||
# 首次同步
|
||
logger.info("执行首次同步...")
|
||
self.sync_now()
|
||
|
||
# 启动监控
|
||
event_handler = FileChangeHandler(self)
|
||
self._observer = Observer()
|
||
self._observer.schedule(event_handler, self.documents_path, recursive=True)
|
||
self._observer.start()
|
||
|
||
self._running = True
|
||
logger.info(f"文件监控已启动，监控目录: {self.documents_path}")
|
||
return True
|
||
|
||
def stop(self):
|
||
"""停止文件监控"""
|
||
if self._observer:
|
||
self._observer.stop()
|
||
self._observer.join()
|
||
self._observer = None
|
||
|
||
self._running = False
|
||
logger.info("文件监控已停止")
|
||
|
||
def is_running(self) -> bool:
|
||
"""检查监控是否在运行"""
|
||
return self._running
|
||
|
||
|
||
# 便捷函数
|
||
def create_sync_service(documents_path: str = None) -> KnowledgeSyncService:
|
||
"""创建同步服务实例"""
|
||
return KnowledgeSyncService(documents_path)
|
||
|
||
|
||
# 测试代码
|
||
if __name__ == "__main__":
|
||
print("=" * 60)
|
||
print("知识库同步服务测试")
|
||
print("=" * 60)
|
||
|
||
# 创建服务
|
||
sync_service = KnowledgeSyncService()
|
||
|
||
# 测试扫描文档
|
||
print("\n[1] 扫描文档...")
|
||
docs = sync_service.scan_documents()
|
||
print(f"找到 {len(docs)} 个文档")
|
||
for doc_id, doc in list(docs.items())[:5]:
|
||
print(f" - {doc['document_name']}: {doc['content_hash'][:8]}...")
|
||
|
||
# 测试变更检测
|
||
print("\n[2] 检测变更...")
|
||
changes = sync_service.detect_changes()
|
||
print(f"检测到 {len(changes)} 个变更")
|
||
for change in changes[:5]:
|
||
print(f" - {change.document_name}: {change.change_type.value}")
|
||
|
||
# 测试同步
|
||
print("\n[3] 执行同步...")
|
||
result = sync_service.sync_now()
|
||
print(f"同步状态: {result.status.value}")
|
||
print(f"处理文档: {result.documents_processed}")
|
||
print(f"新增: {result.documents_added}, 修改: {result.documents_modified}, 删除: {result.documents_deleted}")
|
||
|
||
print("\n" + "=" * 60)
|
||
print("测试完成")
|