my_voice_assistant_v3_jamulix/app/dependencies.py

from dataclasses import dataclass

from app.config import Settings, settings
from app.runtime_config import runtime_settings
from app.errors import UnknownComponentError
from app.audio.router import AudioRouter
from app.audio.endpoints.input.local_default import LocalDefaultInput
from app.audio.endpoints.input.bluetooth import BluetoothInput
from app.audio.endpoints.input.mobile_ws import MobileWebSocketInput
from app.audio.endpoints.input.mobile_webrtc import MobileWebRTCInput
from app.audio.endpoints.output.local_default import LocalDefaultOutput
from app.audio.endpoints.output.bluetooth import BluetoothOutput
from app.audio.endpoints.output.mobile_ws import MobileWebSocketOutput
from app.audio.endpoints.output.mobile_webrtc import MobileWebRTCOutput
from app.audio.endpoints.output.loopback import LoopbackOutput
from app.providers.stt.openrouter import OpenRouterSTTProvider
from app.providers.stt.faster_whisper import FasterWhisperProvider
from app.providers.llm.local_openai_compatible import LocalOpenAICompatibleLLM
from app.providers.llm.openrouter import OpenRouterLLMProvider
from app.providers.llm.tool_calling import ToolCallingLLM
from app.tools.web_search import SonarTool
from app.providers.tts.openrouter import OpenRouterTTSProvider
from app.providers.tts.cartesia import CartesiaTTSProvider
from app.providers.tts.chatterbox import ChatterboxTTSProvider
from app.providers.tts.piper import PiperTTSProvider
from app.providers.fallback import (
    FallbackSTTProvider,
    FallbackLLMProvider,
    FallbackTTSProvider,
)
from app.pipeline.decontextualizer import Decontextualizer
from app.pipeline.fillers import FillerPhrases, make_openrouter_translator
from app.pipeline.input_cleaner import InputCleaner
from app.pipeline.spoken_response_adapter import SpokenResponseAdapter
from app.pipeline.tts_normalizer import TTSNormalizer
from app.core.orchestrator import Orchestrator
from app.store import SQLiteStore, Store, User

# ---------------------------------------------------------------------------
# Persistenz-Store: Modul-Singleton (SQLite). Spaetere Backends implementieren
# dasselbe Store-Interface, ohne die App zu aendern.
# ---------------------------------------------------------------------------
_store: Store | None = None


def get_store() -> Store:
    global _store
    if _store is None:
        _store = SQLiteStore(settings.db_path)
    return _store


# Filler-Sätze (Web-Suche): Modul-Singleton mit Übersetzungs-Cache.
_fillers: FillerPhrases | None = None


def get_fillers(cfg=None) -> FillerPhrases:
    global _fillers
    if _fillers is None:
        cfg = cfg or runtime_settings
        _fillers = FillerPhrases(
            translate_pool=make_openrouter_translator(cfg.openrouter_api_key, cfg.openrouter_llm_model)
        )
    return _fillers

# ---------------------------------------------------------------------------
# Provider-Registries: Modul austauschbar via Name, ohne Kern-Code zu aendern.
# Ein neuer Provider = ein Eintrag. Unbekannter Name -> UnknownComponentError.
# ---------------------------------------------------------------------------
STT_REGISTRY = {
    "openrouter": lambda s: OpenRouterSTTProvider(s.openrouter_api_key, s.openrouter_stt_model),
    "faster-whisper": lambda s: FasterWhisperProvider(),
}

LLM_REGISTRY = {
    "openrouter": lambda s: OpenRouterLLMProvider(s.openrouter_api_key, s.openrouter_llm_model),
    # Tool-fähige Variante desselben OpenRouter-Modells (Weg 2: web_search via Sonar).
    "openrouter-tools": lambda s: ToolCallingLLM(
        s.openrouter_api_key,
        s.openrouter_llm_model,
        tools=[SonarTool(s.openrouter_api_key)],
        decontextualizer=Decontextualizer(s.openrouter_api_key, s.openrouter_llm_model),
    ),
    "local-openai-compatible": lambda s: LocalOpenAICompatibleLLM(
        s.local_llm_base_url,
        s.local_llm_api_key,
        s.local_llm_model,
        system_prompt=s.local_llm_system_prompt,
        disable_reasoning=s.local_llm_disable_reasoning,
        max_tokens=s.local_llm_max_tokens,
        temperature=s.local_llm_temperature,
        top_p=s.local_llm_top_p,
    ),
}

