backend/services/plugin_sdk.py

"""
Plugin SDK — abstract interface for third-party TTS engines.

Allows community contributors to add support for ElevenLabs, XTTS, Bark,
Fish TTS, etc. without modifying core OmniVoice code.

Usage:
    1. Create a Python file in backend/plugins/  (e.g. elevenlabs.py)
    2. Subclass `TTSPlugin` and implement the 4 abstract methods
    3. Register via `@register_plugin` decorator or add to PLUGINS dict
    4. The engine will appear in the frontend Settings → TTS Engine picker

Example:
    from services.plugin_sdk import TTSPlugin, register_plugin

    @register_plugin
    class ElevenLabsPlugin(TTSPlugin):
        id = "elevenlabs"
        display_name = "ElevenLabs"
        ...
"""
from __future__ import annotations

import logging
from abc import ABC, abstractmethod
from typing import Optional

logger = logging.getLogger("omnivoice.plugins")

# ── Plugin registry ──────────────────────────────────────────────────

PLUGINS: dict[str, type["TTSPlugin"]] = {}


def register_plugin(cls: type["TTSPlugin"]) -> type["TTSPlugin"]:
    """Decorator: register a TTS plugin class by its `id`."""
    if not hasattr(cls, "id") or not cls.id:
        raise ValueError(f"Plugin class {cls.__name__} must define a non-empty `id`.")
    PLUGINS[cls.id] = cls
    logger.info("Registered TTS plugin: %s (%s)", cls.id, cls.display_name)
    return cls


def get_plugin(plugin_id: str) -> "TTSPlugin":
    """Instantiate and return a plugin by id."""
    cls = PLUGINS.get(plugin_id)
    if cls is None:
        available = ", ".join(sorted(PLUGINS.keys())) or "none"
        raise KeyError(f"Unknown TTS plugin '{plugin_id}'. Available: {available}")
    return cls()


def list_plugins() -> list[dict]:
    """Return metadata for all registered plugins (for the frontend)."""
    out = []
    for pid, cls in sorted(PLUGINS.items()):
        ok, msg = cls.is_available()
        out.append({
            "id": pid,
            "display_name": cls.display_name,
            "requires_api_key": cls.requires_api_key,
            "is_local": cls.is_local,
            "available": ok,
            "availability_message": msg,
            "supported_languages": cls.supported_languages_hint,
        })
    return out


# ── Abstract base class ─────────────────────────────────────────────


class TTSPlugin(ABC):
    """Base class for all TTS engine plugins.

    Subclass this and implement the abstract methods to add support for
    a new TTS engine (cloud API or local model).
    """

    #: Unique identifier (lowercase, no spaces). Used in API requests.
    id: str = ""

    #: Human-readable name for the UI.
    display_name: str = "Unnamed Plugin"

    #: Whether this engine needs an API key (cloud providers).
    requires_api_key: bool = False

    #: Whether this engine runs locally (no network calls).
    is_local: bool = False

    #: Hint for the UI — list of commonly supported languages.
    supported_languages_hint: list[str] = ["en"]

    @classmethod
    @abstractmethod
    def is_available(cls) -> tuple[bool, str]:
        """Check if the engine can run in the current environment.

        Returns:
            (True, "Ready") if available.
            (False, "pip install ...") with actionable fix instructions.
        """

    @abstractmethod
    def generate(
        self,
        text: str,
        *,
        voice_id: Optional[str] = None,
        language: Optional[str] = None,
        speed: float = 1.0,
        **kwargs,
    ) -> bytes:
        """Generate speech from text.

        Args:
            text: The text to synthesize.
            voice_id: Provider-specific voice identifier.
            language: ISO 639 language code.
            speed: Speech speed multiplier.

        Returns:
            Raw audio bytes (WAV or MP3, depending on provider).
        """

    @abstractmethod
    def list_voices(self) -> list[dict]:
        """Return available voices for this engine.

        Returns:
            List of dicts with at least: {"id": str, "name": str, "language": str}
        """

    def get_sample_rate(self) -> int:
        """Output sample rate. Override if not 24000."""
        return 24000


# ── Built-in plugin: ElevenLabs (example) ────────────────────────────


