Admin MPCZ
77e884d620
- Service: regex stricte '<auteur>_YYYY-MM-DD_avant_patch' (avant: laxiste avec suffixe optionnel)
- Champ is_patchcenter_format ajoute aux snapshots, et auteur seulement si format match
- Router: _get_user_intervenant_name lit JWT 'sub' (correctif - etait 'username' qui n'existe pas)
- UI:
* Nouveau filtre 'Format PatchCenter uniquement' (checkbox, default ON)
* Filtre 'Mes snapshots' marche meme si auteur input vide -> on garde uniquement
ceux dont l'auteur est connu (= snapshots PatchCenter)
* Dates: formattees jj/mm/aaaa HH:MM (fmtDateFR via Date object navigateur)
* Cellule auteur 'inconnu' rendue avec balise <i> proprement (bypass escapeHTML)
* Helper setBusy/clearBusy pour feedback unifie '⏳ Recherche en cours…' / '⏳ Suppression en cours…'
(status + texte du bouton change pendant l'action)
123 lines
4.6 KiB
Python
123 lines
4.6 KiB
Python
"""Router /snapshots — listing + suppression des snapshots VM existants sur les vCenters."""
|
|
import logging
|
|
from fastapi import APIRouter, Request, Depends, Form
|
|
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
|
|
from fastapi.templating import Jinja2Templates
|
|
from sqlalchemy import text
|
|
|
|
from ..dependencies import get_db, get_current_user, get_user_perms, can_view, can_edit, base_context
|
|
from ..config import APP_NAME
|
|
|
|
log = logging.getLogger("patchcenter.snapshots_router")
|
|
|
|
router = APIRouter()
|
|
templates = Jinja2Templates(directory="app/templates")
|
|
|
|
|
|
def _can_view_snaps(perms):
|
|
return can_view(perms, "campaigns") or can_view(perms, "planning") or can_view(perms, "settings")
|
|
|
|
|
|
def _can_delete_snaps(perms):
|
|
return can_edit(perms, "campaigns") or can_edit(perms, "planning") or can_edit(perms, "settings")
|
|
|
|
|
|
def _get_user_intervenant_name(db, user):
|
|
"""Retourne le nom utilisé comme préfixe dans les snapshots PatchCenter de cet utilisateur.
|
|
Priorité : settings 'patcher' > username (sub du JWT) > display_name."""
|
|
try:
|
|
from ..services.secrets_service import get_secret
|
|
patcher = (get_secret(db, "patcher") or "").strip()
|
|
if patcher:
|
|
return patcher
|
|
except Exception:
|
|
pass
|
|
# JWT stocke le username dans 'sub' (cf auth.py)
|
|
return (user.get("sub") or user.get("username") or user.get("display_name") or "").strip()
|
|
|
|
|
|
@router.get("/snapshots", response_class=HTMLResponse)
|
|
async def snapshots_page(request: Request, db=Depends(get_db)):
|
|
user = get_current_user(request)
|
|
if not user:
|
|
return RedirectResponse(url="/login")
|
|
perms = get_user_perms(db, user)
|
|
if not _can_view_snaps(perms):
|
|
return RedirectResponse(url="/dashboard")
|
|
|
|
intervenant = _get_user_intervenant_name(db, user)
|
|
vcenters = db.execute(text(
|
|
"SELECT id, name, endpoint FROM vcenters WHERE is_active = true ORDER BY name"
|
|
)).fetchall()
|
|
|
|
ctx = base_context(request, db, user)
|
|
ctx.update({
|
|
"app_name": APP_NAME,
|
|
"intervenant_default": intervenant,
|
|
"vcenters": vcenters,
|
|
"can_delete": _can_delete_snaps(perms),
|
|
})
|
|
return templates.TemplateResponse("snapshots.html", ctx)
|
|
|
|
|
|
@router.post("/snapshots/list")
|
|
async def snapshots_list(request: Request, db=Depends(get_db),
|
|
vcenter_id: str = Form("")):
|
|
"""Endpoint AJAX : liste les snapshots (tous vCenters ou un seul)."""
|
|
user = get_current_user(request)
|
|
if not user:
|
|
return JSONResponse({"ok": False, "msg": "Non authentifié"}, status_code=401)
|
|
perms = get_user_perms(db, user)
|
|
if not _can_view_snaps(perms):
|
|
return JSONResponse({"ok": False, "msg": "Permission refusée"}, status_code=403)
|
|
|
|
from ..services.snapshot_mgmt_service import list_snapshots
|
|
vc_id = int(vcenter_id) if vcenter_id and vcenter_id.isdigit() else None
|
|
res = list_snapshots(db, vcenter_filter_id=vc_id)
|
|
return JSONResponse(res)
|
|
|
|
|
|
@router.post("/snapshots/delete")
|
|
async def snapshots_delete(request: Request, db=Depends(get_db)):
|
|
"""Endpoint AJAX : supprime une liste de snapshots.
|
|
Body JSON : [{vcenter_id, vm_moid, snap_id}, ...]"""
|
|
user = get_current_user(request)
|
|
if not user:
|
|
return JSONResponse({"ok": False, "msg": "Non authentifié"}, status_code=401)
|
|
perms = get_user_perms(db, user)
|
|
if not _can_delete_snaps(perms):
|
|
return JSONResponse({"ok": False, "msg": "Permission refusée"}, status_code=403)
|
|
|
|
try:
|
|
body = await request.json()
|
|
except Exception:
|
|
return JSONResponse({"ok": False, "msg": "Body JSON invalide"}, status_code=400)
|
|
items = body.get("items") if isinstance(body, dict) else None
|
|
if not isinstance(items, list) or not items:
|
|
return JSONResponse({"ok": False, "msg": "Aucun snapshot ciblé"}, status_code=400)
|
|
|
|
from ..services.snapshot_mgmt_service import delete_snapshot
|
|
results = []
|
|
n_ok = 0
|
|
for it in items:
|
|
try:
|
|
r = delete_snapshot(
|
|
db,
|
|
vcenter_id=int(it["vcenter_id"]),
|
|
vm_moid=str(it["vm_moid"]),
|
|
snap_id=str(it["snap_id"]),
|
|
remove_children=bool(it.get("remove_children", False)),
|
|
)
|
|
except Exception as e:
|
|
r = {"ok": False, "msg": f"Param error: {e}"}
|
|
r["vm_name"] = it.get("vm_name", "?")
|
|
r["snap_name"] = it.get("snap_name", "?")
|
|
results.append(r)
|
|
if r.get("ok"):
|
|
n_ok += 1
|
|
return JSONResponse({
|
|
"ok": n_ok == len(items),
|
|
"summary": f"{n_ok}/{len(items)} snapshots supprimés",
|
|
"results": results,
|
|
})
|