TTS_REGISTRY = {
    "openrouter": lambda s: OpenRouterTTSProvider(
        s.openrouter_api_key, s.openrouter_tts_model, s.openrouter_tts_voice
    ),
    "chatterbox": lambda s: ChatterboxTTSProvider(
        s.chatterbox_base_url,
        s.chatterbox_voice,
        s.chatterbox_lang,
        s.chatterbox_speed,
        s.tts_sample_rate,
        s.chatterbox_timeout,
        voices_dir=s.chatterbox_voices_dir,
    ),
    "cartesia": lambda s: CartesiaTTSProvider(
        s.cartesia_api_key, s.cartesia_voice_id, s.cartesia_tts_model, s.tts_sample_rate,
        voices_str=s.cartesia_voices,
    ),
    "piper": lambda s: PiperTTSProvider(
        s.piper_bin, s.piper_voices_dir, s.piper_voice, s.tts_sample_rate
    ),
}


def _from_registry(registry: dict, name: str, kind: str, cfg: Settings):
    try:
        factory = registry[name]
    except KeyError as exc:
        raise UnknownComponentError(
            f"Unbekannter {kind}-Provider {name!r}. Verfuegbar: {sorted(registry)}"
        ) from exc
    return factory(cfg)


def get_stt_provider(name: str | None = None, cfg=None):
    cfg = cfg or runtime_settings
    return _from_registry(STT_REGISTRY, name or cfg.default_stt_provider, "STT", cfg)


def get_llm_provider(name: str | None = None, cfg=None):
    cfg = cfg or runtime_settings
    return _from_registry(LLM_REGISTRY, name or cfg.default_llm_provider, "LLM", cfg)


def get_tts_provider(name: str | None = None, cfg=None):
    cfg = cfg or runtime_settings
    return _from_registry(TTS_REGISTRY, name or cfg.default_tts_provider, "TTS", cfg)


# ---------------------------------------------------------------------------
# Audio-Router: Modul-Singleton, damit zustandsbehaftete Endpunkte
# (z. B. LoopbackOutput.chunks) ueber Requests hinweg stabil bleiben.
# ---------------------------------------------------------------------------
_audio_router: AudioRouter | None = None


def get_audio_router() -> AudioRouter:
    global _audio_router
    if _audio_router is None:
        _audio_router = AudioRouter(
            inputs=[
                LocalDefaultInput(),
                BluetoothInput(),
                MobileWebSocketInput(),
                MobileWebRTCInput(),
            ],
            outputs=[
                LocalDefaultOutput(),
                BluetoothOutput(),
                MobileWebSocketOutput(),
                MobileWebRTCOutput(),
                LoopbackOutput(),
            ],
        )
    return _audio_router


# ---------------------------------------------------------------------------
# Session-Routing und einheitliche Route-Aufloesung ueber alle Achsen.
# Praezedenz: Settings-Defaults < Session-Route < Request-Overrides.
# ---------------------------------------------------------------------------
ROUTE_KEYS = (
    "input_endpoint",
    "output_endpoint",
    "stt_provider",
    "llm_provider",
    "tts_provider",
    "language",
    "voice_gender",
    "web_search_enabled",
)


def _as_bool(value, default: bool = True) -> bool:
    """Robuste Bool-Auflösung (Prefs/Overrides können Strings sein)."""
    if isinstance(value, bool):
        return value
    if isinstance(value, str):
        return value.strip().lower() in ("1", "true", "yes", "on")
    if value is None:
        return default
    return bool(value)

# Stimm-Auswahl nach Sprache: (effektive) Sprache → beste Piper-Stimme (gender-agnostisch).
LANG_TO_PIPER_VOICE: dict[str, str] = {
    "de":  "de_DE-thorsten-high",
    "en":  "en_US-lessac-high",
    "fr":  "fr_FR-siwis-medium",
    "es":  "es_ES-sharvard-medium",
    "it":  "it_IT-paola-medium",
    "nl":  "nl_NL-mls-medium",
    "pt":  "pt_BR-faber-medium",
    "pl":  "pl_PL-darkman-medium",
    "ar":  "ar_JO-kareem-medium",
    "ru":  "ru_RU-irina-medium",
    "zh":  "zh_CN-huayan-medium",
    "cmn": "zh_CN-huayan-medium",
}

