import html from datetime import datetime from urllib.parse import quote from fastapi import APIRouter, Depends, HTTPException, Query, Request from fastapi.responses import HTMLResponse, PlainTextResponse from app.auth import require_user from app.db.apps import ( get_deployment, get_deployment_filter_options, get_deployment_stats, get_deployments, ) from app.templates.layout import page router = APIRouter() SUCCESS_STATUSES = {"ok", "success", "succeeded", "done", "deployed", "completed"} RUNNING_STATUSES = {"running", "pending", "queued", "in_progress", "starting"} FAILED_STATUSES = {"failed", "failure", "error", "cancelled", "canceled"} def normalize_status(status: str | None) -> str: return (status or "").strip().lower() def is_running(status: str | None) -> bool: return normalize_status(status) in RUNNING_STATUSES def is_failed(status: str | None) -> bool: return normalize_status(status) in FAILED_STATUSES def render_status_pill(status: str | None) -> str: status_value = status or "" normalized = normalize_status(status) if normalized in SUCCESS_STATUSES: class_name = "pill pill-success" elif normalized in RUNNING_STATUSES: class_name = "pill pill-warning" elif normalized in FAILED_STATUSES: class_name = "pill pill-danger" else: class_name = "pill pill-muted" return f'{html.escape(status_value)}' def parse_timestamp(value: str | None) -> datetime | None: if not value: return None normalized = value.strip().replace("Z", "+00:00") formats = [ "%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M:%S.%f", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%dT%H:%M:%S.%f", ] try: return datetime.fromisoformat(normalized) except ValueError: pass for fmt in formats: try: return datetime.strptime(normalized, fmt) except ValueError: continue return None def calculate_duration(started_at: str | None, finished_at: str | None) -> str: if not started_at or not finished_at: return "" started = parse_timestamp(started_at) finished = parse_timestamp(finished_at) if not started or not finished: return "" seconds = int((finished - started).total_seconds()) if seconds < 0: return "" minutes, seconds = divmod(seconds, 60) hours, minutes = divmod(minutes, 60) if hours: return f"{hours}h {minutes}m {seconds}s" if minutes: return f"{minutes}m {seconds}s" return f"{seconds}s" def actor_label(deployment: dict) -> str: display_name = deployment.get("triggered_by_display_name") or "" username = deployment.get("triggered_by_username") or "" if display_name and username: return f"{display_name} (@{username})" return display_name or username or "" def render_options(values: list[str], selected: str | None, empty_label: str) -> str: options = [f''] for value in values: selected_attr = " selected" if selected == value else "" options.append(f'') return "".join(options) @router.get("/deployments", response_class=HTMLResponse) async def deployments_page( request: Request, status: str = Query(""), app_id: str = Query(""), source: str = Query(""), user=Depends(require_user), ): selected_status = status.strip() selected_app = app_id.strip() selected_source = source.strip() deployments = get_deployments( status=selected_status or None, app_id=selected_app or None, source=selected_source or None, ) stats = get_deployment_stats() options = get_deployment_filter_options() refresh = "" if any( is_running(item.get("status")) for item in deployments ) else "" rows = "" for deployment in deployments: deployment_id = html.escape(str(deployment.get("id", ""))) raw_app_id = deployment.get("app_id", "") or "" escaped_app_id = html.escape(raw_app_id) app_url_id = quote(raw_app_id, safe="") kind = html.escape(deployment.get("kind", "") or "") status_pill = render_status_pill(deployment.get("status")) started_at = html.escape(deployment.get("started_at", "") or "") finished_at = html.escape(deployment.get("finished_at", "") or "") source_label = html.escape(deployment.get("trigger_source", "") or "") triggered_by = html.escape(actor_label(deployment)) returncode = deployment.get("returncode") returncode_label = "" if returncode is None else html.escape(str(returncode)) row_class = ' class="running-row"' if is_running(deployment.get("status")) else "" rows += f"""
| ID | Aplikace | Typ | Status | Zdroj | Spustil | Kód | Dokončeno | Akce |
|---|
Detail běhu nasazení včetně oddělených výstupů stdout a stderr.
{failed_notice}| Aplikace | {app_id} |
|---|---|
| Typ | {kind} |
| Status | {status} |
| Trigger source | {trigger_source} |
| Triggered by | {triggered_by} |
| Commit author | {commit_author} |
| Pusher | {pusher} |
| Spuštěno | {started_at} |
| Dokončeno | {finished_at} |
| Duration | {duration} |
| Návratový kód | {returncode_label} |
{stdout}
{stderr}