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" 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) 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""" #{deployment_id}
{started_at} {escaped_app_id} {kind} {status_pill} {source_label} {triggered_by} {returncode_label} {finished_at} Detail """ if not rows: rows = 'Zatím nejsou evidovaná žádná nasazení.' status_values = ["running", "success", "failed"] status_options = render_options(status_values, selected_status, "Všechny statusy") app_options = render_options(options["apps"], selected_app, "Všechny aplikace") source_options = render_options(options["sources"], selected_source, "Všechny zdroje") first_item = offset + 1 if total_deployments else 0 last_item = min(offset + len(deployments), total_deployments) previous_disabled = " disabled" if page_number <= 1 else "" next_disabled = " disabled" if page_number >= total_pages else "" previous_href = pagination_url(max(1, page_number - 1), selected_status, selected_app, selected_source) next_href = pagination_url(min(total_pages, page_number + 1), selected_status, selected_app, selected_source) pagination = f""" """ return page( "Nasazení", f""" {refresh}

Historie nasazení

Přehled posledních běhů nasazení a jejich výsledků.

← Zpět na portál
Total deploys{stats.get("total") or 0}
Failed deploys{stats.get("failed") or 0}
Running deploys{stats.get("running") or 0}
Success rate{stats.get("success_rate") or 0}%

Filtry

Reset

Nasazení

{pagination} {rows}
ID Aplikace Typ Status Zdroj Spustil Kód Dokončeno Akce
{pagination}
""", user=user, ) @router.get("/deployments/{deployment_id}/logs/raw", response_class=PlainTextResponse) async def deployment_raw_logs( deployment_id: int, request: Request, user=Depends(require_user), ): deployment = get_deployment(deployment_id) if not deployment: raise HTTPException(status_code=404, detail="Deployment not found") stdout = deployment.get("stdout", "") or "" stderr = deployment.get("stderr", "") or "" return PlainTextResponse(f"--- stdout ---\n{stdout}\n\n--- stderr ---\n{stderr}") @router.get("/deployments/{deployment_id}", response_class=HTMLResponse) async def deployment_detail_page( deployment_id: int, request: Request, user=Depends(require_user), ): deployment = get_deployment(deployment_id) if not deployment: raise HTTPException(status_code=404, detail="Deployment not found") title = f"Nasazení #{html.escape(str(deployment.get('id', deployment_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 "") status = render_status_pill(deployment.get("status")) started_at = html.escape(deployment.get("started_at", "") or "") finished_at = html.escape(deployment.get("finished_at", "") or "") duration = html.escape(calculate_duration(deployment.get("started_at"), deployment.get("finished_at"))) trigger_source = html.escape(deployment.get("trigger_source", "") or "") triggered_by = html.escape(actor_label(deployment)) commit_author = html.escape(deployment.get("commit_author", "") or "") pusher = html.escape(deployment.get("pusher", "") or "") returncode = deployment.get("returncode") returncode_label = "" if returncode is None else html.escape(str(returncode)) stdout = html.escape(deployment.get("stdout", "") or "") stderr = html.escape(deployment.get("stderr", "") or "") refresh = "" failed_notice = ( f'
Deployment selhal. Návratový kód: {returncode_label}
' if is_failed(deployment.get("status")) else "" ) return page( title, f""" {refresh}

{title}

Detail běhu nasazení včetně oddělených výstupů stdout a stderr.

{failed_notice}

← Zpět na nasazení Detail aplikace Raw logy

Souhrn

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

{stdout}

stderr

{stderr}
""", user=user, )