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"""
"""
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} |
Detail
{delete_button}
|
"""
if not rows:
rows = '| Zatím nejsou evidované žádné plánované skripty. |
'
return page(
"Plánované skripty",
f"""
Seznam
| Název |
Skript |
Frekvence |
Čas |
Zapnuto |
Běží |
Poslední stav |
Poslední běh |
Další běh |
Akce |
{rows}
""",
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"""
{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"""
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)