import html
from pathlib import Path
from fastapi import APIRouter, Depends, Form, Request
from fastapi.responses import HTMLResponse, RedirectResponse
from ..auth import require_user
from ..backups import is_backup_path, list_backups
from ..config import BACKUP_SCRIPT
from ..db.audit import log_audit_event
from ..shell import run_command
from ..templates.layout import page, render_result
router = APIRouter()
@router.get("/backups", response_class=HTMLResponse)
def backups_page(request: Request, user=Depends(require_user)):
rows = ""
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)
restore_cmd = html.escape(
f"sudo /home/jiri/workspace/appfactory-tools/scripts/restore-appfactory.sh --force {path}",
quote=True,
)
rows += f"""
{backup_name} {size_mb:.2f} MB
Obnova je nebezpečná a musí se spustit ručně přes SSH.
"""
if not rows:
rows = '
Nebyly nalezeny žádné zálohy.
'
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.
Obnova zůstává ruční, aby nedošlo k nechtěnému přepsání dat.
Dostupné zálohy
Záloha
Cesta
Příkaz pro obnovu
Akce
{rows}
""",
user=user,
)
@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-trigger")
def restore_trigger(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)
log_audit_event(
user,
action="restore_trigger",
target_type="backup",
target_id=target.name,
metadata={
"backup_path": str(target),
},
)
return RedirectResponse(url="/portal/backups", status_code=303)