import asyncio
import html
from urllib.parse import quote
from fastapi import APIRouter, Depends, Request, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse
from app.auth import current_user, require_user
from app.db.audit import log_audit_event
from app.db.operations import (
get_operations_snapshot,
)
from app.routes.deployments import render_status_pill
from app.routes.incidents import render_incident_history_rows, render_open_incident_rows
from app.routes.jobs import render_job_status
from app.templates.layout import page
router = APIRouter()
@router.websocket("/ws/operations")
async def operations_websocket(websocket: WebSocket):
user = current_user(websocket)
if not user:
await websocket.close(code=1008)
return
await websocket.accept()
try:
while True:
await websocket.send_json(get_operations_snapshot())
await asyncio.sleep(2)
except WebSocketDisconnect:
return
def render_recent_jobs(jobs: list[dict]) -> str:
rows = ""
for job in jobs:
job_id = html.escape(str(job.get("id", "")))
job_type = html.escape(job.get("type", "") or "")
target_type = html.escape(job.get("target_type", "") or "")
target_id = html.escape(job.get("target_id", "") or "")
source = html.escape(job.get("source", "") or "")
created_at = html.escape(job.get("created_at", "") or "")
rows += f"""
| #{job_id} |
{render_job_status(job.get("status"))} |
{job_type} |
{target_type}: {target_id} |
{source} |
{created_at} |
"""
if not rows:
rows = '| Zatím nejsou evidované žádné joby. |
'
return rows
def render_recent_deployments(deployments: list[dict]) -> str:
rows = ""
for deployment in deployments:
deployment_id = html.escape(str(deployment.get("id", "")))
raw_app_id = deployment.get("app_id", "") or ""
app_id = html.escape(raw_app_id)
app_url_id = quote(raw_app_id, safe="")
kind = html.escape(deployment.get("kind", "") or "")
started_at = html.escape(deployment.get("started_at", "") or "")
finished_at = html.escape(deployment.get("finished_at", "") or "")
source = html.escape(deployment.get("trigger_source", "") or "")
rows += f"""
| #{deployment_id} |
{app_id} |
{kind} |
{render_status_pill(deployment.get("status"))} |
{source} |
{started_at} |
{finished_at} |
"""
if not rows:
rows = '| Zatím nejsou evidovaná žádná nasazení. |
'
return rows
def render_recent_audit_events(events: list[dict]) -> str:
rows = ""
for event in events:
rows += f"""
| {html.escape(event.get("created_at", "") or "")} |
{html.escape(event.get("username", "") or "")} |
{html.escape(event.get("action", "") or "")} |
{html.escape(event.get("target_type", "") or "")} |
{html.escape(str(event.get("target_id") or ""))} |
{html.escape(event.get("source", "") or "")} |
"""
if not rows:
rows = '| Zatím nejsou evidované žádné auditní události. |
'
return rows
def render_health_status(status: str | None) -> str:
value = status or ""
normalized = value.lower()
labels = {
"healthy": "zdravá",
"unhealthy": "nezdravá",
"unreachable": "nedostupná",
}
class_name = "pill pill-muted"
if normalized == "healthy":
class_name = "pill pill-success"
elif normalized == "unhealthy":
class_name = "pill pill-warning"
elif normalized == "unreachable":
class_name = "pill pill-danger"
return f'{html.escape(labels.get(normalized, value))}'
def render_health_problems(items: list[dict]) -> str:
rows = ""
for item in items:
service_id = html.escape(item.get("service_id", "") or "")
service_name = html.escape(item.get("service_name") or item.get("service_id", "") or "")
error_text = item.get("error_text", "") or ""
error_preview = error_text if len(error_text) <= 120 else f"{error_text[:117]}..."
rows += f"""
| {service_name} |
{render_health_status(item.get("status"))} |
{html.escape(item.get("checked_at", "") or "")} |
{html.escape(str(item.get("response_time_ms") or ""))} |
{html.escape(error_preview)} |
"""
if not rows:
rows = '| Žádné služby nevyžadují pozornost. |
'
return rows
@router.get("/operations", response_class=HTMLResponse)
def operations_dashboard(request: Request, user=Depends(require_user)):
snapshot = get_operations_snapshot()
log_audit_event(
user,
action="operations.dashboard.view",
target_type="operations",
metadata={
"jobs": {
"queued": snapshot["jobs"]["queued"],
"running": snapshot["jobs"]["running"],
"failed_24h": snapshot["jobs"]["failed_24h"],
},
"workers": {
"online": snapshot["workers"]["online"],
"offline": snapshot["workers"]["offline"],
},
"deployments": {
"running": snapshot["deployments"]["running"],
"failed_24h": snapshot["deployments"]["failed_24h"],
},
"incidents": {
"open": snapshot["incidents"]["open"],
},
},
)
return page(
"Přehled",
f"""
Přehled systému
Rychlá odpověď na otázku, zda jsou služby, úlohy a nasazení v pořádku.
Běžící služby{snapshot["apps"]["active"]}
Problémové služby{snapshot["apps"]["problematic"]}
Zdravé služby{snapshot["health"]["healthy"]}
Nezdravé služby{snapshot["health"]["unhealthy"]}
Nedostupné služby{snapshot["health"]["unreachable"]}
Aktivní incidenty{snapshot["incidents"]["open"]}
Aktivní úlohy{snapshot["jobs"]["running"]}
Nasazení dnes{snapshot["deployments"]["today"]}
Online workery{snapshot["workers"]["online"]}
Selhané úlohy (24 h){snapshot["jobs"]["failed_24h"]}
Služby vyžadující pozornost
| Služba |
Status |
Poslední kontrola |
Odezva |
Chyba |
{render_health_problems(snapshot["health"]["problems"])}
Aktivní incidenty
| Služba |
Název |
Začátek |
Trvání |
{render_open_incident_rows(snapshot["incidents"]["active"])}
Poslední neúspěšná nasazení
| ID |
Služba |
Typ |
Status |
Zdroj |
Spuštěno |
Dokončeno |
{render_recent_deployments(snapshot["deployments"]["recent_failed"])}
Poslední incidenty
| Služba |
Název |
Začátek |
Konec |
Trvání |
Stav |
{render_incident_history_rows(snapshot["incidents"]["recent"])}
Poslední auditní události
| Čas |
Uživatel |
Akce |
Typ cíle |
Cíl |
Zdroj |
{render_recent_audit_events(snapshot["audit"]["recent"])}
""",
user=user,
)