manage/backend/src/media_library_viewer_api/widgets/sources.py

"""Widget source adapters.

Each adapter implements a uniform async interface and translates widget
configuration into data for the dashboard. Adapters reuse existing clients,
machine registries, and environment settings; they never accept arbitrary
commands or store credentials.
"""

from __future__ import annotations

import asyncio
import logging
import shlex
from typing import Any, Protocol

import requests
from starlette.requests import Request

from media_library_viewer_api.config import get_settings
from media_library_viewer_api.dependencies import get_jellyfin_client
from media_library_viewer_api.domain.dashboard import (
    _map_sessions_to_activity_rows,
    build_backup_dashboard_summary,
)
from media_library_viewer_api.routers.tasks import _client_for_machine, _resolve_machine_for_task
from media_library_viewer_api.services.settings_store import get_settings_store

logger = logging.getLogger(__name__)


def _request_with_machine_id(machine_id: str | None = None) -> Request:
    """Build a minimal Starlette Request carrying a machine_id query param."""
    query = f"machine_id={machine_id}".encode() if machine_id else b""
    return Request({"type": "http", "query_string": query})


class WidgetSource(Protocol):
    """Protocol for widget source adapters."""

    source_type: str

    async def fetch(self, config: dict[str, Any]) -> dict[str, Any]: ...


class JellyfinWidgetSource:
    """Fetch Jellyfin sessions and map them to activity rows."""

    source_type = "jellyfin"
    timeout = 10

    async def fetch(self, config: dict[str, Any]) -> dict[str, Any]:
        try:
            request = _request_with_machine_id(config.get("machine_id") or None)
            client = await asyncio.wait_for(
                asyncio.to_thread(get_jellyfin_client, request),
                timeout=self.timeout,
            )
            sessions = await asyncio.wait_for(
                asyncio.to_thread(client.sessions),
                timeout=self.timeout,
            )
            rows = _map_sessions_to_activity_rows(sessions)
            return {"sessions": rows}
        except asyncio.TimeoutError:
            return {"error": "Widget data fetch timed out"}
        except Exception as exc:
            logger.exception("jellyfin adapter failed")
            return {"error": f"Jellyfin data fetch failed: {exc}"}


class BackupsWidgetSource:
    """Compute the backup dashboard summary."""

    source_type = "backups"
    timeout = 10

    async def fetch(self, config: dict[str, Any]) -> dict[str, Any]:
        try:
            store = get_settings_store()
            summary = build_backup_dashboard_summary(store)
            return summary.model_dump()
        except asyncio.TimeoutError:
            return {"error": "Widget data fetch timed out"}
        except Exception as exc:
            logger.exception("backups adapter failed")
            return {"error": f"Backup summary failed: {exc}"}


class GrafanaWidgetSource:
    """Build a Grafana deep-link (no embedding)."""

    source_type = "grafana"
    timeout = 5

    async def fetch(self, config: dict[str, Any]) -> dict[str, Any]:
        try:
            settings = get_settings()
            dashboard_uid = config.get("dashboard_uid")
            if not dashboard_uid:
                return {"error": "dashboard_uid is required"}
            url = f"{settings.grafana_url.rstrip('/')}/d/{dashboard_uid}"
            panel_id = config.get("panel_id")
            if panel_id is not None:
                url = f"{url}?viewPanel={panel_id}"
            return {"url": url}
        except Exception as exc:
            logger.exception("grafana adapter failed")
            return {"error": f"Grafana link failed: {exc}"}


class PrometheusWidgetSource:
    """Run a PromQL instant query against Prometheus."""

    source_type = "prometheus"
    timeout = 10

    async def fetch(self, config: dict[str, Any]) -> dict[str, Any]:
        try:
            settings = get_settings()
            promql = config.get("promql")
            if not promql:
                return {"error": "promql is required"}
            url = f"{settings.prometheus_url.rstrip('/')}/api/v1/query"
            response = await asyncio.wait_for(
                asyncio.to_thread(
                    requests.get,
                    url,
                    params={"query": promql},
                    timeout=self.timeout,
                ),
                timeout=self.timeout,
            )
            response.raise_for_status()
            payload = response.json()
            return {"result": payload.get("data", {})}
        except asyncio.TimeoutError:
            return {"error": "Widget data fetch timed out"}
        except requests.RequestException as exc:
            logger.exception("prometheus adapter failed")
            return {"error": f"Prometheus query failed: {exc}"}
        except Exception as exc:
            logger.exception("prometheus adapter failed")
            return {"error": f"Prometheus query failed: {exc}"}


class SshTaskWidgetSource:
    """Run a saved task from the registry and return its output."""

    source_type = "ssh_task"
    timeout = 30

    async def fetch(self, config: dict[str, Any]) -> dict[str, Any]:
        try:
            store = get_settings_store()
            task_id = config.get("task_id")
            if not task_id:
                return {"error": "task_id is required"}
            task = store.get_task(task_id)
            if not task:
                return {"error": f"Task {task_id} not found"}
            if not task.get("enabled", True):
                return {"error": "Task is disabled"}

            machine = _resolve_machine_for_task(store, task, None)
            if not machine:
                return {"error": "No machine available for this task"}

            client = _client_for_machine(store, machine)
            task_type = str(task.get("task_type") or "shell").lower()
            command = str(task.get("content") or "")
            if task_type == "python":
                command = f"python3 -c {shlex.quote(command)}"
            elif task_type != "shell":
                return {"error": f"Unknown task type: {task_type}"}

            result = await asyncio.wait_for(
                asyncio.to_thread(client.run, command, timeout=self.timeout),
                timeout=self.timeout,
            )
            return {
                "exit_status": result.exit_status,
                "stdout": result.stdout or "",
                "stderr": result.stderr or "",
            }
        except asyncio.TimeoutError:
            return {"error": "Widget data fetch timed out"}
        except Exception as exc:
            logger.exception("ssh_task adapter failed")
            return {"error": f"SSH task failed: {exc}"}


class StaticWidgetSource:
    """Return static text/markdown unchanged."""

    source_type = "static"

    async def fetch(self, config: dict[str, Any]) -> dict[str, Any]:
        return {"text": config.get("text", "")}


SOURCE_REGISTRY: dict[str, WidgetSource] = {
    "jellyfin": JellyfinWidgetSource(),
    "backups": BackupsWidgetSource(),
    "grafana": GrafanaWidgetSource(),
    "prometheus": PrometheusWidgetSource(),
    "ssh_task": SshTaskWidgetSource(),
    "static": StaticWidgetSource(),
}


def get_source_adapter(source_type: str) -> WidgetSource | None:
    """Return the adapter for a source type, or None if unknown."""
    return SOURCE_REGISTRY.get(source_type)