JiriUhlir
646 lines
26 KiB
Python
646 lines
26 KiB
Python
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.logging_config import get_logger
|
||
from app.db.scheduled_scripts import (
|
||
compute_next_run_at,
|
||
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'<option value="{value}"{selected_attr}>{render_schedule_type(value)}</option>')
|
||
return "".join(options)
|
||
|
||
|
||
def render_enabled_pill(value) -> str:
|
||
if value:
|
||
return '<span class="pill pill-success">zapnuto</span>'
|
||
return '<span class="pill pill-muted">vypnuto</span>'
|
||
|
||
|
||
def render_running_pill(value) -> str:
|
||
if value:
|
||
return '<span class="pill pill-warning">běží</span>'
|
||
return '<span class="pill pill-muted">neběží</span>'
|
||
|
||
|
||
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í, ale nesmí být ani tiché.
|
||
try:
|
||
os.chmod(path, 0o755)
|
||
except OSError as exc:
|
||
get_logger(__name__).warning("Nepodařilo se nastavit spustitelná práva na %s: %s", path, exc)
|
||
|
||
|
||
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"""
|
||
<label>Skript</label>
|
||
<input name="script_name" value="{script_name}" placeholder="maintenance.sh" required>
|
||
"""
|
||
else:
|
||
script_name_input = f"""
|
||
<label>Skript</label>
|
||
<input value="{script_name}" disabled>
|
||
"""
|
||
|
||
return f"""
|
||
<form method="post" action="{action}" class="metadata-form">
|
||
<label>Název</label>
|
||
<input name="name" value="{name}" required>
|
||
|
||
<label>Popis</label>
|
||
<textarea name="description" rows="3">{description}</textarea>
|
||
|
||
{script_name_input}
|
||
|
||
<label>Frekvence</label>
|
||
<select name="schedule_type">{render_schedule_options(schedule_type)}</select>
|
||
|
||
<label>Čas</label>
|
||
<input name="schedule_time" value="{schedule_time}" placeholder="např. 02:30">
|
||
|
||
<label>Timeout</label>
|
||
<input name="timeout_seconds" value="{timeout_seconds}" inputmode="numeric" required>
|
||
|
||
<label class="checkbox-label"><input type="checkbox" name="is_enabled" value="1"{enabled_checked}> Zapnuto</label>
|
||
|
||
<div class="form-actions">
|
||
<button type="submit"><i class="fa-solid fa-floppy-disk" aria-hidden="true"></i> Uložit</button>
|
||
</div>
|
||
</form>
|
||
"""
|
||
|
||
|
||
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'<p class="alert">{script_file["warning"]}</p>'
|
||
error = ""
|
||
if script_file.get("error"):
|
||
error = f'<p class="alert alert-danger">{script_file["error"]}</p>'
|
||
|
||
if not is_admin(user):
|
||
readonly_hint = '<p class="muted">Obsah skriptu je dostupný pouze pro čtení. Ukládat může jen administrátor.</p>'
|
||
return f"""
|
||
<div class="card">
|
||
<h2>Obsah skriptu</h2>
|
||
<p><strong>Soubor:</strong> {script_name_html}</p>
|
||
{warning}
|
||
{error}
|
||
{readonly_hint}
|
||
<textarea rows="18" readonly>{content}</textarea>
|
||
</div>
|
||
"""
|
||
|
||
return f"""
|
||
<div class="card">
|
||
<h2>Obsah skriptu</h2>
|
||
<p><strong>Soubor:</strong> {script_name_html}</p>
|
||
{warning}
|
||
{error}
|
||
<form method="post" action="/portal/scheduled-scripts/{html.escape(str(script.get("id")))}/script" class="metadata-form">
|
||
<label>Obsah souboru</label>
|
||
<textarea name="content" rows="22" spellcheck="false">{content}</textarea>
|
||
<div class="form-actions">
|
||
<button type="submit"><i class="fa-solid fa-floppy-disk" aria-hidden="true"></i> Uložit skript</button>
|
||
</div>
|
||
</form>
|
||
</div>
|
||
"""
|
||
|
||
|
||
@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"""
|
||
<form method="post" action="/portal/scheduled-scripts/{script_id}/delete" class="inline-form" onsubmit="return confirm('Smazat plánovaný skript {name}?');">
|
||
<button type="submit" class="icon-action icon-action-danger" title="Smazat" aria-label="Smazat"><i class="fa-solid fa-trash" aria-hidden="true"></i></button>
|
||
</form>
|
||
"""
|
||
|
||
rows += f"""
|
||
<tr>
|
||
<td><strong><a href="/portal/scheduled-scripts/{script_id}">{name}</a></strong></td>
|
||
<td>{script_name}</td>
|
||
<td>{schedule_type}</td>
|
||
<td>{schedule_time}</td>
|
||
<td>{render_enabled_pill(enabled)}</td>
|
||
<td>{render_running_pill(script.get("is_running"))}</td>
|
||
<td>{last_status}</td>
|
||
<td>{last_run_at}</td>
|
||
<td>{next_run_at}</td>
|
||
<td class="actions-cell service-actions">
|
||
<a class="icon-action" href="/portal/scheduled-scripts/{script_id}" title="Detail" aria-label="Detail"><i class="fa-solid fa-eye" aria-hidden="true"></i></a>
|
||
<form method="post" action="/portal/scheduled-scripts/{script_id}/run-now" class="inline-form">
|
||
<button type="submit" class="icon-action" title="Spustit nyní" aria-label="Spustit nyní"><i class="fa-solid fa-play" aria-hidden="true"></i></button>
|
||
</form>
|
||
<form method="post" action="/portal/scheduled-scripts/{script_id}/toggle" class="inline-form">
|
||
<button type="submit" class="icon-action" title="{toggle_label}" aria-label="{toggle_label}"><i class="fa-solid fa-power-off" aria-hidden="true"></i></button>
|
||
</form>
|
||
{delete_button}
|
||
</td>
|
||
</tr>
|
||
"""
|
||
|
||
if not rows:
|
||
rows = '<tr><td colspan="10">Zatím nejsou evidované žádné plánované skripty.</td></tr>'
|
||
|
||
return page(
|
||
"Plánované skripty",
|
||
f"""
|
||
<div class="card">
|
||
<h2><i class="fa-solid fa-calendar-days" aria-hidden="true"></i> Plánované skripty</h2>
|
||
<p class="muted">Přehled skriptů spouštěných schedulerem nebo ručně z portálu.</p>
|
||
<p><a class="btn" href="/portal/scheduled-scripts/new"><i class="fa-solid fa-plus" aria-hidden="true"></i> Nový plánovaný skript</a></p>
|
||
</div>
|
||
|
||
<div class="card">
|
||
<h2>Seznam</h2>
|
||
<table>
|
||
<tr>
|
||
<th>Název</th>
|
||
<th>Skript</th>
|
||
<th>Frekvence</th>
|
||
<th>Čas</th>
|
||
<th>Zapnuto</th>
|
||
<th>Běží</th>
|
||
<th>Poslední stav</th>
|
||
<th>Poslední běh</th>
|
||
<th>Další běh</th>
|
||
<th>Akce</th>
|
||
</tr>
|
||
{rows}
|
||
</table>
|
||
</div>
|
||
""",
|
||
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"""
|
||
<div class="card">
|
||
<h2><i class="fa-solid fa-circle-plus" aria-hidden="true"></i> Nový plánovaný skript</h2>
|
||
<p><a class="btn" href="/portal/scheduled-scripts">← Zpět</a></p>
|
||
</div>
|
||
<div class="card">
|
||
{render_form({"is_enabled": True, "timeout_seconds": 300, "schedule_type": "daily"}, "/portal/scheduled-scripts/new", True)}
|
||
</div>
|
||
""",
|
||
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)
|
||
# Zapnutý skript nesmí skončit s next_run_at = NULL. Když čas nelze spočítat,
|
||
# ukaž adminovi chybu místo tichého uložení nespustitelného skriptu.
|
||
try:
|
||
script_id = create_scheduled_script(metadata)
|
||
except ValueError as exc:
|
||
raise HTTPException(status_code=400, detail=f"Nelze spočítat další běh: {exc}")
|
||
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'<a href="/portal/jobs/{last_job_html}">#{last_job_html}</a>'
|
||
|
||
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"""
|
||
<form method="post" action="/portal/scheduled-scripts/{script_id}/delete" onsubmit="return confirm('Smazat plánovaný skript?');">
|
||
<button type="submit" class="danger"><i class="fa-solid fa-trash" aria-hidden="true"></i> Smazat</button>
|
||
</form>
|
||
"""
|
||
script_content_section = render_script_content_section(script, user)
|
||
|
||
return page(
|
||
html.escape(script.get("name", "") or "Plánovaný skript"),
|
||
f"""
|
||
<div class="card">
|
||
<h2><i class="fa-solid fa-calendar-day" aria-hidden="true"></i> {html.escape(script.get("name", "") or "")}</h2>
|
||
<p>
|
||
<a class="btn" href="/portal/scheduled-scripts">← Zpět na plánované skripty</a>
|
||
<a class="btn btn-secondary" href="/portal/scheduled-scripts/{script_id}/edit"><i class="fa-solid fa-pen-to-square" aria-hidden="true"></i> Upravit</a>
|
||
</p>
|
||
<div class="inline-form">
|
||
<form method="post" action="/portal/scheduled-scripts/{script_id}/run-now">
|
||
<button type="submit"><i class="fa-solid fa-play" aria-hidden="true"></i> Spustit nyní</button>
|
||
</form>
|
||
<form method="post" action="/portal/scheduled-scripts/{script_id}/toggle">
|
||
<button type="submit" class="btn-secondary"><i class="fa-solid fa-power-off" aria-hidden="true"></i> {toggle_label}</button>
|
||
</form>
|
||
{delete_form}
|
||
</div>
|
||
</div>
|
||
|
||
<div class="card">
|
||
<h2>Souhrn</h2>
|
||
<table>
|
||
<tr><th>Název</th><td>{html.escape(script.get("name", "") or "")}</td></tr>
|
||
<tr><th>Popis</th><td>{html.escape(script.get("description", "") or "")}</td></tr>
|
||
<tr><th>Skript</th><td>{html.escape(script.get("script_name", "") or "")}</td></tr>
|
||
<tr><th>Frekvence</th><td>{render_schedule_type(script.get("schedule_type"))}</td></tr>
|
||
<tr><th>Čas</th><td>{html.escape(script.get("schedule_time", "") or "")}</td></tr>
|
||
<tr><th>Timeout</th><td>{html.escape(str(script.get("timeout_seconds") or ""))}</td></tr>
|
||
<tr><th>Zapnuto</th><td>{render_bool(script.get("is_enabled"))}</td></tr>
|
||
<tr><th>Běží</th><td>{render_bool(script.get("is_running"))}</td></tr>
|
||
<tr><th>Poslední stav</th><td>{render_status_pill(script.get("last_status")) if script.get("last_status") else ""}</td></tr>
|
||
<tr><th>Poslední chyba</th><td>{html.escape(script.get("last_error_text", "") or "")}</td></tr>
|
||
<tr><th>Poslední běh</th><td>{html.escape(script.get("last_run_at", "") or "")}</td></tr>
|
||
<tr><th>Další běh</th><td>{html.escape(script.get("next_run_at", "") or "")}</td></tr>
|
||
<tr><th>Poslední úloha</th><td>{last_job}</td></tr>
|
||
</table>
|
||
</div>
|
||
|
||
{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"""
|
||
<div class="card">
|
||
<h2><i class="fa-solid fa-pen-to-square" aria-hidden="true"></i> Upravit plánovaný skript</h2>
|
||
<p><a class="btn" href="/portal/scheduled-scripts/{script_id}">← Zpět</a></p>
|
||
</div>
|
||
<div class="card">
|
||
{render_form(script, f"/portal/scheduled-scripts/{script_id}/edit", False)}
|
||
</div>
|
||
""",
|
||
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)
|
||
# Při editaci se přepočítává next_run_at; neplatnou kombinaci hlásíme adminovi.
|
||
try:
|
||
update_scheduled_script(script_id, metadata)
|
||
except ValueError as exc:
|
||
raise HTTPException(status_code=400, detail=f"Nelze spočítat další běh: {exc}")
|
||
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"))
|
||
# Zapnutí dopočítá next_run_at, vypnutí ho nastaví na NULL (řeší DB vrstva).
|
||
next_run_at = None
|
||
if enabled:
|
||
try:
|
||
next_run_at = compute_next_run_at(
|
||
script.get("schedule_type"), script.get("schedule_time")
|
||
)
|
||
except ValueError as exc:
|
||
raise HTTPException(status_code=400, detail=f"Nelze spočítat další běh: {exc}")
|
||
set_scheduled_script_enabled(script_id, enabled, next_run_at)
|
||
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)
|