# Geschlechtsspezifische Piper-Stimmen für Sprachen mit m+f Varianten.
# Präzedenz: PIPER_VOICE_GENDERED > LANG_TO_PIPER_VOICE.
# ES sharvard ist ein Multi-Speaker-Modell: #0=M, #1=F (speaker_id_map: {'M':0,'F':1})
PIPER_VOICE_GENDERED: dict[tuple[str, str], str] = {
    ("de", "f"): "de_DE-kerstin-low",
    ("de", "m"): "de_DE-thorsten-high",
    ("en", "f"): "en_US-amy-medium",
    ("en", "m"): "en_US-lessac-high",
    ("fr", "f"): "fr_FR-siwis-medium",
    ("fr", "m"): "fr_FR-tom-medium",
    ("es", "f"): "es_ES-sharvard-medium#1",
    ("es", "m"): "es_ES-sharvard-medium#0",
    ("it", "f"): "it_IT-paola-medium",
    ("it", "m"): "it_IT-riccardo-x_low",
    ("ru", "f"): "ru_RU-irina-medium",
    ("ru", "m"): "ru_RU-ruslan-medium",
    ("pl", "f"): "pl_PL-gosia-medium",
    ("pl", "m"): "pl_PL-darkman-medium",
}


def piper_voice_for_language(tts_provider: str, language: str | None) -> str | None:
    """Kompatibilitäts-Alias für voice_for_route ohne gender."""
    return voice_for_route(tts_provider, language, "any")


def voice_for_route(
    tts_provider: str, language: str | None, voice_gender: str = "any"
) -> str | None:
    """Liefert den voice-Parameter für synthesize() passend zum Provider.

    - piper:    Piper-Stimmmodell-Name; geschlechtsspezifisch wenn Varianten vorhanden
    - cartesia: Geschlechts-Code ("m"/"f"/"any") → CartesiaTTSProvider löst UUID auf
    - andere:   None (Provider nutzt seinen eigenen Default)
    """
    if tts_provider == "piper" and language:
        lang = language.lower()
        gender = (voice_gender or "any").lower()
        # Geschlechtsspezifische Stimme: bevorzugte Richtung, dann Gegenteil
        if gender in ("m", "f"):
            candidates = [gender, "f" if gender == "m" else "m"]
        else:  # "any" → weiblich bevorzugt (konsistent mit Cartesia)
            candidates = ["f", "m"]
        for g in candidates:
            voice = PIPER_VOICE_GENDERED.get((lang, g))
            if voice:
                return voice
        return LANG_TO_PIPER_VOICE.get(lang)
    if tts_provider == "cartesia":
        return voice_gender or "any"
    return None


@dataclass
class ResolvedRoute:
    input_endpoint: str
    output_endpoint: str
    stt_provider: str
    llm_provider: str
    tts_provider: str
    language: str
    voice_gender: str = "any"
    web_search_enabled: bool = True

    def as_dict(self) -> dict:
        return {
            "input_endpoint": self.input_endpoint,
            "output_endpoint": self.output_endpoint,
            "stt_provider": self.stt_provider,
            "llm_provider": self.llm_provider,
            "tts_provider": self.tts_provider,
            "language": self.language,
            "voice_gender": self.voice_gender,
            "web_search_enabled": self.web_search_enabled,
        }


def get_session_route(session_id: str | None, user: User | None = None) -> dict:
    """Liefert die gespeicherte Route einer Session des Nutzers (leeres dict sonst).

    Gehoert die Session einem anderen Nutzer, wird SessionOwnershipError ausgeloest.
    """
    if not session_id:
        return {}
    session = get_store().get_session(session_id)
    if session is None:
        return {}
    if user is not None and session.user_id != user.id:
        from app.store import SessionOwnershipError

        raise SessionOwnershipError(
            f"Session {session_id!r} gehoert einem anderen Nutzer"
        )
    return session.data


