"""Widget source adapters. Adapters translate a widget instance into dashboard data. Service-bound widgets are resolved against a :class:`ServiceRecord` (config + decrypted secrets); the built-in widgets (backups, static) take ``service=None``. Adapters never accept arbitrary commands and never store credentials — secrets are decrypted in memory only for the duration of a fetch. """ from __future__ import annotations import asyncio import logging import shlex import time from dataclasses import dataclass, field from typing import Any, Protocol import requests from media_library_viewer_api.clients.jellyfin import JellyfinClient from media_library_viewer_api.clients.ssh import RemoteSSHClient from media_library_viewer_api.config import get_settings from media_library_viewer_api.domain.dashboard import ( _map_sessions_to_activity_rows, build_backup_dashboard_summary, ) from media_library_viewer_api.services.settings_store import SettingsStore, get_settings_store logger = logging.getLogger(__name__) @dataclass class ServiceRecord: """Runtime view of a service instance with decrypted secrets.""" id: str service_type: str name: str config: dict[str, Any] = field(default_factory=dict) secrets: dict[str, str] = field(default_factory=dict) enabled: bool = True def build_service_record(store: SettingsStore, service_row: dict[str, Any]) -> ServiceRecord: """Build a :class:`ServiceRecord`, decrypting secrets in memory.""" from media_library_viewer_api.services.secrets import decrypt_secrets return ServiceRecord( id=service_row["id"], service_type=service_row["service_type"], name=service_row["name"], config=service_row.get("config") or {}, secrets=decrypt_secrets(service_row.get("secrets") or {}), enabled=bool(service_row.get("enabled", True)), ) class WidgetSource(Protocol): """Protocol for widget source adapters.""" async def fetch( self, service: ServiceRecord | None, widget_kind: str, config: dict[str, Any], ) -> dict[str, Any]: ... # --------------------------------------------------------------------------- # Built-in (service-less) adapters # --------------------------------------------------------------------------- class BackupsWidgetSource: """Compute the backup dashboard summary from internal tables.""" async def fetch(self, service: ServiceRecord | None, widget_kind: str, config: dict[str, Any]) -> dict[str, Any]: try: store = get_settings_store() summary = build_backup_dashboard_summary(store) return summary.model_dump() except Exception as exc: logger.exception("backups adapter failed") return {"error": f"Backup summary failed: {exc}"} class StaticWidgetSource: """Return static text/markdown unchanged.""" async def fetch(self, service: ServiceRecord | None, widget_kind: str, config: dict[str, Any]) -> dict[str, Any]: return {"text": config.get("text", "")} # --------------------------------------------------------------------------- # Service-bound adapters # --------------------------------------------------------------------------- class GrafanaWidgetSource: """Build a Grafana deep-link (no embedding).""" async def fetch(self, service: ServiceRecord | None, widget_kind: str, config: dict[str, Any]) -> dict[str, Any]: try: if service is None: return {"error": "Grafana widget is missing its service"} base_url = str(service.config.get("base_url") or "").rstrip("/") dashboard_uid = config.get("dashboard_uid") if not dashboard_uid: return {"error": "dashboard_uid is required"} url = f"{base_url}/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 a Prometheus service.""" async def fetch(self, service: ServiceRecord | None, widget_kind: str, config: dict[str, Any]) -> dict[str, Any]: try: if service is None: return {"error": "Prometheus widget is missing its service"} base_url = str(service.config.get("base_url") or "").rstrip("/") timeout = int(service.config.get("timeout_seconds") or 10) promql = config.get("promql") if not promql: return {"error": "promql is required"} url = f"{base_url}/api/v1/query" response = await asyncio.wait_for( asyncio.to_thread( requests.get, url, params={"query": promql}, timeout=timeout, ), timeout=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 JellyfinWidgetSource: """Fetch Jellyfin sessions and map them to activity rows.""" async def fetch(self, service: ServiceRecord | None, widget_kind: str, config: dict[str, Any]) -> dict[str, Any]: timeout = 10 try: if service is None: return {"error": "Jellyfin widget is missing its service"} base_url = str(service.config.get("base_url") or "") api_key = str(service.secrets.get("api_key") or "") timeout = int(service.config.get("timeout_seconds") or 10) client = await asyncio.wait_for( asyncio.to_thread(JellyfinClient, base_url, api_key, timeout), timeout=timeout, ) sessions = await asyncio.wait_for( asyncio.to_thread(client.sessions), timeout=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 SshTaskWidgetSource: """Run a saved task on an SSH task runner instance and log the run.""" async def fetch(self, service: ServiceRecord | None, widget_kind: str, config: dict[str, Any]) -> dict[str, Any]: timeout = 30 try: if service is None: return {"error": "SSH task widget is missing its service"} store = get_settings_store() task_id = config.get("task_id") or "" 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"} client = _build_ssh_client(store, service) timeout = int(service.config.get("timeout_seconds") or 30) 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}"} start = time.perf_counter() result = await asyncio.wait_for( asyncio.to_thread(client.run, command, timeout), timeout=timeout, ) duration_ms = int((time.perf_counter() - start) * 1000) stdout = result.stdout or "" stderr = result.stderr or "" store.record_service_task_run( { "task_id": task_id, "service_id": service.id, "status": "success" if result.exit_status == 0 else "failure", "exit_status": result.exit_status, "duration_ms": duration_ms, "stdout_tail": stdout, "stderr_tail": stderr, "error": "" if result.exit_status == 0 else (stderr or stdout or "Task failed"), } ) return {"exit_status": result.exit_status, "stdout": stdout, "stderr": stderr} except asyncio.TimeoutError: _record_timeout(service, config, timeout) return {"error": "Widget data fetch timed out"} except Exception as exc: logger.exception("ssh_task adapter failed") store = get_settings_store() store.record_service_task_run( { "task_id": str(config.get("task_id") or ""), "service_id": service.id if service else "", "status": "error", "duration_ms": 0, "error": str(exc)[:1000], } ) return {"error": f"SSH task failed: {exc}"} def _record_timeout(service: ServiceRecord | None, config: dict[str, Any], timeout: int) -> None: try: store = get_settings_store() store.record_service_task_run( { "task_id": str(config.get("task_id") or ""), "service_id": service.id if service else "", "status": "timeout", "duration_ms": timeout * 1000, "error": f"Task timed out after {timeout}s", } ) except Exception: # pragma: no cover - logging best-effort logger.exception("failed to record ssh task timeout") def _build_ssh_client(store: SettingsStore, service: ServiceRecord) -> RemoteSSHClient: """Build an SSH client from an ssh_tasks service instance + referenced key.""" config = service.config host = str(config.get("host") or "").strip() username = str(config.get("username") or "").strip() if not host or not username: raise ValueError("SSH task service is missing host or username") settings = get_settings() private_key = "" key_passphrase = "" ssh_key_id = str(config.get("ssh_key_id") or "").strip() if ssh_key_id: ssh_key = store.get_ssh_key(ssh_key_id) if ssh_key: private_key = str(ssh_key.get("private_key") or "") key_passphrase = str(ssh_key.get("passphrase") or "") # Service-level passphrase secret takes precedence. key_passphrase = str(service.secrets.get("passphrase") or "") or key_passphrase return RemoteSSHClient( host=host, username=username, port=int(config.get("port") or 22), private_key=private_key or None, private_key_passphrase=key_passphrase or None, known_hosts_path=str(settings.ssh_known_hosts_file), timeout=int(config.get("timeout_seconds") or 30), ) # --------------------------------------------------------------------------- # Registries # --------------------------------------------------------------------------- SERVICE_ADAPTERS: dict[str, WidgetSource] = { "grafana": GrafanaWidgetSource(), "prometheus": PrometheusWidgetSource(), "jellyfin": JellyfinWidgetSource(), "ssh_tasks": SshTaskWidgetSource(), } BUILTIN_ADAPTERS: dict[str, WidgetSource] = { "backups": BackupsWidgetSource(), "static": StaticWidgetSource(), } def get_service_adapter(service_type: str) -> WidgetSource | None: return SERVICE_ADAPTERS.get(service_type) def get_builtin_adapter(kind: str) -> WidgetSource | None: return BUILTIN_ADAPTERS.get(kind)