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 (
create_app_variable,
delete_app_variable,
get_app,
get_app_deployments,
get_app_templates,
get_app_variables,
get_apps,
update_app_metadata,
update_app_resources,
update_app_variable,
)
from ..db.audit import log_audit_event
from ..db.health import get_latest_service_health, get_service_health, get_service_health_history
from ..db.incidents import get_service_incidents
from ..db.jobs import create_job, get_jobs, has_active_deploy_job
from ..routes.deployments import render_status_pill
from ..routes.incidents import render_incident_history_rows
from ..shell import run_command
from ..templates.layout import page, render_result
router = APIRouter()
DEFAULT_PAGE_SIZE = 20
METADATA_FIELDS = (
"name",
"description",
"owner",
"template",
"runtime",
"repository_url",
"repository_name",
"default_branch",
"domain",
"health_url",
"container_port",
"is_public",
"is_enabled",
)
def clean_optional(value: str | None) -> str:
return (value or "").strip()
def parse_optional_int(value: str | None) -> int | None:
value = clean_optional(value)
if not value:
return None
try:
return int(value)
except ValueError:
raise HTTPException(status_code=400, detail="Neplatne cislo")
def bool_checked(value) -> str:
return " checked" if value else ""
def render_template_options(templates: list[dict], selected_template: str) -> str:
options = ['']
for template in templates:
name = template.get("name", "") or ""
runtime = template.get("runtime", "") or ""
description = template.get("description", "") or ""
label_parts = [name]
if runtime:
label_parts.append(runtime)
if description:
label_parts.append(description)
selected = " selected" if selected_template == name else ""
options.append(
f''
)
return "".join(options)
def diff_metadata(before: dict, after: dict) -> dict:
changes = {}
for field in METADATA_FIELDS:
old_value = before.get(field)
new_value = after.get(field)
if field in ("is_public", "is_enabled"):
changed = bool(old_value) != bool(new_value)
else:
changed = str(old_value or "") != str(new_value or "")
if changed:
changes[field] = {"old": old_value, "new": new_value}
return changes
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'{html.escape(labels.get(normalized, value or "neznámá"))}'
def render_health_dot(status: str | None, checked_at: str | None) -> str:
normalized = (status or "").lower()
labels = {
"healthy": "zdravá",
"unhealthy": "nezdravá",
"unreachable": "nedostupná",
}
class_name = "health-dot health-dot-muted"
if normalized == "healthy":
class_name = "health-dot health-dot-success"
elif normalized == "unhealthy":
class_name = "health-dot health-dot-warning"
elif normalized == "unreachable":
class_name = "health-dot health-dot-danger"
label = labels.get(normalized, "zdraví neznámé")
title = label
if checked_at:
title = f"{label}, poslední kontrola: {checked_at}"
return f''
@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 index, item in enumerate(apps, start=1):
app_id = html.escape(item.get("id", ""))
app_url_id = quote(item.get("id", ""), safe="")
resource_modal_id = f"resources-{page_number}-{index}"
status = html.escape(item.get("status", ""))
health = latest_health.get(item.get("id", "")) or {}
health_checked_at = html.escape(health.get("checked_at", "") or "")
health_dot = render_health_dot(health.get("status"), health.get("checked_at"))
docs = html.escape(item.get("docs", f"/apps/{app_id}/docs"))
memory = html.escape(item.get("memory", "") or "")
cpus = html.escape(item.get("cpus", "") or "")
memory_label = memory or "výchozí"
cpus_label = cpus or "výchozí"
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")
rows += f"""
{health_dot}
{app_id}
/apps/{app_id}
|
{status} |
Swagger |
Paměť: {memory_label}
CPU: {cpus_label}
|
Příkazy pro klonování
|
Detail
Nasazení
|
"""
if not rows:
rows = '| Zatím nejsou nasazené žádné služby. |
'
status_options = ['']
for value in status_values:
selected = " selected" if selected_status == value else ""
escaped_value = html.escape(value)
status_options.append(f'')
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'Předchozí'
if page_number > 1
else ""
)
next_link = (
f'Další'
if page_number < total_pages
else ""
)
pagination = ""
if total_pages > 1:
pagination = f"""
"""
return page(
"Služby",
f"""
Služby
Vytváření, nasazení, klonování, nastavení prostředků a mazání služeb.
Zálohy
Vytváření záloh a kopírování příkazů pro obnovu. Obnova je záměrně ruční a chráněná.
Spravovat zálohy
Nasazení
Historie posledních běhů nasazení, stavů a výstupů z deploy procesu.
Zobrazit nasazení
Nasazené služby
+ Nová služba
{pagination}
| Služba |
Status |
Dokumentace |
Prostředky i |
Git |
Akce |
{rows}
{pagination}
""",
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 "")
description = html.escape(app.get("description", "") or "")
owner = html.escape(app.get("owner", "") or "")
template_value = html.escape(app.get("template", "") or "")
runtime = html.escape(app.get("runtime", "") or "")
repository_url = html.escape(app.get("repository_url", "") or "")
repository_name = html.escape(app.get("repository_name", "") or "")
default_branch = html.escape(app.get("default_branch", "") or "")
domain = html.escape(app.get("domain", "") or "")
health_url = html.escape(app.get("health_url", "") or "")
container_port = html.escape(str(app.get("container_port") or ""))
is_public = bool(app.get("is_public"))
is_enabled = bool(app.get("is_enabled"))
templates = get_app_templates()
template_options = render_template_options(templates, app.get("template", "") or "")
variables = get_app_variables(app.get("id", ""))
incidents = get_service_incidents(app.get("id", ""), limit=20)
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"""
| #{deployment_id} |
{render_status_pill(deployment.get("status"))} |
{started_at} |
{triggered_by} |
"""
if not rows:
rows = '| Zatím nejsou evidovaná žádná nasazení této služby. |
'
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"""
| #{job_id} |
{render_status_pill(job.get("status"))} |
{html.escape(job.get("type", "") or "")} |
{html.escape(job.get("created_at", "") or "")} |
"""
if not job_rows:
job_rows = '| Zatím nejsou evidované žádné úlohy této služby. |
'
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"""
| {html.escape(item.get("checked_at", "") or "")} |
{render_health_status(item.get("status"))} |
{html.escape(str(item.get("http_status") or ""))} |
{html.escape(str(item.get("response_time_ms") or ""))} |
{html.escape(error_preview)} |
"""
if not health_rows:
health_rows = '| Zatím nejsou evidované žádné kontroly zdraví. |
'
variable_rows = ""
for variable in variables:
variable_id = html.escape(str(variable.get("id", "")))
variable_key = html.escape(variable.get("key", "") or "")
variable_value_raw = variable.get("value", "") or ""
variable_is_secret = bool(variable.get("is_secret"))
variable_value = "---" if variable_is_secret else html.escape(variable_value_raw)
value_input = "" if variable_is_secret else html.escape(variable_value_raw)
secret_checked = bool_checked(variable_is_secret)
secret_hint = ' placeholder="---"' if variable_is_secret else ""
variable_rows += f"""
| {variable_key} |
{variable_value} |
{"Ano" if variable_is_secret else "Ne"} |
|
"""
if not variable_rows:
variable_rows = '| Zatím nejsou evidované žádné proměnné. |
'
return page(
"Detail slu\u017eby",
f"""
Souhrn
| ID | {escaped_app_id} |
| Název | {name} |
| Popis | {description} |
| Vlastník | {owner} |
| Template | {template_value} |
| Runtime | {runtime} |
| Repository URL | {repository_url} |
| Repository Name | {repository_name} |
| Default Branch | {default_branch} |
| Domain | {domain} |
| Health URL | {health_url} |
| Container Port | {container_port} |
| Veřejná služba | {"Ano" if is_public else "Ne"} |
| Aktivní služba | {"Ano" if is_enabled else "Ne"} |
| Jazyk | {language} |
| Verze | {version} |
| Status | {status} |
| Paměť | {memory} |
| CPU | {cpus} |
| Upraveno | {updated_at} |
| Logy | Zobrazit úlohy a logy |
Zdraví služby
| Aktuální status | {health_status} |
| HTTP status | {health_http_status} |
| Odezva | {health_response_time} |
| Poslední kontrola | {health_checked_at} |
Historie kontrol
| Čas |
Status |
HTTP status |
Odezva |
Chyba |
{health_rows}
Incidenty služby
| Název |
Začátek |
Konec |
Trvání |
Stav |
{render_incident_history_rows(incidents, include_service=False)}
Historie nasazení
| ID |
Status |
Čas |
Spustil |
{rows}
Historie úloh
| ID |
Status |
Typ |
Vytvořeno |
{job_rows}
""",
user=user,
)
@router.post("/apps/{app_id}/metadata")
def save_app_metadata(
app_id: str,
name: str = Form(...),
description: str = Form(""),
owner: str = Form(""),
template: str = Form(""),
runtime: str = Form(""),
repository_url: str = Form(""),
repository_name: str = Form(""),
default_branch: str = Form(""),
domain: str = Form(""),
health_url: str = Form(""),
container_port: str = Form(""),
is_public: str | None = Form(None),
is_enabled: str | None = Form(None),
user=Depends(require_user),
):
app = get_app(app_id)
if not app:
raise HTTPException(status_code=404, detail="App not found")
metadata = {
"name": clean_optional(name),
"description": clean_optional(description),
"owner": clean_optional(owner),
"template": clean_optional(template),
"runtime": clean_optional(runtime),
"repository_url": clean_optional(repository_url),
"repository_name": clean_optional(repository_name),
"default_branch": clean_optional(default_branch),
"domain": clean_optional(domain),
"health_url": clean_optional(health_url),
"container_port": parse_optional_int(container_port),
"is_public": bool(is_public),
"is_enabled": bool(is_enabled),
}
changes = diff_metadata(app, metadata)
update_app_metadata(app_id, metadata)
if changes:
log_audit_event(
user,
action="app.metadata.updated",
target_type="app",
target_id=app_id,
metadata={"changes": changes},
)
return RedirectResponse(url=f"/portal/apps/{quote(app_id, safe='')}#metadata", status_code=303)
@router.post("/apps/{app_id}/variables")
def add_app_variable(
app_id: str,
key: str = Form(...),
value: str = Form(""),
is_secret: str | None = Form(None),
user=Depends(require_user),
):
app = get_app(app_id)
if not app:
raise HTTPException(status_code=404, detail="App not found")
variable_key = clean_optional(key)
if not variable_key:
raise HTTPException(status_code=400, detail="Key is required")
create_app_variable(app_id, variable_key, value, bool(is_secret))
log_audit_event(
user,
action="app.variable.created",
target_type="app",
target_id=app_id,
metadata={"key": variable_key, "is_secret": bool(is_secret)},
)
return RedirectResponse(url=f"/portal/apps/{quote(app_id, safe='')}#promenne", status_code=303)
@router.post("/apps/{app_id}/variables/{variable_id}/update")
def save_app_variable(
app_id: str,
variable_id: int,
key: str = Form(...),
value: str = Form(""),
is_secret: str | None = Form(None),
user=Depends(require_user),
):
app = get_app(app_id)
if not app:
raise HTTPException(status_code=404, detail="App not found")
existing = next((item for item in get_app_variables(app_id) if int(item.get("id")) == variable_id), None)
if not existing:
raise HTTPException(status_code=404, detail="Variable not found")
variable_key = clean_optional(key)
if not variable_key:
raise HTTPException(status_code=400, detail="Key is required")
stored_value = None if existing.get("is_secret") and value == "" else value
update_app_variable(variable_id, app_id, variable_key, stored_value, bool(is_secret))
log_audit_event(
user,
action="app.variable.updated",
target_type="app",
target_id=app_id,
metadata={"key": variable_key, "is_secret": bool(is_secret)},
)
return RedirectResponse(url=f"/portal/apps/{quote(app_id, safe='')}#promenne", status_code=303)
@router.post("/apps/{app_id}/variables/{variable_id}/delete")
def remove_app_variable(app_id: str, variable_id: int, user=Depends(require_user)):
app = get_app(app_id)
if not app:
raise HTTPException(status_code=404, detail="App not found")
existing = next((item for item in get_app_variables(app_id) if int(item.get("id")) == variable_id), None)
if not existing:
raise HTTPException(status_code=404, detail="Variable not found")
delete_app_variable(variable_id, app_id)
log_audit_event(
user,
action="app.variable.deleted",
target_type="app",
target_id=app_id,
metadata={"key": existing.get("key"), "is_secret": bool(existing.get("is_secret"))},
)
return RedirectResponse(url=f"/portal/apps/{quote(app_id, safe='')}#promenne", status_code=303)
@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",
"""
""",
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,
)