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