import html import os import subprocess 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.config import ( DEFAULT_GITEA_ORG, get_gitea_admin_token, get_gitea_server_url, read_env_value, ) 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() TOOLS_REPO_DIR = Path("/opt/appfactory/workspace/appfactory-tools") TOOLS_REPO_NAME = "appfactory-tools" MAINTENANCE_DIR = TOOLS_REPO_DIR / "maintenance" GIT_AUTHOR_NAME = "AppFactory Portal" GIT_AUTHOR_EMAIL = "portal@appfactory.local" 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 _git(args: list[str]): env = {**os.environ, "GIT_TERMINAL_PROMPT": "0"} try: return subprocess.run( ["git", "-C", str(TOOLS_REPO_DIR), *args], capture_output=True, text=True, env=env, timeout=30, ) except (subprocess.TimeoutExpired, OSError): raise HTTPException(status_code=500, detail="Git příkaz selhal nebo vypršel limit") def _git_actor(user: dict) -> str: return (user.get("email") or user.get("display_name") or user.get("username") or "neznámý").strip() def _gitea_push_remote() -> tuple[str, str | None]: """Vrátí (remote, token). Remote je autentizovaná gitea URL z tokenů v proměnných, při absenci tokenu fallback na origin. Token vracíme zvlášť, aby šel zamaskovat v chybách.""" token = get_gitea_admin_token() server = get_gitea_server_url() if not token or "://" not in server: return "origin", None scheme, rest = server.split("://", 1) org = read_env_value("GITEA_ORG", DEFAULT_GITEA_ORG) return f"{scheme}://{token}@{rest}/{org}/{TOOLS_REPO_NAME}.git", token def _git_error_detail(result, token: str | None = None) -> str: """Zkombinuje git stderr/stdout do čitelného důvodu chyby (s maskováním tokenu).""" detail = (result.stderr or "").strip() or (result.stdout or "").strip() if token and detail: detail = detail.replace(token, "***") return html.escape(detail) if detail else "git nevrátil žádný výstup" def commit_and_push(script_name: str, message: str, user: dict) -> None: """Zacommituje a pushne změnu jednoho maintenance skriptu do gitea (appfactory-tools).""" rel_path = f"maintenance/{script_name}" add = _git(["add", "--", rel_path]) if add.returncode != 0: raise HTTPException(status_code=500, detail=f"Git add selhal (v {TOOLS_REPO_DIR}): {_git_error_detail(add)}") # Žádná změna oproti HEAD -> přeskočíme, ať nevznikají prázdné commity. if _git(["diff", "--cached", "--quiet", "--", rel_path]).returncode == 0: return commit = _git([ "-c", f"user.name={GIT_AUTHOR_NAME}", "-c", f"user.email={GIT_AUTHOR_EMAIL}", "commit", "-m", f"{message} (portál: {_git_actor(user)})", ]) if commit.returncode != 0: raise HTTPException(status_code=500, detail=f"Git commit selhal (v {TOOLS_REPO_DIR}): {_git_error_detail(commit)}") branch = (_git(["rev-parse", "--abbrev-ref", "HEAD"]).stdout or "").strip() or "main" remote, token = _gitea_push_remote() push = _git(["push", remote, f"HEAD:{branch}"]) if push.returncode != 0: raise HTTPException(status_code=500, detail=f"Git push selhal: {_git_error_detail(push, token)}") 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} |