phd_research_os/backup.py

"""
PhD Research OS — Backup & Recovery (Phase 6)
===============================================
Automated backup and restore for SQLite database.
"""

import os
import sqlite3
import shutil
from datetime import datetime
from pathlib import Path


def backup_database(db_path: str = "data/research_os.db",
                    backup_dir: str = "data/backups") -> str:
    """
    Create a timestamped backup of the database using SQLite's backup API.
    Returns the backup file path.
    """
    os.makedirs(backup_dir, exist_ok=True)
    
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    backup_path = os.path.join(backup_dir, f"research_os_{timestamp}.db")
    
    # Use SQLite's built-in backup (safe, handles WAL)
    source = sqlite3.connect(db_path)
    dest = sqlite3.connect(backup_path)
    source.backup(dest)
    source.close()
    dest.close()
    
    # Verify backup
    conn = sqlite3.connect(backup_path)
    claim_count = conn.execute("SELECT COUNT(*) FROM claims").fetchone()[0]
    conn.close()
    
    print(f"Backup created: {backup_path} ({claim_count} claims)")
    return backup_path


def restore_database(backup_path: str, db_path: str = "data/research_os.db") -> bool:
    """
    Restore database from backup.
    Creates a safety backup of current DB first.
    """
    if not os.path.exists(backup_path):
        print(f"Error: Backup file not found: {backup_path}")
        return False
    
    # Safety backup of current
    if os.path.exists(db_path):
        safety = db_path + ".pre_restore"
        shutil.copy2(db_path, safety)
        print(f"Safety backup: {safety}")
    
    # Restore
    source = sqlite3.connect(backup_path)
    dest = sqlite3.connect(db_path)
    source.backup(dest)
    source.close()
    dest.close()
    
    # Verify
    conn = sqlite3.connect(db_path)
    claim_count = conn.execute("SELECT COUNT(*) FROM claims").fetchone()[0]
    conn.close()
    
    print(f"Restored from {backup_path} ({claim_count} claims)")
    return True


def list_backups(backup_dir: str = "data/backups") -> list:
    """List available backups with metadata."""
    if not os.path.exists(backup_dir):
        return []
    
    backups = []
    for f in sorted(Path(backup_dir).glob("research_os_*.db"), reverse=True):
        conn = sqlite3.connect(str(f))
        try:
            claim_count = conn.execute("SELECT COUNT(*) FROM claims").fetchone()[0]
        except:
            claim_count = -1
        conn.close()
        
        backups.append({
            "path": str(f),
            "size_mb": f.stat().st_size / 1024 / 1024,
            "claims": claim_count,
            "created": datetime.fromtimestamp(f.stat().st_mtime).isoformat()
        })
    
    return backups