#!/usr/bin/env python3 """ TFM aInventory Standalone Server Launcher Starts both backend (FastAPI) and frontend (Next.js) servers Manages processes, logs, and graceful shutdown """ import os import sys import subprocess import signal import time from pathlib import Path class StandaloneServerManager: def __init__(self): self.root_dir = Path(__file__).parent 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...") subprocess.run( ["npm", "run", "build"], cwd=str(frontend_dir), check=True, capture_output=True ) def start_backend(self): """Start FastAPI backend server using uvicorn directly from project root""" self.log(f"Starting backend on port {self.backend_port}...") backend_log = self.logs_dir / "backend.log" backend_dir = self.root_dir / "backend" env = os.environ.copy() env['PATH'] = f"{self.python_exe.parent}:{env['PATH']}" # Use uvicorn CLI from venv to run the app with open(backend_log, 'w') as log_f: self.backend_process = subprocess.Popen( [ str(self.python_exe), "-m", "uvicorn", "main:app", "--host", "0.0.0.0", "--port", str(self.backend_port), "--log-level", self.log_level ], cwd=str(backend_dir), stdout=log_f, stderr=subprocess.STDOUT, env=env, preexec_fn=os.setsid # Create new process group to avoid signal issues ) 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 # Create new process group ) self.log(f"✓ Frontend started (PID: {self.frontend_process.pid})") 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" All logs: tail -f {self.logs_dir}/*.log") print("\nTo stop servers:") print(f" Press Ctrl+C in this terminal") print("="*60 + "\n") def run(self): """Run the server manager""" 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() # Monitor processes self.log("✓ Servers are running! Press Ctrl+C to stop.") while True: # Check if processes are alive 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 cleanup(self): """Gracefully shutdown servers""" import signal 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 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) if __name__ == "__main__": manager = StandaloneServerManager() manager.run()