#!/usr/bin/env python3 """ TFM aInventory - Docker Deployment Script (v1.12.0) Converted from deploy.sh to Python per Decision D-05. """ import os import sys import yaml import argparse import subprocess import time import socket import logging from typing import Dict, Any, List, Optional # Color codes for terminal output class Colors: RED = '\033[0;31m' GREEN = '\033[0;32m' YELLOW = '\033[1;33m' BLUE = '\033[0;34m' CYAN = '\033[0;36m' NC = '\033[0m' # No Color class ColoredFormatter(logging.Formatter): format_str = "%(asctime)s - %(levelname)s - %(message)s" FORMATS = { logging.DEBUG: Colors.CYAN + format_str + Colors.NC, logging.INFO: Colors.BLUE + format_str + Colors.NC, logging.WARNING: Colors.YELLOW + format_str + Colors.NC, logging.ERROR: Colors.RED + format_str + Colors.NC, logging.CRITICAL: Colors.RED + format_str + Colors.NC } def format(self, record): log_fmt = self.FORMATS.get(record.levelno) formatter = logging.Formatter(log_fmt, datefmt='%H:%M:%S') return formatter.format(record) # Setup logging logger = logging.getLogger("deploy") logger.setLevel(logging.INFO) ch = logging.StreamHandler() ch.setFormatter(ColoredFormatter()) logger.addHandler(ch) def run_command(cmd: List[str], capture_output: bool = False, env: Optional[Dict[str, str]] = None) -> subprocess.CompletedProcess: """Run a system command securely with shell=False.""" try: return subprocess.run( cmd, check=True, capture_output=capture_output, text=True, shell=False, env=env or os.environ.copy() ) except subprocess.CalledProcessError as e: logger.error(f"Command failed: {' '.join(cmd)}") if e.stdout: logger.error(f"STDOUT: {e.stdout}") if e.stderr: logger.error(f"STDERR: {e.stderr}") raise def load_yaml(file_path: str) -> Dict[str, Any]: """Load and parse a YAML file.""" if not os.path.exists(file_path): logger.warning(f"Config file {file_path} not found. Using empty defaults.") 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 is_port_in_use(port: int) -> bool: """Check if a port is in use on localhost.""" with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: return s.connect_ex(('localhost', port)) == 0 def check_disk_space(path: str = ".", required_gb: int = 10) -> bool: """Check if there is sufficient disk space.""" stat = os.statvfs(path) free_gb = (stat.f_bavail * stat.f_frsize) / (1024**3) if free_gb < required_gb: logger.warning(f"Low disk space: {free_gb:.2f} GB available (recommended: {required_gb} GB)") return False logger.info(f"Sufficient disk space: {free_gb:.2f} GB available") return True def pre_flight_checks(): """Step 1-5: System and config checks.""" logger.info("Step 1/12: Running pre-flight checks...") # Check Docker try: run_command(["docker", "--version"], capture_output=True) logger.info(" ✓ Docker is installed") except (subprocess.CalledProcessError, FileNotFoundError): logger.critical("Docker not installed. Please install Docker 24.0+") sys.exit(1) # Check Docker Compose try: # Try 'docker compose' first (v2 plugin) run_command(["docker", "compose", "version"], capture_output=True) docker_compose_cmd = ["docker", "compose"] logger.info(" ✓ Docker Compose (v2) is installed") except (subprocess.CalledProcessError, FileNotFoundError): try: # Fallback to 'docker-compose' (v1) run_command(["docker-compose", "--version"], capture_output=True) docker_compose_cmd = ["docker-compose"] logger.info(" ✓ docker-compose (v1) is installed") except (subprocess.CalledProcessError, FileNotFoundError): logger.critical("Docker Compose not installed. Please install Docker Compose 2.0+") sys.exit(1) # Check docker-compose.yml if not os.path.exists("docker-compose.yml"): logger.critical("docker-compose.yml not found in current directory") sys.exit(1) logger.info(" ✓ docker-compose.yml found") # Check configs required_configs = ["config/backend.yaml", "config/network.yaml"] for cfg in required_configs: if not os.path.exists(cfg): logger.error(f"Required config file {cfg} missing.") example = cfg + ".example" if os.path.exists(example): logger.info(f"Suggestion: Copy {example} to {cfg} and customize it.") sys.exit(1) logger.info(f" ✓ {cfg} found") return docker_compose_cmd def main(): parser = argparse.ArgumentParser(description="TFM aInventory Docker Deployment Script") parser.add_argument("environment", nargs="?", default="production", choices=["production", "staging", "development"], help="Deployment environment (default: production)") parser.add_argument("--rebuild", action="store_true", help="Force rebuild of Docker images") parser.add_argument("--verbose", action="store_true", help="Enable debug logging") args = parser.parse_args() if args.verbose: logger.setLevel(logging.DEBUG) logger.info("=== TFM aInventory Deployment Script ===") logger.info(f"Environment: {args.environment}") logger.info(f"Rebuild: {'Yes' if args.rebuild else 'No'}") # 1-5 Pre-flight docker_compose_cmd = pre_flight_checks() # Load Configs docker_cfg = load_yaml("config/docker.yaml") network_cfg = load_yaml("config/network.yaml") backend_cfg = load_yaml("config/backend.yaml") # 6. Port availability check logger.info("Step 6/12: Checking port availability...") ports_cfg = network_cfg.get("ports", {}) ports_to_check = { "Backend HTTP": ports_cfg.get("backend_port", 8916), "Frontend HTTP": ports_cfg.get("frontend_port", 8917), "Backend HTTPS": ports_cfg.get("backend_ssl_port", 8918), "Frontend HTTPS": ports_cfg.get("frontend_ssl_port", 8919), } ports_busy = False for name, port in ports_to_check.items(): if is_port_in_use(port): logger.error(f"Port {port} ({name}) is already in use!") ports_busy = True else: logger.debug(f"Port {port} ({name}) is available") if ports_busy: logger.critical("Some required ports are occupied. Please free them or change config/network.yaml") sys.exit(1) logger.info(" ✓ All required ports are available") # 7. Environment validation logger.info("Step 7/12: Validating backend configuration...") auth_cfg = backend_cfg.get("auth", {}) jwt_secret = auth_cfg.get("jwt_secret_key") if not jwt_secret or jwt_secret == "change_me_in_production": if args.environment == "production": logger.error("JWT_SECRET_KEY is missing or insecure in config/backend.yaml") logger.info("Generate one with: openssl rand -hex 32") sys.exit(1) else: logger.warning("Using insecure JWT_SECRET_KEY (acceptable for non-production)") ai_cfg = backend_cfg.get("ai", {}) if ai_cfg.get("gemini_api_key") == "your-gemini-api-key": logger.warning("Gemini API key is still the default placeholder") if ai_cfg.get("claude_api_key") == "your-claude-api-key": logger.warning("Claude API key is still the default placeholder") logger.info(" ✓ Backend configuration validated") # 4 (Disk space - extra check) check_disk_space() # Prepare environment variables for Docker Compose # We map YAML values to the environment variables expected by docker-compose.yml env = os.environ.copy() env["BACKEND_PORT"] = str(ports_to_check["Backend HTTP"]) env["FRONTEND_PORT"] = str(ports_to_check["Frontend HTTP"]) env["BACKEND_SSL_PORT"] = str(ports_to_check["Backend HTTPS"]) env["FRONTEND_SSL_PORT"] = str(ports_to_check["Frontend HTTPS"]) env["JWT_SECRET_KEY"] = jwt_secret or "change_me_in_production" # 8. Docker Compose Build logger.info("Step 8/12: Building/Pulling Docker images...") build_cmd = docker_compose_cmd + ["build"] if args.rebuild: build_cmd.append("--no-cache") try: run_command(build_cmd, env=env) logger.info(" ✓ Docker images prepared") except Exception: logger.critical("Docker build failed") sys.exit(1) # 9. Preparing data directories logger.info("Step 9/12: Preparing data directories...") dirs = ["data", "logs", "config", "data/caddy_data", "data/caddy_config"] for d in dirs: os.makedirs(d, exist_ok=True) # Note: chmod -R 777 is used in bash script, though slightly insecure, # we'll keep it if it's necessary for the containers to write. # subprocess.run(["chmod", "-R", "777", "data", "logs", "config"]) logger.info(" ✓ Data directories ready") # 10. Start Services logger.info("Step 10/12: Starting Docker services...") up_cmd = docker_compose_cmd + ["up", "-d"] try: run_command(up_cmd, env=env) logger.info(" ✓ Services started in background") except Exception: logger.critical("Failed to start Docker services") sys.exit(1) # 11. Health checks logger.info("Step 11/12: Waiting for services to become healthy...") max_attempts = 30 all_healthy = False for attempt in range(1, max_attempts + 1): try: ps_out = run_command(docker_compose_cmd + ["ps"], capture_output=True, env=env).stdout # Simple check for health status in 'docker compose ps' output # Usually it says '(healthy)' if ps_out.count("(healthy)") >= 3: all_healthy = True break # Newer docker compose versions might just show 'Running' or 'Up' but with health status # If ps doesn't show health clearly, we can try curl except Exception: pass logger.info(f" Waiting... ({attempt}/{max_attempts})") time.sleep(4) if all_healthy: logger.info(" ✓ All services are healthy") else: logger.warning("Services did not report healthy via Docker. Checking via curl...") # Fallback to manual connectivity check # 12. Connectivity & Report logger.info("Step 12/12: Verifying service connectivity...") backend_url = f"http://localhost:{ports_to_check['Backend HTTP']}" frontend_url = f"http://localhost:{ports_to_check['Frontend HTTP']}" backend_ok = False for _ in range(5): try: # We use subprocess curl for simplicity as requested, avoiding external libs like 'requests' res = run_command(["curl", "-sf", f"{backend_url}/health"], capture_output=True) if res.returncode == 0: backend_ok = True break except: time.sleep(2) if backend_ok: logger.info(f" ✓ Backend API responding at {backend_url}/health") else: logger.warning(f" Backend health check failed at {backend_url}/health") # Report print(f"\n{Colors.GREEN}╔════════════════════════════════════════════════════════════╗{Colors.NC}") print(f"{Colors.GREEN}║{Colors.NC} Deployment completed successfully! {Colors.GREEN}║{Colors.NC}") print(f"{Colors.GREEN}╚════════════════════════════════════════════════════════════╝{Colors.NC}\n") print("Access points:") print(f" Frontend (HTTP): {frontend_url}") print(f" Backend (HTTP): {backend_url}") print(f" API Docs: {backend_url}/docs") print(f" Frontend (HTTPS): https://localhost:{ports_to_check['Frontend HTTPS']}") print(f" Backend (HTTPS): https://localhost:{ports_to_check['Backend HTTPS']}") print("\nUseful commands:") print(f" View logs: {' '.join(docker_compose_cmd)} logs -f") print(f" Stop services: {' '.join(docker_compose_cmd)} down") print(f" Status: {' '.join(docker_compose_cmd)} ps") if args.environment == "production": print(f"\n{Colors.YELLOW}Production Notes:{Colors.NC}") print(" • Ensure firewall allows only required ports") print(" • Set up automated backups using scripts/export_prod.py") if __name__ == "__main__": main()