#!/usr/bin/env python3 """ TFM aInventory - Standalone Launcher (v1.12.0) Converted from run_standalone.sh to Python per Decision D-05. """ import os import sys import yaml import argparse import subprocess import signal import time import threading import logging import socket from typing import List, Optional, Dict, Any # Color codes class Colors: RED = '\033[0;31m' GREEN = '\033[0;32m' YELLOW = '\033[1;33m' BLUE = '\033[0;34m' MAGENTA = '\033[0;35m' CYAN = '\033[0;36m' NC = '\033[0m' # Setup logging logging.basicConfig(level=logging.INFO, format=f"{Colors.BLUE}[INFO]{Colors.NC} %(message)s") logger = logging.getLogger("run_standalone") processes: List[subprocess.Popen] = [] shutdown_event = threading.Event() def signal_handler(sig, frame): logger.info("Shutdown signal received. Stopping processes...") shutdown_event.set() for p in processes: try: p.terminate() except: pass # Give them a moment to terminate gracefully time.sleep(1) for p in processes: try: p.kill() except: pass sys.exit(0) signal.signal(signal.SIGINT, signal_handler) signal.signal(signal.SIGTERM, signal_handler) def load_yaml(file_path: str) -> Dict[str, Any]: if not os.path.exists(file_path): logger.warning(f"Config file {file_path} not found. Using 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 {} # PID file for tracking processes PID_FILE = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), ".standalone.pid") def get_running_pids() -> List[int]: if not os.path.exists(PID_FILE): return [] try: with open(PID_FILE, 'r') as f: return [int(line.strip()) for line in f if line.strip()] except: return [] def save_pids(pids: List[int]): with open(PID_FILE, 'w') as f: for pid in pids: f.write(f"{pid}\n") def clear_pids(): if os.path.exists(PID_FILE): try: os.remove(PID_FILE) except: pass def is_pid_running(pid: int) -> bool: try: os.kill(pid, 0) return True except OSError: return False def is_port_in_use(port: int) -> bool: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: return s.connect_ex(('localhost', port)) == 0 def log_stream(stream, prefix, color): for line in iter(stream.readline, ''): if shutdown_event.is_set(): break if line: print(f"{color}[{prefix}]{Colors.NC} {line.strip()}") stream.close() def start_backend(backend_port: int, backend_cfg: Dict[str, Any], background: bool = False): logger.info(f"Starting Backend on port {backend_port}...") app_cfg = backend_cfg.get("application", {}) env = os.environ.copy() env["DATA_DIR"] = app_cfg.get("data_dir", "./data") env["LOGS_DIR"] = app_cfg.get("logs_dir", "./logs") env["LOG_LEVEL"] = backend_cfg.get("logging", {}).get("log_level", "INFO") # Ensure directories exist os.makedirs(env["DATA_DIR"], exist_ok=True) os.makedirs(env["LOGS_DIR"], exist_ok=True) cmd = [ sys.executable, "-m", "uvicorn", "backend.main:app", "--host", "0.0.0.0", "--port", str(backend_port) ] log_file = None if background: log_path = os.path.join(env["LOGS_DIR"], "backend_standalone.log") log_file = open(log_path, "a") logger.info(f"Backend logs redirected to {log_path}") try: p = subprocess.Popen( cmd, stdout=log_file if background else subprocess.PIPE, stderr=log_file if background else subprocess.PIPE, text=True, env=env, bufsize=1, universal_newlines=True, start_new_session=background ) processes.append(p) if not background: threading.Thread(target=log_stream, args=(p.stdout, "BACKEND", Colors.GREEN), daemon=True).start() threading.Thread(target=log_stream, args=(p.stderr, "BACKEND", Colors.GREEN), daemon=True).start() return p except FileNotFoundError: logger.error("uvicorn not found. Please install it with: pip install uvicorn") sys.exit(1) def start_frontend(frontend_port: int, frontend_cfg: Dict[str, Any], network_cfg: Dict[str, Any], background: bool = False): logger.info(f"Starting Frontend on port {frontend_port}...") server_ip = network_cfg.get("application", {}).get("server_ip", "localhost") ssl_enabled = network_cfg.get("ssl", {}).get("ssl_enabled", False) ports = network_cfg.get("ports", {}) # Dynamically construct API URL from master settings if ssl_enabled: back_ssl_port = ports.get("backend_ssl_port", 8918) api_url = f"https://{server_ip}:{back_ssl_port}" else: back_port = ports.get("backend_port", 8916) api_url = f"http://{server_ip}:{back_port}" logger.info(f"Dynamically injecting API URL: {api_url}") env = os.environ.copy() env["NEXT_PUBLIC_API_URL"] = api_url env["PORT"] = str(frontend_port) cmd = ["npm", "run", "dev"] if os.path.exists("frontend/server.js"): cmd = ["node", "frontend/server.js"] log_file = None if background: logs_dir = os.environ.get("LOGS_DIR", "./logs") log_path = os.path.join(logs_dir, "frontend_standalone.log") log_file = open(log_path, "a") logger.info(f"Frontend logs redirected to {log_path}") try: p = subprocess.Popen( cmd, cwd="frontend" if os.path.isdir("frontend") else None, stdout=log_file if background else subprocess.PIPE, stderr=log_file if background else subprocess.PIPE, text=True, env=env, bufsize=1, universal_newlines=True, start_new_session=background ) processes.append(p) if not background: threading.Thread(target=log_stream, args=(p.stdout, "FRONTEND", Colors.MAGENTA), daemon=True).start() threading.Thread(target=log_stream, args=(p.stderr, "FRONTEND", Colors.MAGENTA), daemon=True).start() return p except FileNotFoundError: logger.error("npm or node not found. Please install Node.js") sys.exit(1) def start_proxy(network_cfg: Dict[str, Any], background: bool = False): ssl_cfg = network_cfg.get("ssl", {}) if not ssl_cfg.get("ssl_enabled", False): return None caddyfile = "Caddyfile.standalone" if not os.path.exists(caddyfile): logger.warning(f"{caddyfile} not found. SSL proxy will not start.") return None logger.info("Starting SSL Proxy (Caddy)...") # Check if caddy is installed try: subprocess.run(["caddy", "version"], capture_output=True, check=True) except: logger.error("Caddy not found. Please install Caddy to use SSL proxy.") return None cmd = ["caddy", "run", "--config", caddyfile, "--adapter", "caddyfile"] log_file = None if background: logs_dir = os.environ.get("LOGS_DIR", "./logs") log_path = os.path.join(logs_dir, "proxy_standalone.log") log_file = open(log_path, "a") logger.info(f"Proxy logs redirected to {log_path}") try: p = subprocess.Popen( cmd, stdout=log_file if background else subprocess.PIPE, stderr=log_file if background else subprocess.PIPE, text=True, bufsize=1, universal_newlines=True, start_new_session=background ) processes.append(p) if not background: threading.Thread(target=log_stream, args=(p.stdout, "PROXY", Colors.CYAN), daemon=True).start() threading.Thread(target=log_stream, args=(p.stderr, "PROXY", Colors.CYAN), daemon=True).start() return p except Exception as e: logger.error(f"Failed to start Caddy: {e}") return None def action_status(backend_port: int, frontend_port: int, network_cfg: Dict[str, Any]): logger.info("Checking status...") backend_up = is_port_in_use(backend_port) frontend_up = is_port_in_use(frontend_port) ssl_cfg = network_cfg.get("ssl", {}) ssl_enabled = ssl_cfg.get("ssl_enabled", False) proxy_up = False if ssl_enabled: proxy_up = is_port_in_use(network_cfg.get("ports", {}).get("frontend_ssl_port", 8919)) pids = get_running_pids() running_pids = [pid for pid in pids if is_pid_running(pid)] if backend_up or frontend_up or running_pids: print(f"{Colors.GREEN}● aInventory Standalone is RUNNING{Colors.NC}") print(f" - Backend (Port {backend_port}): {'UP' if backend_up else 'DOWN'}") print(f" - Frontend (Port {frontend_port}): {'UP' if frontend_up else 'DOWN'}") if ssl_enabled: print(f" - SSL Proxy: {'UP' if proxy_up else 'DOWN'}") if running_pids: print(f" - Active PIDs: {', '.join(map(str, running_pids))}") return True else: print(f"{Colors.RED}○ aInventory Standalone is STOPPED{Colors.NC}") return False def action_stop(backend_port: int, frontend_port: int, network_cfg: Dict[str, Any]): pids = get_running_pids() if pids: logger.info(f"Stopping processes with PIDs: {', '.join(map(str, pids))}...") for pid in pids: if is_pid_running(pid): try: # Kill the whole process group if possible os.killpg(os.getpgid(pid), signal.SIGTERM) except: try: os.kill(pid, signal.SIGTERM) except OSError: pass time.sleep(2) for pid in pids: if is_pid_running(pid): try: os.killpg(os.getpgid(pid), signal.SIGKILL) except: try: os.kill(pid, signal.SIGKILL) except OSError: pass clear_pids() # Port-based cleanup (handles orphaned children like npm -> node) cleanup_ports = [backend_port, frontend_port] ports_cfg = network_cfg.get("ports", {}) if network_cfg.get("ssl", {}).get("ssl_enabled"): cleanup_ports.append(ports_cfg.get("backend_ssl_port", 8918)) cleanup_ports.append(ports_cfg.get("frontend_ssl_port", 8919)) for port in cleanup_ports: if is_port_in_use(port): logger.info(f"Cleaning up orphaned process on port {port}...") try: # Use fuser-like logic to find the PID on that port # Note: This is Linux specific cmd = ["lsof", "-t", f"-i:{port}"] result = subprocess.run(cmd, capture_output=True, text=True) orphans = [int(p) for p in result.stdout.splitlines() if p.strip()] for o_pid in orphans: if o_pid != os.getpid(): logger.info(f" - Killing orphan PID {o_pid}") try: os.kill(o_pid, signal.SIGKILL) except OSError: pass except Exception as e: logger.warning(f"Could not perform port-based cleanup: {e}") logger.info("Stopped.") def main(): parser = argparse.ArgumentParser(description="TFM aInventory Standalone Launcher") parser.add_argument("action", nargs="?", choices=["start", "stop", "restart", "status", "run"], default="run", help="Action to perform: run (foreground), start (background), stop, restart, status") parser.add_argument("--no-proxy", action="store_true", help="Disable SSL proxy even if enabled in config") group = parser.add_mutually_exclusive_group() group.add_argument("--backend-only", action="store_true", help="Operation on backend only") group.add_argument("--frontend-only", action="store_true", help="Operation on frontend only") args = parser.parse_args() # Load all configs to find ports and settings backend_cfg = load_yaml("config/backend.yaml") frontend_cfg = load_yaml("config/frontend.yaml") network_cfg = load_yaml("config/network.yaml") ports_cfg = network_cfg.get("ports", {}) backend_port = ports_cfg.get("backend_port", 8916) frontend_port = ports_cfg.get("frontend_port", 8917) if args.action == "status": action_status(backend_port, frontend_port, network_cfg) return if args.action == "stop": action_stop(backend_port, frontend_port, network_cfg) return if args.action == "restart": action_stop(backend_port, frontend_port, network_cfg) # Fall through to start/run logic args.action = "start" if args.action == "restart" else args.action if args.action in ["start", "run"]: background = (args.action == "start") if not args.frontend_only: if is_port_in_use(backend_port): logger.error(f"Port {backend_port} is already in use. Backend cannot start.") if not args.backend_only and args.action == "run": sys.exit(1) else: start_backend(backend_port, backend_cfg, background=background) if not args.backend_only: if is_port_in_use(frontend_port): logger.error(f"Port {frontend_port} is already in use. Frontend cannot start.") if not args.frontend_only and args.action == "run": sys.exit(1) else: start_frontend(frontend_port, frontend_cfg, network_cfg, background=background) # Start SSL Proxy if enabled if not args.no_proxy and not args.backend_only and not args.frontend_only: start_proxy(network_cfg, background=background) if not processes: logger.error("No processes started. Check configuration and port availability.") sys.exit(1) # Save PIDs save_pids([p.pid for p in processes]) if background: logger.info(f"All services started in BACKGROUND. PIDs: {', '.join(str(p.pid) for p in processes)}") logger.info(f"Use 'python3 {sys.argv[0]} status' to check state.") return logger.info("All services started in FOREGROUND. Press Ctrl+C to stop.") # Monitor processes try: while not shutdown_event.is_set(): time.sleep(1) for p in processes: if p.poll() is not None: logger.error(f"Process {p.pid} exited with code {p.returncode}") shutdown_event.set() break except KeyboardInterrupt: pass finally: signal_handler(None, None) clear_pids() if __name__ == "__main__": main()