#!/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()