manage/backend/src/media_library_viewer_api/integrations/base.py

"""Base classes for service integrations.

A *service definition* is a closed, compile-time description of an external service
the app can talk to (Jellyfin, Prometheus, …). Each definition declares:

* its non-secret ``config_schema`` (derived from a Pydantic model),
* the secret fields it accepts (API keys / tokens),
* the widget kinds it can contribute to the dashboard (each with its own
  Pydantic-derived config schema).

Definitions live in :mod:`media_library_viewer_api.integrations` modules and are
assembled into the closed :data:`~media_library_viewer_api.integrations.registry.SERVICE_DEFINITIONS`
map. There is no runtime plugin loading.
"""

from __future__ import annotations

import asyncio
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Annotated, Any, Callable

import requests
from pydantic import BaseModel, BeforeValidator, Field

if TYPE_CHECKING:
    from media_library_viewer_api.services.settings_store import SettingsStore


def _validate_service_base_url(value: Any) -> str:
    """Require an absolute http(s) URL for service ``base_url`` fields.

    Relative hosts (e.g. ``example.com``) break downstream HTTP clients
    because ``requests`` treats them as relative paths, so we fail fast with a
    clear error instead of letting the call silently malfunction.
    """
    if not isinstance(value, str):
        raise ValueError("base_url must be a string starting with http:// or https://")
    text = value.strip()
    if not text:
        raise ValueError("base_url must not be empty")
    lowered = text.lower()
    if not (lowered.startswith("http://") or lowered.startswith("https://")):
        raise ValueError("base_url must start with http:// or https:// (include the schema)")
    return text


#: Shared annotated type for service ``base_url`` fields. applying the validator
#: uniformly across every integration so missing schemas are rejected at the
#: config boundary with a helpful message.
ServiceBaseUrl = Annotated[
    str,
    Field(description="Absolute URL including the http:// or https:// schema."),
    BeforeValidator(_validate_service_base_url),
]


class ServiceConfigBase(BaseModel):
    """Base for per-service non-secret config models.

    Subclass this in each integration module and declare the connection fields.
    The JSON schema is derived via ``model_json_schema()`` and exposed to the UI.

    Connection URLs should use the :data:`ServiceBaseUrl` type so the
    ``http(s)://`` schema is enforced consistently across integrations.
    """


class WidgetConfigBase(BaseModel):
    """Base for per-widget config models.

    Subclass this for each widget kind a service provides. Widget configs never
    hold secrets; credentials live on the parent service record.
    """

    model_config = {"extra": "forbid"}


@dataclass(frozen=True)
class SecretField:
    """A secret field stored encrypted on the service record."""

    key: str
    label: str
    required: bool = False
    helper: str | None = None


@dataclass(frozen=True)
class WidgetKind:
    """A widget kind contributed by a service definition."""

    kind: str
    name: str
    description: str
    config_schema: dict[str, Any]
    default_config: dict[str, Any] = field(default_factory=dict)
    refresh_interval_ms: int = 0
    config_model: type[WidgetConfigBase] | None = None


@dataclass(frozen=True)
class TestResult:
    """Outcome of a credential/connectivity test for a service instance."""

    ok: bool
    detail: str
    evidence: str | None = None


#: A test routine receives (config, secrets, store). The store is needed for
#: ssh_tasks (SSH-key resolution). Other types ignore it.
TestCallable = Callable[[dict[str, Any], dict[str, str], "SettingsStore"], TestResult]


@dataclass(frozen=True)
class ServiceDefinition:
    """Closed description of an external service type."""

    service_type: str
    name: str
    description: str
    config_model: type[ServiceConfigBase]
    secret_fields: list[SecretField]
    widget_kinds: list[WidgetKind]
    test_callable: TestCallable | None = None

    @property
    def config_schema(self) -> dict[str, Any]:
        """JSON schema for the service's non-secret config."""
        return self.config_model.model_json_schema()

    @property
    def secret_keys(self) -> set[str]:
        return {sf.key for sf in self.secret_fields}

    def widget_kind(self, kind: str) -> WidgetKind | None:
        for wk in self.widget_kinds:
            if wk.kind == kind:
                return wk
        return None


def widget_kind(
    kind: str,
    name: str,
    description: str,
    model_cls: type[WidgetConfigBase],
    *,
    default_config: dict[str, Any] | None = None,
    refresh_interval_ms: int = 0,
) -> WidgetKind:
    """Build a :class:`WidgetKind` from a Pydantic widget-config model."""
    schema = model_cls.model_json_schema()
    # Strip Pydantic's title noise so the exposed schema stays clean.
    schema.pop("title", None)
    return WidgetKind(
        kind=kind,
        name=name,
        description=description,
        config_schema=schema,
        default_config=dict(default_config or {}),
        refresh_interval_ms=refresh_interval_ms,
        config_model=model_cls,
    )


def validate_config(model_cls: type[BaseModel], config: dict[str, Any] | None) -> dict[str, Any]:
    """Validate a config dict against a Pydantic model and return the cleaned dict."""
    instance = model_cls.model_validate(config or {})
    return instance.model_dump(exclude_none=True)


def translate_connection_error(exc: Exception, *, context: str = "") -> TestResult:
    """Map a common connection/auth exception to a human-friendly TestResult.

    Handles patterns extracted from ``test_machine_ssh`` (settings.py) plus
    HTTP-client patterns from the widget sources. Each per-type test routine
    calls this for unexpected exceptions, but handles its **type-specific**
    auth failures directly (e.g., qBit ``"Fails."``).
    """
    message = str(exc)
    lowered = message.lower()

    # Auth failures (HTTP 401/403)
    if isinstance(exc, requests.HTTPError):
        status_code = exc.response.status_code if exc.response is not None else 0
        if status_code in (401, 403):
            return TestResult(
                ok=False,
                detail=f"Authentication failed — the service rejected the credentials ({status_code}).",
            )
    if "authentication failed" in lowered or "no authentication methods available" in lowered:
        return TestResult(ok=False, detail="Authentication failed — check the credentials, API key, or SSH key.")

    # Timeout (before OSError check, since requests.Timeout is a subclass of OSError)
    if isinstance(exc, (requests.Timeout, TimeoutError, asyncio.TimeoutError)):
        return TestResult(ok=False, detail="Connection timed out — the service did not respond in time.")

    # Connection refused / DNS / unreachable
    if isinstance(exc, (requests.ConnectionError, ConnectionRefusedError, OSError)):
        if (
            "name or service not known" in lowered
            or "nodename nor servname" in lowered
            or "getaddrinfo failed" in lowered
        ):
            return TestResult(ok=False, detail="Host not found — check the URL/hostname for typos.")
        return TestResult(
            ok=False,
            detail="Connection refused — the service is not reachable at the configured address.",
        )

    # SSL / certificate errors
    if "ssl" in lowered or "certificate" in lowered:
        return TestResult(ok=False, detail="SSL/TLS error — the service's certificate is invalid or untrusted.")

    # SSH banner (from test_machine_ssh pattern)
    if "protocol banner" in lowered:
        return TestResult(
            ok=False,
            detail="SSH banner not received — confirm the SSH service is running and the port is correct.",
        )

    # Fallback
    prefix = f"{context}: " if context else ""
    return TestResult(ok=False, detail=f"{prefix}{message[:200]}")