feat(07-03): convert run_standalone.sh to Python with YAML config parsing

This commit is contained in:
2026-04-23 12:46:50 +03:00
parent 938fd2daf8
commit 1621625bfb

216
scripts/run_standalone.py Executable file
View File

@@ -0,0 +1,216 @@
#!/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 {}
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]):
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)
]
# Add --reload if in dev mode (detectable via environment or flag,
# but the script is for 'run_standalone' which might be used for prod too)
# For now, let's keep it simple.
try:
p = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
env=env,
bufsize=1,
universal_newlines=True
)
processes.append(p)
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]):
logger.info(f"Starting Frontend on port {frontend_port}...")
api_cfg = frontend_cfg.get("api", {})
env = os.environ.copy()
env["NEXT_PUBLIC_API_URL"] = api_cfg.get("backend_url", "http://localhost:8916")
env["PORT"] = str(frontend_port)
# Check if we should use 'npm run dev' or 'node server.js'
# For standalone, we might prefer 'npm run start' if built, or 'npm run dev'
# Bash script used 'npm run start'
cmd = ["npm", "run", "dev"]
if os.path.exists("frontend/server.js"):
# If built for production
cmd = ["node", "frontend/server.js"]
try:
p = subprocess.Popen(
cmd,
cwd="frontend" if os.path.isdir("frontend") else None,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
env=env,
bufsize=1,
universal_newlines=True
)
processes.append(p)
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 main():
parser = argparse.ArgumentParser(description="TFM aInventory Standalone Launcher")
group = parser.add_mutually_exclusive_group()
group.add_argument("--backend-only", action="store_true", help="Start only the backend")
group.add_argument("--frontend-only", action="store_true", help="Start only the frontend")
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 not args.frontend_only:
if is_port_in_use(backend_port):
logger.error(f"Port {backend_port} is already in use. Cannot start backend.")
if not args.backend_only:
# If we are supposed to start both, we fail.
sys.exit(1)
else:
start_backend(backend_port, backend_cfg)
if not args.backend_only:
if is_port_in_use(frontend_port):
logger.error(f"Port {frontend_port} is already in use. Cannot start frontend.")
if not args.frontend_only:
sys.exit(1)
else:
start_frontend(frontend_port, frontend_cfg)
if not processes:
logger.error("No processes started. Check configuration and port availability.")
sys.exit(1)
logger.info("All services started. 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)
if __name__ == "__main__":
main()