import html
from pathlib import Path
from urllib.parse import quote
from fastapi import APIRouter, Depends, Form, HTTPException, Request
from fastapi.responses import FileResponse, HTMLResponse, RedirectResponse
from ..auth import require_user
from ..backups import backup_dir, is_backup_path, list_backups
from ..config import BACKUP_SCRIPT, RESTORE_SCRIPT
from ..db.audit import log_audit_event
from ..shell import run_command
from ..templates.layout import page, render_result
router = APIRouter()
def is_admin(user) -> bool:
return (user.get("role") or "").lower() == "admin"
@router.get("/backups", response_class=HTMLResponse)
def backups_page(request: Request, user=Depends(require_user)):
rows = ""
can_download = is_admin(user)
for path in list_backups():
size_mb = path.stat().st_size / 1024 / 1024
backup_name = html.escape(path.name)
backup_path = html.escape(str(path), quote=True)
backup_url_name = quote(path.name, safe="")
download_action = (
f''
f' Stáhnout'
if can_download
else ""
)
# Restore overwrites live data, so it is admin-only and guarded by an explicit confirm.
restore_action = (
f"""
"""
if can_download
else ""
)
rows += f"""
{backup_name} {size_mb:.2f} MB
{download_action}
{restore_action}
"""
if not rows:
rows = '
Nebyly nalezeny žádné zálohy.
'
restore_hint = (
"Obnovu lze spustit přímo z portálu tlačítkem Obnovit (pouze administrátor)."
if can_download
else "Obnovu zálohy může spustit pouze administrátor."
)
return page(
"Zálohy",
f"""
Správa záloh
Zálohy neobsahují samotný adresář se zálohami.
Obnova starší zálohy by neměla smazat novější soubory .tar.gz.
{restore_hint} Obnova přepíše aktuální data a je nevratná.
Dostupné zálohy
Záloha
Cesta
Akce
{rows}
""",
user=user,
)
@router.get("/backups/download/{backup_name}")
def download_backup(backup_name: str, user=Depends(require_user)):
if not is_admin(user):
raise HTTPException(status_code=403, detail="Only admins can download backups")
target = backup_dir() / backup_name
if not is_backup_path(target):
raise HTTPException(status_code=400, detail="Invalid backup path")
target = target.resolve()
if not target.exists() or not target.is_file():
raise HTTPException(status_code=404, detail="Backup not found")
log_audit_event(
user,
action="backup_download",
target_type="backup",
target_id=target.name,
metadata={"backup_path": str(target)},
)
return FileResponse(
path=target,
filename=target.name,
media_type="application/gzip",
)
@router.post("/backups/create", response_class=HTMLResponse)
def create_backup(user=Depends(require_user)):
result = run_command([BACKUP_SCRIPT])
status = "OK" if result.returncode == 0 else "FAILED"
log_audit_event(
user,
action="backup_create",
target_type="backup",
metadata={
"status": status,
"returncode": result.returncode,
},
)
return render_result(
title=f"Vytvoření zálohy: {status}",
back_url="/portal/backups",
sections=[("Výstup", result.stdout), ("Chyba", result.stderr)],
user=user,
)
@router.post("/backups/delete")
def delete_backup(backup_path: str = Form(...), user=Depends(require_user)):
target = Path(backup_path)
if not is_backup_path(target):
return HTMLResponse("Neplatná cesta k záloze", status_code=400)
target = target.resolve()
if target.exists():
target.unlink()
return RedirectResponse(url="/portal/backups", status_code=303)
@router.post("/backups/restore", response_class=HTMLResponse)
def restore_backup(backup_path: str = Form(...), user=Depends(require_user)):
# Restoring overwrites live data and may restart services, so it is restricted to admins.
if not is_admin(user):
raise HTTPException(status_code=403, detail="Only admins can restore backups")
target = Path(backup_path)
if not is_backup_path(target):
return HTMLResponse("Neplatná cesta k záloze", status_code=400)
target = target.resolve()
if not target.exists() or not target.is_file():
raise HTTPException(status_code=404, detail="Backup not found")
result = run_command([RESTORE_SCRIPT, "--force", str(target)])
status = "OK" if result.returncode == 0 else "FAILED"
log_audit_event(
user,
action="backup_restore",
target_type="backup",
target_id=target.name,
metadata={
"backup_path": str(target),
"status": status,
"returncode": result.returncode,
},
)
return render_result(
title=f"Obnova zálohy: {status}",
back_url="/portal/backups",
sections=[("Výstup", result.stdout), ("Chyba", result.stderr)],
user=user,
)