import html import json from urllib.parse import quote from fastapi import APIRouter, Depends, HTTPException, Request from fastapi.responses import HTMLResponse from app.auth import require_user from app.db.audit import log_audit_event from app.db.workers import get_worker, get_workers from app.templates.layout import page router = APIRouter() def render_worker_badge(online: bool) -> str: if online: return 'ONLINE' return 'OFFLINE' def pretty_json(value: str | None) -> str: if not value: return "{}" try: return json.dumps(json.loads(value), ensure_ascii=False, indent=2, sort_keys=True) except (TypeError, ValueError): return value @router.get("/workers", response_class=HTMLResponse) def workers_page(request: Request, user=Depends(require_user)): workers = get_workers() online_count = sum(1 for worker in workers if worker.get("online")) total_count = len(workers) offline_count = total_count - online_count log_audit_event( user, action="workers.view", target_type="workers", metadata={ "online": online_count, "offline": offline_count, "total": total_count, }, ) rows = "" for worker in workers: worker_id_raw = worker.get("worker_id", "") or "" worker_id = html.escape(worker_id_raw) worker_url_id = quote(worker_id_raw, safe="") status = html.escape(worker.get("status", "") or "") last_seen_at = html.escape(worker.get("last_seen_at", "") or "") started_at = html.escape(worker.get("started_at", "") or "") badge = render_worker_badge(bool(worker.get("online"))) current_job_id = worker.get("current_job_id") current_job = "" if current_job_id is not None: current_job_label = html.escape(str(current_job_id)) current_job = f'#{current_job_label}' rows += f"""
Přehled worker procesů obsluhujících AppFactory joby.
| Worker ID | Status | Last Seen | Current Job | Started At | Akce |
|---|
{metadata}
| Worker ID | {worker_id_html} |
|---|---|
| Status | {badge} {status} |
| Last Seen | {last_seen_at} |
| Current Job | {current_job} |
| Started At | {started_at} |