pengxiao
ddbb79b9f6
单容器 Docker 架构的国土空间规划课程智能问答系统，集成 FastAPI 后端与 Next.js 前端。 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
312 lines
11 KiB
Python
312 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
数据备份脚本 - 备份数据库和向量存储
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import shutil
|
||
import json
|
||
import tarfile
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
import logging
|
||
from typing import Dict, List, Optional
|
||
|
||
# 配置日志
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
||
)
|
||
logger = logging.getLogger(__name__)
|
||
|
||
class DataBackup:
|
||
"""数据备份类"""
|
||
|
||
def __init__(self, backup_dir: str = "./backups"):
|
||
self.backup_dir = Path(backup_dir)
|
||
self.backup_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
# 备份配置
|
||
self.backup_items = {
|
||
"database": {
|
||
"name": "数据库",
|
||
"paths": ["course_agent.db"],
|
||
"enabled": True
|
||
},
|
||
"vector_store": {
|
||
"name": "向量存储",
|
||
"paths": ["vector_store"],
|
||
"enabled": True
|
||
},
|
||
"knowledge_base": {
|
||
"name": "知识库",
|
||
"paths": ["data/knowledge_base"],
|
||
"enabled": True
|
||
},
|
||
"uploads": {
|
||
"name": "上传文件",
|
||
"paths": ["uploads"],
|
||
"enabled": True
|
||
},
|
||
"logs": {
|
||
"name": "日志文件",
|
||
"paths": ["logs"],
|
||
"enabled": False # 默认不备份日志
|
||
}
|
||
}
|
||
|
||
def create_backup(self, backup_name: Optional[str] = None) -> str:
|
||
"""创建备份"""
|
||
if not backup_name:
|
||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
backup_name = f"backup_{timestamp}"
|
||
|
||
backup_path = self.backup_dir / backup_name
|
||
backup_path.mkdir(exist_ok=True)
|
||
|
||
logger.info(f"开始创建备份: {backup_name}")
|
||
|
||
backup_info = {
|
||
"name": backup_name,
|
||
"created_at": datetime.now().isoformat(),
|
||
"items": {}
|
||
}
|
||
|
||
# 备份每个项目
|
||
for item_id, item_config in self.backup_items.items():
|
||
if not item_config["enabled"]:
|
||
continue
|
||
|
||
item_backup_path = backup_path / item_id
|
||
item_backup_path.mkdir(exist_ok=True)
|
||
|
||
item_info = self._backup_item(item_id, item_config, item_backup_path)
|
||
backup_info["items"][item_id] = item_info
|
||
|
||
if item_info["status"] == "success":
|
||
logger.info(f"✅ {item_config['name']}: {item_info['file_count']} 个文件")
|
||
else:
|
||
logger.warning(f"⚠️ {item_config['name']}: {item_info['error']}")
|
||
|
||
# 保存备份信息
|
||
info_file = backup_path / "backup_info.json"
|
||
with open(info_file, "w", encoding="utf-8") as f:
|
||
json.dump(backup_info, f, ensure_ascii=False, indent=2)
|
||
|
||
# 创建压缩包
|
||
tar_path = self._create_tar_archive(backup_path)
|
||
|
||
# 清理临时目录
|
||
shutil.rmtree(backup_path)
|
||
|
||
logger.info(f"✅ 备份创建完成: {tar_path}")
|
||
return str(tar_path)
|
||
|
||
def _backup_item(self, item_id: str, item_config: Dict, backup_path: Path) -> Dict:
|
||
"""备份单个项目"""
|
||
item_info = {
|
||
"name": item_config["name"],
|
||
"paths": item_config["paths"],
|
||
"status": "pending",
|
||
"file_count": 0,
|
||
"total_size": 0
|
||
}
|
||
|
||
try:
|
||
file_count = 0
|
||
total_size = 0
|
||
|
||
for path_str in item_config["paths"]:
|
||
source_path = Path(path_str)
|
||
|
||
if not source_path.exists():
|
||
logger.warning(f"路径不存在: {source_path}")
|
||
continue
|
||
|
||
if source_path.is_file():
|
||
# 备份单个文件
|
||
shutil.copy2(source_path, backup_path / source_path.name)
|
||
file_count += 1
|
||
total_size += source_path.stat().st_size
|
||
else:
|
||
# 备份目录
|
||
for root, dirs, files in os.walk(source_path):
|
||
for file in files:
|
||
source_file = Path(root) / file
|
||
rel_path = source_file.relative_to(source_path)
|
||
target_file = backup_path / rel_path
|
||
|
||
target_file.parent.mkdir(parents=True, exist_ok=True)
|
||
shutil.copy2(source_file, target_file)
|
||
|
||
file_count += 1
|
||
total_size += source_file.stat().st_size
|
||
|
||
item_info.update({
|
||
"status": "success",
|
||
"file_count": file_count,
|
||
"total_size": total_size
|
||
})
|
||
|
||
except Exception as e:
|
||
item_info.update({
|
||
"status": "error",
|
||
"error": str(e)
|
||
})
|
||
|
||
return item_info
|
||
|
||
def _create_tar_archive(self, backup_path: Path) -> Path:
|
||
"""创建tar.gz压缩包"""
|
||
tar_filename = f"{backup_path.name}.tar.gz"
|
||
tar_path = self.backup_dir / tar_filename
|
||
|
||
with tarfile.open(tar_path, "w:gz") as tar:
|
||
# 添加备份信息文件
|
||
info_file = backup_path / "backup_info.json"
|
||
if info_file.exists():
|
||
tar.add(info_file, arcname="backup_info.json")
|
||
|
||
# 添加其他文件
|
||
for item_dir in backup_path.iterdir():
|
||
if item_dir.is_dir():
|
||
tar.add(item_dir, arcname=item_dir.name)
|
||
|
||
return tar_path
|
||
|
||
def list_backups(self) -> List[Dict]:
|
||
"""列出所有备份"""
|
||
backups = []
|
||
|
||
for item in self.backup_dir.iterdir():
|
||
if item.is_file() and item.suffix == ".gz":
|
||
backup_info = self._get_backup_info(item)
|
||
if backup_info:
|
||
backups.append(backup_info)
|
||
|
||
# 按创建时间排序
|
||
backups.sort(key=lambda x: x.get("created_at", ""), reverse=True)
|
||
return backups
|
||
|
||
def _get_backup_info(self, backup_file: Path) -> Optional[Dict]:
|
||
"""从备份文件中获取信息"""
|
||
try:
|
||
with tarfile.open(backup_file, "r:gz") as tar:
|
||
# 查找备份信息文件
|
||
info_member = None
|
||
for member in tar.getmembers():
|
||
if member.name == "backup_info.json":
|
||
info_member = member
|
||
break
|
||
|
||
if info_member:
|
||
# 提取并读取信息文件
|
||
info_data = tar.extractfile(info_member)
|
||
if info_data:
|
||
info = json.load(info_data)
|
||
info["filename"] = backup_file.name
|
||
info["size"] = backup_file.stat().st_size
|
||
return info
|
||
|
||
except Exception as e:
|
||
logger.error(f"读取备份信息失败 {backup_file}: {e}")
|
||
|
||
return None
|
||
|
||
def restore_backup(self, backup_filename: str, restore_dir: str = ".") -> bool:
|
||
"""恢复备份"""
|
||
backup_file = self.backup_dir / backup_filename
|
||
|
||
if not backup_file.exists():
|
||
logger.error(f"备份文件不存在: {backup_filename}")
|
||
return False
|
||
|
||
restore_path = Path(restore_dir)
|
||
restore_path.mkdir(parents=True, exist_ok=True)
|
||
|
||
logger.info(f"开始恢复备份: {backup_filename}")
|
||
|
||
try:
|
||
with tarfile.open(backup_file, "r:gz") as tar:
|
||
# 提取所有文件
|
||
tar.extractall(restore_path)
|
||
logger.info(f"✅ 备份恢复完成到: {restore_path}")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"恢复备份失败: {e}")
|
||
return False
|
||
|
||
def cleanup_old_backups(self, keep_days: int = 30) -> int:
|
||
"""清理旧备份"""
|
||
cutoff_date = datetime.now() - timedelta(days=keep_days)
|
||
deleted_count = 0
|
||
|
||
for backup_info in self.list_backups():
|
||
created_at_str = backup_info.get("created_at")
|
||
if not created_at_str:
|
||
continue
|
||
|
||
try:
|
||
created_at = datetime.fromisoformat(created_at_str.replace("Z", "+00:00"))
|
||
if created_at < cutoff_date:
|
||
backup_file = self.backup_dir / backup_info["filename"]
|
||
backup_file.unlink()
|
||
logger.info(f"删除旧备份: {backup_info['filename']}")
|
||
deleted_count += 1
|
||
except Exception as e:
|
||
logger.error(f"解析备份日期失败: {e}")
|
||
|
||
return deleted_count
|
||
|
||
def main():
|
||
"""主函数"""
|
||
import argparse
|
||
|
||
parser = argparse.ArgumentParser(description="数据备份工具")
|
||
parser.add_argument("action", choices=["create", "list", "restore", "cleanup"],
|
||
help="执行的操作")
|
||
parser.add_argument("--name", help="备份名称（用于create和restore）")
|
||
parser.add_argument("--backup-dir", default="./backups", help="备份目录")
|
||
parser.add_argument("--restore-dir", default=".", help="恢复目录")
|
||
parser.add_argument("--keep-days", type=int, default=30, help="保留备份的天数")
|
||
|
||
args = parser.parse_args()
|
||
|
||
backup = DataBackup(args.backup_dir)
|
||
|
||
if args.action == "create":
|
||
backup_path = backup.create_backup(args.name)
|
||
print(f"备份创建完成: {backup_path}")
|
||
|
||
elif args.action == "list":
|
||
backups = backup.list_backups()
|
||
print(f"\n找到 {len(backups)} 个备份:\n")
|
||
for b in backups:
|
||
print(f"名称: {b['name']}")
|
||
print(f"文件: {b['filename']}")
|
||
print(f"大小: {b['size'] / 1024 / 1024:.2f} MB")
|
||
print(f"时间: {b['created_at']}")
|
||
print(f"项目: {', '.join(b['items'].keys())}")
|
||
print("-" * 40)
|
||
|
||
elif args.action == "restore":
|
||
if not args.name:
|
||
print("错误: 恢复备份需要指定备份文件名")
|
||
sys.exit(1)
|
||
|
||
success = backup.restore_backup(args.name, args.restore_dir)
|
||
if success:
|
||
print("✅ 备份恢复成功")
|
||
else:
|
||
print("❌ 备份恢复失败")
|
||
sys.exit(1)
|
||
|
||
elif args.action == "cleanup":
|
||
deleted = backup.cleanup_old_backups(args.keep_days)
|
||
print(f"清理了 {deleted} 个旧备份")
|
||
|
||
if __name__ == "__main__":
|
||
main()
|