#!/usr/bin/env python3 """Restore Script — restore LeoCRM from a backup directory. Usage: python scripts/restore.py --backup-dir /backups/leocrm_backup_20260723_220000 python scripts/restore.py --backup-dir /backups/leocrm_backup_20260723_220000 --skip-db python scripts/restore.py --backup-dir /backups/leocrm_backup_20260723_220000 --skip-files python scripts/restore.py --backup-dir /backups/leocrm_backup_20260723_220000 --dry-run Restores: 1. Database from database.sql (psql) 2. Files from files/ directory (shutil copy) Environment variables: DATABASE_URL — PostgreSQL connection string STORAGE_PATH — Target path for file restoration (default: /data/storage) Exit codes: 0 — restore successful 1 — restore failed 2 — configuration error """ from __future__ import annotations import argparse import json import os import shutil import subprocess import sys from datetime import datetime, timezone from pathlib import Path from typing import Any # Add project root to path sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) DEFAULT_STORAGE_PATH = os.environ.get("STORAGE_PATH", "/data/storage") def get_db_connection_params() -> dict[str, str]: """Extract PostgreSQL connection params from DATABASE_URL env var.""" db_url = os.environ.get("DATABASE_URL", "") if not db_url: raise ValueError("DATABASE_URL environment variable not set") if db_url.startswith("postgresql+asyncpg://"): db_url = db_url.replace("postgresql+asyncpg://", "postgresql://", 1) from urllib.parse import urlparse parsed = urlparse(db_url) return { "url": db_url, "host": parsed.hostname or "localhost", "port": str(parsed.port or 5432), "database": parsed.path.lstrip("/"), "username": parsed.username or "postgres", "password": parsed.password or "", } def restore_database(backup_dir: Path, db_params: dict[str, str]) -> tuple[bool, str]: """Restore database from database.sql using psql. Returns (success, error_message). """ db_file = backup_dir / "database.sql" if not db_file.is_file(): return False, f"Database dump not found: {db_file}" cmd = [ "psql", "--host", db_params["host"], "--port", db_params["port"], "--username", db_params["username"], "--dbname", db_params["database"], "--file", str(db_file), "--quiet", ] env = os.environ.copy() if db_params["password"]: env["PGPASSWORD"] = db_params["password"] try: result = subprocess.run( cmd, capture_output=True, text=True, env=env, timeout=600, ) if result.returncode != 0: return False, result.stderr or "psql failed" return True, "" except subprocess.TimeoutExpired: return False, "psql timed out after 600s" except FileNotFoundError: return False, "psql command not found" except Exception as exc: return False, str(exc) def restore_files(backup_dir: Path, target_path: str) -> tuple[bool, int, int, str]: """Restore files from backup files/ directory to target path. Returns (success, file_count, total_size_bytes, error_message). """ files_dir = backup_dir / "files" if not files_dir.is_dir(): return True, 0, 0, "" # No files in backup — not an error try: os.makedirs(target_path, exist_ok=True) file_count = 0 total_size = 0 for root, dirs, files in os.walk(files_dir): rel_root = os.path.relpath(root, files_dir) dest_root = os.path.join(target_path, rel_root) if rel_root != "." else target_path os.makedirs(dest_root, exist_ok=True) for fname in files: src_file = os.path.join(root, fname) dest_file = os.path.join(dest_root, fname) shutil.copy2(src_file, dest_file) file_count += 1 total_size += os.path.getsize(src_file) return True, file_count, total_size, "" except Exception as exc: return False, 0, 0, str(exc) def validate_backup_dir(backup_dir: Path) -> dict[str, Any] | None: """Validate that backup directory has a manifest.json.""" manifest_file = backup_dir / "manifest.json" if not manifest_file.is_file(): return None try: with open(manifest_file) as f: return json.load(f) except Exception: return None def run_restore( backup_dir_path: str, target_storage_path: str, skip_db: bool = False, skip_files: bool = False, dry_run: bool = False, ) -> dict[str, Any]: """Run the full restore pipeline and return a report.""" backup_dir = Path(backup_dir_path) report: dict[str, Any] = { "backup_dir": str(backup_dir), "dry_run": dry_run, "started_at": datetime.now(timezone.utc).isoformat(), "database": {"success": False, "error": ""}, "files": {"success": False, "count": 0, "size_bytes": 0, "error": ""}, "overall_success": False, "errors": [], } # Validate backup directory if not backup_dir.is_dir(): report["errors"].append(f"Backup directory not found: {backup_dir}") report["finished_at"] = datetime.now(timezone.utc).isoformat() return report manifest = validate_backup_dir(backup_dir) if manifest: report["manifest"] = manifest print(f" Backup from: {manifest.get('created_at', 'unknown')}") else: print(f" Warning: No manifest.json found in {backup_dir}") if dry_run: report["database"]["error"] = "[DRY-RUN] Would restore database" report["files"]["error"] = "[DRY-RUN] Would restore files" report["overall_success"] = True report["finished_at"] = datetime.now(timezone.utc).isoformat() return report # Step 1: Database restore if not skip_db: print("▶ Step 1: Database restore (psql)") try: db_params = get_db_connection_params() db_success, db_error = restore_database(backup_dir, db_params) report["database"] = {"success": db_success, "error": db_error} if db_success: print(" ✓ Database restored") else: print(f" ✗ Database restore failed: {db_error}") report["errors"].append(f"Database: {db_error}") except Exception as exc: report["database"]["error"] = str(exc) report["errors"].append(f"Database config: {exc}") print(f" ✗ Database config error: {exc}") else: print("▶ Step 1: Database restore — SKIPPED") report["database"] = {"success": True, "error": "skipped"} # Step 2: File restore if not skip_files: print(f"▶ Step 2: File restore → {target_storage_path}") files_success, files_count, files_size, files_error = restore_files(backup_dir, target_storage_path) report["files"] = { "success": files_success, "count": files_count, "size_bytes": files_size, "error": files_error, } if files_success: print(f" ✓ Files restored: {files_count} files, {files_size} bytes") else: print(f" ✗ File restore failed: {files_error}") report["errors"].append(f"Files: {files_error}") else: print("▶ Step 2: File restore — SKIPPED") report["files"] = {"success": True, "count": 0, "size_bytes": 0, "error": "skipped"} # Determine overall success report["overall_success"] = report["database"]["success"] and report["files"]["success"] report["finished_at"] = datetime.now(timezone.utc).isoformat() return report # ─── CLI Entry Point ───────────────────────────────────────────────── def main() -> int: parser = argparse.ArgumentParser( description="LeoCRM Restore — restore database and files from a backup directory." ) parser.add_argument( "--backup-dir", required=True, help="Path to the backup directory (e.g. /backups/leocrm_backup_20260723_220000)", ) parser.add_argument( "--storage-path", default=DEFAULT_STORAGE_PATH, help=f"Target path for file restoration (default: {DEFAULT_STORAGE_PATH})", ) parser.add_argument( "--skip-db", action="store_true", help="Skip database restoration", ) parser.add_argument( "--skip-files", action="store_true", help="Skip file restoration", ) parser.add_argument( "--dry-run", action="store_true", help="Simulate restore without executing", ) parser.add_argument( "--output", default="-", help="Write JSON report to file (default: stdout)", ) args = parser.parse_args() print(f"\n{'='*60}") print(f" LeoCRM Restore Pipeline") print(f" {'[DRY-RUN]' if args.dry_run else '[LIVE]'}") print(f"{'='*60}\n") report = run_restore( backup_dir_path=args.backup_dir, target_storage_path=args.storage_path, skip_db=args.skip_db, skip_files=args.skip_files, dry_run=args.dry_run, ) report_json = json.dumps(report, indent=2, ensure_ascii=False) if args.output == "-": print(f"\n{'='*60}") print(" Restore Report") print(f"{'='*60}") print(report_json) else: with open(args.output, "w", encoding="utf-8") as f: f.write(report_json) print(f"\nReport written to {args.output}", file=sys.stderr) return 0 if report["overall_success"] else 1 if __name__ == "__main__": sys.exit(main())