import html import math from datetime import datetime from urllib.parse import quote, urlencode 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 ( count_deployments, 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"} DEFAULT_PAGE_SIZE = 20 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" labels = { "running": "běží", "pending": "čeká", "queued": "ve frontě", "in_progress": "probíhá", "starting": "startuje", "success": "úspěšné", "ok": "úspěšné", "succeeded": "úspěšné", "done": "hotovo", "deployed": "nasazeno", "completed": "dokončeno", "failed": "selhalo", "failure": "selhalo", "error": "chyba", "cancelled": "zrušeno", "canceled": "zrušeno", } return f'{html.escape(labels.get(normalized, 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) def pagination_url(page_number: int, status: str, app_id: str, source: str) -> str: params = {"page": page_number} if status: params["status"] = status if app_id: params["app_id"] = app_id if source: params["source"] = source return f"/portal/deployments?{urlencode(params)}" @router.get("/deployments", response_class=HTMLResponse) async def deployments_page( request: Request, status: str = Query(""), app_id: str = Query(""), source: str = Query(""), page_number: int = Query(1, alias="page", ge=1), user=Depends(require_user), ): selected_status = status.strip() selected_app = app_id.strip() selected_source = source.strip() offset = (page_number - 1) * DEFAULT_PAGE_SIZE total_deployments = count_deployments( status=selected_status or None, app_id=selected_app or None, source=selected_source or None, ) total_pages = max(1, math.ceil(total_deployments / DEFAULT_PAGE_SIZE)) if page_number > total_pages: page_number = total_pages offset = (page_number - 1) * DEFAULT_PAGE_SIZE deployments = get_deployments( limit=DEFAULT_PAGE_SIZE, offset=offset, 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 = "" 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}| Služba | {app_id} |
|---|---|
| Typ | {kind} |
| Status | {status} |
| Zdroj spuštění | {trigger_source} |
| Spustil | {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}