S.H.O.N.A.R._Desktop_Companion/backend/shonar/services/ai/__init__.py

"""AI provider interfaces (M7).

Two independent axes, both optional and both configured only through
environment variables (never hard-coded keys):

- transcription: none | whisper_http | faster_whisper
- LLM (summary/action items): none | openai_compat | ollama

"none" is a first-class choice: recording, sync, playback, and manual
transcripts work with no AI configured at all. The pipeline treats a
missing provider as "skip this stage", never as an error.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Protocol

from shonar.core.config import Settings


class AIError(Exception):
    """Base for AI failures. Messages must be user-safe: they surface in
    ``processing_error`` and therefore on screens."""


class ProviderConfigError(AIError):
    """Persistent misconfiguration (bad credentials, unknown model, missing
    dependency). Fails the job immediately — retrying cannot help."""


class ProviderTransientError(AIError):
    """May succeed on retry (timeouts, 429/5xx). The worker requeues these
    up to the job's max attempts."""


@dataclass(frozen=True)
class Segment:
    start: float
    end: float
    text: str
    speaker: str | None = None


@dataclass(frozen=True)
class TranscriptResult:
    text: str
    language: str | None
    segments: list[Segment] = field(default_factory=list)
    model: str = ""


class TranscriptionProvider(Protocol):
    name: str

    async def transcribe(
        self,
        audio: bytes,
        mime: str,
        *,
        language_hint: str | None = None,
        on_progress=None,  # optional Callable[[int], None], 0..99
    ) -> TranscriptResult: ...


SUMMARY_KEYS = ("short", "detailed", "key_points", "decisions", "action_items", "questions")


@dataclass(frozen=True)
class SummaryResult:
    short: str = ""
    detailed: str = ""
    key_points: tuple[str, ...] = ()
    decisions: tuple[str, ...] = ()
    action_items: tuple[str, ...] = ()
    questions: tuple[str, ...] = ()
    model: str = ""

    def to_dict(self) -> dict:
        return {
            "short": self.short,
            "detailed": self.detailed,
            "key_points": list(self.key_points),
            "decisions": list(self.decisions),
            "action_items": list(self.action_items),
            "questions": list(self.questions),
        }

    @classmethod
    def from_dict(cls, raw: dict, model: str = "") -> SummaryResult:
        def text(key: str) -> str:
            v = raw.get(key)
            return v if isinstance(v, str) else ""

        def strs(key: str) -> tuple[str, ...]:
            v = raw.get(key)
            if not isinstance(v, list):
                return ()
            return tuple(s for s in v if isinstance(s, str) and s.strip())

        return cls(
            short=text("short"),
            detailed=text("detailed"),
            key_points=strs("key_points"),
            decisions=strs("decisions"),
            action_items=strs("action_items"),
            questions=strs("questions"),
            model=model,
        )


class LlmProvider(Protocol):
    name: str

    async def summarize(self, transcript: str, *, title: str | None = None,
                        tone: str | None = None,
                        on_progress=None) -> SummaryResult: ...  # Callable[[int], None] | None


def get_transcription_provider(settings: Settings) -> TranscriptionProvider | None:
    """None means "transcription stage skipped", never an error."""
    kind = settings.transcription_provider.strip().lower()
    if kind in ("", "none"):
        return None
    if kind == "whisper_http":
        from shonar.services.ai.whisper_http import WhisperHttpProvider

        return WhisperHttpProvider(
            base_url=settings.transcription_base_url,
            model=settings.transcription_model,
            api_key=settings.transcription_api_key,
        )
    if kind == "faster_whisper":
        from shonar.services.ai.faster_whisper import FasterWhisperProvider

        return FasterWhisperProvider(model=settings.transcription_model)
    raise ProviderConfigError(
        f"Unknown transcription provider: {settings.transcription_provider!r}"
    )


def get_llm_provider(settings: Settings) -> LlmProvider | None:
    """None means "summary stage skipped", never an error."""
    kind = settings.llm_provider.strip().lower()
    if kind in ("", "none"):
        return None
    if kind == "openai_compat":
        from shonar.services.ai.openai_compat import OpenAICompatProvider

        return OpenAICompatProvider(
            base_url=settings.llm_base_url,
            model=settings.llm_model,
            api_key=settings.llm_api_key,
        )
    if kind == "ollama":
        from shonar.services.ai.ollama import OllamaProvider

        return OllamaProvider(
            base_url=settings.llm_base_url,
            model=settings.llm_model,
        )
    raise ProviderConfigError(f"Unknown LLM provider: {settings.llm_provider!r}")


def get_llm_fallback_provider(settings: Settings) -> LlmProvider | None:
    """Rescue provider tried when the primary summarize fails mid-job.

    None means "no fallback configured" — the primary's error stands.
    Same kinds as get_llm_provider, built from the llm_fallback_* settings.
    """
    kind = settings.llm_fallback_provider.strip().lower()
    if kind in ("", "none"):
        return None
    if kind == "openai_compat":
        from shonar.services.ai.openai_compat import OpenAICompatProvider

        return OpenAICompatProvider(
            base_url=settings.llm_fallback_base_url,
            model=settings.llm_fallback_model,
            api_key=settings.llm_fallback_api_key,
        )
    if kind == "ollama":
        from shonar.services.ai.ollama import OllamaProvider

        return OllamaProvider(
            base_url=settings.llm_fallback_base_url,
            model=settings.llm_fallback_model,
        )
    raise ProviderConfigError(
        f"Unknown LLM fallback provider: {settings.llm_fallback_provider!r}"
    )