Adriano
2453e552fb
- Extract JSON data to <script> tags instead of tojson_attr in x-data attributes
- Remove literal " from CSS selector in x-data (meta[name=csrf-token])
- Move Alpine.js defer script after extra_js block in base.html
- Add isinstance(resp, dict) guard before .get("error") in measure.py and maker.py
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
275 lines
9.1 KiB
Python
275 lines
9.1 KiB
Python
"""MeasurementTec blueprint - recipe selection and measurement execution."""
|
|
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 with search and barcode support."""
|
|
# Load recipes from API
|
|
resp = api_client.get("/api/recipes", params={"per_page": 100})
|
|
if isinstance(resp, dict) and 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."""
|
|
# 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 isinstance(tasks_resp, dict) and 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."""
|
|
# 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 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 (isinstance(meas_resp, dict) and 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"),
|
|
"input_method": data.get("input_method", "manual"),
|
|
}
|
|
|
|
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
|