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.audit import log_audit_event from app.db.jobs import create_job from app.db.scheduled_scripts import ( create_scheduled_script, delete_scheduled_script, get_scheduled_script, get_scheduled_scripts, set_scheduled_script_enabled, update_scheduled_script, ) from app.routes.deployments import render_status_pill from app.templates.layout import page from app.tools_repo import TOOLS_REPO_DIR, commit_and_push router = APIRouter() MAINTENANCE_DIR = TOOLS_REPO_DIR / "maintenance" MAX_SCRIPT_BYTES = 100 * 1024 DEFAULT_SCRIPT_CONTENT = """#!/usr/bin/env bash set -euo pipefail echo "TODO" """ SCHEDULE_TYPES = ("hourly", "daily", "weekly", "monthly") SCHEDULE_LABELS = { "hourly": "každou hodinu", "daily": "denně", "weekly": "týdně", "monthly": "měsíčně", } 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_bool(value) -> str: return "Ano" if value else "Ne" def render_schedule_type(value: str | None) -> str: value = value or "" return SCHEDULE_LABELS.get(value, html.escape(value)) def render_schedule_options(selected: str) -> str: options = [] for value in SCHEDULE_TYPES: selected_attr = " selected" if selected == value else "" options.append(f'') return "".join(options) def render_enabled_pill(value) -> str: if value: return 'zapnuto' return 'vypnuto' def render_running_pill(value) -> str: if value: return 'běží' return 'neběží' 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 = (MAINTENANCE_DIR / safe_name).resolve() base = MAINTENANCE_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 { "exists": False, "content": DEFAULT_SCRIPT_CONTENT, "error": None, "warning": "Soubor zatím neexistuje. Můžete ho vytvořit z výchozího obsahu.", } if not path.is_file(): return {"exists": False, "content": "", "error": "Cesta není soubor.", "warning": None} try: if path.stat().st_size > MAX_SCRIPT_BYTES: return {"exists": True, "content": "", "error": "Soubor je větší než 100 KB.", "warning": None} content = path.read_text(encoding="utf-8") except Exception: return {"exists": True, "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 {"exists": True, "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") except OSError as exc: reason = exc.strerror or str(exc) raise HTTPException(status_code=500, detail=f"Soubor se nepodařilo uložit: {reason}") # Spustitelná práva nastavujeme best-effort – soubor už je uložený, takže selhání chmod # (jiný vlastník, FS bez podpory práv) nesmí hlásit chybu uložení. try: os.chmod(path, 0o755) except OSError: pass def validate_schedule_type(value: str) -> str: schedule_type = clean_optional(value) if schedule_type not in SCHEDULE_TYPES: raise HTTPException(status_code=400, detail="Neplatná frekvence") return schedule_type def validate_timeout(value: str | int) -> int: try: timeout_seconds = int(value) except (TypeError, ValueError): raise HTTPException(status_code=400, detail="Timeout musí být číslo") if timeout_seconds < 10 or timeout_seconds > 3600: raise HTTPException(status_code=400, detail="Timeout musí být mezi 10 a 3600 sekundami") return timeout_seconds def form_metadata( name: str, description: str, schedule_type: str, schedule_time: str, timeout_seconds: 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), "schedule_type": validate_schedule_type(schedule_type), "schedule_time": clean_optional(schedule_time), "timeout_seconds": validate_timeout(timeout_seconds), "is_enabled": bool(is_enabled), } def render_form(script: dict | None, action: str, include_script_name: bool) -> str: script = script or {} name = html.escape(script.get("name", "") or "") description = html.escape(script.get("description", "") or "") script_name = html.escape(script.get("script_name", "") or "") schedule_type = script.get("schedule_type", "daily") or "daily" schedule_time = html.escape(script.get("schedule_time", "") or "") timeout_seconds = html.escape(str(script.get("timeout_seconds") or 300)) enabled_checked = " checked" if script.get("is_enabled", True) else "" script_name_input = "" if include_script_name: script_name_input = f""" """ else: script_name_input = f""" """ return f"""
{script_name_input}
""" def render_script_content_section(script: dict, user: dict) -> str: script_name = script.get("script_name", "") or "" script_name_html = html.escape(script_name) try: script_file = read_script_file(script_name) except HTTPException: script_file = { "exists": False, "content": "", "error": "Název skriptu není bezpečný.", "warning": None, } content = html.escape(script_file.get("content", "") or "") warning = "" if script_file.get("warning"): warning = f'

{script_file["warning"]}

' error = "" if script_file.get("error"): error = f'

{script_file["error"]}

' if not is_admin(user): readonly_hint = '

Obsah skriptu je dostupný pouze pro čtení. Ukládat může jen administrátor.

' return f"""

Obsah skriptu

Soubor: {script_name_html}

{warning} {error} {readonly_hint}
""" return f"""

Obsah skriptu

Soubor: {script_name_html}

{warning} {error}
""" @router.get("/scheduled-scripts", response_class=HTMLResponse) def scheduled_scripts_page(request: Request, user=Depends(require_user)): scripts = get_scheduled_scripts() rows = "" for script in scripts: script_id_raw = int(script.get("id")) script_id = html.escape(str(script_id_raw)) name = html.escape(script.get("name", "") or "") script_name = html.escape(script.get("script_name", "") or "") schedule_type = render_schedule_type(script.get("schedule_type")) schedule_time = html.escape(script.get("schedule_time", "") or "") last_status = render_status_pill(script.get("last_status")) if script.get("last_status") else "" last_run_at = html.escape(script.get("last_run_at", "") or "") next_run_at = html.escape(script.get("next_run_at", "") or "") enabled = bool(script.get("is_enabled")) toggle_label = "Zakázat" if enabled else "Povolit" delete_button = "" if not script.get("is_running"): delete_button = f"""
""" rows += f""" {name} {script_name} {schedule_type} {schedule_time} {render_enabled_pill(enabled)} {render_running_pill(script.get("is_running"))} {last_status} {last_run_at} {next_run_at}
{delete_button} """ if not rows: rows = 'Zatím nejsou evidované žádné plánované skripty.' return page( "Plánované skripty", f"""

Plánované skripty

Přehled skriptů spouštěných schedulerem nebo ručně z portálu.

Nový plánovaný skript

Seznam

{rows}
Název Skript Frekvence Čas Zapnuto Běží Poslední stav Poslední běh Další běh Akce
""", user=user, ) @router.get("/scheduled-scripts/new", response_class=HTMLResponse) def new_scheduled_script_form(request: Request, user=Depends(require_user)): return page( "Nový plánovaný skript", f"""

Nový plánovaný skript

← Zpět

{render_form({"is_enabled": True, "timeout_seconds": 300, "schedule_type": "daily"}, "/portal/scheduled-scripts/new", True)}
""", user=user, ) @router.post("/scheduled-scripts/new") def create_scheduled_script_action( name: str = Form(...), description: str = Form(""), script_name: str = Form(...), schedule_type: str = Form(...), schedule_time: str = Form(""), timeout_seconds: str = Form("300"), is_enabled: str | None = Form(None), user=Depends(require_user), ): metadata = form_metadata(name, description, schedule_type, schedule_time, timeout_seconds, is_enabled) metadata["script_name"] = validate_script_name(script_name) script_id = create_scheduled_script(metadata) created_name = metadata["script_name"] if not script_path(created_name).exists(): save_script_file(created_name, DEFAULT_SCRIPT_CONTENT) commit_and_push(f"maintenance/{created_name}", f"Vytvořen skript {created_name}", user) log_audit_event( user, action="scheduled_script.created", target_type="scheduled_script", target_id=script_id, metadata={"script_name": metadata["script_name"]}, ) return RedirectResponse(url=f"/portal/scheduled-scripts/{script_id}", status_code=303) @router.get("/scheduled-scripts/{script_id}", response_class=HTMLResponse) def scheduled_script_detail(script_id: int, request: Request, user=Depends(require_user)): script = get_scheduled_script(script_id) if not script: raise HTTPException(status_code=404, detail="Scheduled script not found") last_job_id = script.get("last_job_id") last_job = "" if last_job_id: last_job_html = html.escape(str(last_job_id)) last_job = f'#{last_job_html}' enabled = bool(script.get("is_enabled")) toggle_label = "Zakázat" if enabled else "Povolit" delete_form = "" if not script.get("is_running"): delete_form = f"""
""" script_content_section = render_script_content_section(script, user) return page( html.escape(script.get("name", "") or "Plánovaný skript"), f"""

{html.escape(script.get("name", "") or "")}

← Zpět na plánované skripty Upravit

{delete_form}

Souhrn

Název{html.escape(script.get("name", "") or "")}
Popis{html.escape(script.get("description", "") or "")}
Skript{html.escape(script.get("script_name", "") or "")}
Frekvence{render_schedule_type(script.get("schedule_type"))}
Čas{html.escape(script.get("schedule_time", "") or "")}
Timeout{html.escape(str(script.get("timeout_seconds") or ""))}
Zapnuto{render_bool(script.get("is_enabled"))}
Běží{render_bool(script.get("is_running"))}
Poslední stav{render_status_pill(script.get("last_status")) if script.get("last_status") else ""}
Poslední chyba{html.escape(script.get("last_error_text", "") or "")}
Poslední běh{html.escape(script.get("last_run_at", "") or "")}
Další běh{html.escape(script.get("next_run_at", "") or "")}
Poslední úloha{last_job}
{script_content_section} """, user=user, ) @router.get("/scheduled-scripts/{script_id}/edit", response_class=HTMLResponse) def edit_scheduled_script_form(script_id: int, request: Request, user=Depends(require_user)): script = get_scheduled_script(script_id) if not script: raise HTTPException(status_code=404, detail="Scheduled script not found") return page( "Upravit plánovaný skript", f"""

Upravit plánovaný skript

← Zpět

{render_form(script, f"/portal/scheduled-scripts/{script_id}/edit", False)}
""", user=user, ) @router.post("/scheduled-scripts/{script_id}/edit") def update_scheduled_script_action( script_id: int, name: str = Form(...), description: str = Form(""), schedule_type: str = Form(...), schedule_time: str = Form(""), timeout_seconds: str = Form("300"), is_enabled: str | None = Form(None), user=Depends(require_user), ): script = get_scheduled_script(script_id) if not script: raise HTTPException(status_code=404, detail="Scheduled script not found") metadata = form_metadata(name, description, schedule_type, schedule_time, timeout_seconds, is_enabled) update_scheduled_script(script_id, metadata) log_audit_event( user, action="scheduled_script.updated", target_type="scheduled_script", target_id=script_id, metadata={"script_name": script.get("script_name")}, ) return RedirectResponse(url=f"/portal/scheduled-scripts/{script_id}", status_code=303) @router.post("/scheduled-scripts/{script_id}/run-now") def run_scheduled_script_now(script_id: int, user=Depends(require_user)): script = get_scheduled_script(script_id) if not script: raise HTTPException(status_code=404, detail="Scheduled script not found") job_id = create_job( job_type="run_script", target_type="scheduled_script", target_id=script.get("script_name"), payload={ "scheduled_script_id": script.get("id"), "script_name": script.get("script_name"), "timeout_seconds": script.get("timeout_seconds"), }, user=user, source="portal_manual", ) log_audit_event( user, action="scheduled_script.run_now", target_type="scheduled_script", target_id=script_id, metadata={"script_name": script.get("script_name"), "job_id": job_id}, ) return RedirectResponse(url=f"/portal/jobs/{job_id}", status_code=303) @router.post("/scheduled-scripts/{script_id}/script") def update_scheduled_script_file( script_id: int, content: str = Form(...), user=Depends(require_user), ): if not is_admin(user): raise HTTPException(status_code=403, detail="Skript může upravit pouze administrátor") script = get_scheduled_script(script_id) if not script: raise HTTPException(status_code=404, detail="Scheduled script not found") script_name = validate_script_name(script.get("script_name", "") or "") save_script_file(script_name, content) commit_and_push(f"maintenance/{script_name}", f"Úprava skriptu {script_name}", user) log_audit_event( user, action="scheduled_script.file_updated", target_type="scheduled_script", target_id=script_id, metadata={ "scheduled_script_id": script_id, "script_name": script_name, }, ) return RedirectResponse(url=f"/portal/scheduled-scripts/{script_id}", status_code=303) @router.post("/scheduled-scripts/{script_id}/toggle") def toggle_scheduled_script(script_id: int, user=Depends(require_user)): script = get_scheduled_script(script_id) if not script: raise HTTPException(status_code=404, detail="Scheduled script not found") enabled = not bool(script.get("is_enabled")) set_scheduled_script_enabled(script_id, enabled) log_audit_event( user, action="scheduled_script.enabled" if enabled else "scheduled_script.disabled", target_type="scheduled_script", target_id=script_id, metadata={"script_name": script.get("script_name")}, ) return RedirectResponse(url=f"/portal/scheduled-scripts/{script_id}", status_code=303) @router.post("/scheduled-scripts/{script_id}/delete") def delete_scheduled_script_action(script_id: int, user=Depends(require_user)): script = get_scheduled_script(script_id) if not script: raise HTTPException(status_code=404, detail="Scheduled script not found") if script.get("is_running"): raise HTTPException(status_code=409, detail="Běžící skript nelze smazat") if not delete_scheduled_script(script_id): raise HTTPException(status_code=409, detail="Skript nelze smazat") script_name = clean_optional(script.get("script_name")) if script_name: path = script_path(script_name) if path.exists(): path.unlink() commit_and_push(f"maintenance/{script_name}", f"Smazán skript {script_name}", user) log_audit_event( user, action="scheduled_script.deleted", target_type="scheduled_script", target_id=script_id, metadata={"script_name": script.get("script_name")}, ) return RedirectResponse(url="/portal/scheduled-scripts", status_code=303)