@register_plugin
class ElevenLabsPlugin(TTSPlugin):
    """ElevenLabs cloud TTS — high-quality voice synthesis.

    Requires: ELEVENLABS_API_KEY environment variable.
    Install:  pip install elevenlabs
    """

    id = "elevenlabs"
    display_name = "ElevenLabs"
    requires_api_key = True
    is_local = False
    supported_languages_hint = [
        "en", "es", "fr", "de", "it", "pt", "pl", "hi", "ar", "zh",
        "ja", "ko", "nl", "tr", "ru", "sv", "id", "fil", "ms", "ro",
        "uk", "el", "cs", "da", "fi", "bg", "hr", "sk", "ta",
    ]

    @classmethod
    def is_available(cls) -> tuple[bool, str]:
        import os
        if not os.environ.get("ELEVENLABS_API_KEY"):
            return False, "Set ELEVENLABS_API_KEY environment variable."
        try:
            import elevenlabs  # noqa: F401
            return True, "Ready"
        except ImportError:
            return False, "pip install elevenlabs"

    def generate(self, text, *, voice_id=None, language=None, speed=1.0, **kw) -> bytes:
        import os
        from elevenlabs import ElevenLabs

        client = ElevenLabs(api_key=os.environ["ELEVENLABS_API_KEY"])
        audio_iter = client.text_to_speech.convert(
            text=text,
            voice_id=voice_id or "JBFqnCBsd6RMkjVDRZzb",  # George default
            model_id="eleven_multilingual_v2",
            output_format="mp3_44100_128",
        )
        return b"".join(audio_iter)

    def list_voices(self) -> list[dict]:
        import os
        try:
            from elevenlabs import ElevenLabs
            client = ElevenLabs(api_key=os.environ.get("ELEVENLABS_API_KEY", ""))
            voices = client.voices.get_all()
            return [
                {"id": v.voice_id, "name": v.name, "language": "multi"}
                for v in voices.voices
            ]
        except Exception as e:
            logger.warning("ElevenLabs list_voices failed: %s", e)
            return []

    def get_sample_rate(self) -> int:
        return 44100


# ── Built-in plugin: Bark (local) ────────────────────────────────────


@register_plugin
class BarkPlugin(TTSPlugin):
    """Suno Bark — open-source local TTS with music/effects support.

    Install: pip install suno-bark
    """

    id = "bark"
    display_name = "Bark (Suno)"
    requires_api_key = False
    is_local = True
    supported_languages_hint = ["en", "es", "fr", "de", "it", "pt", "ru", "zh", "ja", "ko"]

    @classmethod
    def is_available(cls) -> tuple[bool, str]:
        try:
            from bark import SAMPLE_RATE  # noqa: F401
            return True, "Ready"
        except ImportError:
            return False, "pip install suno-bark"

    def generate(self, text, *, voice_id=None, language=None, speed=1.0, **kw) -> bytes:
        import io
        import numpy as np
        from bark import generate_audio, SAMPLE_RATE
        import scipy.io.wavfile

        speaker = voice_id or "v2/en_speaker_6"
        audio_array = generate_audio(text, history_prompt=speaker)

        buf = io.BytesIO()
        scipy.io.wavfile.write(buf, SAMPLE_RATE, (audio_array * 32767).astype(np.int16))
        return buf.getvalue()

    def list_voices(self) -> list[dict]:
        return [
            {"id": f"v2/en_speaker_{i}", "name": f"English Speaker {i}", "language": "en"}
            for i in range(10)
        ]

    def get_sample_rate(self) -> int:
        return 24000


# ── Auto-discover plugins from backend/plugins/ directory ────────────

def discover_plugins():
    """Import all .py files in backend/plugins/ to trigger @register_plugin."""
    import importlib
    import pathlib

    plugins_dir = pathlib.Path(__file__).parent.parent / "plugins"
    if not plugins_dir.exists():
        return

    for path in plugins_dir.glob("*.py"):
        if path.name.startswith("_"):
            continue
        module_name = f"plugins.{path.stem}"
        try:
            importlib.import_module(module_name)
            logger.info("Loaded plugin module: %s", module_name)
        except Exception as e:
            logger.warning("Failed to load plugin %s: %s", module_name, e)


# Run discovery on import
discover_plugins()