Build [v1.14.22]

scripts/restore_prod.py

@@ -0,0 +1,134 @@
+#!/usr/bin/env python3
+"""
+TFM aInventory - Production Restore Script (v1.12.0)
+Converted from restore.sh to Python per Decision D-05.
+"""
+
+import os
+import sys
+import yaml
+import argparse
+import tarfile
+import logging
+import shutil
+import subprocess
+import time
+from datetime import datetime
+from typing import Dict, Any
+
+# Color codes
+class Colors:
+    RED = '\033[0;31m'
+    GREEN = '\033[0;32m'
+    YELLOW = '\033[1;33m'
+    BLUE = '\033[0;34m'
+    NC = '\033[0m'
+
+# Setup logging
+logging.basicConfig(level=logging.INFO, format=f"{Colors.BLUE}[INFO]{Colors.NC} %(message)s")
+logger = logging.getLogger("restore_prod")
+
+def load_yaml(file_path: str) -> Dict[str, Any]:
+    if not os.path.exists(file_path):
+        return {}
+    try:
+        with open(file_path, 'r') as f:
+            return yaml.safe_load(f) or {}
+    except Exception as e:
+        logger.error(f"Failed to parse {file_path}: {e}")
+        return {}
+
+def confirm_action(prompt: str, required_text: str = "RESTORE") -> bool:
+    print(f"{Colors.YELLOW}⚠️  WARNING: {prompt}{Colors.NC}")
+    for i in range(1, 4):
+        response = input(f"Type '{required_text}' to confirm ({i}/3): ")
+        if response != required_text:
+            return False
+    return True
+
+def is_docker_running() -> bool:
+    try:
+        subprocess.run(["docker", "ps"], capture_output=True, check=True)
+        return True
+    except:
+        return False
+
+def main():
+    parser = argparse.ArgumentParser(description="TFM aInventory Production Restore Tool")
+    parser.add_argument("backup_file", help="Path to the tar.gz backup file")
+    parser.add_argument("--force", action="store_true", help="Skip confirmation (DANGEROUS)")
+    
+    args = parser.parse_args()
+    
+    if not os.path.exists(args.backup_file):
+        logger.error(f"Backup file not found: {args.backup_file}")
+        sys.exit(1)
+        
+    if not args.force:
+        if not confirm_action(f"This will overwrite current data with content from {args.backup_file}!"):
+            logger.error("Restore cancelled by user.")
+            sys.exit(1)
+            
+    # Load config to find data directory
+    backend_cfg = load_yaml("config/backend.yaml")
+    app_cfg = backend_cfg.get("application", {})
+    data_dir = app_cfg.get("data_dir", "./data")
+    
+    # 1. Detect deployment mode and stop services
+    docker_mode = os.path.exists("docker-compose.yml") and is_docker_running()
+    
+    if docker_mode:
+        logger.info("Docker mode detected. Stopping services...")
+        subprocess.run(["docker-compose", "down"], check=True)
+    else:
+        logger.info("Standalone mode detected. Ensuring services are stopped...")
+        # Try to use run_standalone.py if available
+        launcher = "scripts/run_standalone.py"
+        if os.path.exists(launcher):
+            subprocess.run([sys.executable, launcher, "stop"], capture_output=True)
+        else:
+            # Fallback to pkill
+            subprocess.run(["pkill", "-f", "uvicorn"], capture_output=True)
+            subprocess.run(["pkill", "-f", "next start"], capture_output=True)
+            
+    # 2. Safety backup of current data
+    safety_dir = os.path.join(data_dir, "backups_before_restore")
+    os.makedirs(safety_dir, exist_ok=True)
+    safety_backup = os.path.join(safety_dir, f"pre_restore_{int(time.time())}.tar.gz")
+    
+    logger.info(f"Creating safety backup of current data: {safety_backup}")
+    try:
+        with tarfile.open(safety_backup, "w:gz") as tar:
+            if os.path.exists(data_dir):
+                tar.add(data_dir)
+            if os.path.exists("config"):
+                tar.add("config")
+    except Exception as e:
+        logger.warning(f"Safety backup failed (continuing anyway): {e}")
+        
+    # 3. Extract backup
+    logger.info(f"Extracting {args.backup_file}...")
+    try:
+        with tarfile.open(args.backup_file, "r:gz") as tar:
+            tar.extractall(path=".")
+        logger.info(f"{Colors.GREEN}✓ Extraction successful!{Colors.NC}")
+    except Exception as e:
+        logger.error(f"Extraction failed: {e}")
+        sys.exit(1)
+        
+    # 4. Verification
+    db_path = os.path.join(data_dir, "inventory.db")
+    if not os.path.exists(db_path):
+        logger.error(f"CRITICAL: Database not found at {db_path} after restore!")
+        sys.exit(1)
+        
+    # 5. Restart services
+    if docker_mode:
+        logger.info("Restarting Docker services...")
+        subprocess.run(["docker-compose", "up", "-d"], check=True)
+        logger.info(f"{Colors.GREEN}✓ Services started. Check 'docker-compose ps' for status.{Colors.NC}")
+    else:
+        logger.info(f"{Colors.YELLOW}Restore complete. Please start services manually using run_standalone.py{Colors.NC}")
+
+if __name__ == "__main__":
+    main()