import html
import os
from pathlib import Path
from fastapi import APIRouter, Depends, Form, HTTPException, Request
from fastapi.responses import HTMLResponse, RedirectResponse
from app.auth import require_user
from app.db.alerting import (
create_alert_rule,
delete_alert_rule,
get_alert_event,
get_alert_events,
get_alert_rule,
get_alert_rules,
get_rule_alert_events,
set_alert_rule_enabled,
update_alert_rule,
)
from app.db.audit import log_audit_event
from app.routes.jobs import pretty_json
from app.templates.layout import page
router = APIRouter()
ALERTS_DIR = Path("/opt/appfactory/workspace/appfactory-tools/alerts")
MAX_SCRIPT_BYTES = 100 * 1024
EVENT_TYPES = ("incident.opened", "incident.resolved")
DEFAULT_SCRIPT_CONTENT = """#!/usr/bin/env bash
set -euo pipefail
echo "TODO"
"""
def clean_optional(value: str | None) -> str:
return (value or "").strip()
def is_admin(user) -> bool:
return (user.get("role") or "").lower() == "admin"
def render_enabled_pill(value) -> str:
if value:
return 'zapnuto'
return 'vypnuto'
def render_status(value: str | None) -> str:
status = (value or "").lower()
if status in {"success", "processed", "done", "ok"}:
return 'zpracováno'
if status in {"queued", "running", "pending"}:
return 'čeká'
if status in {"failed", "error"}:
return 'selhalo'
if not status:
return 'bez stavu'
return f'{html.escape(value or "")}'
def render_event_type_options(selected: str) -> str:
options = []
for event_type in EVENT_TYPES:
selected_attr = " selected" if selected == event_type else ""
escaped = html.escape(event_type)
options.append(f'')
return "".join(options)
def validate_event_type(value: str) -> str:
event_type = clean_optional(value)
if event_type not in EVENT_TYPES:
raise HTTPException(status_code=400, detail="Neplatný typ události")
return event_type
def validate_script_name(value: str) -> str:
script_name = clean_optional(value)
if not script_name:
raise HTTPException(status_code=400, detail="Název skriptu je povinný")
if not script_name.endswith(".sh"):
raise HTTPException(status_code=400, detail="Název skriptu musí končit .sh")
if "/" in script_name or "\\" in script_name or ".." in script_name:
raise HTTPException(status_code=400, detail="Název skriptu nesmí obsahovat cestu")
return script_name
def script_path(script_name: str) -> Path:
safe_name = validate_script_name(script_name)
path = (ALERTS_DIR / safe_name).resolve()
base = ALERTS_DIR.resolve()
try:
path.relative_to(base)
except ValueError:
raise HTTPException(status_code=400, detail="Neplatný název skriptu")
return path
def read_script_file(script_name: str) -> dict:
path = script_path(script_name)
if not path.exists():
return {
"content": DEFAULT_SCRIPT_CONTENT,
"error": None,
"warning": "Soubor zatím neexistuje. Administrátor ho může vytvořit z výchozího obsahu.",
}
if not path.is_file():
return {"content": "", "error": "Cesta není soubor.", "warning": None}
try:
if path.stat().st_size > MAX_SCRIPT_BYTES:
return {"content": "", "error": "Soubor je větší než 100 KB.", "warning": None}
content = path.read_text(encoding="utf-8")
except Exception:
return {"content": "", "error": "Soubor se nepodařilo načíst.", "warning": None}
warning = None
if not content.startswith("#!/usr/bin/env bash"):
warning = "První řádek by měl být #!/usr/bin/env bash."
return {"content": content, "error": None, "warning": warning}
def normalize_script_content(content: str) -> str:
value = content.replace("\r\n", "\n").replace("\r", "\n")
if not value.strip():
raise HTTPException(status_code=400, detail="Obsah skriptu nesmí být prázdný")
if len(value.encode("utf-8")) > MAX_SCRIPT_BYTES:
raise HTTPException(status_code=400, detail="Obsah skriptu je větší než 100 KB")
if not value.startswith("#!/usr/bin/env bash"):
value = "#!/usr/bin/env bash\n" + value.lstrip("\n")
if len(value.encode("utf-8")) > MAX_SCRIPT_BYTES:
raise HTTPException(status_code=400, detail="Obsah skriptu je větší než 100 KB")
return value
def save_script_file(script_name: str, content: str) -> None:
path = script_path(script_name)
value = normalize_script_content(content)
try:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(value, encoding="utf-8", newline="\n")
os.chmod(path, 0o755)
except Exception:
raise HTTPException(status_code=500, detail="Soubor se nepodařilo uložit")
def form_metadata(
name: str,
description: str,
event_type: str,
service_id: str,
script_name: str,
is_enabled: str | None,
) -> dict:
name_value = clean_optional(name)
if not name_value:
raise HTTPException(status_code=400, detail="Název je povinný")
return {
"name": name_value,
"description": clean_optional(description),
"event_type": validate_event_type(event_type),
"service_id": clean_optional(service_id),
"script_name": validate_script_name(script_name),
"is_enabled": bool(is_enabled),
}
def render_rule_form(rule: dict | None, action: str) -> str:
rule = rule or {}
name = html.escape(rule.get("name", "") or "")
description = html.escape(rule.get("description", "") or "")
service_id = html.escape(rule.get("service_id", "") or "")
script_name = html.escape(rule.get("script_name", "log-alert.sh") or "log-alert.sh")
event_type = rule.get("event_type", EVENT_TYPES[0]) or EVENT_TYPES[0]
enabled_checked = " checked" if rule.get("is_enabled", True) else ""
return f"""
"""
def render_event_rows(events: list[dict], empty_colspan: int = 9, include_rule: bool = True) -> str:
rows = ""
for event in events:
event_id = html.escape(str(event.get("id", "")))
rule_id = event.get("rule_id")
rule = ""
if include_rule:
if rule_id:
rule_name = html.escape(event.get("rule_name") or f"#{rule_id}")
rule = f'{rule_name} | '
else:
rule = " | "
job_id = event.get("job_id")
job_link = ""
if job_id:
job_id_html = html.escape(str(job_id))
job_link = f'#{job_id_html}'
rows += f"""
| #{event_id} |
{rule}
{html.escape(event.get("event_type", "") or "")} |
{html.escape(event.get("service_id", "") or "")} |
{html.escape(str(event.get("incident_id") or ""))} |
{render_status(event.get("status"))} |
{job_link} |
{html.escape(event.get("created_at", "") or "")} |
Detail |
"""
if not rows:
rows = f'| Zatím nejsou evidované žádné alert eventy. |
'
return rows
def render_script_editor(rule: dict, user: dict) -> str:
script_name = rule.get("script_name", "") or ""
script_name_html = html.escape(script_name)
try:
script_file = read_script_file(script_name)
except HTTPException:
script_file = {"content": "", "error": "Název skriptu není bezpečný.", "warning": None}
content = html.escape(script_file.get("content", "") or "")
warning = f'{html.escape(script_file["warning"])}
' if script_file.get("warning") else ""
error = f'{html.escape(script_file["error"])}
' if script_file.get("error") else ""
if not is_admin(user):
return f"""
Obsah alert skriptu
Soubor: {script_name_html}
{warning}
{error}
Obsah skriptu je pouze pro čtení. Ukládat ho může jen administrátor.
"""
return f"""
Editor alert skriptu
Soubor: {script_name_html}
{warning}
{error}
"""
@router.get("/alerting", response_class=HTMLResponse)
def alerting_redirect(user=Depends(require_user)):
return RedirectResponse(url="/portal/alerting/rules", status_code=303)
@router.get("/alerting/rules", response_class=HTMLResponse)
def alert_rules_page(request: Request, user=Depends(require_user)):
rows = ""
for rule in get_alert_rules():
rule_id = html.escape(str(rule.get("id", "")))
enabled = bool(rule.get("is_enabled"))
toggle_label = "Zakázat" if enabled else "Povolit"
last_event = ""
if rule.get("last_event_id"):
event_id = html.escape(str(rule.get("last_event_id")))
last_event = f'#{event_id} {render_status(rule.get("last_event_status"))}'
rows += f"""
| {html.escape(rule.get("name", "") or "")} |
{html.escape(rule.get("event_type", "") or "")} |
{html.escape(rule.get("service_id", "") or "")} |
{html.escape(rule.get("script_name", "") or "")} |
{render_enabled_pill(enabled)} |
{last_event} |
{html.escape(rule.get("updated_at", "") or "")} |
Detail
Upravit
|
"""
if not rows:
rows = '| Zatím nejsou evidovaná žádná alert pravidla. |
'
return page(
"Alert pravidla",
f"""
Pravidla
| Název |
Typ události |
Služba |
Skript |
Stav |
Poslední event |
Upraveno |
Akce |
{rows}
""",
user=user,
)
@router.get("/alerting/rules/new", response_class=HTMLResponse)
def new_alert_rule_form(request: Request, user=Depends(require_user)):
return page(
"Nové alert pravidlo",
f"""
{render_rule_form({"is_enabled": True, "event_type": EVENT_TYPES[0], "script_name": "log-alert.sh"}, "/portal/alerting/rules/new")}
""",
user=user,
)
@router.post("/alerting/rules/new")
def create_alert_rule_action(
name: str = Form(...),
description: str = Form(""),
event_type: str = Form(...),
service_id: str = Form(""),
script_name: str = Form(...),
is_enabled: str | None = Form(None),
user=Depends(require_user),
):
metadata = form_metadata(name, description, event_type, service_id, script_name, is_enabled)
rule_id = create_alert_rule(metadata)
log_audit_event(
user,
action="alert_rule.created",
target_type="alert_rule",
target_id=rule_id,
metadata={"event_type": metadata["event_type"], "script_name": metadata["script_name"]},
)
return RedirectResponse(url=f"/portal/alerting/rules/{rule_id}", status_code=303)
@router.get("/alerting/rules/{rule_id}", response_class=HTMLResponse)
def alert_rule_detail(rule_id: int, request: Request, user=Depends(require_user)):
rule = get_alert_rule(rule_id)
if not rule:
raise HTTPException(status_code=404, detail="Alert pravidlo nenalezeno")
enabled = bool(rule.get("is_enabled"))
toggle_label = "Zakázat" if enabled else "Povolit"
events = get_rule_alert_events(rule_id, limit=20)
return page(
rule.get("name", "") or "Alert pravidlo",
f"""
Souhrn
| Název | {html.escape(rule.get("name", "") or "")} |
| Popis | {html.escape(rule.get("description", "") or "")} |
| Typ události | {html.escape(rule.get("event_type", "") or "")} |
| Služba | {html.escape(rule.get("service_id", "") or "")} |
| Skript | {html.escape(rule.get("script_name", "") or "")} |
| Zapnuto | {"Ano" if enabled else "Ne"} |
| Vytvořeno | {html.escape(rule.get("created_at", "") or "")} |
| Upraveno | {html.escape(rule.get("updated_at", "") or "")} |
Poslední eventy
| ID |
Typ události |
Služba |
Incident |
Stav |
Úloha |
Vytvořeno |
Akce |
{render_event_rows(events, empty_colspan=8, include_rule=False)}
{render_script_editor(rule, user)}
""",
user=user,
)
@router.get("/alerting/rules/{rule_id}/edit", response_class=HTMLResponse)
def edit_alert_rule_form(rule_id: int, request: Request, user=Depends(require_user)):
rule = get_alert_rule(rule_id)
if not rule:
raise HTTPException(status_code=404, detail="Alert pravidlo nenalezeno")
return page(
"Upravit alert pravidlo",
f"""
{render_rule_form(rule, f"/portal/alerting/rules/{rule_id}/edit")}
""",
user=user,
)
@router.post("/alerting/rules/{rule_id}/edit")
def update_alert_rule_action(
rule_id: int,
name: str = Form(...),
description: str = Form(""),
event_type: str = Form(...),
service_id: str = Form(""),
script_name: str = Form(...),
is_enabled: str | None = Form(None),
user=Depends(require_user),
):
rule = get_alert_rule(rule_id)
if not rule:
raise HTTPException(status_code=404, detail="Alert pravidlo nenalezeno")
metadata = form_metadata(name, description, event_type, service_id, script_name, is_enabled)
update_alert_rule(rule_id, metadata)
log_audit_event(
user,
action="alert_rule.updated",
target_type="alert_rule",
target_id=rule_id,
metadata={"event_type": metadata["event_type"], "script_name": metadata["script_name"]},
)
return RedirectResponse(url=f"/portal/alerting/rules/{rule_id}", status_code=303)
@router.post("/alerting/rules/{rule_id}/toggle")
def toggle_alert_rule(rule_id: int, user=Depends(require_user)):
rule = get_alert_rule(rule_id)
if not rule:
raise HTTPException(status_code=404, detail="Alert pravidlo nenalezeno")
enabled = not bool(rule.get("is_enabled"))
set_alert_rule_enabled(rule_id, enabled)
log_audit_event(
user,
action="alert_rule.enabled" if enabled else "alert_rule.disabled",
target_type="alert_rule",
target_id=rule_id,
metadata={"script_name": rule.get("script_name")},
)
return RedirectResponse(url=f"/portal/alerting/rules/{rule_id}", status_code=303)
@router.post("/alerting/rules/{rule_id}/delete")
def delete_alert_rule_action(rule_id: int, user=Depends(require_user)):
rule = get_alert_rule(rule_id)
if not rule:
raise HTTPException(status_code=404, detail="Alert pravidlo nenalezeno")
if not delete_alert_rule(rule_id):
raise HTTPException(status_code=409, detail="Alert pravidlo nelze smazat")
log_audit_event(
user,
action="alert_rule.deleted",
target_type="alert_rule",
target_id=rule_id,
metadata={"script_name": rule.get("script_name")},
)
return RedirectResponse(url="/portal/alerting/rules", status_code=303)
@router.post("/alerting/rules/{rule_id}/script")
def update_alert_script(rule_id: int, content: str = Form(...), user=Depends(require_user)):
if not is_admin(user):
raise HTTPException(status_code=403, detail="Alert skript může upravit pouze administrátor")
rule = get_alert_rule(rule_id)
if not rule:
raise HTTPException(status_code=404, detail="Alert pravidlo nenalezeno")
script_name = validate_script_name(rule.get("script_name", "") or "")
save_script_file(script_name, content)
log_audit_event(
user,
action="alert_script.updated",
target_type="alert_script",
target_id=script_name,
metadata={"alert_rule_id": rule_id, "script_name": script_name},
)
return RedirectResponse(url=f"/portal/alerting/rules/{rule_id}", status_code=303)
@router.get("/alerting/events", response_class=HTMLResponse)
def alert_events_page(request: Request, user=Depends(require_user)):
events = get_alert_events(limit=200)
return page(
"Alert eventy",
f"""
Alert eventy
Historie vyvolaných alertů a jejich zpracování.
Alert pravidla
Eventy
| ID |
Pravidlo |
Typ události |
Služba |
Incident |
Stav |
Úloha |
Vytvořeno |
Akce |
{render_event_rows(events)}
""",
user=user,
)
@router.get("/alerting/events/{event_id}", response_class=HTMLResponse)
def alert_event_detail(event_id: int, request: Request, user=Depends(require_user)):
event = get_alert_event(event_id)
if not event:
raise HTTPException(status_code=404, detail="Alert event nenalezeno")
rule_link = ""
if event.get("rule_id"):
rule_id = html.escape(str(event.get("rule_id")))
rule_name = html.escape(event.get("rule_name") or f"#{rule_id}")
rule_link = f'{rule_name}'
job_link = ""
if event.get("job_id"):
job_id = html.escape(str(event.get("job_id")))
job_link = f'#{job_id}'
error_text = html.escape(event.get("error_text", "") or "")
payload = html.escape(pretty_json(event.get("payload_json")))
return page(
f"Alert event #{event_id}",
f"""
Souhrn
| Pravidlo | {rule_link} |
| Typ události | {html.escape(event.get("event_type", "") or "")} |
| Služba | {html.escape(event.get("service_id", "") or "")} |
| Incident | {html.escape(str(event.get("incident_id") or ""))} |
| Stav | {render_status(event.get("status"))} |
| Úloha | {job_link} |
| Vytvořeno | {html.escape(event.get("created_at", "") or "")} |
| Zpracováno | {html.escape(event.get("processed_at", "") or "")} |
""",
user=user,
)