fix(minio): 增强知识库删除功能，添加 MinIO 文件清理逻辑 Fixes 关于删除知识库时，所涉及到的上传文件问题
Fixes #533
This commit is contained in:
Wenjie Zhang
parent
39d8a21bd7
commit
df80829bb2
@ -18,7 +18,7 @@ from src.knowledge.indexing import SUPPORTED_FILE_EXTENSIONS, is_supported_file_
|
||||
from src.knowledge.utils import calculate_content_hash
|
||||
from src.models.embed import test_all_embedding_models_status, test_embedding_model_status
|
||||
from src.storage.postgres.models_business import User
|
||||
from src.storage.minio.client import StorageError, aupload_file_to_minio, get_minio_client
|
||||
from src.storage.minio.client import MinIOClient, StorageError, aupload_file_to_minio, get_minio_client
|
||||
from src.utils import logger
|
||||
|
||||
knowledge = APIRouter(prefix="/knowledge", tags=["knowledge"])
|
||||
@ -685,7 +685,7 @@ async def batch_delete_documents(
|
||||
# 尝试从MinIO删除文件，如果失败（例如旧知识库没有MinIO实例），则忽略
|
||||
try:
|
||||
minio_client = get_minio_client()
|
||||
await minio_client.adelete_file("ref-" + db_id.replace("_", "-"), file_name)
|
||||
await minio_client.adelete_file(MinIOClient.get_ref_bucket_name(db_id), file_name)
|
||||
logger.debug(f"成功从MinIO删除文件: {file_name}")
|
||||
except Exception as minio_error:
|
||||
logger.warning(f"从MinIO删除文件失败（可能是旧知识库）: {minio_error}")
|
||||
@ -728,7 +728,7 @@ async def delete_document(db_id: str, doc_id: str, current_user: User = Depends(
|
||||
# 尝试从MinIO删除文件，如果失败（例如旧知识库没有MinIO实例），则忽略
|
||||
try:
|
||||
minio_client = get_minio_client()
|
||||
await minio_client.adelete_file("ref-" + db_id.replace("_", "-"), file_name)
|
||||
await minio_client.adelete_file(MinIOClient.get_ref_bucket_name(db_id), file_name)
|
||||
logger.debug(f"成功从MinIO删除文件: {file_name}")
|
||||
except Exception as minio_error:
|
||||
logger.warning(f"从MinIO删除文件失败（可能是旧知识库）: {minio_error}")
|
||||
|
||||
@ -478,15 +478,43 @@ class KnowledgeBase(ABC):
|
||||
"""
|
||||
if db_id in self.databases_meta:
|
||||
from src.repositories.knowledge_base_repository import KnowledgeBaseRepository
|
||||
from src.storage.minio import get_minio_client
|
||||
from src.knowledge.utils.kb_utils import parse_minio_url
|
||||
|
||||
# 删除相关文件记录
|
||||
minio_client = get_minio_client()
|
||||
|
||||
# 1. 删除文件元数据中记录的 MinIO 文件
|
||||
files_to_delete = [fid for fid, finfo in self.files_meta.items() if finfo.get("database_id") == db_id]
|
||||
for file_id in files_to_delete:
|
||||
file_path = self.files_meta[file_id].get("path")
|
||||
if file_path and file_path.startswith(("http://", "https://")):
|
||||
try:
|
||||
bucket_name, object_name = parse_minio_url(file_path)
|
||||
await minio_client.adelete_file(bucket_name, object_name)
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to delete MinIO file {file_path}: {e}")
|
||||
|
||||
# 删除解析后的 markdown 文件 (kb-parsed/{db_id}/{file_id}/parsed.md)
|
||||
parsed_object = f"{db_id}/{file_id}/parsed.md"
|
||||
await minio_client.adelete_file(minio_client.KB_BUCKETS["parsed"], parsed_object)
|
||||
|
||||
del self.files_meta[file_id]
|
||||
|
||||
# 删除数据库记录
|
||||
# 2. 并行删除所有知识库 bucket 中该 db_id 下的文件
|
||||
prefix = f"{db_id}/"
|
||||
await asyncio.gather(
|
||||
minio_client.adelete_objects_by_prefix(minio_client.KB_BUCKETS["parsed"], prefix),
|
||||
minio_client.adelete_objects_by_prefix(minio_client.KB_BUCKETS["documents"], prefix),
|
||||
minio_client.adelete_objects_by_prefix(minio_client.KB_BUCKETS["images"], prefix),
|
||||
# 删除 ref bucket 中的文件（两种命名格式）
|
||||
minio_client.adelete_objects_by_prefix(minio_client.get_ref_bucket_name(db_id), prefix),
|
||||
minio_client.adelete_objects_by_prefix(minio_client.get_ref_bucket_name_full(db_id), prefix),
|
||||
)
|
||||
|
||||
# 3. 删除数据库记录
|
||||
del self.databases_meta[db_id]
|
||||
await KnowledgeBaseRepository().delete(db_id)
|
||||
kb_repo = KnowledgeBaseRepository()
|
||||
await kb_repo.delete(db_id)
|
||||
await self._save_metadata()
|
||||
|
||||
# 删除工作目录
|
||||
|
||||
@ -43,6 +43,23 @@ class MinIOClient:
|
||||
|
||||
PUBLIC_READ_BUCKETS = {"generated-images", "avatar", "kb-images"}
|
||||
|
||||
# 知识库相关的 bucket 名称
|
||||
KB_BUCKETS = {
|
||||
"documents": "kb-documents",
|
||||
"parsed": "kb-parsed",
|
||||
"images": "kb-images",
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def get_ref_bucket_name(db_id: str) -> str:
|
||||
"""获取 ref bucket 名称（截断32位）"""
|
||||
return f"ref-{db_id[:32].replace('_', '-')}"
|
||||
|
||||
@staticmethod
|
||||
def get_ref_bucket_name_full(db_id: str) -> str:
|
||||
"""获取 ref bucket 名称（不截断）"""
|
||||
return f"ref-{db_id.replace('_', '-')}"
|
||||
|
||||
def __init__(self):
|
||||
"""初始化 MinIO 客户端"""
|
||||
self.endpoint = os.getenv("MINIO_URI") or "http://milvus-minio:9000"
|
||||
@ -237,6 +254,35 @@ class MinIOClient:
|
||||
)
|
||||
return result
|
||||
|
||||
async def adelete_objects_by_prefix(self, bucket_name: str, prefix: str) -> int:
|
||||
"""
|
||||
按前缀删除对象
|
||||
|
||||
Args:
|
||||
bucket_name: bucket 名称
|
||||
prefix: 对象前缀
|
||||
|
||||
Returns:
|
||||
删除的对象数量
|
||||
"""
|
||||
deleted_count = 0
|
||||
|
||||
def _delete_objects():
|
||||
nonlocal deleted_count
|
||||
try:
|
||||
objects = self.client.list_objects(bucket_name, prefix=prefix, recursive=True)
|
||||
for obj in objects:
|
||||
try:
|
||||
self.client.remove_object(bucket_name, obj.object_name)
|
||||
deleted_count += 1
|
||||
except S3Error as e:
|
||||
logger.warning(f"Failed to delete {bucket_name}/{obj.object_name}: {e}")
|
||||
except S3Error as e:
|
||||
logger.warning(f"Failed to list objects in {bucket_name}/{prefix}: {e}")
|
||||
|
||||
await asyncio.to_thread(_delete_objects)
|
||||
return deleted_count
|
||||
|
||||
def file_exists(self, bucket_name: str, object_name: str) -> bool:
|
||||
"""检查文件是否存在"""
|
||||
try:
|
||||
|
||||
132
test/test_kb_minio_cleanup.py
Normal file
132
test/test_kb_minio_cleanup.py
Normal file
@ -0,0 +1,132 @@
|
||||
"""
|
||||
测试知识库删除时 MinIO 文件清理功能
|
||||
|
||||
测试步骤：
|
||||
1. 创建测试知识库
|
||||
2. 上传测试文件到 MinIO（模拟文件上传流程）
|
||||
3. 删除知识库
|
||||
4. 验证 MinIO 文件是否被清理
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import os
|
||||
import sys
|
||||
|
||||
# 添加项目根目录到 Python 路径
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
|
||||
async def test_delete_knowledge_base_cleanup():
|
||||
"""测试删除知识库时 MinIO 文件清理"""
|
||||
from src.storage.minio import get_minio_client
|
||||
from src.knowledge import knowledge_base
|
||||
from src.knowledge.utils.kb_utils import is_minio_url
|
||||
|
||||
# 初始化
|
||||
minio_client = get_minio_client()
|
||||
kb_manager = knowledge_base
|
||||
|
||||
# 测试参数
|
||||
test_db_name = "test_cleanup_db"
|
||||
test_file_content = b"This is test content for cleanup verification"
|
||||
|
||||
try:
|
||||
# 1. 创建知识库
|
||||
print("1. 创建测试知识库...")
|
||||
kb_info = await kb_manager.create_database(
|
||||
database_name=test_db_name,
|
||||
description="Test knowledge base for MinIO cleanup",
|
||||
kb_type="lightrag",
|
||||
)
|
||||
db_id = kb_info["db_id"]
|
||||
print(f" 知识库创建成功: db_id={db_id}")
|
||||
|
||||
# 2. 上传测试文件到 MinIO（模拟文件上传）
|
||||
print("2. 上传测试文件到 MinIO...")
|
||||
|
||||
# 确保 bucket 存在
|
||||
minio_client.ensure_bucket_exists("kb-documents")
|
||||
minio_client.ensure_bucket_exists("kb-parsed")
|
||||
|
||||
# 上传到 kb-documents (原始文件)
|
||||
test_object_name = f"{db_id}/test_file_{db_id[:8]}.txt"
|
||||
upload_result = minio_client.upload_file(
|
||||
bucket_name="kb-documents",
|
||||
object_name=test_object_name,
|
||||
data=test_file_content,
|
||||
content_type="text/plain",
|
||||
)
|
||||
test_file_url = upload_result.url
|
||||
print(f" 文件上传到 kb-documents: {test_object_name}")
|
||||
|
||||
# 上传到 kb-parsed (解析后的 markdown)
|
||||
test_file_id = f"file_test_{db_id[:8]}"
|
||||
parsed_object_name = f"{db_id}/{test_file_id}/parsed.md"
|
||||
parsed_content = b"# Test Parsed Content\n\nThis is parsed markdown."
|
||||
upload_result_parsed = minio_client.upload_file(
|
||||
bucket_name="kb-parsed",
|
||||
object_name=parsed_object_name,
|
||||
data=parsed_content,
|
||||
content_type="text/markdown",
|
||||
)
|
||||
print(f" 文件上传到 kb-parsed: {parsed_object_name}")
|
||||
|
||||
# 3. 验证文件存在
|
||||
print("3. 验证文件存在...")
|
||||
assert minio_client.file_exists("kb-documents", test_object_name), "原始文件应该存在"
|
||||
assert minio_client.file_exists("kb-parsed", parsed_object_name), "解析后文件应该存在"
|
||||
print(" 文件存在验证通过")
|
||||
|
||||
# 4. 删除知识库
|
||||
print("4. 删除知识库...")
|
||||
result = await kb_manager.delete_database(db_id)
|
||||
print(f" 删除结果: {result}")
|
||||
|
||||
# 5. 验证 MinIO 文件已清理
|
||||
print("5. 验证 MinIO 文件已清理...")
|
||||
|
||||
# 检查 kb-documents 中的文件
|
||||
doc_exists = minio_client.file_exists("kb-documents", test_object_name)
|
||||
print(f" kb-documents/{test_object_name} 存在: {doc_exists}")
|
||||
|
||||
# 检查 kb-parsed 中的文件
|
||||
parsed_exists = minio_client.file_exists("kb-parsed", parsed_object_name)
|
||||
print(f" kb-parsed/{parsed_object_name} 存在: {parsed_exists}")
|
||||
|
||||
# 6. 输出测试结果
|
||||
print("\n" + "=" * 50)
|
||||
if not doc_exists and not parsed_exists:
|
||||
print("✅ 测试通过！MinIO 文件已成功清理")
|
||||
return True
|
||||
else:
|
||||
print("❌ 测试失败！以下文件未被清理:")
|
||||
if doc_exists:
|
||||
print(f" - kb-documents/{test_object_name}")
|
||||
if parsed_exists:
|
||||
print(f" - kb-parsed/{parsed_object_name}")
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
print(f"\n❌ 测试异常: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return False
|
||||
|
||||
finally:
|
||||
# 清理测试数据（确保清理干净）
|
||||
print("\n清理测试数据...")
|
||||
try:
|
||||
# 尝试删除可能残留的文件（忽略错误）
|
||||
for obj in [test_object_name, parsed_object_name]:
|
||||
try:
|
||||
bucket = "kb-documents" if "kb-documents" in str(obj) else "kb-parsed"
|
||||
await minio_client.adelete_file(bucket, obj)
|
||||
except Exception:
|
||||
pass
|
||||
except Exception as e:
|
||||
print(f" 清理时出错: {e}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
result = asyncio.run(test_delete_knowledge_base_cleanup())
|
||||
sys.exit(0 if result else 1)
|
||||
Loading…
Reference in New Issue
Block a user