135 lines
4.7 KiB
Python
135 lines
4.7 KiB
Python
#!/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()
|