Developer
1cd8e926de
PR 2 of 4 for configurable dashboard widgets.
- Add grafana_url and prometheus_url settings (config.py + compose/env).
- Create WidgetSource protocol and adapters for jellyfin, backups, grafana,
prometheus, ssh_task, and static sources.
- Add GET /api/widgets/instances/{id}/data endpoint.
- Extract shared dashboard helpers into domain/dashboard.py so widgets and
the dashboard router reuse the same logic.
- Add adapter and data-endpoint tests.
- Update apply-progress.md.
Verification: ruff clean; backend pytest 200 passed; frontend lint/build green.
92 lines
3.2 KiB
Python
92 lines
3.2 KiB
Python
"""Dashboard domain helpers shared between routers and widget adapters."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from typing import Any
|
|
|
|
from media_library_viewer_api.models.backups import BackupDashboardSummary
|
|
from media_library_viewer_api.services.settings_store import SettingsStore
|
|
|
|
|
|
def _map_sessions_to_activity_rows(sessions: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
"""Normalize Jellyfin sessions into dashboard activity rows."""
|
|
results: list[dict[str, Any]] = []
|
|
for session in sessions:
|
|
item = session.get("NowPlayingItem") or {}
|
|
play_state = session.get("PlayState") or {}
|
|
transcoding = session.get("TranscodingInfo") or {}
|
|
|
|
has_item = bool(item)
|
|
series = item.get("SeriesName") or ""
|
|
title = (
|
|
(f"{series} - {item.get('Name', '')}" if series else item.get("Name", "Unknown"))
|
|
if has_item
|
|
else "(idle)"
|
|
)
|
|
|
|
if not has_item:
|
|
state_label = "idle"
|
|
else:
|
|
state_label = "paused" if play_state.get("IsPaused") else "playing"
|
|
|
|
is_transcoding = bool(transcoding)
|
|
transcode_type: list[str] = []
|
|
if is_transcoding:
|
|
if transcoding.get("IsVideoDirect") is False:
|
|
transcode_type.append("video")
|
|
if transcoding.get("IsAudioDirect") is False:
|
|
transcode_type.append("audio")
|
|
if not transcode_type:
|
|
transcode_type.append("active")
|
|
|
|
results.append(
|
|
{
|
|
"user": session.get("UserName") or "Unknown",
|
|
"title": title,
|
|
"type": item.get("Type", "") if has_item else "",
|
|
"state": state_label,
|
|
"transcoding": "yes" if is_transcoding else "no",
|
|
"transcoding_type": ", ".join(transcode_type),
|
|
"device": session.get("DeviceName") or session.get("Client") or "",
|
|
"session_id": session.get("Id") or "",
|
|
}
|
|
)
|
|
return results
|
|
|
|
|
|
def build_backup_dashboard_summary(store: SettingsStore) -> BackupDashboardSummary:
|
|
"""Compute the backup summary shown on the dashboard."""
|
|
jobs = store.list_backup_jobs()
|
|
total_jobs = len(jobs)
|
|
|
|
cutoff = int(time.time()) - (24 * 60 * 60)
|
|
recent_runs = []
|
|
for job in jobs:
|
|
runs = store.list_backup_runs(job_id=job["id"], limit=1)
|
|
if runs and runs[0]["started_at"] >= cutoff:
|
|
recent_runs.append(runs[0])
|
|
|
|
successful = sum(1 for r in recent_runs if r["status"] == "success")
|
|
success_rate = (successful / len(recent_runs) * 100) if recent_runs else 100.0
|
|
|
|
alerts = store.list_backup_alerts(acknowledged=False)
|
|
active_alerts = len(alerts)
|
|
|
|
failed_runs = []
|
|
for job in jobs:
|
|
runs = store.list_backup_runs(job_id=job["id"], status="failure", limit=1)
|
|
if runs:
|
|
failed_runs.append(runs[0])
|
|
|
|
last_failed_at = None
|
|
if failed_runs:
|
|
last_failed_at = max(r["started_at"] for r in failed_runs)
|
|
|
|
return BackupDashboardSummary(
|
|
total_jobs=total_jobs,
|
|
success_rate_24h=round(success_rate, 1),
|
|
active_alerts=active_alerts,
|
|
last_failed_at=last_failed_at,
|
|
)
|