| """ |
| PA/POS Database Backup Script. |
| |
| Creates atomic, point-in-time backups of: |
| • app.db (SQLite: chats, notes, profile facts, files) |
| • embeddings/embeddings.db (SQLite: vector embeddings) |
| |
| Uses the SQLite Online Backup API (sqlite3.backup) to avoid database lock issues. |
| Retains the last N backups (default: 7) and purges older archives. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import logging |
| import shutil |
| import sqlite3 |
| import tarfile |
| from datetime import UTC, datetime |
| from pathlib import Path |
|
|
| logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s") |
| logger = logging.getLogger("backup") |
|
|
| ROOT_DIR = Path(__file__).resolve().parents[1] |
| BACKUPS_DIR = ROOT_DIR / "data" / "backups" |
| APP_DB = ROOT_DIR / "app.db" |
| EMBEDDINGS_DB = ROOT_DIR / "embeddings" / "embeddings.db" |
|
|
|
|
| def backup_sqlite(source_path: Path, dest_path: Path) -> bool: |
| """Use SQLite Online Backup API for an atomic, non-blocking backup.""" |
| if not source_path.exists(): |
| logger.warning("Source database '%s' does not exist yet. Skipping.", source_path) |
| return False |
|
|
| dest_path.parent.mkdir(parents=True, exist_ok=True) |
| try: |
| src = sqlite3.connect(str(source_path)) |
| dst = sqlite3.connect(str(dest_path)) |
| with dst: |
| src.backup(dst, pages=100) |
| dst.close() |
| src.close() |
| logger.info("Successfully backed up '%s' -> '%s'", source_path.name, dest_path) |
| return True |
| except Exception as exc: |
| logger.error("Failed to backup '%s': %s", source_path, exc) |
| return False |
|
|
|
|
| def run_backup(retention: int = 7) -> Path: |
| """Run full backup and rotate old archives.""" |
| BACKUPS_DIR.mkdir(parents=True, exist_ok=True) |
| timestamp = datetime.now(UTC).strftime("%Y%m%d_%H%M%S") |
| temp_dir = BACKUPS_DIR / f"temp_{timestamp}" |
| temp_dir.mkdir(parents=True, exist_ok=True) |
|
|
| backed_up_any = False |
| if APP_DB.exists(): |
| backed_up_any |= backup_sqlite(APP_DB, temp_dir / "app.db") |
| if EMBEDDINGS_DB.exists(): |
| backed_up_any |= backup_sqlite(EMBEDDINGS_DB, temp_dir / "embeddings.db") |
|
|
| archive_path = BACKUPS_DIR / f"pabot_backup_{timestamp}.tar.gz" |
| with tarfile.open(archive_path, "w:gz") as tar: |
| for file in temp_dir.glob("*.db"): |
| tar.add(file, arcname=file.name) |
|
|
| shutil.rmtree(temp_dir, ignore_errors=True) |
| logger.info("Created backup archive: %s (%d bytes)", archive_path.name, archive_path.stat().st_size) |
|
|
| |
| all_backups = sorted(BACKUPS_DIR.glob("pabot_backup_*.tar.gz"), key=lambda p: p.stat().st_mtime) |
| if len(all_backups) > retention: |
| to_delete = all_backups[:-retention] |
| for old_file in to_delete: |
| old_file.unlink(missing_ok=True) |
| logger.info("Purged old backup: %s", old_file.name) |
|
|
| return archive_path |
|
|
|
|
| if __name__ == "__main__": |
| parser = argparse.ArgumentParser(description="Backup PA/POS SQLite databases") |
| parser.add_argument("--retention", type=int, default=7, help="Number of backups to keep (default: 7)") |
| args = parser.parse_args() |
| archive = run_backup(retention=args.retention) |
| print(f"Backup complete: {archive}") |
|
|
