457 lines
16 KiB
Python
457 lines
16 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
RustFS 存储结构迁移脚本 - 完整版
|
||
|
||
从旧结构迁移到新结构：
|
||
旧桶: moldinsight-storage
|
||
新桶: moldinsight/
|
||
├── stp-files/{uuid}.stp
|
||
├── geometry/{file_hash}.json
|
||
├── mold-cavities/{file_hash}.json
|
||
├── html/{file_hash}.json
|
||
└── user-files/{uuid}.{ext}
|
||
|
||
注意：这是rustFS，而不是minio，只是用了minio的通用S3接口
|
||
"""
|
||
|
||
import asyncio
|
||
import sys
|
||
import os
|
||
from pathlib import Path
|
||
from typing import Dict, List, Optional
|
||
import json
|
||
from datetime import datetime
|
||
import hashlib
|
||
|
||
# 添加项目根目录和 src 目录到 Python 路径
|
||
project_root = Path(__file__).parent
|
||
src_root = project_root / "src"
|
||
sys.path.insert(0, str(project_root))
|
||
sys.path.insert(0, str(src_root))
|
||
|
||
from storage.rustfs_storage import RustFSManager
|
||
from config.settings import settings
|
||
from utils.logger import get_logger
|
||
|
||
logger = get_logger(__name__)
|
||
|
||
|
||
class RustFSMigration:
|
||
"""RustFS 存储结构迁移器"""
|
||
|
||
def __init__(self):
|
||
self.rustfs = RustFSManager()
|
||
self.is_connected = False
|
||
|
||
# 旧桶名称
|
||
self.old_bucket = 'moldinsight-storage'
|
||
|
||
# 新桶名称
|
||
self.new_bucket = 'moldinsight'
|
||
|
||
# 文件类型前缀映射
|
||
self.file_type_mapping = {
|
||
'stp-files': 'stp_files',
|
||
'geometry': 'geometry_data',
|
||
'mold-cavities': 'mold_cavities',
|
||
'html': 'html_files',
|
||
'user-files': 'user_files'
|
||
}
|
||
|
||
async def connect(self):
|
||
"""连接到 RustFS"""
|
||
try:
|
||
await self.rustfs.connect(
|
||
endpoint=settings.RUSTFS_ENDPOINT,
|
||
access_key=settings.RUSTFS_ACCESS_KEY,
|
||
secret_key=settings.RUSTFS_SECRET_KEY,
|
||
timeout=settings.RUSTFS_TIMEOUT
|
||
)
|
||
self.is_connected = True
|
||
logger.info("RustFS 连接成功")
|
||
return True
|
||
except Exception as e:
|
||
logger.error(f"RustFS 连接失败: {e}")
|
||
return False
|
||
|
||
async def close(self):
|
||
"""关闭连接"""
|
||
await self.rustfs.close()
|
||
self.is_connected = False
|
||
logger.info("RustFS 连接已关闭")
|
||
|
||
async def list_all_buckets(self) -> List[str]:
|
||
"""列出所有桶"""
|
||
if not self.is_connected:
|
||
raise RuntimeError("RustFS 未连接")
|
||
|
||
try:
|
||
buckets = self.rustfs.client.list_buckets()
|
||
bucket_names = [bucket.name for bucket in buckets]
|
||
logger.info(f"当前存在的桶: {bucket_names}")
|
||
return bucket_names
|
||
except Exception as e:
|
||
logger.error(f"列出桶失败: {e}")
|
||
return []
|
||
|
||
async def check_old_bucket_files(self) -> List[Dict]:
|
||
"""检查旧桶中的所有文件"""
|
||
if not self.is_connected:
|
||
raise RuntimeError("RustFS 未连接")
|
||
|
||
try:
|
||
# 检查旧桶是否存在
|
||
if not self.rustfs.client.bucket_exists(self.old_bucket):
|
||
logger.info(f"旧桶不存在: {self.old_bucket}")
|
||
return []
|
||
|
||
# 列出所有文件
|
||
objects = self.rustfs.client.list_objects(self.old_bucket, recursive=True)
|
||
files = []
|
||
|
||
for obj in objects:
|
||
files.append({
|
||
'object_key': obj.object_name,
|
||
'size': obj.size,
|
||
'last_modified': obj.last_modified,
|
||
'etag': obj.etag
|
||
})
|
||
|
||
logger.info(f"旧桶 {self.old_bucket} 包含 {len(files)} 个文件")
|
||
return files
|
||
|
||
except Exception as e:
|
||
logger.error(f"检查旧桶文件失败: {e}")
|
||
return []
|
||
|
||
async def ensure_new_bucket(self):
|
||
"""确保新桶存在"""
|
||
if not self.is_connected:
|
||
raise RuntimeError("RustFS 未连接")
|
||
|
||
try:
|
||
if not self.rustfs.client.bucket_exists(self.new_bucket):
|
||
self.rustfs.client.make_bucket(self.new_bucket)
|
||
logger.info(f"创建新桶: {self.new_bucket}")
|
||
else:
|
||
logger.info(f"新桶已存在: {self.new_bucket}")
|
||
except Exception as e:
|
||
logger.error(f"确保新桶存在失败: {e}")
|
||
raise
|
||
|
||
async def migrate_file(self, old_object_key: str) -> bool:
|
||
"""迁移单个文件到新结构"""
|
||
try:
|
||
# 下载旧文件
|
||
response = self.rustfs.client.get_object(self.old_bucket, old_object_key)
|
||
file_data = response.read()
|
||
response.close()
|
||
response.release_conn()
|
||
|
||
# 解析旧对象键，确定文件类型
|
||
# 旧格式: moldinsight/{文件类型}/{文件名} 或 {文件类型}/{文件名}
|
||
parts = old_object_key.split('/')
|
||
|
||
# 确定文件类型和新对象键
|
||
if len(parts) >= 2:
|
||
# 可能是 moldinsight/{类型}/{文件} 或 {类型}/{文件}
|
||
if parts[0] == 'moldinsight' and len(parts) >= 3:
|
||
# moldinsight/{类型}/{文件}
|
||
old_type = parts[1]
|
||
filename = parts[2]
|
||
elif parts[0] in self.file_type_mapping:
|
||
# {类型}/{文件}
|
||
old_type = parts[0]
|
||
filename = parts[1]
|
||
else:
|
||
# 无法识别的格式，使用默认
|
||
old_type = 'misc'
|
||
filename = parts[-1]
|
||
else:
|
||
old_type = 'misc'
|
||
filename = parts[-1]
|
||
|
||
# 根据旧类型确定新类型
|
||
new_type = old_type # 默认保持不变
|
||
|
||
# 生成新对象键
|
||
if old_type in self.file_type_mapping:
|
||
# 直接使用类型名作为目录
|
||
new_object_key = f"{old_type}/{filename}"
|
||
else:
|
||
# 其他文件放到misc目录
|
||
new_object_key = f"misc/{filename}"
|
||
|
||
# 上传到新桶
|
||
self.rustfs.client.put_object(
|
||
self.new_bucket,
|
||
new_object_key,
|
||
data=file_data,
|
||
length=len(file_data),
|
||
content_type='application/octet-stream'
|
||
)
|
||
|
||
logger.info(f"文件迁移成功: {self.old_bucket}/{old_object_key} -> {self.new_bucket}/{new_object_key}")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"文件迁移失败 {old_object_key}: {e}")
|
||
return False
|
||
|
||
async def migrate_all_files(self) -> Dict[str, any]:
|
||
"""迁移所有文件"""
|
||
if not self.is_connected:
|
||
raise RuntimeError("RustFS 未连接")
|
||
|
||
migration_result = {
|
||
'total_files': 0,
|
||
'successful': 0,
|
||
'failed': 0,
|
||
'failed_files': []
|
||
}
|
||
|
||
try:
|
||
# 获取所有文件
|
||
files = await self.check_old_bucket_files()
|
||
migration_result['total_files'] = len(files)
|
||
|
||
if len(files) == 0:
|
||
logger.info("旧桶中没有文件需要迁移")
|
||
return migration_result
|
||
|
||
logger.info(f"开始迁移 {len(files)} 个文件...")
|
||
|
||
for file_info in files:
|
||
old_object_key = file_info['object_key']
|
||
success = await self.migrate_file(old_object_key)
|
||
|
||
if success:
|
||
migration_result['successful'] += 1
|
||
else:
|
||
migration_result['failed'] += 1
|
||
migration_result['failed_files'].append(old_object_key)
|
||
|
||
logger.info(f"迁移完成: 成功 {migration_result['successful']}, 失败 {migration_result['failed']}")
|
||
|
||
except Exception as e:
|
||
logger.error(f"文件迁移失败: {e}")
|
||
|
||
return migration_result
|
||
|
||
async def delete_old_bucket(self) -> bool:
|
||
"""删除旧桶及其所有文件"""
|
||
if not self.is_connected:
|
||
raise RuntimeError("RustFS 未连接")
|
||
|
||
try:
|
||
# 检查桶是否存在
|
||
if not self.rustfs.client.bucket_exists(self.old_bucket):
|
||
logger.info(f"旧桶不存在，无需删除: {self.old_bucket}")
|
||
return True
|
||
|
||
# 列出所有文件
|
||
objects = list(self.rustfs.client.list_objects(self.old_bucket, recursive=True))
|
||
|
||
if len(objects) > 0:
|
||
logger.info(f"删除旧桶中的 {len(objects)} 个文件...")
|
||
|
||
# 删除所有文件
|
||
for obj in objects:
|
||
self.rustfs.client.remove_object(self.old_bucket, obj.object_name)
|
||
logger.debug(f"删除文件: {obj.object_name}")
|
||
|
||
# 删除空桶
|
||
self.rustfs.client.remove_bucket(self.old_bucket)
|
||
|
||
logger.info(f"旧桶删除成功: {self.old_bucket}")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"删除旧桶失败 {self.old_bucket}: {e}")
|
||
return False
|
||
|
||
async def list_new_bucket_structure(self) -> Dict[str, List[str]]:
|
||
"""列出新桶的文件结构"""
|
||
if not self.is_connected:
|
||
raise RuntimeError("RustFS 未连接")
|
||
|
||
try:
|
||
if not self.rustfs.client.bucket_exists(self.new_bucket):
|
||
return {}
|
||
|
||
objects = self.rustfs.client.list_objects(self.new_bucket, recursive=True)
|
||
structure = {
|
||
'stp-files': [],
|
||
'geometry': [],
|
||
'mold-cavities': [],
|
||
'html': [],
|
||
'user-files': [],
|
||
'misc': []
|
||
}
|
||
|
||
for obj in objects:
|
||
parts = obj.object_name.split('/')
|
||
if len(parts) >= 2:
|
||
file_type = parts[0]
|
||
if file_type in structure:
|
||
structure[file_type].append(obj.object_name)
|
||
else:
|
||
structure['misc'].append(obj.object_name)
|
||
|
||
return structure
|
||
|
||
except Exception as e:
|
||
logger.error(f"列出新桶结构失败: {e}")
|
||
return {}
|
||
|
||
async def run_migration(self, delete_old_bucket: bool = False) -> Dict[str, any]:
|
||
"""运行完整迁移流程"""
|
||
migration_summary = {
|
||
'start_time': datetime.now().isoformat(),
|
||
'connection_status': False,
|
||
'all_buckets': [],
|
||
'old_bucket_files': [],
|
||
'migration_results': {},
|
||
'new_bucket_structure': {},
|
||
'deletion_result': False,
|
||
'end_time': None,
|
||
'status': 'failed'
|
||
}
|
||
|
||
try:
|
||
# 1. 连接
|
||
logger.info("=== RustFS 存储结构迁移开始 ===")
|
||
connection_result = await self.connect()
|
||
if not connection_result:
|
||
raise RuntimeError("无法连接到 RustFS")
|
||
|
||
migration_summary['connection_status'] = True
|
||
|
||
# 2. 列出所有桶
|
||
logger.info("1. 检查桶状态...")
|
||
all_buckets = await self.list_all_buckets()
|
||
migration_summary['all_buckets'] = all_buckets
|
||
|
||
# 3. 检查旧桶文件
|
||
logger.info("2. 检查旧桶文件...")
|
||
old_bucket_files = await self.check_old_bucket_files()
|
||
migration_summary['old_bucket_files'] = old_bucket_files
|
||
|
||
if not old_bucket_files:
|
||
logger.info("旧桶中没有文件，跳过迁移")
|
||
migration_summary['status'] = 'completed'
|
||
migration_summary['end_time'] = datetime.now().isoformat()
|
||
return migration_summary
|
||
|
||
# 4. 确保新桶存在
|
||
logger.info("3. 确保新桶存在...")
|
||
await self.ensure_new_bucket()
|
||
|
||
# 5. 执行迁移
|
||
logger.info("4. 开始迁移文件...")
|
||
migration_results = await self.migrate_all_files()
|
||
migration_summary['migration_results'] = migration_results
|
||
|
||
# 6. 检查新桶结构
|
||
logger.info("5. 检查新桶结构...")
|
||
new_bucket_structure = await self.list_new_bucket_structure()
|
||
migration_summary['new_bucket_structure'] = new_bucket_structure
|
||
|
||
# 7. 可选：删除旧桶
|
||
if delete_old_bucket:
|
||
logger.info("6. 删除旧桶...")
|
||
deletion_result = await self.delete_old_bucket()
|
||
migration_summary['deletion_result'] = deletion_result
|
||
else:
|
||
logger.info("6. 保留旧桶（跳过删除）")
|
||
|
||
# 8. 完成
|
||
migration_summary['status'] = 'completed'
|
||
migration_summary['end_time'] = datetime.now().isoformat()
|
||
|
||
logger.info("=== RustFS 存储结构迁移完成 ===")
|
||
|
||
return migration_summary
|
||
|
||
except Exception as e:
|
||
migration_summary['error'] = str(e)
|
||
migration_summary['end_time'] = datetime.now().isoformat()
|
||
logger.error(f"迁移失败: {e}")
|
||
return migration_summary
|
||
|
||
finally:
|
||
await self.close()
|
||
|
||
|
||
async def main():
|
||
"""主函数"""
|
||
migration = RustFSMigration()
|
||
|
||
print("=== RustFS 存储结构迁移工具 ===")
|
||
print("注意：这是rustFS，而不是minio，只是用了minio的通用S3接口")
|
||
print()
|
||
print("从旧结构迁移到新结构：")
|
||
print(" 旧桶: moldinsight-storage")
|
||
print(" 新桶: moldinsight/")
|
||
print(" ├── stp-files/")
|
||
print(" ├── geometry/")
|
||
print(" ├── mold-cavities/")
|
||
print(" ├── html/")
|
||
print(" └── user-files/")
|
||
print()
|
||
|
||
# 询问是否删除旧桶
|
||
delete_old = input("是否在迁移完成后删除旧桶 moldinsight-storage？(y/N): ").strip().lower() == 'y'
|
||
|
||
print("\n开始迁移...")
|
||
|
||
# 运行迁移
|
||
result = await migration.run_migration(delete_old_bucket=delete_old)
|
||
|
||
# 输出结果摘要
|
||
print("\n=== 迁移结果摘要 ===")
|
||
print(f"状态: {result['status']}")
|
||
print(f"开始时间: {result['start_time']}")
|
||
print(f"结束时间: {result['end_time']}")
|
||
|
||
if 'error' in result:
|
||
print(f"错误: {result['error']}")
|
||
|
||
# 桶列表
|
||
print("\n--- 当前桶列表 ---")
|
||
for bucket in result['all_buckets']:
|
||
print(f" - {bucket}")
|
||
|
||
# 旧桶文件统计
|
||
print(f"\n--- 旧桶文件统计 ---")
|
||
print(f"旧桶 {migration.old_bucket} 包含 {len(result['old_bucket_files'])} 个文件")
|
||
|
||
# 迁移结果
|
||
print("\n--- 迁移结果 ---")
|
||
migration_result = result['migration_results']
|
||
print(f"总文件数: {migration_result['total_files']}")
|
||
print(f"成功: {migration_result['successful']}")
|
||
print(f"失败: {migration_result['failed']}")
|
||
|
||
if migration_result['failed'] > 0:
|
||
print("\n失败的文件:")
|
||
for failed_file in migration_result['failed_files']:
|
||
print(f" - {failed_file}")
|
||
|
||
# 新桶结构
|
||
print("\n--- 新桶文件结构 ---")
|
||
for file_type, files in result['new_bucket_structure'].items():
|
||
if files:
|
||
print(f"{file_type}: {len(files)} 个文件")
|
||
|
||
# 删除结果
|
||
if 'deletion_result' in result:
|
||
status = "成功" if result['deletion_result'] else "失败"
|
||
print(f"\n--- 旧桶删除结果 ---")
|
||
print(f"旧桶 {migration.old_bucket} 删除: {status}")
|
||
|
||
print("\n=== 迁移完成 ===")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main()) |