#!/usr/bin/env python3 """ TFM aInventory Standalone Server Launcher Starts both backend (FastAPI) and frontend (Next.js) servers Manages processes, logs, and graceful shutdown Usage: python3 start_servers.py # Start in foreground (interactive) python3 start_servers.py start # Start in background python3 start_servers.py stop # Stop background servers python3 start_servers.py restart # Restart background servers python3 start_servers.py status # Show server status """ import os import sys import subprocess import signal import time import argparse import json from pathlib import Path class StandaloneServerManager: def __init__(self): self.root_dir = Path(__file__).parent self.pid_file = self.root_dir / ".servers.pid" self.backend_process = None self.frontend_process = None self.load_config() self.setup_directories() def load_config(self): """Load configuration from inventory.env""" env_file = self.root_dir / "inventory.env" if not env_file.exists(): self.error(f"inventory.env not found at {env_file}") self.config = {} with open(env_file) as f: for line in f: line = line.strip() if line and not line.startswith('#') and '=' in line: key, value = line.split('=', 1) self.config[key.strip()] = value.strip().strip('"').strip("'") self.backend_port = int(self.config.get('BACKEND_PORT', 8000)) self.frontend_port = int(self.config.get('FRONTEND_PORT', 3000)) self.data_dir = self.root_dir / self.config.get('DATA_DIR', 'data') self.logs_dir = self.root_dir / self.config.get('LOGS_DIR', 'logs') self.log_level = self.config.get('LOG_LEVEL', 'INFO').lower() self.log(f"✓ Configuration loaded (Backend: {self.backend_port}, Frontend: {self.frontend_port})") def setup_directories(self): """Create necessary directories""" self.logs_dir.mkdir(parents=True, exist_ok=True) self.data_dir.mkdir(parents=True, exist_ok=True) (self.data_dir / "temp").mkdir(parents=True, exist_ok=True) self.log(f"✓ Directories ready") def check_ports(self): """Check if required ports are available""" import socket for port in [self.backend_port, self.frontend_port]: try: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) result = sock.connect_ex(('127.0.0.1', port)) sock.close() if result == 0: self.error(f"Port {port} is already in use") except Exception as e: self.log(f"⚠ Could not check port {port}: {e}") self.log(f"✓ Required ports available") def setup_backend(self): """Setup Python virtual environment and dependencies""" venv_dir = self.root_dir / ".venv" backend_dir = self.root_dir / "backend" # Create venv if needed if not venv_dir.exists(): self.log("Creating Python virtual environment...") subprocess.run([sys.executable, "-m", "venv", str(venv_dir)], check=True) # Install dependencies pip_exe = venv_dir / "bin" / "pip" python_exe = venv_dir / "bin" / "python" requirements = backend_dir / "requirements.txt" if not python_exe.exists(): self.error(f"Virtual environment failed - python not found at {python_exe}") # Check if fastapi is installed try: subprocess.run( [str(python_exe), "-c", "import fastapi"], check=True, capture_output=True ) except subprocess.CalledProcessError: self.log("Installing Python dependencies...") subprocess.run([str(pip_exe), "install", "-q", "-r", str(requirements)], check=True) self.python_exe = python_exe def setup_frontend(self): """Setup frontend dependencies""" frontend_dir = self.root_dir / "frontend" # Install dependencies if needed if not (frontend_dir / "node_modules").exists(): self.log("Installing npm dependencies...") subprocess.run( ["npm", "ci", "--silent"], cwd=str(frontend_dir), check=True, capture_output=True ) # Build if needed if not (frontend_dir / ".next" / "standalone").exists(): self.log("Building Next.js application...") result = subprocess.run( ["npm", "run", "build"], cwd=str(frontend_dir), capture_output=False ) if result.returncode != 0: self.error(f"Frontend build failed with exit code {result.returncode}") def start_backend(self): """Start FastAPI backend server using uvicorn from project root""" self.log(f"Starting backend on port {self.backend_port}...") backend_log = self.logs_dir / "backend.log" env = os.environ.copy() env['PATH'] = f"{self.python_exe.parent}:{env['PATH']}" with open(backend_log, 'w') as log_f: self.backend_process = subprocess.Popen( [ str(self.python_exe), "-m", "uvicorn", "backend.main:app", "--host", "0.0.0.0", "--port", str(self.backend_port), "--log-level", self.log_level ], cwd=str(self.root_dir), stdout=log_f, stderr=subprocess.STDOUT, env=env, preexec_fn=os.setsid ) self.log(f"✓ Backend started (PID: {self.backend_process.pid})") def start_frontend(self): """Start Next.js frontend server""" self.log(f"Starting frontend on port {self.frontend_port}...") frontend_log = self.logs_dir / "frontend.log" frontend_dir = self.root_dir / "frontend" env = os.environ.copy() env['PORT'] = str(self.frontend_port) env['NODE_ENV'] = 'production' with open(frontend_log, 'w') as log_f: self.frontend_process = subprocess.Popen( ["node", ".next/standalone/server.js"], cwd=str(frontend_dir), stdout=log_f, stderr=subprocess.STDOUT, env=env, preexec_fn=os.setsid ) self.log(f"✓ Frontend started (PID: {self.frontend_process.pid})") def save_pids(self): """Save process IDs to file for stop/restart commands""" pids = { 'backend': self.backend_process.pid if self.backend_process else None, 'frontend': self.frontend_process.pid if self.frontend_process else None, 'timestamp': time.time() } with open(self.pid_file, 'w') as f: json.dump(pids, f) self.log(f"✓ Process IDs saved to {self.pid_file}") def load_pids(self): """Load process IDs from file""" if not self.pid_file.exists(): return None try: with open(self.pid_file) as f: return json.load(f) except Exception: return None def wait_for_startup(self): """Wait a bit and check if servers started""" time.sleep(3) # Check backend try: import urllib.request urllib.request.urlopen(f"http://localhost:{self.backend_port}/health", timeout=2) self.log(f"✓ Backend responding at http://localhost:{self.backend_port}/health") except Exception: self.log(f"⚠ Backend not responding yet (may still be initializing)") # Check frontend try: import urllib.request urllib.request.urlopen(f"http://localhost:{self.frontend_port}/", timeout=2) self.log(f"✓ Frontend responding at http://localhost:{self.frontend_port}") except Exception: self.log(f"⚠ Frontend not responding yet (may still be initializing)") def print_summary(self): """Print summary of running servers""" print("\n" + "="*60) print(" ✓ Standalone servers started successfully!") print("="*60 + "\n") print("Access points:") print(f" Frontend: http://localhost:{self.frontend_port}") print(f" Backend: http://localhost:{self.backend_port}") print(f" API Docs: http://localhost:{self.backend_port}/docs") print("\nProcess IDs:") print(f" Backend: {self.backend_process.pid}") print(f" Frontend: {self.frontend_process.pid}") print("\nLog files:") print(f" Backend: {self.logs_dir / 'backend.log'}") print(f" Frontend: {self.logs_dir / 'frontend.log'}") print("\nMonitoring:") print(f" Backend: tail -f {self.logs_dir / 'backend.log'}") print(f" Frontend: tail -f {self.logs_dir / 'frontend.log'}") print(f" Status: python3 start_servers.py status") print("\nTo stop servers:") print(f" python3 start_servers.py stop") print("="*60 + "\n") def run_foreground(self): """Run servers in foreground (interactive)""" try: print("\n=== TFM aInventory Standalone Server Start ===\n") self.check_ports() self.setup_backend() self.setup_frontend() self.start_backend() self.start_frontend() self.wait_for_startup() self.print_summary() self.log("✓ Servers are running! Press Ctrl+C to stop.") while True: if self.backend_process.poll() is not None: self.log(f"⚠ Backend process exited with code {self.backend_process.returncode}") self.log(" Check logs/backend.log for details") if self.frontend_process.poll() is not None: self.log(f"⚠ Frontend process exited with code {self.frontend_process.returncode}") self.log(" Check logs/frontend.log for details") time.sleep(5) except KeyboardInterrupt: self.log("Received interrupt signal, shutting down...") self.cleanup() except Exception as e: self.error(str(e)) def run_background(self): """Run servers in background (detached)""" try: print("\n=== TFM aInventory Standalone Server Start (Background) ===\n") self.check_ports() self.setup_backend() self.setup_frontend() self.start_backend() self.start_frontend() self.save_pids() self.wait_for_startup() self.print_summary() print(f"✓ Servers running in background") print(f" Stop: python3 start_servers.py stop") print(f" Status: python3 start_servers.py status\n") except Exception as e: self.error(str(e)) def stop_servers(self): """Stop background servers""" pids = self.load_pids() if not pids: print("[ERROR] No running servers found") return False backend_pid = pids.get('backend') frontend_pid = pids.get('frontend') success = True if backend_pid: try: os.killpg(os.getpgid(backend_pid), signal.SIGTERM) print(f"[INFO] Backend stopped (PID {backend_pid})") time.sleep(1) except Exception as e: print(f"[WARN] Failed to stop backend: {e}") success = False if frontend_pid: try: os.killpg(os.getpgid(frontend_pid), signal.SIGTERM) print(f"[INFO] Frontend stopped (PID {frontend_pid})") time.sleep(1) except Exception as e: print(f"[WARN] Failed to stop frontend: {e}") success = False if self.pid_file.exists(): self.pid_file.unlink() print("[INFO] ✓ Servers stopped") return success def show_status(self): """Show status of running servers""" pids = self.load_pids() if not pids: print("[INFO] No running servers") return backend_pid = pids.get('backend') frontend_pid = pids.get('frontend') print("\n" + "="*60) print(" Server Status") print("="*60 + "\n") # Check backend if backend_pid: try: os.kill(backend_pid, 0) print(f"✓ Backend: RUNNING (PID {backend_pid})") print(f" http://localhost:{self.backend_port}") except ProcessLookupError: print(f"✗ Backend: STOPPED (was PID {backend_pid})") else: print(f"✗ Backend: NOT FOUND") # Check frontend if frontend_pid: try: os.kill(frontend_pid, 0) print(f"✓ Frontend: RUNNING (PID {frontend_pid})") print(f" http://localhost:{self.frontend_port}") except ProcessLookupError: print(f"✗ Frontend: STOPPED (was PID {frontend_pid})") else: print(f"✗ Frontend: NOT FOUND") print("\nLog files:") print(f" {self.logs_dir / 'backend.log'}") print(f" {self.logs_dir / 'frontend.log'}") print("="*60 + "\n") def cleanup(self): """Gracefully shutdown servers""" if self.backend_process: try: os.killpg(os.getpgid(self.backend_process.pid), signal.SIGTERM) self.backend_process.wait(timeout=5) except Exception: try: self.backend_process.kill() except Exception: pass if self.frontend_process: try: os.killpg(os.getpgid(self.frontend_process.pid), signal.SIGTERM) self.frontend_process.wait(timeout=5) except Exception: try: self.frontend_process.kill() except Exception: pass if self.pid_file.exists(): self.pid_file.unlink() self.log("✓ Servers stopped") sys.exit(0) def log(self, message): """Print log message""" print(f"[INFO] {message}") def error(self, message): """Print error and exit""" print(f"[ERROR] {message}", file=sys.stderr) sys.exit(1) def main(): parser = argparse.ArgumentParser( description='TFM aInventory Standalone Server Launcher', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: python3 start_servers.py # Start in foreground (interactive, Ctrl+C to stop) python3 start_servers.py start # Start in background python3 start_servers.py stop # Stop background servers python3 start_servers.py restart # Restart background servers python3 start_servers.py status # Show server status """ ) parser.add_argument( 'command', nargs='?', choices=['start', 'stop', 'restart', 'status'], default='foreground', help='Command to execute (default: foreground mode)' ) args = parser.parse_args() manager = StandaloneServerManager() if args.command == 'start': manager.run_background() elif args.command == 'stop': manager.stop_servers() elif args.command == 'restart': manager.stop_servers() time.sleep(2) manager.run_background() elif args.command == 'status': manager.show_status() else: # Foreground mode (default) manager.run_foreground() if __name__ == "__main__": main()