"""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 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.domain.dashboard import ( _map_sessions_to_activity_rows, build_backup_dashboard_summary, ) from media_library_viewer_api.integrations.alertmanager import summarize_alerts from media_library_viewer_api.services.settings_store import SettingsStore, get_settings_store from media_library_viewer_api.services.task_runner import run_saved_task from media_library_viewer_api.widgets.prometheus_range import ( WINDOW_PRESETS, normalize_prometheus_matrix, step_for_window, ) 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 or query datasource for a chart.""" 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("/") api_key = str(service.secrets.get("api_key") or "") timeout = int(service.config.get("timeout_seconds") or 10) if widget_kind == "chart": return await self._fetch_chart(base_url, api_key, timeout, config) # Default: deep-link 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}"} async def _fetch_chart(self, base_url: str, api_key: str, timeout: int, config: dict[str, Any]) -> dict[str, Any]: """Query Grafana /api/ds/query and normalize to {series: [...]}.""" if not api_key: return {"error": "Grafana api_key is required for chart queries"} query = config.get("query", "") if not query: return {"error": "query is required"} datasource_uid = config.get("datasource_uid", "prometheus") body = { "queries": [ { "datasource": {"uid": datasource_uid, "type": "prometheus"}, "expr": query, "format": "time_series", "intervalMs": int(config.get("interval_ms", 30_000)), "maxDataPoints": int(config.get("max_data_points", 100)), "refId": "A", } ], "from": config.get("from_ts", "now-1h"), "to": config.get("to_ts", "now"), } def _do_post() -> dict[str, Any]: resp = requests.post( f"{base_url}/api/ds/query", json=body, headers={ "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", }, timeout=timeout, ) resp.raise_for_status() return resp.json() try: raw = await asyncio.wait_for(asyncio.to_thread(_do_post), timeout=timeout) except asyncio.TimeoutError: return {"error": "Grafana query timed out"} except requests.RequestException as exc: return {"error": f"Grafana query failed: {exc}"} # Normalize Grafana's /api/ds/query response into series. series: list[dict[str, Any]] = [] results = raw.get("results", {}) seen_labels: dict[str, int] = {} for ref_id, ref_data in results.items(): for frame in ref_data.get("frames", []): values = frame.get("data", {}).get("values", []) if len(values) < 2: continue timestamps = values[0] vals = values[1] # Derive a meaningful series label from the frame metadata. # Prometheus frames carry metric labels in schema.fields[-1].labels. fields = frame.get("schema", {}).get("fields", []) value_field = fields[-1] if fields else {} # Prefer displayName (explicitly set in Grafana), then Prometheus # labels (e.g. {instance: "server:9100", mode: "iowait"}), then # the field name as a last resort. display_name = value_field.get("config", {}).get("displayName") or value_field.get("displayName") frame_labels = value_field.get("labels") or {} if display_name: label = str(display_name) elif frame_labels: # Build a readable label from the Prometheus labels, excluding # redundant ones like __name__. parts = [f"{k}={v}" for k, v in sorted(frame_labels.items()) if not k.startswith("__")] label = " ".join(parts) if parts else "value" else: label = value_field.get("name", "value") # Ensure unique labels when multiple series share the same name. if label in seen_labels: seen_labels[label] += 1 label = f"{label} ({seen_labels[label]})" else: seen_labels[label] = 0 points = [{"t": int(t), "v": float(v) if v is not None else None} for t, v in zip(timestamps, vals)] series.append({"label": label, "points": points}) return {"series": series} class PrometheusWidgetSource: """Run PromQL queries against a Prometheus service (instant + range).""" 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) if widget_kind == "chart": return await self._fetch_chart(base_url, timeout, config) # Default: instant-query metric path (unchanged). 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}"} async def _range_query(self, base_url: str, timeout: int, promql: str, window: int) -> dict[str, Any]: """Run a Prometheus ``/api/v1/query_range`` over a window (seconds). Shared by the ``chart`` (SC-101) and ``mean`` widget kinds. Returns ``{"matrix": result}`` on success or ``{"error": str}`` (never raises, per SC-103). """ step = step_for_window(window) end = int(time.time()) start = end - window try: response = await asyncio.wait_for( asyncio.to_thread( requests.get, f"{base_url}/api/v1/query_range", params={"query": promql, "start": start, "end": end, "step": step}, timeout=timeout, ), timeout=timeout, ) response.raise_for_status() payload = response.json() except asyncio.TimeoutError: return {"error": "Prometheus query timed out"} except requests.RequestException as exc: logger.exception("prometheus range query failed") return {"error": f"Prometheus query failed: {exc}"} result = payload.get("data", {}).get("result", []) return {"matrix": result} async def _fetch_chart(self, base_url: str, timeout: int, config: dict[str, Any]) -> dict[str, Any]: """Range query → ``{series}`` for the chart widget (SC-101..SC-104).""" promql = config.get("promql") if not promql: return {"error": "promql is required"} window = WINDOW_PRESETS.get(config.get("window", "1h"), WINDOW_PRESETS["1h"]) raw = await self._range_query(base_url, timeout, promql, window) if "error" in raw: return raw return {"series": normalize_prometheus_matrix(raw["matrix"])} class AlertmanagerWidgetSource: """Fetch firing alerts from an Alertmanager service and summarize them.""" async def fetch(self, service: ServiceRecord | None, widget_kind: str, config: dict[str, Any]) -> dict[str, Any]: try: if service is None: return {"error": "Alertmanager widget is missing its service"} base_url = str(service.config.get("base_url") or "").rstrip("/") timeout = int(service.config.get("timeout_seconds") or 5) severity_filter = config.get("severity_filter") or None headers: dict[str, str] = {} api_key = str(service.secrets.get("api_key") or "") if api_key: headers["Authorization"] = f"Bearer {api_key}" response = await asyncio.wait_for( asyncio.to_thread( requests.get, f"{base_url}/api/v1/alerts", headers=headers, timeout=timeout, ), timeout=timeout, ) response.raise_for_status() payload = response.json() alerts = payload.get("data", []) if isinstance(payload, dict) else [] return summarize_alerts(alerts, severity_filter=severity_filter) except asyncio.TimeoutError: return {"error": "Widget data fetch timed out"} except requests.RequestException as exc: logger.exception("alertmanager adapter failed") return {"error": f"Alertmanager query failed: {exc}"} except Exception as exc: logger.exception("alertmanager adapter failed") return {"error": f"Alertmanager 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, ) if widget_kind == "now_playing": sessions = [ s for s in sessions if s.get("NowPlayingItem") and not s.get("PlayState", {}).get("IsPaused", True) ] 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"} timeout = int(service.config.get("timeout_seconds") or 30) result = await asyncio.wait_for( asyncio.to_thread(run_saved_task, store, task, service), timeout=timeout, ) return {"exit_status": result.exit_status, "stdout": result.stdout, "stderr": result.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") 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") # --------------------------------------------------------------------------- # Registries # --------------------------------------------------------------------------- SERVICE_ADAPTERS: dict[str, WidgetSource] = { "grafana": GrafanaWidgetSource(), "prometheus": PrometheusWidgetSource(), "alertmanager": AlertmanagerWidgetSource(), "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)