Developer
459 lines
19 KiB
Python
459 lines
19 KiB
Python
"""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.clients.qbittorrent import QbittorrentClient
|
|
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.qbittorrent_store import QbittorrentSampleStore
|
|
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 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)
|
|
if widget_kind == "gauge":
|
|
return await self._fetch_gauge(base_url, timeout, config)
|
|
if widget_kind == "mean":
|
|
return await self._fetch_mean(base_url, timeout, config)
|
|
# Default: instant-query metric path (unchanged).
|
|
raw = await self._instant_query(base_url, timeout, config.get("promql", ""))
|
|
return raw
|
|
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 _instant_query(self, base_url: str, timeout: int, promql: str) -> dict[str, Any]:
|
|
"""Run a Prometheus ``/api/v1/query`` instant query.
|
|
|
|
Shared by the ``metric`` and ``gauge`` widget kinds. Returns
|
|
``{"result": data}`` on success or ``{"error": str}`` (never raises,
|
|
per SC-103).
|
|
"""
|
|
if not promql:
|
|
return {"error": "promql is required"}
|
|
try:
|
|
response = await asyncio.wait_for(
|
|
asyncio.to_thread(
|
|
requests.get,
|
|
f"{base_url}/api/v1/query",
|
|
params={"query": promql},
|
|
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 instant query failed")
|
|
return {"error": f"Prometheus query failed: {exc}"}
|
|
return {"result": payload.get("data", {})}
|
|
|
|
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"])}
|
|
|
|
async def _fetch_gauge(self, base_url: str, timeout: int, config: dict[str, Any]) -> dict[str, Any]:
|
|
"""Instant query → scalar for the gauge widget (SC-109, SC-110, SC-111).
|
|
|
|
Scalar-only: a multi-series query returns an error (SC-111). Threshold
|
|
config (``warn_at``/``crit_at``/``min``/``max``/``unit``) is passed
|
|
through for the frontend renderer.
|
|
"""
|
|
raw = await self._instant_query(base_url, timeout, config.get("promql") or "")
|
|
if "error" in raw:
|
|
return raw
|
|
result = raw["result"].get("result", [])
|
|
if len(result) != 1:
|
|
return {"error": "Gauge requires a single-series query; refine your PromQL"}
|
|
try:
|
|
value = float(result[0]["value"][1])
|
|
except (KeyError, IndexError, ValueError, TypeError):
|
|
return {"error": "Gauge query returned no scalar value"}
|
|
return {
|
|
"value": value,
|
|
"warn_at": config.get("warn_at"),
|
|
"crit_at": config.get("crit_at"),
|
|
"min": config.get("min"),
|
|
"max": config.get("max"),
|
|
"unit": config.get("unit"),
|
|
}
|
|
|
|
async def _fetch_mean(self, base_url: str, timeout: int, config: dict[str, Any]) -> dict[str, Any]:
|
|
"""Range query → client-side mean for the mean widget (SC-112..SC-114).
|
|
|
|
Runs ``query_range`` over the configured window preset, averages all
|
|
non-null numeric samples of the single series, and returns a scalar.
|
|
Scalar-only: a multi-series query returns an error (SC-114).
|
|
"""
|
|
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
|
|
result = raw["matrix"]
|
|
if len(result) != 1:
|
|
return {"error": "Mean requires a single-series query; refine your PromQL"}
|
|
points = result[0].get("values") or []
|
|
nums: list[float] = []
|
|
for _, v in points:
|
|
if v in (None, "NaN", "+Inf", "-Inf"):
|
|
continue
|
|
try:
|
|
nums.append(float(v))
|
|
except (TypeError, ValueError):
|
|
continue
|
|
if not nums:
|
|
return {"error": "Mean query returned no numeric samples in the window"}
|
|
mean = sum(nums) / len(nums)
|
|
return {"value": mean, "unit": config.get("unit")}
|
|
|
|
|
|
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")
|
|
|
|
|
|
class QbittorrentWidgetSource:
|
|
"""Fetch qBittorrent data for totals, active, and speed widgets."""
|
|
|
|
async def fetch(self, service: ServiceRecord | None, widget_kind: str, config: dict[str, Any]) -> dict[str, Any]:
|
|
try:
|
|
if service is None:
|
|
return {"error": "qBittorrent widget is missing its service"}
|
|
base_url = str(service.config.get("base_url") or "")
|
|
username = str(service.secrets.get("username") or "")
|
|
password = str(service.secrets.get("password") or "")
|
|
timeout = int(service.config.get("timeout_seconds") or 10)
|
|
if not base_url or not username or not password:
|
|
return {"error": "qBittorrent service is missing base_url, username, or password"}
|
|
|
|
client = QbittorrentClient(base_url, username, password, timeout)
|
|
data = await asyncio.wait_for(asyncio.to_thread(client.maindata), timeout=timeout)
|
|
server_state = data.get("server_state", {})
|
|
torrents = data.get("torrents", {})
|
|
|
|
if widget_kind == "totals":
|
|
by_state: dict[str, int] = {}
|
|
for t in torrents.values():
|
|
state = str(t.get("state", "unknown"))
|
|
by_state[state] = by_state.get(state, 0) + 1
|
|
return {"total": len(torrents), "by_state": by_state}
|
|
|
|
if widget_kind == "active":
|
|
active = [
|
|
{
|
|
"name": t.get("name"),
|
|
"state": t.get("state"),
|
|
"size": t.get("size"),
|
|
"progress": t.get("progress"),
|
|
"dl_speed": t.get("dlspeed"),
|
|
"up_speed": t.get("upspeed"),
|
|
}
|
|
for t in torrents.values()
|
|
if str(t.get("state", "")) in {"downloading", "uploading"}
|
|
]
|
|
return {"torrents": active}
|
|
|
|
if widget_kind == "speed":
|
|
dl = int(server_state.get("dl_info_speed", 0))
|
|
up = int(server_state.get("up_info_speed", 0))
|
|
ts = int(time.time())
|
|
store = QbittorrentSampleStore()
|
|
store.append(service.id, ts, dl, up)
|
|
samples = store.window(service.id)
|
|
series = [
|
|
{"label": "download", "points": [{"t": s["ts"] * 1000, "v": s["dl_speed"]} for s in samples]},
|
|
{"label": "upload", "points": [{"t": s["ts"] * 1000, "v": s["up_speed"]} for s in samples]},
|
|
]
|
|
return {"series": series}
|
|
|
|
return {"error": f"Unknown qBittorrent widget kind: {widget_kind}"}
|
|
except asyncio.TimeoutError:
|
|
return {"error": "qBittorrent data fetch timed out"}
|
|
except Exception as exc:
|
|
logger.exception("qbittorrent adapter failed")
|
|
return {"error": f"qBittorrent fetch failed: {exc}"}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Registries
|
|
# ---------------------------------------------------------------------------
|
|
|
|
SERVICE_ADAPTERS: dict[str, WidgetSource] = {
|
|
"prometheus": PrometheusWidgetSource(),
|
|
"qbittorrent": QbittorrentWidgetSource(),
|
|
"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)
|