Add phd_research_os/backup.py

phd_research_os/backup.py

@@ -0,0 +1,93 @@
+"""
+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