Files
appfactory-portal/app/routes/apps.py
T
JiriUhlir ed194093e6 Přidal jsem Health Monitoring do portálu:
Přehled má nové karty: Zdravé služby, Nezdravé služby, Nedostupné služby.
Přehled zobrazuje sekci Služby vyžadující pozornost pro unhealthy a unreachable.
Snapshot /portal/ws/operations teď obsahuje health data a stránka je aktualizuje přes existující realtime vrstvu.
Seznam služeb zobrazuje aktuální health status a poslední kontrolu, včetně řazení Podle zdraví.
Detail služby má sekce Zdraví služby a Historie kontrol s posledními 50 záznamy.
Přidal jsem audit event service.health.view.
Browser title teď používá formát CSBot Services Portal - ..., takže detail služby odpovídá požadavku.
Nepoužil jsem ORM, vše je přes SQLite dotazy.
2026-05-29 11:57:57 +02:00

637 lines
23 KiB
Python

import html
import math
from urllib.parse import quote, urlencode
from fastapi import APIRouter, Depends, Form, HTTPException, Query, Request
from fastapi.responses import HTMLResponse, RedirectResponse
from ..auth import require_user
from ..config import (
DEFAULT_APPFACTORY_HOST,
DEFAULT_GITEA_ORG,
DELETE_APP_SCRIPT,
DEPLOY_SCRIPT,
GENERATE_COMPOSE_SCRIPT,
NEW_APP_SCRIPT,
read_env_value,
)
from ..db.apps import get_app, get_app_deployments, get_apps, update_app_resources
from ..db.audit import log_audit_event
from ..db.health import get_latest_service_health, get_service_health, get_service_health_history
from ..db.jobs import create_job, get_jobs, has_active_deploy_job
from ..routes.deployments import render_status_pill
from ..shell import run_command
from ..templates.layout import page, render_result
router = APIRouter()
DEFAULT_PAGE_SIZE = 20
def render_health_status(status: str | None) -> str:
value = status or ""
normalized = value.lower()
labels = {
"healthy": "zdravá",
"unhealthy": "nezdravá",
"unreachable": "nedostupná",
}
class_name = "pill pill-muted"
if normalized == "healthy":
class_name = "pill pill-success"
elif normalized == "unhealthy":
class_name = "pill pill-warning"
elif normalized == "unreachable":
class_name = "pill pill-danger"
return f'<span class="{class_name}">{html.escape(labels.get(normalized, value or "neznámá"))}</span>'
@router.get("/")
def portal_home(user=Depends(require_user)):
return RedirectResponse(url="/portal/operations", status_code=303)
@router.get("/apps", response_class=HTMLResponse)
def apps_page(
request: Request,
q: str = Query(""),
status: str = Query(""),
page_number: int = Query(1, alias="page", ge=1),
user=Depends(require_user),
):
apps = get_apps()
latest_health = get_latest_service_health()
query = q.strip()
selected_status = status.strip()
if query:
apps = [
item for item in apps
if query.lower() in (item.get("id", "") or "").lower()
or query.lower() in (item.get("name", "") or "").lower()
]
if selected_status:
apps = [item for item in apps if (item.get("status", "") or "") == selected_status]
sort = request.query_params.get("sort", "").strip()
if sort == "health":
order = {"unreachable": 0, "unhealthy": 1, "healthy": 2}
apps = sorted(
apps,
key=lambda item: (
order.get((latest_health.get(item.get("id", "")) or {}).get("status"), 3),
item.get("id", ""),
),
)
status_values = sorted({item.get("status", "") for item in get_apps() if item.get("status")})
total_apps = len(apps)
total_pages = max(1, math.ceil(total_apps / DEFAULT_PAGE_SIZE))
if page_number > total_pages:
page_number = total_pages
offset = (page_number - 1) * DEFAULT_PAGE_SIZE
apps = apps[offset : offset + DEFAULT_PAGE_SIZE]
gitea_url = read_env_value("GITEA_URL", "")
gitea_org = read_env_value("GITEA_ORG", DEFAULT_GITEA_ORG)
host = read_env_value("APPFACTORY_HOST", DEFAULT_APPFACTORY_HOST)
rows = ""
for item in apps:
app_id = html.escape(item.get("id", ""))
app_url_id = quote(item.get("id", ""), safe="")
status = html.escape(item.get("status", ""))
health = latest_health.get(item.get("id", "")) or {}
health_status = render_health_status(health.get("status"))
health_checked_at = html.escape(health.get("checked_at", "") or "")
docs = html.escape(item.get("docs", f"/apps/{app_id}/docs"))
memory = item.get("memory", "")
cpus = item.get("cpus", "")
http_clone = html.escape(f"git clone {gitea_url}/{gitea_org}/{app_id}.git")
ssh_clone = html.escape(f"git clone ssh://git@{host}:2222/{gitea_org}/{app_id}.git")
memory_options = ""
for value, label in [
("", "Výchozí"),
("256m", "256 MB - malá služba"),
("512m", "512 MB - běžná služba"),
("1g", "1 GB - větší služba"),
("2g", "2 GB - náročná služba"),
]:
selected = "selected" if memory == value else ""
memory_options += f'<option value="{value}" {selected}>{label}</option>'
cpu_options = ""
for value, label in [
("", "Výchozí"),
("0.25", "0,25 CPU - velmi malá služba"),
("0.50", "0,50 CPU - běžná služba"),
("1.00", "1 CPU - celé jádro"),
("2.00", "2 CPU - náročná služba"),
]:
selected = "selected" if cpus == value else ""
cpu_options += f'<option value="{value}" {selected}>{label}</option>'
rows += f"""
<tr>
<td>
<strong>{app_id}</strong><br>
<span class="muted">/apps/{app_id}</span>
</td>
<td><span class="pill">{status}</span></td>
<td>{health_status}<br><span class="muted">{health_checked_at}</span></td>
<td><a href="{docs}">Swagger</a></td>
<td>
<form method="post" action="/portal/update-resources">
<input type="hidden" name="app_id" value="{app_id}">
<div class="inline-form">
<select name="memory">{memory_options}</select>
<select name="cpus">{cpu_options}</select>
<button type="submit">Použít</button>
</div>
<div class="resource-help">
Paměť je limit RAM. CPU určuje maximální podíl výpočetního výkonu.
Příklad: 0,50 = polovina jádra, 1,00 = celé jádro.
</div>
</form>
</td>
<td>
<details>
<summary>Příkazy pro klonování</summary>
<label class="muted">HTTP</label>
<div class="cmd-row">
<input readonly value="{http_clone}" id="http-{app_id}">
<button type="button" onclick="return copyText(this)">Kopírovat</button>
</div>
<label class="muted">SSH</label>
<div class="cmd-row">
<input readonly value="{ssh_clone}" id="ssh-{app_id}">
<button type="button" onclick="return copyText(this)">Kopírovat</button>
</div>
</details>
</td>
<td>
<p><a class="btn" href="/portal/apps/{app_url_id}">Detail</a></p>
<p><a class="btn btn-secondary" href="/portal/deployments?app_id={app_url_id}">Nasazení</a></p>
<form method="post" action="/portal/apps/{app_url_id}/redeploy" onsubmit="return confirm('Spustit nové nasazení služby {app_id}?');">
<button type="submit" class="btn-secondary">Nasadit znovu</button>
</form>
<form method="post" action="/portal/delete-app" onsubmit="return confirm('Smazat {app_id}? Tím se odstraní kontejner, image, workspace, záznam v katalogu a Gitea repozitář.');">
<input type="hidden" name="app_id" value="{app_id}">
<button type="submit" class="danger">Smazat</button>
</form>
</td>
</tr>
"""
if not rows:
rows = '<tr><td colspan="7">Zatím nejsou nasazené žádné služby.</td></tr>'
status_options = ['<option value="">Všechny stavy</option>']
for value in status_values:
selected = " selected" if selected_status == value else ""
escaped_value = html.escape(value)
status_options.append(f'<option value="{escaped_value}"{selected}>{escaped_value}</option>')
first_item = offset + 1 if total_apps else 0
last_item = min(offset + len(apps), total_apps)
def page_url(page: int) -> str:
params = {"page": page}
if query:
params["q"] = query
if selected_status:
params["status"] = selected_status
if sort:
params["sort"] = sort
return f"/portal/apps?{urlencode(params)}"
previous_link = (
f'<a class="btn btn-secondary" href="{page_url(page_number - 1)}">Předchozí</a>'
if page_number > 1
else ""
)
next_link = (
f'<a class="btn btn-secondary" href="{page_url(page_number + 1)}">Další</a>'
if page_number < total_pages
else ""
)
pagination = ""
if total_pages > 1:
pagination = f"""
<div class="pagination">
<span>Zobrazeno {first_item}-{last_item} z {total_apps}</span>
<div class="pagination-actions">
{previous_link}
<span class="page-indicator">Strana {page_number} / {total_pages}</span>
{next_link}
</div>
</div>
"""
return page(
"Služby",
f"""
<div class="grid">
<div class="card">
<h2>Služby</h2>
<p class="muted">Vytváření, nasazení, klonování, nastavení prostředků a mazání služeb.</p>
</div>
<div class="card">
<h2>Zálohy</h2>
<p class="muted">Vytváření záloh a kopírování příkazů pro obnovu. Obnova je záměrně ruční a chráněná.</p>
<a class="btn" href="/portal/backups">Spravovat zálohy</a>
</div>
<div class="card">
<h2>Nasazení</h2>
<p class="muted">Historie posledních běhů nasazení, stavů a výstupů z deploy procesu.</p>
<a class="btn" href="/portal/deployments">Zobrazit nasazení</a>
</div>
</div>
<div class="card">
<h2>Nasazené služby</h2>
<p><a class="btn" href="/portal/new-app">+ Nová služba</a></p>
<form method="get" action="/portal/apps" class="filter-form">
<input name="q" value="{html.escape(query)}" placeholder="Název nebo ID služby">
<select name="status">{"".join(status_options)}</select>
<select name="sort">
<option value="">Výchozí řazení</option>
<option value="health"{" selected" if sort == "health" else ""}>Podle zdraví</option>
</select>
<input type="hidden" name="page" value="1">
<button type="submit">Filtrovat</button>
<a class="btn btn-secondary" href="/portal/apps">Reset</a>
</form>
{pagination}
<table>
<tr>
<th>Služba</th>
<th>Status</th>
<th>Zdraví</th>
<th>Dokumentace</th>
<th>Prostředky</th>
<th>Git</th>
<th>Akce</th>
</tr>
{rows}
</table>
{pagination}
</div>
""",
user=user,
)
@router.get("/apps/{app_id}", response_class=HTMLResponse)
def app_detail(app_id: str, request: Request, user=Depends(require_user)):
app = get_app(app_id)
if not app:
raise HTTPException(status_code=404, detail="App not found")
log_audit_event(
user,
action="service.health.view",
target_type="service",
target_id=app_id,
)
escaped_app_id = html.escape(app.get("id", ""))
app_url_id = quote(app.get("id", ""), safe="")
name = html.escape(app.get("name", "") or "")
language = html.escape(app.get("language", "") or "")
version = html.escape(app.get("version", "") or "")
status = html.escape(app.get("status", "") or "")
memory = html.escape(app.get("memory", "") or "")
cpus = html.escape(app.get("cpus", "") or "")
updated_at = html.escape(app.get("updated_at", "") or "")
current_health = get_service_health(app.get("id", ""))
health_history = get_service_health_history(app.get("id", ""), limit=50)
rows = ""
for deployment in get_app_deployments(app.get("id", ""), limit=10):
deployment_id = html.escape(str(deployment.get("id", "")))
started_at = html.escape(deployment.get("started_at", "") or "")
triggered_by = html.escape(
deployment.get("triggered_by_display_name")
or deployment.get("triggered_by_username")
or ""
)
rows += f"""
<tr>
<td><a href="/portal/deployments/{deployment_id}">#{deployment_id}</a></td>
<td>{render_status_pill(deployment.get("status"))}</td>
<td>{started_at}</td>
<td>{triggered_by}</td>
</tr>
"""
if not rows:
rows = '<tr><td colspan="4">Zatím nejsou evidovaná žádná nasazení této služby.</td></tr>'
job_rows = ""
for job in get_jobs(limit=100, target=app.get("id", "")):
if job.get("target_type") != "app" or job.get("target_id") != app.get("id", ""):
continue
job_id = html.escape(str(job.get("id", "")))
job_rows += f"""
<tr>
<td><a href="/portal/jobs/{job_id}">#{job_id}</a></td>
<td>{render_status_pill(job.get("status"))}</td>
<td>{html.escape(job.get("type", "") or "")}</td>
<td>{html.escape(job.get("created_at", "") or "")}</td>
</tr>
"""
if not job_rows:
job_rows = '<tr><td colspan="4">Zatím nejsou evidované žádné úlohy této služby.</td></tr>'
health_status = render_health_status(current_health.get("status") if current_health else None)
health_http_status = html.escape(str(current_health.get("http_status") or "")) if current_health else ""
health_response_time = html.escape(str(current_health.get("response_time_ms") or "")) if current_health else ""
health_checked_at = html.escape(current_health.get("checked_at", "") or "") if current_health else ""
health_rows = ""
for item in health_history:
error_text = item.get("error_text", "") or ""
error_preview = error_text if len(error_text) <= 140 else f"{error_text[:137]}..."
health_rows += f"""
<tr>
<td>{html.escape(item.get("checked_at", "") or "")}</td>
<td>{render_health_status(item.get("status"))}</td>
<td>{html.escape(str(item.get("http_status") or ""))}</td>
<td>{html.escape(str(item.get("response_time_ms") or ""))}</td>
<td>{html.escape(error_preview)}</td>
</tr>
"""
if not health_rows:
health_rows = '<tr><td colspan="5">Zatím nejsou evidované žádné kontroly zdraví.</td></tr>'
return page(
name or escaped_app_id,
f"""
<div class="card">
<h2>{escaped_app_id}</h2>
<p class="muted">{name}</p>
<p>
<a class="btn" href="/portal/apps">&larr; Zpět na služby</a>
<a class="btn btn-secondary" href="/portal/deployments?app_id={app_url_id}">Nasazení služby</a>
<a class="btn btn-secondary" href="/portal/jobs?target={app_url_id}">Úlohy služby</a>
</p>
<form method="post" action="/portal/apps/{app_url_id}/redeploy" onsubmit="return confirm('Spustit nové nasazení služby {escaped_app_id}?');">
<button type="submit">Nasadit znovu</button>
</form>
</div>
<div class="card">
<h2>Souhrn</h2>
<table>
<tr><th>ID</th><td>{escaped_app_id}</td></tr>
<tr><th>Název</th><td>{name}</td></tr>
<tr><th>Jazyk</th><td>{language}</td></tr>
<tr><th>Verze</th><td>{version}</td></tr>
<tr><th>Status</th><td><span class="pill">{status}</span></td></tr>
<tr><th>Paměť</th><td>{memory}</td></tr>
<tr><th>CPU</th><td>{cpus}</td></tr>
<tr><th>Upraveno</th><td>{updated_at}</td></tr>
<tr><th>Logy</th><td><a href="/portal/jobs?target={app_url_id}">Zobrazit úlohy a logy</a></td></tr>
</table>
</div>
<div class="card">
<h2>Zdraví služby</h2>
<table>
<tr><th>Aktuální status</th><td>{health_status}</td></tr>
<tr><th>HTTP status</th><td>{health_http_status}</td></tr>
<tr><th>Odezva</th><td>{health_response_time}</td></tr>
<tr><th>Poslední kontrola</th><td>{health_checked_at}</td></tr>
</table>
</div>
<div class="card">
<h2>Historie kontrol</h2>
<table>
<tr>
<th>Čas</th>
<th>Status</th>
<th>HTTP status</th>
<th>Odezva</th>
<th>Chyba</th>
</tr>
{health_rows}
</table>
</div>
<div class="card">
<h2>Historie nasazení</h2>
<table>
<tr>
<th>ID</th>
<th>Status</th>
<th>Čas</th>
<th>Spustil</th>
</tr>
{rows}
</table>
</div>
<div class="card">
<h2>Historie úloh</h2>
<table>
<tr>
<th>ID</th>
<th>Status</th>
<th>Typ</th>
<th>Vytvořeno</th>
</tr>
{job_rows}
</table>
</div>
""",
user=user,
)
@router.post("/apps/{app_id}/redeploy")
def redeploy_app(app_id: str, user=Depends(require_user)):
app = get_app(app_id)
if not app:
raise HTTPException(status_code=404, detail="App not found")
active_job = has_active_deploy_job("app", app_id)
if active_job:
return RedirectResponse(url=f"/portal/jobs/{active_job['id']}", status_code=303)
job_id = create_job(
job_type="deploy_app",
target_type="app",
target_id=app_id,
payload={},
user=user,
source="portal",
)
log_audit_event(
user,
action="app.redeploy.queued",
target_type="app",
target_id=app_id,
metadata={"job_id": job_id},
)
return RedirectResponse(url=f"/portal/jobs/{job_id}", status_code=303)
@router.get("/new-app", response_class=HTMLResponse)
def new_app_form(request: Request, user=Depends(require_user)):
return page(
"Nová služba",
"""
<div class="card">
<h2>Vytvořit novou službu</h2>
<p class="muted">Vytvoří Gitea repozitář, webhook, lokální workspace, první commit a nasadí službu.</p>
<form method="post" action="/portal/new-app">
<p>
<label>ID služby</label><br>
<input name="app_id" placeholder="gmail-service" required>
</p>
<p>
<label>Název služby</label><br>
<input name="app_name" placeholder="Gmail služba" required>
</p>
<p>
<label>Šablona</label><br>
<select name="template">
<option value="python-fastapi">Python FastAPI</option>
</select>
</p>
<button type="submit">Vytvořit službu</button>
</form>
<p><a href="/portal/apps">&larr; Zpět</a></p>
</div>
""",
user=user,
)
@router.post("/new-app", response_class=HTMLResponse)
def create_app(
app_id: str = Form(...),
app_name: str = Form(...),
template: str = Form(...),
user=Depends(require_user),
):
if template != "python-fastapi":
return HTMLResponse("Nepodporovaná šablona", status_code=400)
create_result = run_command([NEW_APP_SCRIPT, app_id, app_name])
deploy_result = run_command([DEPLOY_SCRIPT, app_id])
status = "OK" if create_result.returncode == 0 and deploy_result.returncode == 0 else "FAILED"
log_audit_event(
user,
action="create_app",
target_type="app",
target_id=app_id,
metadata={
"app_id": app_id,
"app_name": app_name,
"template": template,
"status": status,
"create_returncode": create_result.returncode,
"deploy_returncode": deploy_result.returncode,
},
)
return render_result(
title=f"Vytvoření služby: {status}",
back_url="/portal/apps",
sections=[
("Výstup vytvoření", create_result.stdout),
("Chyba vytvoření", create_result.stderr),
("Výstup nasazení", deploy_result.stdout),
("Chyba nasazení", deploy_result.stderr),
],
extra_link=f"/apps/{html.escape(app_id)}/docs",
extra_label="Otevřít Swagger",
user=user,
)
@router.post("/delete-app", response_class=HTMLResponse)
def delete_app(app_id: str = Form(...), user=Depends(require_user)):
result = run_command([DELETE_APP_SCRIPT, app_id])
status = "OK" if result.returncode == 0 else "FAILED"
log_audit_event(
user,
action="delete_app",
target_type="app",
target_id=app_id,
metadata={
"app_id": app_id,
"status": status,
"returncode": result.returncode,
},
)
return render_result(
title=f"Smazání služby: {status}",
back_url="/portal/apps",
sections=[("Výstup", result.stdout), ("Chyba", result.stderr)],
user=user,
)
@router.post("/update-resources", response_class=HTMLResponse)
def update_resources(
app_id: str = Form(...),
memory: str = Form(""),
cpus: str = Form(""),
user=Depends(require_user),
):
memory = memory.strip()
cpus = cpus.strip()
update_app_resources(app_id, memory, cpus)
catalog_result = run_command(["/tools/generate-catalog.sh"])
compose_result = run_command([GENERATE_COMPOSE_SCRIPT])
deploy_result = run_command([DEPLOY_SCRIPT, app_id])
status = (
"OK"
if catalog_result.returncode == 0 and compose_result.returncode == 0 and deploy_result.returncode == 0
else "FAILED"
)
log_audit_event(
user,
action="update_resources",
target_type="app",
target_id=app_id,
metadata={
"app_id": app_id,
"memory": memory,
"cpus": cpus,
"status": status,
"catalog_returncode": catalog_result.returncode,
"compose_returncode": compose_result.returncode,
"deploy_returncode": deploy_result.returncode,
},
)
return render_result(
title=f"Úprava prostředků: {status}",
back_url="/portal/apps",
sections=[
("Výstup katalogu", catalog_result.stdout),
("Chyba katalogu", catalog_result.stderr),
("Výstup compose", compose_result.stdout),
("Chyba compose", compose_result.stderr),
("Výstup nasazení", deploy_result.stdout),
("Chyba nasazení", deploy_result.stderr),
],
user=user,
)