Files
appfactory-portal/app/routes/scheduled_scripts.py
T
2026-06-15 11:50:33 +02:00

688 lines
27 KiB
Python

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'<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&ecaron;&zcaron;&iacute;</span>'
return '<span class="pill pill-muted">neb&ecaron;&zcaron;&iacute;</span>'
def validate_script_name(value: str) -> str:
script_name = clean_optional(value)
if not script_name:
raise HTTPException(status_code=400, detail="N&aacute;zev skriptu je povinn&yacute;")
if not script_name.endswith(".sh"):
raise HTTPException(status_code=400, detail="N&aacute;zev skriptu mus&iacute; kon&ccaron;it .sh")
if "/" in script_name or "\\" in script_name or ".." in script_name:
raise HTTPException(status_code=400, detail="N&aacute;zev skriptu nesm&iacute; 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&yacute; n&aacute;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&iacute;m neexistuje. M&uring;&zcaron;ete ho vytvo&rcaron;it z v&yacute;choz&iacute;ho obsahu.",
}
if not path.is_file():
return {"exists": False, "content": "", "error": "Cesta nen&iacute; soubor.", "warning": None}
try:
if path.stat().st_size > MAX_SCRIPT_BYTES:
return {"exists": True, "content": "", "error": "Soubor je v&ecaron;t&scaron;&iacute; ne&zcaron; 100 KB.", "warning": None}
content = path.read_text(encoding="utf-8")
except Exception:
return {"exists": True, "content": "", "error": "Soubor se nepoda&rcaron;ilo na&ccaron;&iacute;st.", "warning": None}
warning = None
if not content.startswith("#!/usr/bin/env bash"):
warning = "Prvn&iacute; &rcaron;&aacute;dek by m&ecaron;l b&yacute;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&iacute; b&yacute;t pr&aacute;zdn&yacute;")
if len(value.encode("utf-8")) > MAX_SCRIPT_BYTES:
raise HTTPException(status_code=400, detail="Obsah skriptu je v&ecaron;t&scaron;&iacute; ne&zcaron; 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&ecaron;t&scaron;&iacute; ne&zcaron; 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&rcaron;ilo ulo&zcaron;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&rcaron;&iacute;kaz selhal nebo vypr&scaron;el limit")
def _git_actor(user: dict) -> str:
return (user.get("email") or user.get("display_name") or user.get("username") or "nezn&aacute;m&yacute;").strip()
def _gitea_push_remote() -> tuple[str, str | None]:
"""Vr&aacute;t&iacute; (remote, token). Remote je autentizovan&aacute; gitea URL z token&uring; v prom&ecaron;nn&yacute;ch,
p&rcaron;i absenci tokenu fallback na origin. Token vrac&iacute;me zvl&aacute;&scaron;&tcaron;, aby &scaron;el zamaskovat v chyb&aacute;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 commit_and_push(script_name: str, message: str, user: dict) -> None:
"""Zacommituje a pushne zm&ecaron;nu jednoho maintenance skriptu do gitea (appfactory-tools)."""
rel_path = f"maintenance/{script_name}"
if _git(["add", "--", rel_path]).returncode != 0:
raise HTTPException(status_code=500, detail="Git: soubor se nepoda&rcaron;ilo p&rcaron;idat do indexu")
# &Zcaron;&aacute;dn&aacute; zm&ecaron;na oproti HEAD -> p&rcaron;esko&ccaron;&iacute;me, a&tcaron; nevznikaj&iacute; pr&aacute;zdn&eacute; 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&aacute;l: {_git_actor(user)})",
])
if commit.returncode != 0:
raise HTTPException(status_code=500, detail="Git commit selhal")
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:
detail = (push.stderr or "").strip()
if token:
detail = detail.replace(token, "***")
raise HTTPException(status_code=500, detail=f"Git push selhal: {html.escape(detail)}")
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&aacute; 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&iacute; b&yacute;t &ccaron;&iacute;slo")
if timeout_seconds < 10 or timeout_seconds > 3600:
raise HTTPException(status_code=400, detail="Timeout mus&iacute; b&yacute;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&aacute;zev je povinn&yacute;")
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&aacute;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>&Ccaron;as</label>
<input name="schedule_time" value="{schedule_time}" placeholder="nap&rcaron;. 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">Ulo&zcaron;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&aacute;zev skriptu nen&iacute; bezpe&ccaron;n&yacute;.",
"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&yacute; pouze pro &ccaron;ten&iacute;. Ukl&aacute;dat m&uring;&zcaron;e jen administr&aacute;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">Ulo&zcaron;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&aacute;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&aacute;novan&yacute; skript {name}?');">
<button type="submit" class="danger">Smazat</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="btn" href="/portal/scheduled-scripts/{script_id}">Detail</a>
<form method="post" action="/portal/scheduled-scripts/{script_id}/run-now" class="inline-form">
<button type="submit" class="btn btn-secondary">Spustit nyn&iacute;</button>
</form>
<form method="post" action="/portal/scheduled-scripts/{script_id}/toggle" class="inline-form">
<button type="submit" class="btn btn-secondary">{toggle_label}</button>
</form>
{delete_button}
</td>
</tr>
"""
if not rows:
rows = '<tr><td colspan="10">Zat&iacute;m nejsou evidovan&eacute; &zcaron;&aacute;dn&eacute; pl&aacute;novan&eacute; skripty.</td></tr>'
return page(
"Pl&aacute;novan&eacute; skripty",
f"""
<div class="card">
<h2>Pl&aacute;novan&eacute; skripty</h2>
<p class="muted">P&rcaron;ehled skript&uring; spou&scaron;t&ecaron;n&yacute;ch schedulerem nebo ru&ccaron;n&ecaron; z port&aacute;lu.</p>
<p><a class="btn" href="/portal/scheduled-scripts/new">+ Nov&yacute; pl&aacute;novan&yacute; skript</a></p>
</div>
<div class="card">
<h2>Seznam</h2>
<table>
<tr>
<th>N&aacute;zev</th>
<th>Skript</th>
<th>Frekvence</th>
<th>&Ccaron;as</th>
<th>Zapnuto</th>
<th>B&ecaron;&zcaron;&iacute;</th>
<th>Posledn&iacute; stav</th>
<th>Posledn&iacute; b&ecaron;h</th>
<th>Dal&scaron;&iacute; b&ecaron;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&yacute; pl&aacute;novan&yacute; skript",
f"""
<div class="card">
<h2>Nov&yacute; pl&aacute;novan&yacute; skript</h2>
<p><a class="btn" href="/portal/scheduled-scripts">&larr; Zp&ecaron;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)
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(created_name, f"Vytvo&rcaron;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&aacute;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&aacute;novan&yacute; skript?');">
<button type="submit" class="danger">Smazat</button>
</form>
"""
script_content_section = render_script_content_section(script, user)
return page(
html.escape(script.get("name", "") or "Pl&aacute;novan&yacute; skript"),
f"""
<div class="card">
<h2>{html.escape(script.get("name", "") or "")}</h2>
<p>
<a class="btn" href="/portal/scheduled-scripts">&larr; Zp&ecaron;t na pl&aacute;novan&eacute; skripty</a>
<a class="btn btn-secondary" href="/portal/scheduled-scripts/{script_id}/edit">Upravit</a>
</p>
<div class="inline-form">
<form method="post" action="/portal/scheduled-scripts/{script_id}/run-now">
<button type="submit">Spustit nyn&iacute;</button>
</form>
<form method="post" action="/portal/scheduled-scripts/{script_id}/toggle">
<button type="submit" class="btn-secondary">{toggle_label}</button>
</form>
{delete_form}
</div>
</div>
<div class="card">
<h2>Souhrn</h2>
<table>
<tr><th>N&aacute;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>&Ccaron;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&ecaron;&zcaron;&iacute;</th><td>{render_bool(script.get("is_running"))}</td></tr>
<tr><th>Posledn&iacute; stav</th><td>{render_status_pill(script.get("last_status")) if script.get("last_status") else ""}</td></tr>
<tr><th>Posledn&iacute; chyba</th><td>{html.escape(script.get("last_error_text", "") or "")}</td></tr>
<tr><th>Posledn&iacute; b&ecaron;h</th><td>{html.escape(script.get("last_run_at", "") or "")}</td></tr>
<tr><th>Dal&scaron;&iacute; b&ecaron;h</th><td>{html.escape(script.get("next_run_at", "") or "")}</td></tr>
<tr><th>Posledn&iacute; &uacute;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&aacute;novan&yacute; skript",
f"""
<div class="card">
<h2>Upravit pl&aacute;novan&yacute; skript</h2>
<p><a class="btn" href="/portal/scheduled-scripts/{script_id}">&larr; Zp&ecaron;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)
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&uring;&zcaron;e upravit pouze administr&aacute;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(script_name, f"&Uacute;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&ecaron;&zcaron;&iacute;c&iacute; 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(script_name, f"Smaz&aacute;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)