Adriano
57376556d2
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Jbf7qKuxsnAVjxthRy7dnB
286 lines
10 KiB
Python
286 lines
10 KiB
Python
"""Admin blueprint - user management."""
|
|
from urllib.parse import urlparse # #88: validazione url camera
|
|
import requests as http_requests # #88: lettura sorgenti dall'acquisitore (come maker.py)
|
|
from flask import Blueprint, flash, jsonify, redirect, render_template, request, session, url_for
|
|
from flask_babel import gettext as _
|
|
|
|
from blueprints.auth import login_required
|
|
from services.api_client import api_client
|
|
|
|
admin_bp = Blueprint("admin", __name__, url_prefix="/admin")
|
|
|
|
|
|
def admin_required(f):
|
|
"""Decorator to require admin privileges."""
|
|
from functools import wraps
|
|
@wraps(f)
|
|
def decorated(*args, **kwargs):
|
|
user = session.get("user", {})
|
|
if not user.get("is_admin"):
|
|
flash(_("Accesso non autorizzato"), "error")
|
|
return redirect(url_for("measure.select_recipe"))
|
|
return f(*args, **kwargs)
|
|
return decorated
|
|
|
|
|
|
# ============================================================================
|
|
# PAGINE (GET)
|
|
# ============================================================================
|
|
|
|
@admin_bp.route("/users")
|
|
@login_required
|
|
@admin_required
|
|
def user_list():
|
|
"""#50: la gestione utenti vive nella tab «Utenti» di Impostazioni."""
|
|
return redirect(url_for("admin.settings_page", tab="utenti"))
|
|
|
|
|
|
@admin_bp.route("/settings")
|
|
@login_required
|
|
@admin_required
|
|
def settings_page():
|
|
"""System settings page (admin only), con la tab «Utenti» (#50)."""
|
|
resp = api_client.get("/api/settings")
|
|
# Solo un dict valido sono impostazioni: un errore o altro tipo = nessuna
|
|
settings = resp if isinstance(resp, dict) and not resp.get("error") else {}
|
|
|
|
users_resp = api_client.get("/api/users")
|
|
if isinstance(users_resp, dict) and users_resp.get("error"):
|
|
flash(_("Errore nel caricamento degli utenti: %(error)s", error=users_resp.get("detail", "")), "error")
|
|
users = []
|
|
elif isinstance(users_resp, list):
|
|
users = users_resp
|
|
else:
|
|
users = []
|
|
|
|
tab = "utenti" if request.args.get("tab") == "utenti" else "generale"
|
|
return render_template("admin/settings.html", settings=settings, users=users, active_tab=tab)
|
|
|
|
|
|
@admin_bp.route("/api/settings", methods=["PUT"])
|
|
@login_required
|
|
@admin_required
|
|
def api_update_settings():
|
|
"""Proxy: Update system settings."""
|
|
data = request.get_json(silent=True) or {}
|
|
resp = api_client.put("/api/settings", data=data)
|
|
if isinstance(resp, dict) and resp.get("error"):
|
|
return jsonify(resp), resp.get("status_code", 500)
|
|
return jsonify(resp), 200
|
|
|
|
|
|
@admin_bp.route("/stations")
|
|
@login_required
|
|
@admin_required
|
|
def station_list():
|
|
"""Station management page."""
|
|
resp = api_client.get("/api/stations")
|
|
if isinstance(resp, dict) and resp.get("error"):
|
|
flash(_("Errore nel caricamento delle stazioni: %(error)s", error=resp.get("detail", "")), "error")
|
|
stations = []
|
|
elif isinstance(resp, list):
|
|
stations = resp
|
|
else:
|
|
stations = []
|
|
|
|
recipes_resp = api_client.get("/api/recipes")
|
|
if isinstance(recipes_resp, list):
|
|
all_recipes = recipes_resp
|
|
elif isinstance(recipes_resp, dict) and isinstance(recipes_resp.get("items"), list):
|
|
all_recipes = recipes_resp["items"]
|
|
else:
|
|
all_recipes = []
|
|
|
|
return render_template("admin/stations.html", stations=stations, all_recipes=all_recipes)
|
|
|
|
|
|
# ============================================================================
|
|
# API PROXY AJAX (JSON)
|
|
# ============================================================================
|
|
|
|
@admin_bp.route("/api/users", methods=["POST"])
|
|
@login_required
|
|
@admin_required
|
|
def api_create_user():
|
|
"""Proxy: Create new user."""
|
|
data = request.get_json(silent=True) or {}
|
|
resp = api_client.post("/api/users", data=data)
|
|
|
|
if isinstance(resp, dict) and resp.get("error"):
|
|
return jsonify(resp), resp.get("status_code", 500)
|
|
|
|
return jsonify(resp), 201
|
|
|
|
|
|
@admin_bp.route("/api/users/<int:user_id>", methods=["PUT"])
|
|
@login_required
|
|
@admin_required
|
|
def api_update_user(user_id: int):
|
|
"""Proxy: Update user."""
|
|
data = request.get_json(silent=True) or {}
|
|
resp = api_client.put(f"/api/users/{user_id}", data=data)
|
|
|
|
if isinstance(resp, dict) and resp.get("error"):
|
|
return jsonify(resp), resp.get("status_code", 500)
|
|
|
|
return jsonify(resp), 200
|
|
|
|
|
|
@admin_bp.route("/api/users/<int:user_id>/password", methods=["PUT"])
|
|
@login_required
|
|
@admin_required
|
|
def api_change_password(user_id: int):
|
|
"""Proxy: Change user password."""
|
|
data = request.get_json(silent=True) or {}
|
|
resp = api_client.put(f"/api/users/{user_id}/password", data=data)
|
|
|
|
if isinstance(resp, dict) and resp.get("error"):
|
|
return jsonify(resp), resp.get("status_code", 500)
|
|
|
|
return jsonify(resp), 200
|
|
|
|
|
|
@admin_bp.route("/api/users/<int:user_id>/pin", methods=["PUT"])
|
|
@login_required
|
|
@admin_required
|
|
def api_set_pin(user_id: int):
|
|
"""Proxy: set (4 digits) or remove ({"pin": null}) a user's PIN (#59)."""
|
|
data = request.get_json(silent=True) or {}
|
|
# Forward only the pin field; backend answers 422 for admins / malformed PINs.
|
|
resp = api_client.put(f"/api/users/{user_id}/pin", data={"pin": data.get("pin")})
|
|
|
|
if isinstance(resp, dict) and resp.get("error"):
|
|
return jsonify(resp), resp.get("status_code", 500)
|
|
|
|
return jsonify(resp), 200
|
|
|
|
|
|
@admin_bp.route("/api/users/<int:user_id>/toggle-active", methods=["POST"])
|
|
@login_required
|
|
@admin_required
|
|
def api_toggle_active(user_id: int):
|
|
"""Proxy: Toggle user active status."""
|
|
data = request.get_json(silent=True) or {}
|
|
active = data.get("active", True)
|
|
resp = api_client.put(f"/api/users/{user_id}", data={"active": active})
|
|
|
|
if isinstance(resp, dict) and resp.get("error"):
|
|
return jsonify(resp), resp.get("status_code", 500)
|
|
|
|
return jsonify(resp), 200
|
|
|
|
|
|
# --- Stations ---
|
|
|
|
@admin_bp.route("/api/camera/sources", methods=["GET"])
|
|
@login_required
|
|
@admin_required
|
|
def api_camera_sources():
|
|
"""Proxy (#88): elenco sorgenti di un acquisitore, GET <url>/camera/sources."""
|
|
url = (request.args.get("url") or "").strip()
|
|
parsed = urlparse(url)
|
|
if parsed.scheme not in ("http", "https") or not parsed.netloc:
|
|
return jsonify({"error": True, "detail": _("Indirizzo della camera non valido")}), 400
|
|
try:
|
|
resp = http_requests.get(f"{url.rstrip('/')}/camera/sources", timeout=15) # #106: era 5, la prima enumerazione dopo l'avvio ha preso 3,7 s
|
|
except http_requests.RequestException as exc:
|
|
return jsonify({
|
|
"error": True,
|
|
"detail": _("Servizio camera non raggiungibile: %(detail)s", detail=str(exc)),
|
|
}), 502
|
|
invalid = (jsonify({"error": True, "detail": _("Risposta non valida dal servizio camera")}), 502)
|
|
if resp.status_code != 200:
|
|
return invalid
|
|
try:
|
|
sources = resp.json().get("sources")
|
|
except (ValueError, AttributeError):
|
|
return invalid
|
|
if not isinstance(sources, list):
|
|
return invalid
|
|
# solo i tre campi che la UI usa, sempre stringhe
|
|
clean = [
|
|
{k: str(s.get(k, "")) for k in ("name", "label", "kind")}
|
|
for s in sources if isinstance(s, dict)
|
|
]
|
|
return jsonify({"sources": clean}), 200
|
|
|
|
|
|
@admin_bp.route("/api/stations", methods=["POST"])
|
|
@login_required
|
|
@admin_required
|
|
def api_create_station():
|
|
"""Proxy: Create a new station."""
|
|
data = request.get_json(silent=True) or {}
|
|
resp = api_client.post("/api/stations", data=data)
|
|
if isinstance(resp, dict) and resp.get("error"):
|
|
return jsonify(resp), resp.get("status_code", 500)
|
|
return jsonify(resp), 201
|
|
|
|
|
|
@admin_bp.route("/api/stations/<int:station_id>", methods=["PUT"])
|
|
@login_required
|
|
@admin_required
|
|
def api_update_station(station_id: int):
|
|
"""Proxy: Update a station."""
|
|
data = request.get_json(silent=True) or {}
|
|
resp = api_client.put(f"/api/stations/{station_id}", data=data)
|
|
if isinstance(resp, dict) and resp.get("error"):
|
|
return jsonify(resp), resp.get("status_code", 500)
|
|
return jsonify(resp), 200
|
|
|
|
|
|
@admin_bp.route("/api/stations/<int:station_id>", methods=["DELETE"])
|
|
@login_required
|
|
@admin_required
|
|
def api_delete_station(station_id: int):
|
|
"""Proxy: Delete a station."""
|
|
resp = api_client.delete(f"/api/stations/{station_id}")
|
|
if isinstance(resp, dict) and resp.get("error"):
|
|
return jsonify(resp), resp.get("status_code", 500)
|
|
return jsonify({"deleted": True}), 200
|
|
|
|
|
|
@admin_bp.route("/api/stations/<int:station_id>/recipes", methods=["GET"])
|
|
@login_required
|
|
@admin_required
|
|
def api_list_station_recipes(station_id: int):
|
|
"""Proxy: List recipes assigned to a station."""
|
|
resp = api_client.get(f"/api/stations/{station_id}/recipes")
|
|
if isinstance(resp, dict) and resp.get("error"):
|
|
return jsonify(resp), resp.get("status_code", 500)
|
|
return jsonify(resp), 200
|
|
|
|
|
|
@admin_bp.route("/api/stations/<int:station_id>/recipes", methods=["POST"])
|
|
@login_required
|
|
@admin_required
|
|
def api_assign_recipe(station_id: int):
|
|
"""Proxy: Assign a recipe to a station."""
|
|
data = request.get_json(silent=True) or {}
|
|
resp = api_client.post(f"/api/stations/{station_id}/recipes", data=data)
|
|
if isinstance(resp, dict) and resp.get("error"):
|
|
return jsonify(resp), resp.get("status_code", 500)
|
|
return jsonify(resp), 201
|
|
|
|
|
|
@admin_bp.route("/api/stations/<int:station_id>/recipes/<int:recipe_id>", methods=["DELETE"])
|
|
@login_required
|
|
@admin_required
|
|
def api_unassign_recipe(station_id: int, recipe_id: int):
|
|
"""Proxy: Remove a recipe assignment from a station."""
|
|
resp = api_client.delete(f"/api/stations/{station_id}/recipes/{recipe_id}")
|
|
if isinstance(resp, dict) and resp.get("error"):
|
|
return jsonify(resp), resp.get("status_code", 500)
|
|
return jsonify({"deleted": True}), 200
|
|
|
|
|
|
@admin_bp.route("/api/stations/<int:station_id>/recipes", methods=["DELETE"])
|
|
@login_required
|
|
@admin_required
|
|
def api_reset_station_recipes(station_id: int):
|
|
"""Proxy: Clear every recipe assignment of a station."""
|
|
resp = api_client.delete(f"/api/stations/{station_id}/recipes")
|
|
if isinstance(resp, dict) and resp.get("error"):
|
|
return jsonify(resp), resp.get("status_code", 500)
|
|
return jsonify(resp), 200
|