备份Chatwise SQLiteDB文件脚本

backup_chatwaise_sqlitedb.py

#!/usr/bin/env python3
"""SQLite数据库备份恢复工具"""
from __future__ import annotations

import argparse
import os
import sqlite3
import sys
from datetime import datetime
from pathlib import Path

# 默认配置
DEFAULT_DB_PATH = os.path.expanduser("~/Library/Application Support/app.chatwise/app.db")
DEFAULT_BACKUP_DIR = os.path.expanduser("~/Library/Mobile Documents/com~apple~CloudDocs/SoftwareData/chatwise")
MAX_BACKUPS = 10  # 最多保留的备份文件数量


def cleanup_old_backups(backup_dir: str, prefix: str, max_count: int = MAX_BACKUPS) -> int:
    """清理旧备份，保留最新的max_count份

    Args:
        backup_dir: 备份目录
        prefix: 备份文件前缀（如 'app'）
        max_count: 最多保留的备份数量

    Returns:
        删除的文件数量
    """
    backup_path = Path(backup_dir)
    if not backup_path.exists():
        return 0

    # 获取匹配的备份文件：prefix_backup_*.db
    pattern = f"{prefix}_backup_*.db"
    backup_files = list(backup_path.glob(pattern))

    if len(backup_files) <= max_count:
        return 0

    # 按修改时间排序（最新的在前）
    backup_files.sort(key=lambda f: f.stat().st_mtime, reverse=True)

    # 删除超出数量的旧文件
    deleted_count = 0
    for old_file in backup_files[max_count:]:
        try:
            old_file.unlink()
            print(f"已删除旧备份: {old_file}")
            deleted_count += 1
        except OSError as e:
            print(f"删除旧备份失败: {old_file}, 错误: {e}")

    return deleted_count


def get_latest_backup(backup_dir: str, prefix: str = "app") -> str | None:
    """获取最新的备份文件路径

    Args:
        backup_dir: 备份目录
        prefix: 备份文件前缀（如 'app'）

    Returns:
        最新备份文件的路径，如果没有找到则返回None
    """
    backup_path = Path(backup_dir)
    if not backup_path.exists():
        return None

    # 获取匹配的备份文件：prefix_backup_*.db
    pattern = f"{prefix}_backup_*.db"
    backup_files = list(backup_path.glob(pattern))

    if not backup_files:
        return None

    # 按修改时间排序，返回最新的
    backup_files.sort(key=lambda f: f.stat().st_mtime, reverse=True)
    return str(backup_files[0])


def confirm_overwrite(path: str) -> bool:
    """确认是否覆盖已存在的文件"""
    response = input(f"文件 '{path}' 已存在，是否覆盖? [y/N]: ").strip().lower()
    return response in ('y', 'yes')


def backup_db(source: str, dest: str, force: bool = False) -> bool:
    """备份SQLite数据库到指定路径，使用sqlite3 backup API确保一致性"""
    source_path = Path(source)
    dest_path = Path(dest)

    # 检查源文件
    if not source_path.exists():
        print(f"错误: 源数据库文件不存在: {source}")
        return False

    if not source_path.is_file():
        print(f"错误: 源路径不是文件: {source}")
        return False

    # 如果目标是目录，则在目录下创建带时间戳的备份文件
    if dest_path.is_dir():
        timestamp = datetime.now().strftime("%Y%m%d")
        dest_path = dest_path / f"{source_path.stem}_backup_{timestamp}.db"

    # 检查目标文件是否存在
    if dest_path.exists() and not force:
        if not confirm_overwrite(str(dest_path)):
            print("操作已取消")
            return False

    # 确保目标目录存在
    dest_path.parent.mkdir(parents=True, exist_ok=True)

    try:
        # 使用sqlite3 backup API进行备份
        src_conn = sqlite3.connect(source)
        dst_conn = sqlite3.connect(str(dest_path))
        src_conn.backup(dst_conn)
        src_conn.close()
        dst_conn.close()
        print(f"备份成功: {source} -> {dest_path}")

        # 清理旧备份
        cleanup_old_backups(str(dest_path.parent), source_path.stem)

        return True
    except sqlite3.Error as e:
        print(f"备份失败: {e}")
        return False


