maester_enterprise/registry.py

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