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
| ID |
Status |
Type |
Target |
Source |
Created At |
{render_recent_jobs(recent_jobs)}
Recent Deployments
| ID |
Aplikace |
Typ |
Status |
Zdroj |
Spuštěno |
Dokončeno |
{render_recent_deployments(recent_deployments)}
Recent Audit Events
| Čas |
Uživatel |
Akce |
Typ cíle |
Cíl |
Zdroj |
{render_recent_audit_events(recent_audit_events)}
""",
user=user,
)