feat: FASE 3 - Flusso MeasurementTec (selezione ricetta, esecuzione misure, riepilogo)
Implementazione completa del flusso operativo per il ruolo MeasurementTec:
Blueprint measure.py:
- select_recipe: selezione ricetta con ricerca e barcode
- task_list: lista task con conteggi subtask e allegati
- task_execute: esecuzione misure con numpad, calibro USB, feedback real-time
- task_complete: riepilogo con statistiche pass/fail e export CSV
- API AJAX: lookup-barcode, save-traceability, save-measurement
- Autorizzazione role_required("MeasurementTec") su tutte le route
Componenti riutilizzabili:
- numpad.html/js/css: tastierino numerico touch-friendly con keyboard support
- caliper_status.html + caliper.js: integrazione calibro USB via Web Serial API
- barcode_scanner.html + barcode.js: scansione barcode con html5-qrcode
- measurement_feedback.html: feedback visivo pass/warning/fail in tempo reale
- next_measurement.html: indicatore prossima misurazione
- annotation-viewer.js: visualizzatore canvas con marker su disegni tecnici
- csv-export.js: export CSV con locale italiano (delimitatore ;, decimale ,)
Sicurezza:
- Decoratore role_required(*roles) per autorizzazione basata su ruoli
- CSRF token su tutti i POST AJAX
- |tojson per prevenire XSS su annotations_json
- Validazione input lato client e server
i18n: 23+ nuove chiavi tradotte IT/EN per tutti i template FASE 3
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Adriano
@@ -1,7 +1,7 @@
|
||||
"""Authentication blueprint - login, logout, profile."""
|
||||
from functools import wraps
|
||||
|
||||
from flask import Blueprint, flash, redirect, render_template, request, session, url_for
|
||||
from flask import Blueprint, abort, flash, redirect, render_template, request, session, url_for
|
||||
from flask_babel import gettext as _
|
||||
|
||||
from services.api_client import api_client
|
||||
@@ -20,6 +20,25 @@ def login_required(f):
|
||||
return decorated
|
||||
|
||||
|
||||
def role_required(*roles):
|
||||
"""Decorator to require one of the specified roles.
|
||||
|
||||
Usage:
|
||||
@role_required("MeasurementTec", "Metrologist")
|
||||
def my_view(): ...
|
||||
"""
|
||||
def decorator(f):
|
||||
@wraps(f)
|
||||
def decorated(*args, **kwargs):
|
||||
user = session.get("user", {})
|
||||
user_roles = user.get("roles", [])
|
||||
if not any(r in user_roles for r in roles):
|
||||
abort(403)
|
||||
return f(*args, **kwargs)
|
||||
return decorated
|
||||
return decorator
|
||||
|
||||
|
||||
@auth_bp.route("/login", methods=["GET", "POST"])
|
||||
def login():
|
||||
"""Login page and form handler."""
|
||||
|
||||
+252
-15
@@ -1,36 +1,273 @@
|
||||
"""MeasurementTec blueprint - recipe selection and measurement execution."""
|
||||
from flask import Blueprint, render_template, session, redirect, url_for
|
||||
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, role_required
|
||||
from services.api_client import api_client
|
||||
|
||||
measure_bp = Blueprint("measure", __name__)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Route: Recipe selection
|
||||
# ---------------------------------------------------------------------------
|
||||
@measure_bp.route("/select")
|
||||
@login_required
|
||||
@role_required("MeasurementTec")
|
||||
def select_recipe():
|
||||
"""Recipe selection page."""
|
||||
if "user" not in session:
|
||||
return redirect(url_for("auth.login"))
|
||||
return render_template("measure/select_recipe.html")
|
||||
"""Recipe selection page with search and barcode support."""
|
||||
# Load recipes from API
|
||||
resp = api_client.get("/api/recipes", params={"per_page": 100})
|
||||
if resp.get("error"):
|
||||
flash(
|
||||
_("Errore nel caricamento delle ricette: %(detail)s",
|
||||
detail=resp.get("detail", "")),
|
||||
"error",
|
||||
)
|
||||
recipes = []
|
||||
else:
|
||||
# API may return paginated envelope or plain list
|
||||
recipes = resp.get("items", resp) if isinstance(resp, dict) else resp
|
||||
|
||||
# Auto-fill from query params
|
||||
auto_recipe_code = request.args.get("recipe", "")
|
||||
auto_lot = request.args.get("lot", session.get("lot_number", ""))
|
||||
auto_serial = request.args.get("serial", session.get("serial_number", ""))
|
||||
|
||||
return render_template(
|
||||
"measure/select_recipe.html",
|
||||
recipes=recipes,
|
||||
auto_recipe_code=auto_recipe_code,
|
||||
auto_lot=auto_lot,
|
||||
auto_serial=auto_serial,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Route: Task list for a recipe
|
||||
# ---------------------------------------------------------------------------
|
||||
@measure_bp.route("/tasks/<int:recipe_id>")
|
||||
@login_required
|
||||
@role_required("MeasurementTec")
|
||||
def task_list(recipe_id: int):
|
||||
"""Task list for selected recipe."""
|
||||
if "user" not in session:
|
||||
return redirect(url_for("auth.login"))
|
||||
return render_template("measure/task_list.html", recipe_id=recipe_id)
|
||||
# Persist lot/serial from query params into session
|
||||
lot_number = request.args.get(
|
||||
"lot_number", session.get("lot_number", ""),
|
||||
)
|
||||
serial_number = request.args.get(
|
||||
"serial_number", session.get("serial_number", ""),
|
||||
)
|
||||
if lot_number:
|
||||
session["lot_number"] = lot_number
|
||||
if serial_number:
|
||||
session["serial_number"] = serial_number
|
||||
|
||||
# Load recipe details
|
||||
recipe_resp = api_client.get(f"/api/recipes/{recipe_id}")
|
||||
if recipe_resp.get("error"):
|
||||
flash(
|
||||
_("Ricetta non trovata: %(detail)s",
|
||||
detail=recipe_resp.get("detail", "")),
|
||||
"error",
|
||||
)
|
||||
return redirect(url_for("measure.select_recipe"))
|
||||
|
||||
# Load tasks for this recipe
|
||||
tasks_resp = api_client.get(f"/api/recipes/{recipe_id}/tasks")
|
||||
if tasks_resp.get("error"):
|
||||
flash(
|
||||
_("Errore nel caricamento dei task: %(detail)s",
|
||||
detail=tasks_resp.get("detail", "")),
|
||||
"error",
|
||||
)
|
||||
tasks = []
|
||||
else:
|
||||
tasks = tasks_resp if isinstance(tasks_resp, list) else tasks_resp.get("items", [])
|
||||
|
||||
return render_template(
|
||||
"measure/task_list.html",
|
||||
recipe=recipe_resp,
|
||||
tasks=tasks,
|
||||
lot_number=lot_number,
|
||||
serial_number=serial_number,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Route: Task execution (measurement input)
|
||||
# ---------------------------------------------------------------------------
|
||||
@measure_bp.route("/execute/<int:task_id>")
|
||||
@login_required
|
||||
@role_required("MeasurementTec")
|
||||
def task_execute(task_id: int):
|
||||
"""Execute measurements for a task."""
|
||||
if "user" not in session:
|
||||
return redirect(url_for("auth.login"))
|
||||
return render_template("measure/task_execute.html", task_id=task_id)
|
||||
# Load task + subtasks
|
||||
task_resp = api_client.get(f"/api/tasks/{task_id}")
|
||||
if task_resp.get("error"):
|
||||
flash(
|
||||
_("Task non trovato: %(detail)s",
|
||||
detail=task_resp.get("detail", "")),
|
||||
"error",
|
||||
)
|
||||
return redirect(url_for("measure.select_recipe"))
|
||||
|
||||
lot_number = session.get("lot_number", "")
|
||||
serial_number = session.get("serial_number", "")
|
||||
|
||||
return render_template(
|
||||
"measure/task_execute.html",
|
||||
task=task_resp,
|
||||
lot_number=lot_number,
|
||||
serial_number=serial_number,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Route: Task completion summary
|
||||
# ---------------------------------------------------------------------------
|
||||
@measure_bp.route("/complete/<int:recipe_id>")
|
||||
@login_required
|
||||
@role_required("MeasurementTec")
|
||||
def task_complete(recipe_id: int):
|
||||
"""Task completion summary."""
|
||||
if "user" not in session:
|
||||
return redirect(url_for("auth.login"))
|
||||
return render_template("measure/task_complete.html", recipe_id=recipe_id)
|
||||
"""Task completion summary with all measurements."""
|
||||
# Retrieve version_id from query params
|
||||
version_id = request.args.get("version_id")
|
||||
|
||||
# Load recipe for context
|
||||
recipe_resp = api_client.get(f"/api/recipes/{recipe_id}")
|
||||
if recipe_resp.get("error"):
|
||||
flash(
|
||||
_("Ricetta non trovata: %(detail)s",
|
||||
detail=recipe_resp.get("detail", "")),
|
||||
"error",
|
||||
)
|
||||
return redirect(url_for("measure.select_recipe"))
|
||||
|
||||
# Load measurements if version_id provided
|
||||
measurements = []
|
||||
if version_id:
|
||||
meas_resp = api_client.get(
|
||||
"/api/measurements",
|
||||
params={"version_id": version_id},
|
||||
)
|
||||
if not meas_resp.get("error"):
|
||||
measurements = (
|
||||
meas_resp if isinstance(meas_resp, list)
|
||||
else meas_resp.get("items", [])
|
||||
)
|
||||
|
||||
lot_number = session.get("lot_number", "")
|
||||
serial_number = session.get("serial_number", "")
|
||||
|
||||
return render_template(
|
||||
"measure/task_complete.html",
|
||||
recipe=recipe_resp,
|
||||
measurements=measurements,
|
||||
lot_number=lot_number,
|
||||
serial_number=serial_number,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Route: Barcode lookup (AJAX)
|
||||
# ---------------------------------------------------------------------------
|
||||
@measure_bp.route("/lookup-barcode", methods=["POST"])
|
||||
@login_required
|
||||
@role_required("MeasurementTec")
|
||||
def lookup_barcode():
|
||||
"""Look up a recipe by barcode/code. Returns JSON for AJAX calls."""
|
||||
data = request.get_json(silent=True) or {}
|
||||
code = data.get("code", "").strip()
|
||||
|
||||
if not code:
|
||||
return jsonify({"error": True, "detail": _("Codice non fornito")}), 400
|
||||
|
||||
resp = api_client.get(f"/api/recipes/code/{code}")
|
||||
if resp.get("error"):
|
||||
return jsonify({
|
||||
"error": True,
|
||||
"detail": resp.get("detail", _("Ricetta non trovata")),
|
||||
}), 404
|
||||
|
||||
return jsonify(resp)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Route: Save lot/serial to session (AJAX)
|
||||
# ---------------------------------------------------------------------------
|
||||
@measure_bp.route("/save-traceability", methods=["POST"])
|
||||
@login_required
|
||||
@role_required("MeasurementTec")
|
||||
def save_traceability():
|
||||
"""Save lot_number and serial_number to session."""
|
||||
data = request.get_json(silent=True) or {}
|
||||
lot = data.get("lot_number", "").strip()
|
||||
serial = data.get("serial_number", "").strip()
|
||||
|
||||
if lot:
|
||||
session["lot_number"] = lot
|
||||
if serial:
|
||||
session["serial_number"] = serial
|
||||
|
||||
return jsonify({"ok": True})
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Route: Save measurement (AJAX proxy to FastAPI)
|
||||
# ---------------------------------------------------------------------------
|
||||
@measure_bp.route("/save-measurement", methods=["POST"])
|
||||
@login_required
|
||||
@role_required("MeasurementTec")
|
||||
def save_measurement():
|
||||
"""Save a single measurement value via API proxy.
|
||||
|
||||
Expects JSON body:
|
||||
subtask_id: int
|
||||
task_id: int
|
||||
value: float
|
||||
pass_fail: str ('pass' | 'warning' | 'fail')
|
||||
deviation: float
|
||||
lot_number: str (optional)
|
||||
serial_number: str (optional)
|
||||
|
||||
Returns JSON with the created measurement or error.
|
||||
"""
|
||||
data = request.get_json(silent=True) or {}
|
||||
|
||||
# Validate required fields
|
||||
subtask_id = data.get("subtask_id")
|
||||
task_id = data.get("task_id")
|
||||
value = data.get("value")
|
||||
|
||||
if subtask_id is None or task_id is None or value is None:
|
||||
return jsonify({
|
||||
"error": True,
|
||||
"detail": _("Dati mancanti: subtask_id, task_id e value sono obbligatori"),
|
||||
}), 400
|
||||
|
||||
# Build payload for the FastAPI backend
|
||||
payload = {
|
||||
"subtask_id": subtask_id,
|
||||
"task_id": task_id,
|
||||
"value": value,
|
||||
"pass_fail": data.get("pass_fail", ""),
|
||||
"deviation": data.get("deviation"),
|
||||
"lot_number": data.get("lot_number", session.get("lot_number", "")),
|
||||
"serial_number": data.get("serial_number", session.get("serial_number", "")),
|
||||
"measured_by": session.get("user_id"),
|
||||
}
|
||||
|
||||
resp = api_client.post("/api/measurements", data=payload)
|
||||
|
||||
if resp.get("error"):
|
||||
status_code = resp.get("status_code", 500)
|
||||
return jsonify({
|
||||
"error": True,
|
||||
"detail": resp.get("detail", _("Errore nel salvataggio")),
|
||||
}), status_code if status_code >= 400 else 500
|
||||
|
||||
return jsonify(resp), 201
|
||||
|
||||
Reference in New Issue
Block a user