def resolve_route(
    user: User | None = None,
    session_id: str | None = None,
    overrides: dict | None = None,
    cfg=None,
) -> ResolvedRoute:
    cfg = cfg or runtime_settings
    """Loest die effektive Route auf.

    Praezedenz (hoeher gewinnt): Defaults < Nutzer-Prefs < Session-Route < Request.
    """
    resolved = {
        "input_endpoint": cfg.default_input_endpoint,
        "output_endpoint": cfg.default_output_endpoint,
        "stt_provider": cfg.default_stt_provider,
        "llm_provider": cfg.default_llm_provider,
        "tts_provider": cfg.default_tts_provider,
        "language": cfg.default_language,
        "web_search_enabled": cfg.web_search_enabled,
    }

    user_prefs = user.prefs if user is not None else {}
    session_route = get_session_route(session_id, user)
    request_overrides = overrides or {}

    for layer in (user_prefs, session_route, request_overrides):
        for key in ROUTE_KEYS:
            value = layer.get(key)
            if value is not None:
                resolved[key] = value

    # web_search_enabled kann als String aus Prefs/Overrides kommen -> robust nach bool.
    resolved["web_search_enabled"] = _as_bool(resolved["web_search_enabled"], cfg.web_search_enabled)
    route = ResolvedRoute(**resolved)
    # Admin-Vorgabe „erlaubte Sprachen pro Nutzer": auf eine erlaubte Sprache klemmen.
    allowed = [s.strip() for s in str(user_prefs.get("allowed_languages") or "").split(",") if s.strip()]
    if allowed and route.language not in allowed:
        route.language = allowed[0]
    return route


_FALLBACK_CLASS = {
    "stt": FallbackSTTProvider,
    "llm": FallbackLLMProvider,
    "tts": FallbackTTSProvider,
}


def _provider_chain(registry, primary: str, fallback_csv: str, module: str, cfg: Settings):
    """Baut primaeren Provider + optionale Fallback-Kette (dedupliziert, Reihenfolge erhalten)."""
    names = [primary] + [n.strip() for n in (fallback_csv or "").split(",") if n.strip()]
    seen, ordered = set(), []
    for name in names:
        if name not in seen:
            seen.add(name)
            ordered.append(name)
    entries = [(name, _from_registry(registry, name, module.upper(), cfg)) for name in ordered]
    if len(entries) == 1:
        return entries[0][1]
    return _FALLBACK_CLASS[module](module, entries)


def _resolve_normalize_level(tts_provider: str, cfg: Settings) -> str:
    """auto -> piper bekommt 'full', Cloud-TTS 'light' (macht Zahlen/Abk. selbst gut)."""
    level = (cfg.tts_normalize_level or "auto").lower()
    if level == "auto":
        return "full" if tts_provider == "piper" else "light"
    return level


def build_orchestrator(route: ResolvedRoute, cfg=None) -> Orchestrator:
    cfg = cfg or runtime_settings
    # web_search (Weg 2): tool-fähige Variante desselben OpenRouter-Modells wählen.
    # Nur wenn der aufgelöste LLM-Provider "openrouter" ist — lokale Modelle können
    # (über diesen Pfad) kein Tool-Calling. Die llm_fallback-Kette bleibt erhalten.
    llm_name = route.llm_provider
    if route.web_search_enabled and llm_name == "openrouter":
        llm_name = "openrouter-tools"
    return Orchestrator(
        stt=_provider_chain(STT_REGISTRY, route.stt_provider, cfg.stt_fallback, "stt", cfg),
        llm=_provider_chain(LLM_REGISTRY, llm_name, cfg.llm_fallback, "llm", cfg),
        tts=_provider_chain(TTS_REGISTRY, route.tts_provider, cfg.tts_fallback, "tts", cfg),
        input_cleaner=InputCleaner(),
        spoken_adapter=SpokenResponseAdapter(),
        tts_normalizer=TTSNormalizer(),
        normalize_level=_resolve_normalize_level(route.tts_provider, cfg),
        fillers=get_fillers(cfg),
    )


async def resolve_output_endpoint(route: ResolvedRoute):
    return await get_audio_router().select_output(route.output_endpoint)


async def resolve_input_endpoint(route: ResolvedRoute):
    return await get_audio_router().select_input(route.input_endpoint)