| """ |
| Database Backup Verification Script. |
| |
| Tests recovery procedures by: |
| 1. Locating the newest (or user-specified) backup archive (.tar.gz). |
| 2. Extracting into an isolated temporary directory. |
| 3. Running SQLite 'PRAGMA integrity_check;' on both app.db and embeddings.db. |
| 4. Querying and verifying table structures and row counts. |
| 5. Generating an evidence-based restoration verification report. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import logging |
| import shutil |
| import sqlite3 |
| import sys |
| import tarfile |
| import tempfile |
| from pathlib import Path |
|
|
| logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s") |
| logger = logging.getLogger("verify_backup") |
|
|
| ROOT_DIR = Path(__file__).resolve().parents[1] |
| BACKUPS_DIR = ROOT_DIR / "data" / "backups" |
|
|
|
|
| def verify_sqlite_db(db_path: Path, required_tables: list[str]) -> tuple[bool, str, dict[str, int]]: |
| """Runs integrity check and inspects table counts.""" |
| if not db_path.exists(): |
| return False, f"File {db_path.name} not found in archive", {} |
|
|
| try: |
| conn = sqlite3.connect(str(db_path)) |
| cursor = conn.cursor() |
|
|
| |
| cursor.execute("PRAGMA integrity_check;") |
| res = cursor.fetchone() |
| if not res or res[0] != "ok": |
| conn.close() |
| return False, f"Integrity check failed: {res}", {} |
|
|
| |
| cursor.execute("SELECT name FROM sqlite_master WHERE type='table';") |
| tables = {row[0] for row in cursor.fetchall()} |
|
|
| table_counts: dict[str, int] = {} |
| for req in required_tables: |
| if req in tables: |
| cursor.execute(f"SELECT COUNT(*) FROM {req};") |
| cnt = cursor.fetchone()[0] |
| table_counts[req] = cnt |
| else: |
| table_counts[req] = 0 |
|
|
| conn.close() |
| return True, "Integrity check OK", table_counts |
| except Exception as exc: |
| return False, f"Exception during verification: {exc}", {} |
|
|
|
|
| def test_backup_restoration(archive_path: Path | None = None) -> bool: |
| """Extracts backup and verifies integrity and content.""" |
| if archive_path is None: |
| if not BACKUPS_DIR.exists(): |
| logger.error("Backups directory '%s' does not exist.", BACKUPS_DIR) |
| return False |
| archives = sorted(BACKUPS_DIR.glob("pabot_backup_*.tar.gz"), key=lambda p: p.stat().st_mtime) |
| if not archives: |
| logger.error("No backup archives found in '%s'.", BACKUPS_DIR) |
| return False |
| archive_path = archives[-1] |
|
|
| if not archive_path.exists(): |
| logger.error("Archive '%s' does not exist.", archive_path) |
| return False |
|
|
| logger.info("Verifying backup archive: %s (size: %d bytes)", archive_path.name, archive_path.stat().st_size) |
|
|
| temp_dir = Path(tempfile.mkdtemp(prefix="pabot_backup_verify_")) |
| try: |
| |
| with tarfile.open(archive_path, "r:gz") as tar: |
| tar.extractall(path=temp_dir) |
|
|
| |
| app_db_path = temp_dir / "app.db" |
| app_tables = ["chats", "messages", "notes", "profile_facts", "files", "auth_config"] |
| app_ok, app_msg, app_counts = verify_sqlite_db(app_db_path, app_tables) |
| logger.info("app.db integrity: %s (%s)", "PASS" if app_ok else "FAIL", app_msg) |
| for tbl, count in app_counts.items(): |
| logger.info(" └─ Table '%s': %d records", tbl, count) |
|
|
| |
| emb_db_path = temp_dir / "embeddings.db" |
| emb_tables = ["embeddings"] |
| emb_ok, emb_msg, emb_counts = verify_sqlite_db(emb_db_path, emb_tables) |
| logger.info("embeddings.db integrity: %s (%s)", "PASS" if emb_ok else "FAIL", emb_msg) |
| for tbl, count in emb_counts.items(): |
| logger.info(" └─ Table '%s': %d records", tbl, count) |
|
|
| success = app_ok and emb_ok |
| if success: |
| logger.info("RESTORATION VERIFICATION PASSED: Backup archive is valid, coherent, and restorable.") |
| else: |
| logger.error("RESTORATION VERIFICATION FAILED: Deficiencies found in backup archive.") |
| return success |
| finally: |
| shutil.rmtree(temp_dir, ignore_errors=True) |
|
|
|
|
| if __name__ == "__main__": |
| parser = argparse.ArgumentParser(description="Verify PA/POS backup restoration integrity") |
| parser.add_argument("--archive", type=str, help="Specific backup archive to verify (default: latest)") |
| args = parser.parse_args() |
|
|
| target = Path(args.archive) if args.archive else None |
| passed = test_backup_restoration(target) |
| sys.exit(0 if passed else 1) |
|
|