def restore_db(source: str, dest_dir: str, force: bool = False) -> bool:
    """从备份文件恢复SQLite数据库到指定目录"""
    source_path = Path(source)
    dest_dir_path = Path(dest_dir)

    # 检查源文件
    if not source_path.exists():
        print(f"错误: 备份文件不存在: {source}")
        return False

    if not source_path.is_file():
        print(f"错误: 源路径不是文件: {source}")
        return False

    # 确保目标目录存在
    dest_dir_path.mkdir(parents=True, exist_ok=True)

    # 恢复后的文件名（去除 _backup_时间戳 后缀）
    dest_name = source_path.name
    if "_backup_" in dest_name:
        dest_name = dest_name.split("_backup_")[0] + ".db"
    dest_path = dest_dir_path / dest_name

    # 检查目标文件是否存在
    if dest_path.exists() and not force:
        if not confirm_overwrite(str(dest_path)):
            print("操作已取消")
            return False

    try:
        # 使用sqlite3 backup API进行恢复
        src_conn = sqlite3.connect(source)
        dst_conn = sqlite3.connect(str(dest_path))
        src_conn.backup(dst_conn)
        src_conn.close()
        dst_conn.close()
        print(f"恢复成功: {source} -> {dest_path}")
        return True
    except sqlite3.Error as e:
        print(f"恢复失败: {e}")
        return False


def main():
    parser = argparse.ArgumentParser(
        description="SQLite数据库备份恢复工具",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
示例:
  # 使用默认配置备份ChatWise数据库到/tmp目录
  %(prog)s backup

  # 备份数据库到指定文件
  %(prog)s backup --source /path/to/app.db --dest /path/to/backup.db

  # 备份数据库到目录(自动添加时间戳)
  %(prog)s backup --source /path/to/app.db --dest /path/to/backup_dir/

  # 强制覆盖备份
  %(prog)s backup --source /path/to/app.db --dest /path/to/backup.db -f

  # 使用默认配置恢复(从最新备份恢复到默认路径)
  %(prog)s restore

  # 恢复数据库到指定目录
  %(prog)s restore --source /path/to/backup.db --dest /path/to/restore_dir/

  # 强制覆盖恢复
  %(prog)s restore --source /path/to/backup.db --dest /path/to/restore_dir/ -f

默认配置:
  源数据库: ~/Library/Application Support/app.chatwise/app.db
  备份目录: ~/Library/Mobile Documents/com~apple~CloudDocs/SoftwareData/chatwise
  最大备份数: 10
"""
    )

    subparsers = parser.add_subparsers(dest="command", help="可用命令")

    # backup 子命令
    backup_parser = subparsers.add_parser("backup", help="备份SQLite数据库")
    backup_parser.add_argument(
        "--source", "-s", default=DEFAULT_DB_PATH,
        help=f"源数据库文件路径 (默认: {DEFAULT_DB_PATH})"
    )
    backup_parser.add_argument(
        "--dest", "-d", default=DEFAULT_BACKUP_DIR,
        help=f"备份目标路径(文件或目录) (默认: {DEFAULT_BACKUP_DIR})"
    )
    backup_parser.add_argument(
        "--force", "-f", action="store_true", help="强制覆盖已存在的文件"
    )

    # restore 子命令
    restore_parser = subparsers.add_parser("restore", help="恢复SQLite数据库")
    restore_parser.add_argument(
        "--source", "-s", default=None,
        help=f"备份文件路径 (默认: {DEFAULT_BACKUP_DIR} 中的最新备份)"
    )
    restore_parser.add_argument(
        "--dest", "-d", default=os.path.dirname(DEFAULT_DB_PATH),
        help=f"恢复目标目录 (默认: {os.path.dirname(DEFAULT_DB_PATH)})"
    )
    restore_parser.add_argument(
        "--force", "-f", action="store_true", help="强制覆盖已存在的文件"
    )

    args = parser.parse_args()

    if args.command is None:
        parser.print_help()
        sys.exit(1)

    if args.command == "backup":
        success = backup_db(args.source, args.dest, args.force)
    elif args.command == "restore":
        # 如果未指定source，自动查找最新备份
        source = args.source
        if source is None:
            source = get_latest_backup(DEFAULT_BACKUP_DIR)
            if source is None:
                print(f"错误: 在 {DEFAULT_BACKUP_DIR} 中未找到备份文件")
                sys.exit(1)
            print(f"使用最新备份: {source}")
        success = restore_db(source, args.dest, args.force)
    else:
        parser.print_help()
        sys.exit(1)

    sys.exit(0 if success else 1)


if __name__ == "__main__":
    main()