293 lines
19 KiB
Python
293 lines
19 KiB
Python
"""Interface protégée du module Matériel & Entretien.
|
|
Les agents ne sont jamais des utilisateurs : ils sont uniquement sélectionnés
|
|
comme destinataires via la table staff.
|
|
"""
|
|
from datetime import date
|
|
from decimal import Decimal
|
|
from flask import Blueprint, flash, redirect, render_template, request, url_for
|
|
from flask_login import current_user, login_required
|
|
from ..extensions import db
|
|
from ..core.models import (
|
|
ProductCategory, ProductGeneric, CommercialProduct, ProductPackaging,
|
|
StockLocation, StockLot, StockLotBalance, StockMovement, Staff,
|
|
RequestProfile, StockInventory, StockInventoryLine,
|
|
Equipment, MaterialAssignment,
|
|
ProductDocument,
|
|
)
|
|
from .services.stock import StockError, receive_stock, issue_stock, transfer_stock, transvasement, dilute
|
|
from .services.forecast import forecast_product
|
|
|
|
cleaning_bp = Blueprint("cleaning", __name__, url_prefix="/cleaning", template_folder="templates")
|
|
|
|
def _date(value, default=None):
|
|
try:
|
|
return date.fromisoformat(value) if value else default
|
|
except ValueError:
|
|
return default
|
|
|
|
@cleaning_bp.route("/")
|
|
@login_required
|
|
def dashboard():
|
|
products = ProductGeneric.query.filter_by(is_active=True).order_by(ProductGeneric.display_order, ProductGeneric.name).all()
|
|
low = [p for p in products if p.stock_total <= Decimal(str(p.stock_minimum or 0))]
|
|
expiring = StockLot.query.filter(StockLot.expiry_date.isnot(None), StockLot.expiry_date <= date.today()).order_by(StockLot.expiry_date).all()
|
|
return render_template("cleaning/dashboard.html", products=products, low_stock=low, expiring=expiring, recent=StockMovement.query.order_by(StockMovement.created_at.desc()).limit(15).all())
|
|
|
|
@cleaning_bp.route("/products")
|
|
@login_required
|
|
def products():
|
|
return render_template("cleaning/products.html", products=ProductGeneric.query.order_by(ProductGeneric.display_order, ProductGeneric.name).all(), categories=ProductCategory.query.order_by(ProductCategory.name).all())
|
|
|
|
@cleaning_bp.route("/products/new", methods=["GET", "POST"])
|
|
@login_required
|
|
def product_new():
|
|
categories = ProductCategory.query.order_by(ProductCategory.name).all()
|
|
if request.method == "POST":
|
|
name = (request.form.get("name") or "").strip()
|
|
try:
|
|
minimum = Decimal(request.form.get("stock_minimum") or 0)
|
|
security = Decimal(request.form.get("stock_security") or 0)
|
|
daily = Decimal(request.form.get("forecast_daily_quantity") or 0)
|
|
except Exception:
|
|
flash("Les quantités doivent être numériques.", "danger")
|
|
return render_template("cleaning/product_form.html", product=None, categories=categories)
|
|
if not name:
|
|
flash("Le nom du produit est obligatoire.", "danger")
|
|
elif ProductGeneric.query.filter_by(name=name).first():
|
|
flash("Ce produit générique existe déjà.", "warning")
|
|
else:
|
|
product = ProductGeneric(name=name, description=request.form.get("description"), category_id=request.form.get("category_id") or None, product_type=request.form.get("product_type") or "consommable", reference_unit=request.form.get("reference_unit") or "unité", stock_minimum=minimum, stock_security=security, forecast_daily_quantity=daily)
|
|
db.session.add(product); db.session.commit()
|
|
flash("Produit générique créé.", "success")
|
|
return redirect(url_for("cleaning.products"))
|
|
return render_template("cleaning/product_form.html", product=None, categories=categories)
|
|
|
|
@cleaning_bp.route("/products/<int:id>/edit", methods=["GET", "POST"])
|
|
@login_required
|
|
def product_edit(id):
|
|
product = ProductGeneric.query.get_or_404(id)
|
|
if request.method == "POST":
|
|
product.name = (request.form.get("name") or product.name).strip()
|
|
product.description = request.form.get("description")
|
|
product.category_id = request.form.get("category_id") or None
|
|
product.reference_unit = request.form.get("reference_unit") or product.reference_unit
|
|
product.stock_minimum = Decimal(request.form.get("stock_minimum") or 0)
|
|
product.stock_security = Decimal(request.form.get("stock_security") or 0)
|
|
product.forecast_daily_quantity = Decimal(request.form.get("forecast_daily_quantity") or 0)
|
|
product.is_active = request.form.get("is_active") == "on"
|
|
db.session.commit()
|
|
flash("Produit générique mis à jour.", "success")
|
|
return redirect(url_for("cleaning.products"))
|
|
return render_template("cleaning/product_form.html", product=product, categories=ProductCategory.query.order_by(ProductCategory.name).all())
|
|
|
|
@cleaning_bp.route("/references")
|
|
@login_required
|
|
def references():
|
|
return render_template("cleaning/references.html", references=CommercialProduct.query.order_by(CommercialProduct.commercial_name).all(), products=ProductGeneric.query.order_by(ProductGeneric.name).all())
|
|
|
|
|
|
@cleaning_bp.route("/documents")
|
|
@login_required
|
|
def documents():
|
|
return render_template("cleaning/documents.html", documents=ProductDocument.query.order_by(ProductDocument.uploaded_at.desc()).all())
|
|
|
|
|
|
@cleaning_bp.route("/labels/<int:movement_id>")
|
|
@login_required
|
|
def label(movement_id):
|
|
movement = StockMovement.query.get_or_404(movement_id)
|
|
if not movement.lot or not movement.commercial_product:
|
|
flash("Impossible de générer une étiquette : lot ou référence commerciale manquant.", "danger")
|
|
return redirect(url_for("cleaning.movements"))
|
|
return render_template("cleaning/label.html", movement=movement)
|
|
|
|
@cleaning_bp.route("/references/new", methods=["GET", "POST"])
|
|
@login_required
|
|
def reference_new():
|
|
products = ProductGeneric.query.filter_by(is_active=True).order_by(ProductGeneric.name).all()
|
|
if request.method == "POST":
|
|
ref = CommercialProduct(generic_product_id=request.form.get("generic_product_id"), commercial_name=(request.form.get("commercial_name") or "").strip(), brand=request.form.get("brand"), manufacturer_reference=request.form.get("manufacturer_reference"), supplier_reference=request.form.get("supplier_reference"), is_concentrated=request.form.get("is_concentrated") == "on")
|
|
if not ref.generic_product_id or not ref.commercial_name:
|
|
flash("Produit générique et nom commercial sont obligatoires.", "danger")
|
|
else:
|
|
db.session.add(ref); db.session.commit()
|
|
flash("Référence commerciale créée.", "success")
|
|
return redirect(url_for("cleaning.references"))
|
|
return render_template("cleaning/reference_form.html", products=products)
|
|
|
|
|
|
@cleaning_bp.route("/packaging/new", methods=["GET", "POST"])
|
|
@login_required
|
|
def packaging_new():
|
|
products = ProductGeneric.query.filter_by(is_active=True).order_by(ProductGeneric.name).all()
|
|
references = CommercialProduct.query.filter_by(is_active=True).order_by(CommercialProduct.commercial_name).all()
|
|
if request.method == "POST":
|
|
try:
|
|
item = ProductPackaging(generic_product_id=int(request.form["generic_product_id"]),
|
|
commercial_product_id=request.form.get("commercial_product_id") or None,
|
|
name=(request.form.get("name") or "").strip(), purchase_unit=request.form.get("purchase_unit") or "conditionnement",
|
|
units_per_package=Decimal(request.form.get("units_per_package") or 1),
|
|
reference_quantity_per_package=Decimal(request.form["reference_quantity_per_package"]),
|
|
is_current=request.form.get("is_current") == "on")
|
|
if not item.name: raise ValueError("Le nom du conditionnement est obligatoire")
|
|
db.session.add(item); db.session.commit(); flash("Conditionnement enregistré.", "success"); return redirect(url_for("cleaning.references"))
|
|
except (KeyError, ValueError, ArithmeticError) as exc:
|
|
db.session.rollback(); flash(str(exc), "danger")
|
|
return render_template("cleaning/packaging_form.html", products=products, references=references)
|
|
|
|
@cleaning_bp.route("/stock")
|
|
@login_required
|
|
def stock():
|
|
balances = StockLotBalance.query.join(StockLot).join(ProductGeneric).join(StockLocation).order_by(ProductGeneric.name, StockLot.expiry_date).all()
|
|
return render_template("cleaning/stock.html", balances=balances, locations=StockLocation.query.filter_by(is_active=True).order_by(StockLocation.name).all())
|
|
|
|
@cleaning_bp.route("/locations", methods=["GET", "POST"])
|
|
@login_required
|
|
def locations():
|
|
if request.method == "POST":
|
|
name = (request.form.get("name") or "").strip()
|
|
if name:
|
|
db.session.add(StockLocation(name=name, notes=request.form.get("notes"))); db.session.commit(); flash("Emplacement créé.", "success")
|
|
return redirect(url_for("cleaning.locations"))
|
|
return render_template("cleaning/locations.html", locations=StockLocation.query.order_by(StockLocation.name).all())
|
|
|
|
@cleaning_bp.route("/receive", methods=["GET", "POST"])
|
|
@login_required
|
|
def receive():
|
|
products = ProductGeneric.query.filter_by(is_active=True).order_by(ProductGeneric.name).all()
|
|
if request.method == "POST":
|
|
try:
|
|
lot = receive_stock(generic_product_id=int(request.form["generic_product_id"]), commercial_product_id=int(request.form["commercial_product_id"]), packaging_id=int(request.form["packaging_id"]), packages=request.form["packages"], lot_number=request.form["lot_number"], received_at=_date(request.form.get("received_at"), date.today()), location_id=int(request.form["location_id"]), user_id=current_user.id, expiry_date=_date(request.form.get("expiry_date")))
|
|
db.session.commit()
|
|
flash(f"Réception enregistrée ({lot.quantity_received} unité(s) de référence).", "success")
|
|
return redirect(url_for("cleaning.stock"))
|
|
except (KeyError, ValueError, StockError) as exc:
|
|
db.session.rollback(); flash(str(exc), "danger")
|
|
return render_template("cleaning/receive.html", products=products, references=CommercialProduct.query.filter_by(is_active=True).all(), packagings=ProductPackaging.query.filter_by(is_active=True).all(), locations=StockLocation.query.filter_by(is_active=True).all())
|
|
|
|
@cleaning_bp.route("/issue", methods=["GET", "POST"])
|
|
@login_required
|
|
def issue():
|
|
if request.method == "POST":
|
|
try:
|
|
issue_stock(generic_product_id=int(request.form["generic_product_id"]), quantity=request.form["quantity"], source_location_id=int(request.form["source_location_id"]), staff_id=request.form.get("staff_id") or None, user_id=current_user.id, comment=request.form.get("comment"))
|
|
db.session.commit(); flash("Sortie FEFO enregistrée.", "success")
|
|
return redirect(url_for("cleaning.movements"))
|
|
except (KeyError, ValueError, StockError) as exc:
|
|
db.session.rollback(); flash(str(exc), "danger")
|
|
return render_template("cleaning/issue.html", products=ProductGeneric.query.filter_by(is_active=True).all(), locations=StockLocation.query.filter_by(is_active=True).all(), staff=Staff.query.filter_by(is_active=True).order_by(Staff.last_name).all())
|
|
|
|
@cleaning_bp.route("/transfer", methods=["GET", "POST"])
|
|
@login_required
|
|
def transfer():
|
|
if request.method == "POST":
|
|
try:
|
|
transfer_stock(generic_product_id=int(request.form["generic_product_id"]), quantity=request.form["quantity"], source_location_id=int(request.form["source_location_id"]), destination_location_id=int(request.form["destination_location_id"]), user_id=current_user.id, comment=request.form.get("comment")); db.session.commit()
|
|
flash("Transfert enregistré.", "success"); return redirect(url_for("cleaning.stock"))
|
|
except (KeyError, ValueError, StockError) as exc:
|
|
db.session.rollback(); flash(str(exc), "danger")
|
|
return render_template("cleaning/transfer.html", products=ProductGeneric.query.filter_by(is_active=True).all(), locations=StockLocation.query.filter_by(is_active=True).all())
|
|
|
|
@cleaning_bp.route("/movements")
|
|
@login_required
|
|
def movements():
|
|
return render_template("cleaning/movements.html", movements=StockMovement.query.order_by(StockMovement.created_at.desc()).limit(300).all())
|
|
|
|
|
|
@cleaning_bp.route("/transvasement", methods=["GET", "POST"])
|
|
@login_required
|
|
def transvasement_page():
|
|
if request.method == "POST":
|
|
try:
|
|
fn = dilute if request.form.get("dilution") == "on" else transvasement
|
|
kwargs = dict(generic_product_id=int(request.form["generic_product_id"]), quantity_liters=request.form["quantity_liters"], source_location_id=int(request.form["source_location_id"]), container_id=request.form.get("container_id") or None, user_id=current_user.id, comment=request.form.get("comment"))
|
|
if fn is dilute: kwargs["total_liters"] = kwargs.pop("quantity_liters"); kwargs["ratio_percent"] = request.form["ratio_percent"]
|
|
fn(**kwargs); db.session.commit(); flash("Transvasement/dilution enregistré.", "success"); return redirect(url_for("cleaning.movements"))
|
|
except (KeyError, ValueError, StockError) as exc:
|
|
db.session.rollback(); flash(str(exc), "danger")
|
|
return render_template("cleaning/transvasement.html", products=ProductGeneric.query.filter_by(is_active=True).all(), locations=StockLocation.query.filter_by(is_active=True).all(), containers=ReusableContainer.query.filter_by(state="disponible").all())
|
|
|
|
@cleaning_bp.route("/forecast/<int:product_id>")
|
|
@login_required
|
|
def forecast(product_id):
|
|
product = ProductGeneric.query.get_or_404(product_id)
|
|
result = forecast_product(product, _date(request.args.get("target_date"), date.today()))
|
|
return render_template("cleaning/forecast.html", product=product, result=result)
|
|
|
|
|
|
@cleaning_bp.route("/forecast-config", methods=["GET", "POST"])
|
|
@login_required
|
|
def forecast_config():
|
|
config = CleaningForecastConfig.query.filter_by(is_active=True).first()
|
|
if request.method == "POST":
|
|
if not config: config = CleaningForecastConfig(); db.session.add(config)
|
|
config.normal_coefficient = Decimal(request.form.get("normal_coefficient") or 1)
|
|
config.permanence_coefficient = Decimal(request.form.get("permanence_coefficient") or 0)
|
|
config.closed_coefficient = Decimal(request.form.get("closed_coefficient") or 0)
|
|
config.exceptional_coefficient = Decimal(request.form.get("exceptional_coefficient") or 1)
|
|
db.session.commit(); flash("Coefficients de prévision enregistrés.", "success"); return redirect(url_for("cleaning.forecast_config"))
|
|
return render_template("cleaning/forecast_config.html", config=config,
|
|
values={"normal_coefficient": config.normal_coefficient if config else 1,
|
|
"permanence_coefficient": config.permanence_coefficient if config else Decimal("0.4"),
|
|
"closed_coefficient": config.closed_coefficient if config else 0,
|
|
"exceptional_coefficient": config.exceptional_coefficient if config else 1})
|
|
|
|
@cleaning_bp.route("/inventory", methods=["GET", "POST"])
|
|
@login_required
|
|
def inventory():
|
|
if request.method == "POST":
|
|
location = StockLocation.query.get_or_404(int(request.form["location_id"]))
|
|
inv = StockInventory(location_id=location.id, user_id=current_user.id, status="valide")
|
|
db.session.add(inv); db.session.flush()
|
|
for raw in request.form.getlist("line_lot_id"):
|
|
lot = StockLot.query.get(int(raw)); balance = StockLotBalance.query.filter_by(lot_id=lot.id, location_id=location.id).first()
|
|
if not balance: continue
|
|
counted = Decimal(request.form.get(f"counted_{lot.id}") or 0); theoretical = Decimal(str(balance.quantity or 0))
|
|
db.session.add(StockInventoryLine(inventory_id=inv.id, lot_id=lot.id, theoretical_quantity=theoretical, counted_quantity=counted))
|
|
diff = counted - theoretical
|
|
if diff:
|
|
balance.quantity = counted
|
|
db.session.add(StockMovement(generic_product_id=lot.generic_product_id, commercial_product_id=lot.commercial_product_id, lot_id=lot.id, movement_type="correction_inventaire", quantity_reference=diff, destination_location_id=location.id, user_id=current_user.id, comment="Correction après inventaire"))
|
|
db.session.commit(); flash("Inventaire validé et écarts tracés.", "success"); return redirect(url_for("cleaning.inventory"))
|
|
locations = StockLocation.query.filter_by(is_active=True).all(); selected = request.args.get("location_id", type=int)
|
|
rows = StockLotBalance.query.filter_by(location_id=selected).join(StockLot).all() if selected else []
|
|
return render_template("cleaning/inventory.html", locations=locations, rows=rows, selected=selected)
|
|
|
|
@cleaning_bp.route("/requests")
|
|
@login_required
|
|
def request_profiles():
|
|
return render_template("cleaning/request_profiles.html", profiles=RequestProfile.query.order_by(RequestProfile.name).all())
|
|
|
|
|
|
@cleaning_bp.route("/requests/new", methods=["GET", "POST"])
|
|
@login_required
|
|
def request_profile_new():
|
|
products = ProductGeneric.query.filter_by(is_active=True, request_enabled=True).order_by(ProductGeneric.name).all()
|
|
if request.method == "POST":
|
|
profile = RequestProfile(name=(request.form.get("name") or "").strip(), description=request.form.get("description"))
|
|
if not profile.name: flash("Le nom du profil est obligatoire.", "danger")
|
|
else:
|
|
db.session.add(profile); db.session.flush()
|
|
for order, raw in enumerate(request.form.getlist("product_ids")):
|
|
db.session.add(RequestProfileItem(profile_id=profile.id, product_id=int(raw), display_order=order))
|
|
db.session.commit(); flash("Profil de demande créé.", "success"); return redirect(url_for("cleaning.request_profiles"))
|
|
return render_template("cleaning/request_profile_form.html", products=products)
|
|
|
|
@cleaning_bp.route("/requests/print/<int:profile_id>")
|
|
@login_required
|
|
def request_print(profile_id):
|
|
profile = RequestProfile.query.get_or_404(profile_id)
|
|
staff = Staff.query.filter_by(is_active=True).order_by(Staff.last_name, Staff.first_name).all()
|
|
return render_template("cleaning/request_print.html", profile=profile, staff=staff, today=date.today())
|
|
|
|
@cleaning_bp.route("/material")
|
|
@login_required
|
|
def material():
|
|
return render_template("cleaning/material.html", assignments=MaterialAssignment.query.order_by(MaterialAssignment.assigned_at.desc()).all(), equipment=Equipment.query.filter_by(is_deleted=False).order_by(Equipment.name).all(), staff=Staff.query.filter_by(is_active=True).all())
|
|
|
|
|
|
@cleaning_bp.route("/material/assign", methods=["POST"])
|
|
@login_required
|
|
def material_assign():
|
|
assignment = MaterialAssignment(equipment_id=int(request.form["equipment_id"]), staff_id=request.form.get("staff_id") or None, zone_id=request.form.get("zone_id") or None, room_id=request.form.get("room_id") or None, status=request.form.get("status") or "affecte", notes=request.form.get("notes"))
|
|
db.session.add(assignment); db.session.commit(); flash("Affectation de matériel enregistrée.", "success"); return redirect(url_for("cleaning.material"))
|