from __future__ import annotations import importlib.util import os import sys from contextlib import contextmanager from dataclasses import dataclass from pathlib import Path from typing import Iterator from fastapi import FastAPI @dataclass(frozen=True) class ServiceDefinition: key: str title: str path: str directory: str module_file: str app_object: str description: str enabled_env: str SERVICES = [ ServiceDefinition( key="ktts", title="Kokoro TTS", path="/services/ktts", directory="ktts", module_file="app.py", app_object="app", description="ONNX Kokoro text-to-speech service.", enabled_env="MAESTER_ENABLE_KTTS", ), ServiceDefinition( key="musicgen", title="MusicGen", path="/services/musicgen", directory="musicgen", module_file="app.py", app_object="app", description="Prompt-to-music generation service.", enabled_env="MAESTER_ENABLE_MUSICGEN", ), ServiceDefinition( key="whisper", title="Whisper Operator", path="/services/whisper", directory="whisper", module_file="main.py", app_object="app", description="Transcription, clipping, publishing, and video automation service.", enabled_env="MAESTER_ENABLE_WHISPER", ), ServiceDefinition( key="ffmpeg_automation", title="FFmpeg Automation Hub", path="/services/ffmpeg", directory="ffmpeg_automation", module_file="app.py", app_object="app", description="Common FFmpeg task automation API.", enabled_env="MAESTER_ENABLE_FFMPEG_AUTOMATION", ), ServiceDefinition( key="render_engine", title="Ava2lon Render Engine", path="/services/render", directory="render_engine", module_file="app.py", app_object="app", description="CPU-first rendering engine and studio dashboard.", enabled_env="MAESTER_ENABLE_RENDER_ENGINE", ), ] @contextmanager def _service_import_path(service_root: Path) -> Iterator[None]: original_path = list(sys.path) sys.path.insert(0, str(service_root)) try: yield finally: sys.path[:] = original_path def service_enabled(service: ServiceDefinition) -> bool: return os.getenv(service.enabled_env, "true").strip().lower() in {"1", "true", "yes", "on"} def load_service_app(base_services_dir: Path, service: ServiceDefinition) -> FastAPI: service_root = base_services_dir / service.directory module_path = service_root / service.module_file if not module_path.exists(): raise RuntimeError(f"Missing module for {service.key}: {module_path}") module_name = f"maester_embedded_{service.key}" spec = importlib.util.spec_from_file_location(module_name, module_path) if not spec or not spec.loader: raise RuntimeError(f"Cannot import {service.key} from {module_path}") module = importlib.util.module_from_spec(spec) with _service_import_path(service_root): spec.loader.exec_module(module) app = getattr(module, service.app_object, None) if app is None: raise RuntimeError(f"{service.key} does not expose {service.app_object}") return app