import html from urllib.parse import quote from fastapi import APIRouter, Depends, Request from fastapi.responses import HTMLResponse from app.auth import require_user from app.db.audit import log_audit_event from app.db.operations import ( get_operations_summary, get_recent_audit_events, get_recent_deployments, get_recent_jobs, ) from app.routes.deployments import render_status_pill from app.routes.jobs import render_job_status from app.templates.layout import page router = APIRouter() 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 @router.get("/operations", response_class=HTMLResponse) def operations_dashboard(request: Request, user=Depends(require_user)): summary = get_operations_summary() log_audit_event( user, action="operations.dashboard.view", target_type="operations", metadata=summary, ) recent_jobs = get_recent_jobs() recent_deployments = get_recent_deployments() recent_audit_events = get_recent_audit_events() return page( "Operations", f"""

Operations Dashboard

Provozní přehled workerů, jobů, deploymentů a auditních událostí.

Workers Online{summary.get("workers_online") or 0}
Running Jobs{summary.get("running_jobs") or 0}
Queued Jobs{summary.get("queued_jobs") or 0}
Failed Jobs (24h){summary.get("failed_jobs_24h") or 0}
Deployments Today{summary.get("deployments_today") or 0}
Active Applications{summary.get("active_applications") or 0}

Recent Jobs

{render_recent_jobs(recent_jobs)}
ID Status Type Target Source Created At

Recent Deployments

{render_recent_deployments(recent_deployments)}
ID Aplikace Typ Status Zdroj Spuštěno Dokončeno

Recent Audit Events

{render_recent_audit_events(recent_audit_events)}
Čas Uživatel Akce Typ cíle Cíl Zdroj
""", user=user, )