my_voice_assistant_v3_jamulix/app/api/config.py

import asyncio
import time

import httpx
from fastapi import APIRouter

from app.config import settings, active_profile
from app.capabilities import resolve_gpus
from app.providers.tts.cartesia import _UNSUPPORTED_LANGS as _CARTESIA_UNSUPPORTED
from app.dependencies import (
    resolve_route,
    get_audio_router,
    STT_REGISTRY,
    LLM_REGISTRY,
    TTS_REGISTRY,
)

router = APIRouter()

# /api/config probt externe Dienste (u. a. einen KOSTENPFLICHTIGEN OpenRouter-TTS-Aufruf).
# Daher serverseitig kurz cachen: das Frontend ruft den Endpunkt mehrfach (Laden + Menü-Öffnen).
_CACHE: dict | None = None
_CACHE_TS: float = 0.0
_CACHE_TTL = 30.0
_CACHE_LOCK = asyncio.Lock()


async def _chatterbox_reachable() -> bool:
    try:
        async with httpx.AsyncClient(timeout=1.5) as client:
            r = await client.get(f"{settings.chatterbox_base_url.rstrip('/')}/status")
            return r.status_code < 500
    except Exception:
        return False


async def _cartesia_available() -> bool:
    key = settings.cartesia_api_key.strip()
    voice_id = settings.cartesia_voice_id.strip()
    if not key or not voice_id:
        return False
    try:
        async with httpx.AsyncClient(timeout=5.0) as client:
            r = await client.get(
                "https://api.cartesia.ai/voices",
                headers={"X-API-Key": key, "Cartesia-Version": "2024-06-10"},
            )
            return r.status_code < 400
    except Exception:
        return False


async def _openrouter_tts_available() -> bool:
    """Prüft ob das konfigurierte TTS-Modell auf OpenRouter via /audio/speech erreichbar ist."""
    key = settings.openrouter_api_key.strip()
    model = settings.openrouter_tts_model.strip()
    if not key or not model:
        return False
    try:
        async with httpx.AsyncClient(timeout=5.0) as client:
            r = await client.post(
                "https://openrouter.ai/api/v1/audio/speech",
                headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"},
                json={"model": model, "input": "Test", "voice": settings.openrouter_tts_voice or "alloy",
                      "response_format": "mp3"},
            )
            # 200 = Audio OK, 4xx mit "does not exist" = Modell fehlt
            if r.status_code == 200:
                return True
            body = r.text
            return "does not exist" not in body and r.status_code < 500
    except Exception:
        return False


async def _build_config() -> dict:
    """Baut die /api/config-Antwort (eine Quelle der Wahrheit für die Menü-Sichtbarkeit)."""
    route = resolve_route()
    audio_router = get_audio_router()
    chatterbox_ok, openrouter_tts_ok, cartesia_ok, gpus = (
        await _chatterbox_reachable(),
        await _openrouter_tts_available(),
        await _cartesia_available(),
        await resolve_gpus(),
    )
    # Fertige Sichtbarkeits-Entscheidung je TTS-Provider -> Frontend dupliziert keine Regeln.
    # "device" ist client_only: nur das Frontend kann "Mobilgerät?" beantworten (isMobile()).
    # "piper" bleibt der garantierte, immer sichtbare Default (Menü nie leer).
    tts_menu = [
        {"provider": "piper", "visible": True, "client_only": False},
        {"provider": "chatterbox", "visible": chatterbox_ok, "client_only": False},
        {"provider": "cartesia", "visible": cartesia_ok, "client_only": False},
        {"provider": "openrouter", "visible": openrouter_tts_ok, "client_only": False},
        {"provider": "device", "visible": True, "client_only": True},
    ]
    return {
        "profile": active_profile(),
        "app_env": settings.app_env,
        "default_route": route.as_dict(),
        "tts_menu": tts_menu,
        "available": {
            "stt_providers": sorted(STT_REGISTRY),
            "llm_providers": sorted(LLM_REGISTRY),
            "tts_providers": sorted(TTS_REGISTRY),
            "input_endpoints": [c.model_dump() for c in await audio_router.list_inputs()],
            "output_endpoints": [c.model_dump() for c in await audio_router.list_outputs()],
            "chatterbox_reachable": chatterbox_ok,
            "openrouter_tts_available": openrouter_tts_ok,
            "cartesia_available": cartesia_ok,
            "cartesia_unsupported_languages": sorted(_CARTESIA_UNSUPPORTED),
            "gpus": gpus,
        },
        "secrets": {
            "openrouter_api_key_set": bool(settings.openrouter_api_key.strip()),
        },
    }


@router.get("/config")
async def get_config():
    """Zeigt aktives Profil, aufgeloeste Default-Route und verfuegbare Bausteine (gecacht)."""
    global _CACHE, _CACHE_TS
    if _CACHE is not None and (time.monotonic() - _CACHE_TS) < _CACHE_TTL:
        return _CACHE
    async with _CACHE_LOCK:
        if _CACHE is not None and (time.monotonic() - _CACHE_TS) < _CACHE_TTL:
            return _CACHE
        _CACHE = await _build_config()
        _CACHE_TS = time.monotonic()
        return _CACHE