ai-qpid-pvz/scripts/backup_db.py

#!/usr/bin/env python3
"""
Automated SQLite Database Backup & Pruning Utility for Plants vs. Zombies
Zero external dependencies.

Features:
- Online non-blocking SQLite backup via sqlite3.Connection.backup() API
- Timestamped snapshots formatted as pvz_YYYYMMDD_HHMMSS.db
- Automatic rotation/pruning (retains the most recent 7 backups by default)
- Usable as a standalone CLI script or programmatic Python library
"""

import os
import sys
import sqlite3
import argparse
import re
from datetime import datetime

BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
DEFAULT_DB_PATH = os.path.join(BASE_DIR, "data", "pvz.db")
DEFAULT_BACKUP_DIR = os.path.join(BASE_DIR, "backups")
DEFAULT_MAX_BACKUPS = 7
BACKUP_FILENAME_PATTERN = re.compile(r"^pvz_\d{8}_\d{6}\.db$")


class BackupResult(str):
    """
    String subclass containing the backup file path.
    Also provides dictionary-like and attribute access for maximum interoperability.
    """
    def __new__(cls, path, pruned=None, success=True, size_bytes=0):
        obj = super().__new__(cls, str(path))
        obj.path = str(path)
        obj.backup_path = str(path)
        obj.pruned = pruned or []
        obj.success = success
        obj.size_bytes = size_bytes
        return obj

    def __getitem__(self, item):
        if item in ("path", "backup_path"):
            return self.path
        elif item == "pruned":
            return self.pruned
        elif item == "success":
            return self.success
        elif item in ("size", "size_bytes"):
            return self.size_bytes
        return super().__getitem__(item)

    def get(self, key, default=None):
        if key in ("path", "backup_path"):
            return self.path
        elif key == "pruned":
            return self.pruned
        elif key == "success":
            return self.success
        elif key in ("size", "size_bytes"):
            return self.size_bytes
        return default


def get_existing_backups(backup_dir=DEFAULT_BACKUP_DIR):
    """
    Returns a sorted list (oldest to newest) of valid backup filenames in the directory.
    """
    if not os.path.isdir(backup_dir):
        return []

    files = []
    for f in os.listdir(backup_dir):
        # Match standard pvz_YYYYMMDD_HHMMSS.db or generic pvz_*.db pattern
        if BACKUP_FILENAME_PATTERN.match(f) or (f.startswith("pvz_") and f.endswith(".db")):
            full_path = os.path.join(backup_dir, f)
            if os.path.isfile(full_path):
                files.append(f)

    # Sort ascending: YYYYMMDD_HHMMSS lexicographical sort matches chronological sort
    files.sort()
    return files


def prune_old_backups(backup_dir=DEFAULT_BACKUP_DIR, max_backups=DEFAULT_MAX_BACKUPS):
    """
    Prunes old backups, keeping only the most recent `max_backups` files.
    Returns the list of deleted backup file paths.
    """
    if max_backups <= 0:
        return []

    existing = get_existing_backups(backup_dir)
    pruned = []

    if len(existing) > max_backups:
        excess_count = len(existing) - max_backups
        to_delete = existing[:excess_count]
        for fname in to_delete:
            fpath = os.path.join(backup_dir, fname)
            try:
                os.remove(fpath)
                pruned.append(fpath)
            except OSError as e:
                sys.stderr.write(f"Warning: Failed to prune old backup {fpath}: {e}\n")

    return pruned


def create_backup(backup_dir=None, source_db=None, max_backups=DEFAULT_MAX_BACKUPS):
    """
    Creates a timestamped online backup of the SQLite database and prunes older backups.

    Args:
        backup_dir (str, optional): Target backup directory. Defaults to /home/kopri/agy/backups.
        source_db (str, optional): Source SQLite database file. Defaults to /home/kopri/agy/data/pvz.db.
        max_backups (int, optional): Number of recent backups to keep. Defaults to 7.

    Returns:
        BackupResult: String path of created backup with metadata attributes/dict access.
    """
    if backup_dir is None:
        backup_dir = DEFAULT_BACKUP_DIR
    if source_db is None:
        source_db = DEFAULT_DB_PATH

    backup_dir = os.path.abspath(backup_dir)
    source_db = os.path.abspath(source_db)

    # Ensure destination directory exists
    os.makedirs(backup_dir, exist_ok=True)

    if not os.path.exists(source_db):
        raise FileNotFoundError(f"Source database file not found at: {source_db}")

    # Generate timestamped filename: pvz_YYYYMMDD_HHMMSS.db
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    target_filename = f"pvz_{timestamp}.db"
    target_path = os.path.join(backup_dir, target_filename)

    # Handle collision if multiple backups are triggered in the exact same second
    counter = 1
    while os.path.exists(target_path):
        target_filename = f"pvz_{timestamp}_{counter}.db"
        target_path = os.path.join(backup_dir, target_filename)
        counter += 1

    # Perform online SQLite backup
    src_conn = sqlite3.connect(source_db)
    dest_conn = sqlite3.connect(target_path)
    try:
        src_conn.backup(dest_conn)
    finally:
        dest_conn.close()
        src_conn.close()

    size_bytes = os.path.getsize(target_path) if os.path.exists(target_path) else 0

    # Prune old backups, preserving only the most recent ones
    pruned = prune_old_backups(backup_dir=backup_dir, max_backups=max_backups)

    return BackupResult(target_path, pruned=pruned, success=True, size_bytes=size_bytes)


def main():
    parser = argparse.ArgumentParser(description="PvZ SQLite Database Backup & Pruning Utility")
    parser.add_argument("--db", dest="source_db", default=DEFAULT_DB_PATH,
                        help=f"Path to SQLite database file (default: {DEFAULT_DB_PATH})")
    parser.add_argument("--dir", dest="backup_dir", default=DEFAULT_BACKUP_DIR,
                        help=f"Path to backup directory (default: {DEFAULT_BACKUP_DIR})")
    parser.add_argument("--keep", dest="max_backups", type=int, default=DEFAULT_MAX_BACKUPS,
                        help=f"Number of recent backups to keep (default: {DEFAULT_MAX_BACKUPS})")

    args = parser.parse_args()

    try:
        result = create_backup(
            backup_dir=args.backup_dir,
            source_db=args.source_db,
            max_backups=args.max_backups
        )
        print(f"[OK] Database backup created successfully:")
        print(f"     Destination: {result}")
        print(f"     File Size:   {result.size_bytes} bytes")
        if result.pruned:
            print(f"[INFO] Pruned {len(result.pruned)} old backup(s):")
            for p in result.pruned:
                print(f"     - {os.path.basename(p)}")
        else:
            print(f"[INFO] Total backups kept: {len(get_existing_backups(args.backup_dir))} (limit: {args.max_backups})")
        return 0
    except Exception as e:
        sys.stderr.write(f"[ERROR] Backup failed: {e}\n")
        return 1


if __name__ == "__main__":
    sys.exit(main())