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 router = APIRouter() MAINTENANCE_DIR = Path("/opt/appfactory/workspace/appfactory-tools/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.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 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"""
""" 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"""Soubor: {script_name_html}
{warning} {error} {readonly_hint}Soubor: {script_name_html}
{warning} {error}Přehled skriptů spouštěných schedulerem nebo ručně z portálu.
| Název | Skript | Frekvence | Čas | Zapnuto | Běží | Poslední stav | Poslední běh | Další běh | Akce |
|---|
← Zpět na plánované skripty Upravit
| 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} |