Ajouter le module matériel et produits d'entretien
Some checks are pending
CI - Tests et Syntax / lint-and-test (push) Waiting to run

This commit is contained in:
root 2026-08-21 14:13:00 +00:00
parent 5ddfb3d70f
commit 374e5e5a54
34 changed files with 1017 additions and 1 deletions

View file

@ -132,6 +132,10 @@ def create_app(config_name='default'):
from .parts.routes import parts_bp
app.register_blueprint(parts_bp, url_prefix='/parts')
# Matériel et produits d'entretien (stock, FEFO, distributions, prévisions)
from .cleaning import cleaning_bp
app.register_blueprint(cleaning_bp)
from .services_module.routes import services_bp
app.register_blueprint(services_bp, url_prefix='/services')

View file

@ -0,0 +1,3 @@
from .routes import cleaning_bp
__all__ = ["cleaning_bp"]

276
app_new/cleaning/routes.py Normal file
View file

@ -0,0 +1,276 @@
"""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,
)
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("/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"))

View file

@ -0,0 +1,4 @@
from .stock import StockError, receive_stock, issue_stock, transfer_stock, transvasement, dilute
from .forecast import forecast_product
__all__ = ["StockError", "receive_stock", "issue_stock", "transfer_stock", "transvasement", "dilute", "forecast_product"]

View file

@ -0,0 +1,71 @@
"""Prévision déterministe tenant compte du calendrier réel du collège."""
from datetime import date, timedelta
from decimal import Decimal
from ...core.models import ProductGeneric, ProductPackaging, StockLot, StockLotBalance, CollegeClosure, ClosureWorkDay, CleaningForecastConfig
def _day_kind(day, closures, permanence_dates):
if day in permanence_dates:
return "permanence"
for closure in closures:
if closure.start_date <= day <= closure.end_date:
return "ferme"
if day.weekday() >= 5:
return "ferme"
return "scolaire"
def forecast_product(product, target_date, *, start_date=None, config=None):
"""Retourne un dictionnaire explicable de besoin et de conditionnement."""
start_date = start_date or date.today()
if target_date < start_date:
raise ValueError("La date cible doit être postérieure au début")
config = config or CleaningForecastConfig.query.filter_by(is_active=True).first() or CleaningForecastConfig()
# Les valeurs de colonne SQLAlchemy ne sont appliquées qu'à l'insertion ;
# conserver les mêmes valeurs par défaut pour un calcul sans configuration.
normal_coefficient = config.normal_coefficient if config.normal_coefficient is not None else Decimal("1")
permanence_coefficient = config.permanence_coefficient if config.permanence_coefficient is not None else Decimal("0.4")
closed_coefficient = config.closed_coefficient if config.closed_coefficient is not None else Decimal("0")
closures = CollegeClosure.query.filter(CollegeClosure.end_date >= start_date,
CollegeClosure.start_date <= target_date).all()
closure_ids = [c.id for c in closures]
permanence_dates = {d.work_date for d in ClosureWorkDay.query.filter(
ClosureWorkDay.work_date >= start_date, ClosureWorkDay.work_date <= target_date,
ClosureWorkDay.closure_id.in_(closure_ids)).all()} if closure_ids else set()
normal = Decimal(str(product.forecast_daily_quantity or 0)) * Decimal(str(normal_coefficient))
permanence = Decimal(str(product.forecast_daily_quantity or 0)) * Decimal(str(permanence_coefficient))
closed = Decimal(str(product.forecast_daily_quantity or 0)) * Decimal(str(closed_coefficient))
quantities = {"scolaire": Decimal("0"), "permanence": Decimal("0"), "ferme": Decimal("0")}
day = start_date
while day <= target_date:
kind = _day_kind(day, closures, permanence_dates)
quantities[kind] += {"scolaire": normal, "permanence": permanence, "ferme": closed}[kind]
day += timedelta(days=1)
total_need = sum(quantities.values(), Decimal("0"))
# Stock réellement utile selon la chronologie FEFO : un lot expiré avant
# la fin de période n'est retenu qu'à concurrence de la consommation avant DLU.
usable = Decimal("0")
expiring = Decimal("0")
for lot in (StockLot.query.filter_by(generic_product_id=product.id)
.filter(StockLot.status.notin_(("bloqué", "detruit", "expire", "epuise")))
.order_by(StockLot.expiry_date.is_(None), StockLot.expiry_date, StockLot.received_at).all()):
quantity = sum((Decimal(str(b.quantity or 0)) for b in lot.balances), Decimal("0"))
if not quantity:
continue
if lot.expiry_date and lot.expiry_date < target_date:
# consommation estimée jusqu'à la DLU (approximation jour par jour)
before = forecast_product(product, lot.expiry_date, start_date=start_date, config=config)["consumption_total"] if lot.expiry_date >= start_date else Decimal("0")
usable += min(quantity, before)
expiring += max(quantity - before, Decimal("0"))
else:
usable += quantity
net = max(total_need + Decimal(str(product.stock_security or 0)) - usable, Decimal("0"))
packaging = ProductPackaging.query.filter_by(generic_product_id=product.id, is_active=True, is_current=True).order_by(ProductPackaging.reference_quantity_per_package).first()
packages = (net / Decimal(str(packaging.reference_quantity_per_package))).to_integral_value(rounding="ROUND_CEILING") if packaging and net else Decimal("0")
proposed = packages * Decimal(str(packaging.reference_quantity_per_package)) if packaging else net
return {"start_date": start_date, "target_date": target_date, "school_days_consumption": quantities["scolaire"],
"permanence_consumption": quantities["permanence"], "closed_consumption": quantities["ferme"],
"consumption_total": total_need, "usable_stock": usable, "expiring_stock": expiring,
"safety_stock": Decimal(str(product.stock_security or 0)), "net_need": net,
"packaging": packaging, "packages_to_order": packages, "proposed_quantity": proposed}

View file

@ -0,0 +1,173 @@
"""Opérations transactionnelles et FEFO du stock d'entretien."""
from datetime import date
from decimal import Decimal, ROUND_HALF_UP
from sqlalchemy import or_, case
from ...extensions import db
from ...core.models import (
ProductGeneric, CommercialProduct, ProductPackaging, StockLocation,
StockLot, StockLotBalance, StockMovement, ReusableContainer, ContainerFill,
)
Q = Decimal("0.000001")
class StockError(ValueError):
pass
def dec(value):
try:
return Decimal(str(value)).quantize(Q, rounding=ROUND_HALF_UP)
except Exception as exc:
raise StockError("Quantité invalide") from exc
def _balance(lot_id, location_id, create=False):
query = StockLotBalance.query.filter_by(lot_id=lot_id, location_id=location_id)
balance = query.with_for_update().first()
if not balance and create:
balance = StockLotBalance(lot_id=lot_id, location_id=location_id, quantity=0)
db.session.add(balance)
db.session.flush()
return balance
def _movement(**kwargs):
movement = StockMovement(**kwargs)
db.session.add(movement)
return movement
def receive_stock(*, generic_product_id, commercial_product_id, packaging_id, packages,
lot_number, received_at, location_id, user_id=None, expiry_date=None,
manufactured_at=None, comment=None):
"""Réceptionne un nombre de conditionnements et crée le lot + mouvement."""
packages = dec(packages)
if packages <= 0:
raise StockError("La quantité reçue doit être positive")
packaging = ProductPackaging.query.get(packaging_id)
if not packaging or packaging.generic_product_id != generic_product_id:
raise StockError("Conditionnement inconnu pour ce produit")
if packaging.commercial_product_id and packaging.commercial_product_id != commercial_product_id:
raise StockError("Conditionnement associé à une autre référence")
quantity = (packages * dec(packaging.reference_quantity_per_package)).quantize(Q)
lot = StockLot(generic_product_id=generic_product_id, commercial_product_id=commercial_product_id,
lot_number=(lot_number or "").strip() or "Sans numéro", received_at=received_at,
manufactured_at=manufactured_at, expiry_date=expiry_date,
quantity_received=quantity)
db.session.add(lot)
db.session.flush()
balance = _balance(lot.id, location_id, create=True)
balance.quantity = quantity
_movement(generic_product_id=generic_product_id, commercial_product_id=commercial_product_id,
lot_id=lot.id, movement_type="reception_fournisseur", quantity_reference=quantity,
destination_location_id=location_id, user_id=user_id, comment=comment)
return lot
def _usable_lots(generic_product_id, location_id, today=None):
today = today or date.today()
return (StockLot.query.join(StockLotBalance)
.filter(StockLot.generic_product_id == generic_product_id,
StockLotBalance.location_id == location_id,
StockLotBalance.quantity > 0,
StockLot.status.notin_(("bloqué", "detruit", "expire", "epuise")),
or_(StockLot.expiry_date.is_(None), StockLot.expiry_date >= today))
.order_by(case((StockLot.expiry_date.is_(None), 1), else_=0), StockLot.expiry_date, StockLot.received_at, StockLot.id)
.with_for_update().all())
def issue_stock(*, generic_product_id, quantity, source_location_id, user_id=None,
staff_id=None, movement_type="sortie_agent", comment=None, today=None):
"""Sortie FEFO atomique. Retourne les mouvements créés."""
requested = dec(quantity)
if requested <= 0:
raise StockError("La quantité doit être positive")
remaining = requested
movements = []
for lot in _usable_lots(generic_product_id, source_location_id, today=today):
balance = _balance(lot.id, source_location_id)
available = dec(balance.quantity)
taken = min(available, remaining)
if taken <= 0:
continue
balance.quantity = available - taken
if balance.quantity == 0 and all(dec(b.quantity) == 0 for b in lot.balances):
lot.status = "epuise"
movements.append(_movement(generic_product_id=generic_product_id,
commercial_product_id=lot.commercial_product_id, lot_id=lot.id,
movement_type=movement_type, quantity_reference=-taken,
source_location_id=source_location_id, staff_id=staff_id, user_id=user_id, comment=comment))
remaining -= taken
if remaining <= 0:
break
if remaining > 0:
raise StockError(f"Stock disponible insuffisant (manque {remaining} unité(s) de référence)")
return movements
def transfer_stock(*, generic_product_id, quantity, source_location_id, destination_location_id,
user_id=None, comment=None):
if source_location_id == destination_location_id:
raise StockError("Les emplacements source et destination doivent être différents")
requested = dec(quantity)
if requested <= 0:
raise StockError("La quantité doit être positive")
remaining = requested
total = Decimal("0")
for lot in _usable_lots(generic_product_id, source_location_id):
source = _balance(lot.id, source_location_id)
taken = min(dec(source.quantity), remaining)
if taken <= 0:
continue
source.quantity = dec(source.quantity) - taken
destination = _balance(lot.id, destination_location_id, create=True)
destination.quantity = dec(destination.quantity) + taken
_movement(generic_product_id=generic_product_id,
commercial_product_id=lot.commercial_product_id, lot_id=lot.id,
movement_type="transfert", quantity_reference=-taken,
source_location_id=source_location_id, destination_location_id=destination_location_id,
user_id=user_id, comment=comment)
total += taken
remaining -= taken
if remaining <= 0:
break
if remaining > 0:
raise StockError(f"Stock disponible insuffisant (manque {remaining} unité(s) de référence)")
return total
def transvasement(*, generic_product_id, quantity_liters, source_location_id,
container_id=None, user_id=None, staff_id=None, comment=None):
movements = issue_stock(generic_product_id=generic_product_id, quantity=quantity_liters,
source_location_id=source_location_id, user_id=user_id,
staff_id=staff_id, movement_type="transvasement", comment=comment)
total = sum((-dec(m.quantity_reference) for m in movements), Decimal("0"))
if container_id:
container = ReusableContainer.query.get(container_id)
if not container:
raise StockError("Flacon inconnu")
if container.dedicated_product_id and container.dedicated_product_id != generic_product_id:
raise StockError("Ce flacon est dédié à un autre produit")
if total > dec(container.volume_liters):
raise StockError("Le volume dépasse la capacité du flacon")
container.current_product_id = generic_product_id
container.last_filled_at = db.func.now()
db.session.flush()
db.session.add(ContainerFill(container_id=container.id, movement_id=movements[0].id,
volume_liters=total))
return movements
def dilute(*, generic_product_id, total_liters, ratio_percent, source_location_id,
user_id=None, staff_id=None, container_id=None, comment=None):
total = dec(total_liters)
ratio = dec(ratio_percent)
if total <= 0 or ratio <= 0 or ratio >= 100:
raise StockError("Volume et taux de dilution invalides")
concentrate = (total * ratio / Decimal("100")).quantize(Q)
return transvasement(generic_product_id=generic_product_id, quantity_liters=concentrate,
source_location_id=source_location_id, container_id=container_id,
user_id=user_id, staff_id=staff_id, comment=f"Dilution {ratio}%{('' + comment) if comment else ''}")

View file

@ -0,0 +1 @@
{% extends "base.html" %}{% block title %}Matériel & Entretien{% endblock %}{% block content %}<div class="container-fluid"><h1>Matériel &amp; Produits d'entretien</h1><div class="row g-3"><div class="col-md-4"><div class="card"><div class="card-body"><h5>Produits</h5><strong class="fs-3">{{ products|length }}</strong><a class="btn btn-sm btn-primary float-end" href="{{ url_for('cleaning.products') }}">Gérer</a></div></div></div><div class="col-md-4"><div class="card border-warning"><div class="card-body"><h5>Sous stock minimum</h5><strong class="fs-3">{{ low_stock|length }}</strong></div></div></div><div class="col-md-4"><div class="card border-danger"><div class="card-body"><h5>DLU expirées</h5><strong class="fs-3">{{ expiring|length }}</strong></div></div></div></div><div class="mt-4 d-flex flex-wrap gap-2"><a class="btn btn-primary" href="{{ url_for('cleaning.receive') }}">Réception</a><a class="btn btn-outline-primary" href="{{ url_for('cleaning.issue') }}">Sortie agent</a><a class="btn btn-outline-primary" href="{{ url_for('cleaning.transfer') }}">Transfert</a><a class="btn btn-outline-secondary" href="{{ url_for('cleaning.stock') }}">État du stock</a><a class="btn btn-outline-secondary" href="{{ url_for('cleaning.movements') }}">Mouvements</a><a class="btn btn-outline-secondary" href="{{ url_for('cleaning.inventory') }}">Inventaire</a></div><h3 class="mt-4">Mouvements récents</h3><div class="table-responsive"><table class="table table-sm"><tr><th>Date</th><th>Type</th><th>Produit</th><th>Quantité</th></tr>{% for m in recent %}<tr><td>{{ m.created_at|datetime_fmt }}</td><td>{{ m.movement_type }}</td><td>{{ m.generic_product.name }}</td><td>{{ m.quantity_reference }}</td></tr>{% else %}<tr><td colspan="4" class="text-muted">Aucun mouvement.</td></tr>{% endfor %}</table></div></div>{% endblock %}

View file

@ -0,0 +1 @@
{% extends "base.html" %}{% block title %}Prévision{% endblock %}{% block content %}<div class="container"><h1>Prévision — {{ product.name }}</h1><form class="row g-2 mb-3"><div class="col-auto"><input class="form-control" type="date" name="target_date" value="{{ result.target_date.isoformat() }}"></div><div class="col-auto"><button class="btn btn-primary">Recalculer</button></div></form><table class="table"><tr><th>Élément</th><th>Quantité</th></tr><tr><td>Jours scolaires</td><td>{{ result.school_days_consumption }}</td></tr><tr><td>Permanences</td><td>{{ result.permanence_consumption }}</td></tr><tr><td>Fermetures</td><td>{{ result.closed_consumption }}</td></tr><tr><td>Consommation totale</td><td>{{ result.consumption_total }}</td></tr><tr><td>Stock utilisable</td><td>{{ result.usable_stock }}</td></tr><tr><td>Stock risquant d'expirer</td><td>{{ result.expiring_stock }}</td></tr><tr><td>Stock de sécurité</td><td>{{ result.safety_stock }}</td></tr><tr class="table-primary"><th>Besoin net</th><th>{{ result.net_need }}</th></tr><tr class="table-success"><th>Commande proposée</th><th>{{ result.packages_to_order }} conditionnement(s), soit {{ result.proposed_quantity }}</th></tr></table></div>{% endblock %}

View file

@ -0,0 +1 @@
{% extends "base.html" %}{% block title %}Coefficients de prévision{% endblock %}{% block content %}<div class="container"><h1>Coefficients calendrier</h1><p>1 = journée scolaire normale ; les permanences et fermetures restent configurables.</p><form method="post"><input type="hidden" name="csrf_token" value="{{ csrf_token() }}"><label>Journée scolaire</label><input class="form-control mb-2" type="number" step="0.00001" name="normal_coefficient" value="{{ values.normal_coefficient }}"><label>Permanence</label><input class="form-control mb-2" type="number" step="0.00001" name="permanence_coefficient" value="{{ values.permanence_coefficient }}"><label>Fermeture</label><input class="form-control mb-2" type="number" step="0.00001" name="closed_coefficient" value="{{ values.closed_coefficient }}"><label>Journée exceptionnelle</label><input class="form-control mb-2" type="number" step="0.00001" name="exceptional_coefficient" value="{{ values.exceptional_coefficient }}"><button class="btn btn-primary">Enregistrer</button></form></div>{% endblock %}

View file

@ -0,0 +1 @@
{% extends "base.html" %}{% block title %}Inventaire{% endblock %}{% block content %}<div class="container"><h1>Inventaire physique</h1><form method="get" class="mb-3"><select class="form-select" name="location_id" onchange="this.form.submit()"><option value="">Choisir un emplacement</option>{% for l in locations %}<option value="{{ l.id }}" {% if selected==l.id %}selected{% endif %}>{{ l.name }}</option>{% endfor %}</select></form>{% if selected %}<form method="post"><input type="hidden" name="csrf_token" value="{{ csrf_token() }}"><input type="hidden" name="location_id" value="{{ selected }}"><table class="table"><tr><th>Lot</th><th>Théorique</th><th>Compté</th></tr>{% for b in rows %}<tr><td>{{ b.lot.generic_product.name }} — {{ b.lot.lot_number }}<input type="hidden" name="line_lot_id" value="{{ b.lot_id }}"></td><td>{{ b.quantity }}</td><td><input class="form-control" name="counted_{{ b.lot_id }}" value="{{ b.quantity }}" step="0.000001" type="number"></td></tr>{% endfor %}</table><button class="btn btn-primary">Valider et tracer les écarts</button></form>{% endif %}</div>{% endblock %}

View file

@ -0,0 +1 @@
{% extends "base.html" %}{% block title %}Sortie agent{% endblock %}{% block content %}<div class="container"><h1>Sortie de produits</h1><form method="post"><input type="hidden" name="csrf_token" value="{{ csrf_token() }}"><label>Produit</label><select class="form-select mb-2" name="generic_product_id" required>{% for p in products %}<option value="{{ p.id }}">{{ p.name }}</option>{% endfor %}</select><label>Quantité (unité de référence)</label><input class="form-control mb-2" name="quantity" type="number" step="0.000001" required><label>Emplacement source</label><select class="form-select mb-2" name="source_location_id" required>{% for l in locations %}<option value="{{ l.id }}">{{ l.name }}</option>{% endfor %}</select><label>Agent destinataire (pas de compte GMAO)</label><select class="form-select mb-2" name="staff_id"><option value=""></option>{% for s in staff %}<option value="{{ s.id }}">{{ s.full_name }}</option>{% endfor %}</select><textarea class="form-control mb-3" name="comment" placeholder="Commentaire"></textarea><button class="btn btn-primary">Valider la sortie FEFO</button></form></div>{% endblock %}

View file

@ -0,0 +1 @@
{% extends "base.html" %}{% block title %}Emplacements de stock{% endblock %}{% block content %}<div class="container"><h1>Emplacements</h1><form method="post" class="row g-2"><input type="hidden" name="csrf_token" value="{{ csrf_token() }}"><div class="col"><input class="form-control" name="name" placeholder="Local ménage bâtiment A" required></div><div class="col"><input class="form-control" name="notes" placeholder="Notes"></div><div class="col-auto"><button class="btn btn-primary">Ajouter</button></div></form><ul class="list-group mt-3">{% for l in locations %}<li class="list-group-item">{{ l.name }}</li>{% endfor %}</ul></div>{% endblock %}

View file

@ -0,0 +1 @@
{% extends "base.html" %}{% block title %}Matériel durable{% endblock %}{% block content %}<div class="container-fluid"><h1>Matériel durable de ménage</h1><p class="text-muted">Les équipements existants sont réutilisés ; cette vue conserve leurs affectations aux agents/zones.</p><table class="table"><tr><th>Équipement</th><th>Statut</th><th>Agent</th><th>Zone/salle</th></tr>{% for a in assignments %}<tr><td>{{ a.equipment.name }}</td><td>{{ a.status }}</td><td>{{ a.staff.full_name if a.staff else '—' }}</td><td>{{ a.room.name if a.room else (a.zone.name if a.zone else '—') }}</td></tr>{% else %}<tr><td colspan="4">Aucune affectation.</td></tr>{% endfor %}</table></div>{% endblock %}

View file

@ -0,0 +1 @@
{% extends "base.html" %}{% block title %}Mouvements de stock{% endblock %}{% block content %}<div class="container-fluid"><h1>Journal immuable des mouvements</h1><table class="table table-sm"><tr><th>Date</th><th>Type</th><th>Produit</th><th>Lot</th><th>Quantité</th><th>Agent</th><th>Utilisateur</th></tr>{% for m in movements %}<tr><td>{{ m.created_at|datetime_fmt }}</td><td>{{ m.movement_type }}</td><td>{{ m.generic_product.name }}</td><td>{{ m.lot.lot_number if m.lot else '—' }}</td><td>{{ m.quantity_reference }}</td><td>{{ m.staff.full_name if m.staff else '—' }}</td><td>{{ m.user.username if m.user else '—' }}</td></tr>{% endfor %}</table></div>{% endblock %}

View file

@ -0,0 +1 @@
{% extends "base.html" %}{% block title %}Conditionnement{% endblock %}{% block content %}<div class="container"><h1>Conditionnement d'achat</h1><form method="post"><input type="hidden" name="csrf_token" value="{{ csrf_token() }}"><label>Produit générique</label><select class="form-select mb-2" name="generic_product_id" required>{% for p in products %}<option value="{{ p.id }}">{{ p.name }} ({{ p.reference_unit }})</option>{% endfor %}</select><label>Référence commerciale</label><select class="form-select mb-2" name="commercial_product_id"><option value="">Toutes références</option>{% for r in references %}<option value="{{ r.id }}">{{ r.commercial_name }}</option>{% endfor %}</select><label>Nom</label><input class="form-control mb-2" name="name" placeholder="Carton de 10 rouleaux" required><label>Unité d'achat</label><input class="form-control mb-2" name="purchase_unit" value="conditionnement"><label>Unités par conditionnement</label><input class="form-control mb-2" name="units_per_package" value="1" step="0.000001" type="number"><label>Quantité en unité de référence par conditionnement</label><input class="form-control mb-2" name="reference_quantity_per_package" step="0.000001" type="number" required><label><input type="checkbox" name="is_current" checked> Conditionnement actuel</label><br><button class="btn btn-primary mt-3">Enregistrer</button></form></div>{% endblock %}

View file

@ -0,0 +1 @@
{% extends "base.html" %}{% block title %}Produit générique{% endblock %}{% block content %}<div class="container"><h1>{{ 'Modifier' if product else 'Nouveau' }} produit générique</h1><form method="post"><input type="hidden" name="csrf_token" value="{{ csrf_token() }}"><div class="mb-3"><label class="form-label">Nom générique</label><input class="form-control" required name="name" value="{{ product.name if product else '' }}"></div><div class="mb-3"><label class="form-label">Description</label><textarea class="form-control" name="description">{{ product.description if product else '' }}</textarea></div><div class="row"><div class="col-md-4"><label class="form-label">Catégorie</label><select class="form-select" name="category_id"><option value=""></option>{% for c in categories %}<option value="{{ c.id }}" {% if product and product.category_id==c.id %}selected{% endif %}>{{ c.name }}</option>{% endfor %}</select></div><div class="col-md-4"><label class="form-label">Unité de référence</label><input class="form-control" name="reference_unit" value="{{ product.reference_unit if product else 'unité' }}"></div><div class="col-md-4"><label class="form-label">Type</label><select class="form-select" name="product_type"><option>consommable</option><option>liquide</option><option>matériel</option></select></div></div><div class="row mt-3"><div class="col-md-4"><label>Stock minimum</label><input class="form-control" type="number" step="0.000001" name="stock_minimum" value="{{ product.stock_minimum if product else 0 }}"></div><div class="col-md-4"><label>Stock de sécurité</label><input class="form-control" type="number" step="0.000001" name="stock_security" value="{{ product.stock_security if product else 0 }}"></div><div class="col-md-4"><label>Consommation prévue/jour</label><input class="form-control" type="number" step="0.000001" name="forecast_daily_quantity" value="{{ product.forecast_daily_quantity if product else 0 }}"></div></div><button class="btn btn-primary mt-3">Enregistrer</button></form></div>{% endblock %}

View file

@ -0,0 +1 @@
{% extends "base.html" %}{% block title %}Produits génériques{% endblock %}{% block content %}<div class="container-fluid"><div class="d-flex justify-content-between"><h1>Produits génériques</h1><a class="btn btn-primary" href="{{ url_for('cleaning.product_new') }}">Nouveau produit</a></div><p class="text-muted">Le produit générique reste stable ; les références fournisseurs sont gérées séparément.</p><table class="table table-hover"><thead><tr><th>Nom</th><th>Catégorie</th><th>Unité</th><th>Prévision/jour</th><th>Stock</th><th></th></tr></thead><tbody>{% for p in products %}<tr><td>{{ p.name }}</td><td>{{ p.category.name if p.category else '—' }}</td><td>{{ p.reference_unit }}</td><td>{{ p.forecast_daily_quantity }}</td><td>{{ p.stock_total }}</td><td><a class="btn btn-sm btn-outline-secondary" href="{{ url_for('cleaning.product_edit', id=p.id) }}">Modifier</a> <a class="btn btn-sm btn-outline-info" href="{{ url_for('cleaning.forecast', product_id=p.id) }}">Prévoir</a></td></tr>{% else %}<tr><td colspan="6">Aucun produit.</td></tr>{% endfor %}</tbody></table><a href="{{ url_for('cleaning.references') }}">Références commerciales</a></div>{% endblock %}

View file

@ -0,0 +1 @@
{% extends "base.html" %}{% block title %}Réception{% endblock %}{% block content %}<div class="container"><h1>Réception fournisseur</h1><form method="post"><input type="hidden" name="csrf_token" value="{{ csrf_token() }}"><label>Produit</label><select class="form-select mb-2" name="generic_product_id" required>{% for p in products %}<option value="{{ p.id }}">{{ p.name }}</option>{% endfor %}</select><label>Référence commerciale</label><select class="form-select mb-2" name="commercial_product_id" required>{% for r in references %}<option value="{{ r.id }}">{{ r.generic_product.name }} — {{ r.commercial_name }}</option>{% endfor %}</select><label>Conditionnement</label><select class="form-select mb-2" name="packaging_id" required>{% for p in packagings %}<option value="{{ p.id }}">{{ p.name }} ({{ p.reference_quantity_per_package }})</option>{% endfor %}</select><label>Nombre de conditionnements</label><input class="form-control mb-2" name="packages" type="number" step="0.000001" required><label>Numéro de lot</label><input class="form-control mb-2" name="lot_number" required><label>Date réception</label><input class="form-control mb-2" type="date" name="received_at"><label>DLU</label><input class="form-control mb-2" type="date" name="expiry_date"><label>Emplacement</label><select class="form-select mb-3" name="location_id" required>{% for l in locations %}<option value="{{ l.id }}">{{ l.name }}</option>{% endfor %}</select><button class="btn btn-primary">Valider la réception</button></form></div>{% endblock %}

View file

@ -0,0 +1 @@
{% extends "base.html" %}{% block title %}Référence commerciale{% endblock %}{% block content %}<div class="container"><h1>Nouvelle référence commerciale</h1><form method="post"><input type="hidden" name="csrf_token" value="{{ csrf_token() }}"><label>Produit générique</label><select class="form-select mb-3" name="generic_product_id" required>{% for p in products %}<option value="{{ p.id }}">{{ p.name }}</option>{% endfor %}</select><label>Nom commercial</label><input class="form-control mb-3" name="commercial_name" required><label>Marque</label><input class="form-control mb-3" name="brand"><label>Référence fabricant</label><input class="form-control mb-3" name="manufacturer_reference"><label>Référence fournisseur</label><input class="form-control mb-3" name="supplier_reference"><label><input type="checkbox" name="is_concentrated"> Produit concentré</label><br><button class="btn btn-primary mt-3">Enregistrer</button></form></div>{% endblock %}

View file

@ -0,0 +1 @@
{% extends "base.html" %}{% block title %}Références commerciales{% endblock %}{% block content %}<div class="container-fluid"><div class="d-flex justify-content-between"><h1>Références commerciales</h1><a class="btn btn-primary" href="{{ url_for('cleaning.reference_new') }}">Nouvelle référence</a></div><table class="table"><tr><th>Produit générique</th><th>Nom commercial</th><th>Marque</th><th>Actif</th></tr>{% for r in references %}<tr><td>{{ r.generic_product.name }}</td><td>{{ r.commercial_name }}</td><td>{{ r.brand or '—' }}</td><td>{{ 'Oui' if r.is_active else 'Non' }}</td></tr>{% else %}<tr><td colspan="4">Aucune référence.</td></tr>{% endfor %}</table></div>{% endblock %}

View file

@ -0,0 +1 @@
{% extends "base.html" %}{% block title %}Fiche de demande{% endblock %}{% block content %}<div class="container"><h1>Fiche papier — {{ profile.name }}</h1><p>Établissement : ____________________ &nbsp; Date : {{ today.strftime('%d/%m/%Y') }}</p><label>Agent</label><select class="form-select mb-3">{% for s in staff %}<option>{{ s.full_name }}</option>{% endfor %}</select><table class="table table-bordered"><tr><th>Produit générique</th><th>Quantité demandée</th><th>Commentaire</th></tr>{% for item in profile.items %}<tr><td>{{ item.product.name }}</td><td style="height:40px"></td><td></td></tr>{% endfor %}</table><p>Signature : ______________________________</p><button class="btn btn-primary d-print-none" onclick="window.print()">Imprimer</button></div>{% endblock %}

View file

@ -0,0 +1 @@
{% extends "base.html" %}{% block title %}Profil de demande{% endblock %}{% block content %}<div class="container"><h1>Nouveau profil de demande papier</h1><form method="post"><input type="hidden" name="csrf_token" value="{{ csrf_token() }}"><input class="form-control mb-2" name="name" placeholder="Entretien général" required><textarea class="form-control mb-3" name="description" placeholder="Description"></textarea><h5>Produits génériques affichés</h5>{% for p in products %}<label class="d-block"><input type="checkbox" name="product_ids" value="{{ p.id }}"> {{ p.name }}</label>{% endfor %}<button class="btn btn-primary mt-3">Créer le profil</button></form></div>{% endblock %}

View file

@ -0,0 +1 @@
{% extends "base.html" %}{% block title %}Fiches de demande{% endblock %}{% block content %}<div class="container"><h1>Profils de fiches papier</h1>{% for p in profiles %}<div class="card mb-2"><div class="card-body"><strong>{{ p.name }}</strong> — {{ p.items|length }} produit(s)<a class="btn btn-sm btn-outline-primary float-end" href="{{ url_for('cleaning.request_print', profile_id=p.id) }}">Imprimer</a></div></div>{% else %}<p>Aucun profil configuré.</p>{% endfor %}</div>{% endblock %}

View file

@ -0,0 +1 @@
{% extends "base.html" %}{% block title %}État du stock{% endblock %}{% block content %}<div class="container-fluid"><h1>État du stock par lot et emplacement</h1><table class="table table-sm"><tr><th>Produit</th><th>Lot</th><th>DLU</th><th>Emplacement</th><th>Quantité</th><th>Statut</th></tr>{% for b in balances %}<tr class="{% if b.lot.expiry_date and b.lot.expiry_date <= current_date %}table-danger{% endif %}"><td>{{ b.lot.generic_product.name }}</td><td>{{ b.lot.lot_number }}</td><td>{{ b.lot.expiry_date or '—' }}</td><td>{{ b.location.name }}</td><td>{{ b.quantity }} {{ b.lot.generic_product.reference_unit }}</td><td>{{ b.lot.status }}</td></tr>{% else %}<tr><td colspan="6">Aucun stock.</td></tr>{% endfor %}</table></div>{% endblock %}

View file

@ -0,0 +1 @@
{% extends "base.html" %}{% block title %}Transfert{% endblock %}{% block content %}<div class="container"><h1>Transfert entre emplacements</h1><form method="post"><input type="hidden" name="csrf_token" value="{{ csrf_token() }}"><label>Produit</label><select class="form-select mb-2" name="generic_product_id" required>{% for p in products %}<option value="{{ p.id }}">{{ p.name }}</option>{% endfor %}</select><label>Quantité</label><input class="form-control mb-2" name="quantity" type="number" step="0.000001" required><div class="row"><div class="col"><label>Source</label><select class="form-select" name="source_location_id">{% for l in locations %}<option value="{{ l.id }}">{{ l.name }}</option>{% endfor %}</select></div><div class="col"><label>Destination</label><select class="form-select" name="destination_location_id">{% for l in locations %}<option value="{{ l.id }}">{{ l.name }}</option>{% endfor %}</select></div></div><button class="btn btn-primary mt-3">Transférer</button></form></div>{% endblock %}

View file

@ -0,0 +1 @@
{% extends "base.html" %}{% block title %}Transvasement{% endblock %}{% block content %}<div class="container"><h1>Transvasement / dilution</h1><form method="post"><input type="hidden" name="csrf_token" value="{{ csrf_token() }}"><label>Produit générique</label><select class="form-select mb-2" name="generic_product_id">{% for p in products %}<option value="{{ p.id }}">{{ p.name }}</option>{% endfor %}</select><label>Volume de produit prélevé (L)</label><input class="form-control mb-2" name="quantity_liters" type="number" step="0.000001" required><label>Emplacement source</label><select class="form-select mb-2" name="source_location_id">{% for l in locations %}<option value="{{ l.id }}">{{ l.name }}</option>{% endfor %}</select><label>Flacon réutilisable</label><select class="form-select mb-2" name="container_id"><option value="">Aucun</option>{% for c in containers %}<option value="{{ c.id }}">{{ c.identifier }} — {{ c.volume_liters }} L</option>{% endfor %}</select><label><input type="checkbox" name="dilution"> Dilution</label><input class="form-control mb-2" name="ratio_percent" placeholder="Taux en % (ex. 2)"><textarea class="form-control mb-3" name="comment" placeholder="Commentaire"></textarea><button class="btn btn-primary">Enregistrer</button></form></div>{% endblock %}

View file

@ -73,7 +73,7 @@ PATRIMOINE_BLUEPRINTS = {
"equipments_scheduled", "buildings", "zones", "rooms", "room_types", "wizard", "lots",
}
PLANNING_BLUEPRINTS = {"planning", "scheduler", "interventions_planning"}
STOCK_BLUEPRINTS = {"parts", "meters"}
STOCK_BLUEPRINTS = {"parts", "meters", "cleaning"}
CONTRACT_BLUEPRINTS = {"companies", "contracts", "services"}
PREVENTION_BLUEPRINTS = {"trainings", "constraints", "prevention"}

View file

@ -19,6 +19,13 @@ from .planning import (
from .settings import AppSettings
from .audit import AuditLog, TemplateAuditMark
from .prevention import PreventionWorkLog, StaffAuthorization, RiskAssessment, PreventionAction, SafetyRegisterEntry
from .cleaning import (
ProductCategory, ProductGeneric, CommercialProduct, ProductPackaging,
StockLocation, StockLot, StockLotBalance, StockMovement,
MaterialAssignment, ReusableContainer, ContainerFill, ProductDocument,
RequestProfile, RequestProfileItem, StockInventory, StockInventoryLine,
CleaningForecastConfig,
)
__all__ = [
'User', 'Staff',
@ -33,4 +40,9 @@ __all__ = [
'TechnicianAvailability', 'AdminTask', 'ZoneAccessRule',
'AppSettings', 'AuditLog', 'TemplateAuditMark',
'PreventionWorkLog', 'StaffAuthorization', 'RiskAssessment', 'PreventionAction', 'SafetyRegisterEntry',
'ProductCategory', 'ProductGeneric', 'CommercialProduct', 'ProductPackaging',
'StockLocation', 'StockLot', 'StockLotBalance', 'StockMovement',
'MaterialAssignment', 'ReusableContainer', 'ContainerFill', 'ProductDocument',
'RequestProfile', 'RequestProfileItem', 'StockInventory', 'StockInventoryLine',
'CleaningForecastConfig',
]

View file

@ -0,0 +1,288 @@
"""Modèles du module Matériel & Produits d'entretien.
Les quantités sont toujours exprimées dans l'unité de référence du produit et
stockées en DECIMAL. Les mouvements constituent l'historique immuable ; les
soldes par lot/emplacement sont des projections verrouillées pendant les
opérations transactionnelles.
"""
from datetime import datetime, timezone
from decimal import Decimal
from ...extensions import db
def utcnow():
return datetime.now(timezone.utc)
class ProductCategory(db.Model):
__tablename__ = "cleaning_product_categories"
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(120), nullable=False, unique=True)
description = db.Column(db.Text)
is_active = db.Column(db.Boolean, nullable=False, default=True)
products = db.relationship("ProductGeneric", back_populates="category")
class ProductGeneric(db.Model):
__tablename__ = "cleaning_products_generic"
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(200), nullable=False, unique=True)
description = db.Column(db.Text)
category_id = db.Column(db.Integer, db.ForeignKey("cleaning_product_categories.id"), index=True)
product_type = db.Column(db.String(40), nullable=False, default="consommable")
reference_unit = db.Column(db.String(30), nullable=False, default="unité")
is_active = db.Column(db.Boolean, nullable=False, default=True)
stock_minimum = db.Column(db.Numeric(14, 6), nullable=False, default=0)
stock_security = db.Column(db.Numeric(14, 6), nullable=False, default=0)
forecast_daily_quantity = db.Column(db.Numeric(14, 6), nullable=False, default=0)
display_order = db.Column(db.Integer, nullable=False, default=0)
usage_zones = db.Column(db.String(500))
request_enabled = db.Column(db.Boolean, nullable=False, default=True)
allowed_distribution_modes = db.Column(db.String(255), nullable=False, default="standard")
created_at = db.Column(db.DateTime, nullable=False, default=utcnow)
updated_at = db.Column(db.DateTime, nullable=False, default=utcnow, onupdate=utcnow)
category = db.relationship("ProductCategory", back_populates="products")
commercial_products = db.relationship("CommercialProduct", back_populates="generic_product", cascade="all, delete-orphan")
packagings = db.relationship("ProductPackaging", back_populates="generic_product", cascade="all, delete-orphan")
lots = db.relationship("StockLot", back_populates="generic_product")
@property
def stock_total(self):
return sum((b.quantity for lot in self.lots for b in lot.balances), Decimal("0"))
class CommercialProduct(db.Model):
__tablename__ = "cleaning_commercial_products"
id = db.Column(db.Integer, primary_key=True)
generic_product_id = db.Column(db.Integer, db.ForeignKey("cleaning_products_generic.id"), nullable=False, index=True)
commercial_name = db.Column(db.String(200), nullable=False)
brand = db.Column(db.String(120))
manufacturer_id = db.Column(db.Integer, db.ForeignKey("companies.id"), index=True)
supplier_id = db.Column(db.Integer, db.ForeignKey("companies.id"), index=True)
manufacturer_reference = db.Column(db.String(120))
supplier_reference = db.Column(db.String(120))
barcode = db.Column(db.String(80), index=True)
valid_from = db.Column(db.Date)
valid_to = db.Column(db.Date)
is_active = db.Column(db.Boolean, nullable=False, default=True)
photo_path = db.Column(db.String(500))
dosage_information = db.Column(db.Text)
is_concentrated = db.Column(db.Boolean, nullable=False, default=False)
after_opening_days = db.Column(db.Integer)
safety_notes = db.Column(db.Text)
pictograms = db.Column(db.String(500))
warning_statement = db.Column(db.String(120))
hazard_statements = db.Column(db.Text)
precautionary_statements = db.Column(db.Text)
required_ppe = db.Column(db.String(500))
storage_conditions = db.Column(db.Text)
incompatibilities = db.Column(db.Text)
created_at = db.Column(db.DateTime, nullable=False, default=utcnow)
generic_product = db.relationship("ProductGeneric", back_populates="commercial_products")
manufacturer = db.relationship("Company", foreign_keys=[manufacturer_id])
supplier = db.relationship("Company", foreign_keys=[supplier_id])
lots = db.relationship("StockLot", back_populates="commercial_product")
packagings = db.relationship("ProductPackaging", back_populates="commercial_product")
documents = db.relationship("ProductDocument", back_populates="commercial_product", cascade="all, delete-orphan")
class ProductPackaging(db.Model):
__tablename__ = "cleaning_product_packagings"
id = db.Column(db.Integer, primary_key=True)
generic_product_id = db.Column(db.Integer, db.ForeignKey("cleaning_products_generic.id"), nullable=False, index=True)
commercial_product_id = db.Column(db.Integer, db.ForeignKey("cleaning_commercial_products.id"), index=True)
name = db.Column(db.String(160), nullable=False)
purchase_unit = db.Column(db.String(40), nullable=False, default="conditionnement")
units_per_package = db.Column(db.Numeric(14, 6), nullable=False, default=1)
reference_quantity_per_package = db.Column(db.Numeric(14, 6), nullable=False)
is_active = db.Column(db.Boolean, nullable=False, default=True)
is_current = db.Column(db.Boolean, nullable=False, default=True)
generic_product = db.relationship("ProductGeneric", back_populates="packagings")
commercial_product = db.relationship("CommercialProduct", back_populates="packagings")
class StockLocation(db.Model):
__tablename__ = "cleaning_stock_locations"
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(160), nullable=False, unique=True)
building_id = db.Column(db.Integer, db.ForeignKey("buildings.id"), index=True)
zone_id = db.Column(db.Integer, db.ForeignKey("zones.id"), index=True)
room_id = db.Column(db.Integer, db.ForeignKey("rooms.id"), index=True)
is_active = db.Column(db.Boolean, nullable=False, default=True)
notes = db.Column(db.Text)
balances = db.relationship("StockLotBalance", back_populates="location")
class StockLot(db.Model):
__tablename__ = "cleaning_stock_lots"
id = db.Column(db.Integer, primary_key=True)
generic_product_id = db.Column(db.Integer, db.ForeignKey("cleaning_products_generic.id"), nullable=False, index=True)
commercial_product_id = db.Column(db.Integer, db.ForeignKey("cleaning_commercial_products.id"), nullable=False, index=True)
lot_number = db.Column(db.String(120), nullable=False, index=True)
received_at = db.Column(db.Date, nullable=False)
manufactured_at = db.Column(db.Date)
expiry_date = db.Column(db.Date, index=True)
quantity_received = db.Column(db.Numeric(14, 6), nullable=False)
opened_at = db.Column(db.Date)
after_opening_days = db.Column(db.Integer)
status = db.Column(db.String(30), nullable=False, default="disponible", index=True)
notes = db.Column(db.Text)
generic_product = db.relationship("ProductGeneric", back_populates="lots")
commercial_product = db.relationship("CommercialProduct", back_populates="lots")
balances = db.relationship("StockLotBalance", back_populates="lot", cascade="all, delete-orphan")
movements = db.relationship("StockMovement", back_populates="lot")
class StockLotBalance(db.Model):
__tablename__ = "cleaning_stock_lot_balances"
__table_args__ = (db.UniqueConstraint("lot_id", "location_id", name="uq_cleaning_lot_location"),)
id = db.Column(db.Integer, primary_key=True)
lot_id = db.Column(db.Integer, db.ForeignKey("cleaning_stock_lots.id", ondelete="RESTRICT"), nullable=False, index=True)
location_id = db.Column(db.Integer, db.ForeignKey("cleaning_stock_locations.id", ondelete="RESTRICT"), nullable=False, index=True)
quantity = db.Column(db.Numeric(14, 6), nullable=False, default=0)
lot = db.relationship("StockLot", back_populates="balances")
location = db.relationship("StockLocation", back_populates="balances")
class StockMovement(db.Model):
__tablename__ = "cleaning_stock_movements"
id = db.Column(db.BigInteger, primary_key=True, autoincrement=True)
generic_product_id = db.Column(db.Integer, db.ForeignKey("cleaning_products_generic.id"), nullable=False, index=True)
commercial_product_id = db.Column(db.Integer, db.ForeignKey("cleaning_commercial_products.id"), index=True)
lot_id = db.Column(db.Integer, db.ForeignKey("cleaning_stock_lots.id"), index=True)
movement_type = db.Column(db.String(30), nullable=False, index=True)
quantity_reference = db.Column(db.Numeric(14, 6), nullable=False)
source_location_id = db.Column(db.Integer, db.ForeignKey("cleaning_stock_locations.id"), index=True)
destination_location_id = db.Column(db.Integer, db.ForeignKey("cleaning_stock_locations.id"), index=True)
staff_id = db.Column(db.Integer, db.ForeignKey("staff.id"), index=True)
user_id = db.Column(db.Integer, db.ForeignKey("users.id", ondelete="SET NULL"), index=True)
comment = db.Column(db.Text)
dilution_ratio = db.Column(db.Numeric(8, 5))
created_at = db.Column(db.DateTime, nullable=False, default=utcnow, index=True)
lot = db.relationship("StockLot", back_populates="movements")
generic_product = db.relationship("ProductGeneric")
commercial_product = db.relationship("CommercialProduct")
source_location = db.relationship("StockLocation", foreign_keys=[source_location_id])
destination_location = db.relationship("StockLocation", foreign_keys=[destination_location_id])
staff = db.relationship("Staff")
user = db.relationship("User")
class MaterialAssignment(db.Model):
__tablename__ = "cleaning_material_assignments"
id = db.Column(db.Integer, primary_key=True)
equipment_id = db.Column(db.Integer, db.ForeignKey("equipments.id", ondelete="RESTRICT"), nullable=False, index=True)
staff_id = db.Column(db.Integer, db.ForeignKey("staff.id"), index=True)
zone_id = db.Column(db.Integer, db.ForeignKey("zones.id"), index=True)
room_id = db.Column(db.Integer, db.ForeignKey("rooms.id"), index=True)
status = db.Column(db.String(30), nullable=False, default="affecte")
assigned_at = db.Column(db.DateTime, nullable=False, default=utcnow)
returned_at = db.Column(db.DateTime)
notes = db.Column(db.Text)
equipment = db.relationship("Equipment")
staff = db.relationship("Staff")
zone = db.relationship("Zone")
room = db.relationship("Room")
class ReusableContainer(db.Model):
__tablename__ = "cleaning_reusable_containers"
id = db.Column(db.Integer, primary_key=True)
identifier = db.Column(db.String(80), nullable=False, unique=True)
container_type = db.Column(db.String(80), nullable=False)
volume_liters = db.Column(db.Numeric(12, 6), nullable=False)
dedicated_product_id = db.Column(db.Integer, db.ForeignKey("cleaning_products_generic.id"))
current_product_id = db.Column(db.Integer, db.ForeignKey("cleaning_products_generic.id"))
staff_id = db.Column(db.Integer, db.ForeignKey("staff.id"))
zone_id = db.Column(db.Integer, db.ForeignKey("zones.id"))
state = db.Column(db.String(30), nullable=False, default="disponible")
last_filled_at = db.Column(db.DateTime)
dedicated_product = db.relationship("ProductGeneric", foreign_keys=[dedicated_product_id])
current_product = db.relationship("ProductGeneric", foreign_keys=[current_product_id])
staff = db.relationship("Staff")
zone = db.relationship("Zone")
fills = db.relationship("ContainerFill", back_populates="container", cascade="all, delete-orphan")
class ContainerFill(db.Model):
__tablename__ = "cleaning_container_fills"
id = db.Column(db.Integer, primary_key=True)
container_id = db.Column(db.Integer, db.ForeignKey("cleaning_reusable_containers.id", ondelete="RESTRICT"), nullable=False, index=True)
movement_id = db.Column(db.BigInteger, db.ForeignKey("cleaning_stock_movements.id", ondelete="RESTRICT"), nullable=False)
volume_liters = db.Column(db.Numeric(12, 6), nullable=False)
dilution_ratio = db.Column(db.Numeric(8, 5))
prepared_at = db.Column(db.DateTime, nullable=False, default=utcnow)
container = db.relationship("ReusableContainer", back_populates="fills")
movement = db.relationship("StockMovement")
class ProductDocument(db.Model):
__tablename__ = "cleaning_product_documents"
id = db.Column(db.Integer, primary_key=True)
commercial_product_id = db.Column(db.Integer, db.ForeignKey("cleaning_commercial_products.id", ondelete="RESTRICT"), nullable=False, index=True)
document_type = db.Column(db.String(30), nullable=False, default="fds")
title = db.Column(db.String(255), nullable=False)
filename = db.Column(db.String(255), nullable=False)
filepath = db.Column(db.String(500), nullable=False)
version = db.Column(db.String(50))
published_at = db.Column(db.Date)
uploaded_at = db.Column(db.DateTime, nullable=False, default=utcnow)
uploaded_by_id = db.Column(db.Integer, db.ForeignKey("users.id", ondelete="SET NULL"))
is_current = db.Column(db.Boolean, nullable=False, default=True)
commercial_product = db.relationship("CommercialProduct", back_populates="documents")
uploaded_by = db.relationship("User")
class RequestProfile(db.Model):
__tablename__ = "cleaning_request_profiles"
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(160), nullable=False, unique=True)
description = db.Column(db.Text)
is_active = db.Column(db.Boolean, nullable=False, default=True)
items = db.relationship("RequestProfileItem", back_populates="profile", cascade="all, delete-orphan", order_by="RequestProfileItem.display_order")
class RequestProfileItem(db.Model):
__tablename__ = "cleaning_request_profile_items"
id = db.Column(db.Integer, primary_key=True)
profile_id = db.Column(db.Integer, db.ForeignKey("cleaning_request_profiles.id", ondelete="CASCADE"), nullable=False, index=True)
product_id = db.Column(db.Integer, db.ForeignKey("cleaning_products_generic.id", ondelete="RESTRICT"), nullable=False, index=True)
display_order = db.Column(db.Integer, nullable=False, default=0)
profile = db.relationship("RequestProfile", back_populates="items")
product = db.relationship("ProductGeneric")
class StockInventory(db.Model):
__tablename__ = "cleaning_stock_inventories"
id = db.Column(db.Integer, primary_key=True)
location_id = db.Column(db.Integer, db.ForeignKey("cleaning_stock_locations.id"), nullable=False, index=True)
counted_at = db.Column(db.DateTime, nullable=False, default=utcnow)
status = db.Column(db.String(20), nullable=False, default="brouillon")
user_id = db.Column(db.Integer, db.ForeignKey("users.id", ondelete="SET NULL"))
notes = db.Column(db.Text)
location = db.relationship("StockLocation")
user = db.relationship("User")
lines = db.relationship("StockInventoryLine", back_populates="inventory", cascade="all, delete-orphan")
class StockInventoryLine(db.Model):
__tablename__ = "cleaning_stock_inventory_lines"
id = db.Column(db.Integer, primary_key=True)
inventory_id = db.Column(db.Integer, db.ForeignKey("cleaning_stock_inventories.id", ondelete="CASCADE"), nullable=False, index=True)
lot_id = db.Column(db.Integer, db.ForeignKey("cleaning_stock_lots.id", ondelete="RESTRICT"), nullable=False)
theoretical_quantity = db.Column(db.Numeric(14, 6), nullable=False)
counted_quantity = db.Column(db.Numeric(14, 6), nullable=False)
inventory = db.relationship("StockInventory", back_populates="lines")
lot = db.relationship("StockLot")
class CleaningForecastConfig(db.Model):
__tablename__ = "cleaning_forecast_configs"
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(120), nullable=False, default="Configuration générale")
normal_coefficient = db.Column(db.Numeric(8, 5), nullable=False, default=1)
permanence_coefficient = db.Column(db.Numeric(8, 5), nullable=False, default=Decimal("0.4"))
closed_coefficient = db.Column(db.Numeric(8, 5), nullable=False, default=0)
exceptional_coefficient = db.Column(db.Numeric(8, 5), nullable=False, default=1)
is_active = db.Column(db.Boolean, nullable=False, default=True)

View file

@ -478,6 +478,7 @@ class ScheduledTask(db.Model):
equipment_id = db.Column(db.Integer, db.ForeignKey("equipments.id"))
room_id = db.Column(db.Integer, db.ForeignKey("rooms.id"))
company_id = db.Column(db.Integer, db.ForeignKey("companies.id"))
contract_id = db.Column(db.Integer, db.ForeignKey("contracts.id"), nullable=True)
scheduled_date = db.Column(db.Date)
scheduled_start = db.Column(db.Time)
@ -511,6 +512,7 @@ class ScheduledTask(db.Model):
room = db.relationship("Room")
assigned_to = db.relationship("User")
company = db.relationship("Company")
contract = db.relationship("Contract")
intervention = db.relationship("Intervention")
@property

View file

@ -221,6 +221,17 @@
<li><a class="dropdown-item" href="{{ url_for('lots.index') }}"><i class="bi bi-boxes"></i> Lots</a></li>
<li><a class="dropdown-item" href="{{ url_for('services.index') }}"><i class="bi bi-building-gear"></i> Services</a></li>
<li><a class="dropdown-item" href="{{ url_for('parts.index') }}"><i class="bi bi-box-seam"></i> Pièces</a></li>
<li><hr class="dropdown-divider" style="border-color:rgba(255,255,255,0.15)"></li>
<li><a class="dropdown-item" href="{{ url_for('cleaning.dashboard') }}"><i class="bi bi-droplet-half"></i> Matériel &amp; Entretien</a></li>
<li><a class="dropdown-item" href="{{ url_for('cleaning.products') }}"><i class="bi bi-flask"></i> Produits génériques</a></li>
<li><a class="dropdown-item" href="{{ url_for('cleaning.references') }}"><i class="bi bi-upc-scan"></i> Références / conditionnements</a></li>
<li><a class="dropdown-item" href="{{ url_for('cleaning.stock') }}"><i class="bi bi-boxes"></i> État du stock</a></li>
<li><a class="dropdown-item" href="{{ url_for('cleaning.receive') }}"><i class="bi bi-box-arrow-in-down"></i> Réceptions</a></li>
<li><a class="dropdown-item" href="{{ url_for('cleaning.issue') }}"><i class="bi bi-box-arrow-up"></i> Sorties agents</a></li>
<li><a class="dropdown-item" href="{{ url_for('cleaning.transvasement_page') }}"><i class="bi bi-droplet"></i> Transvasements / dilutions</a></li>
<li><a class="dropdown-item" href="{{ url_for('cleaning.request_profiles') }}"><i class="bi bi-printer"></i> Fiches papier</a></li>
<li><a class="dropdown-item" href="{{ url_for('cleaning.material') }}"><i class="bi bi-tools"></i> Matériel durable</a></li>
<li><a class="dropdown-item" href="{{ url_for('cleaning.forecast_config') }}"><i class="bi bi-sliders"></i> Prévision / permanences</a></li>
<li><a class="dropdown-item" href="{{ url_for('meters.index') }}"><i class="bi bi-graph-up-arrow"></i> Historique des compteurs</a></li>
<li><hr class="dropdown-divider" style="border-color:rgba(255,255,255,0.15)"></li>
<li><a class="dropdown-item" href="{{ url_for('equipments.to_trash') }}"><i class="bi bi-trash"></i> À jeter</a></li>

View file

@ -0,0 +1,33 @@
"""Ajouter le module Matériel & Produits d'entretien.
Revision ID: 5784fc27cb2a
Revises: a07b8c9d0e2f
"""
from alembic import op
from app_new.core.models.cleaning import (
ProductCategory, ProductGeneric, CommercialProduct, ProductPackaging,
StockLocation, StockLot, StockLotBalance, StockMovement,
MaterialAssignment, ReusableContainer, ContainerFill, ProductDocument,
RequestProfile, RequestProfileItem, StockInventory, StockInventoryLine,
CleaningForecastConfig,
)
revision = "5784fc27cb2a"
down_revision = "a07b8c9d0e2f"
branch_labels = None
depends_on = None
_TABLES = [
CleaningForecastConfig.__table__, ProductCategory.__table__,
RequestProfile.__table__, ProductGeneric.__table__, CommercialProduct.__table__,
RequestProfileItem.__table__, ProductDocument.__table__, ProductPackaging.__table__,
ReusableContainer.__table__, StockLot.__table__, StockLocation.__table__,
MaterialAssignment.__table__, StockInventory.__table__, StockLotBalance.__table__,
StockMovement.__table__, ContainerFill.__table__, StockInventoryLine.__table__,
]
def upgrade():
bind = op.get_bind()
for table in _TABLES:
table.create(bind=bind, checkfirst=True)
def downgrade():
bind = op.get_bind()
for table in reversed(_TABLES):
table.drop(bind=bind, checkfirst=True)

View file

@ -0,0 +1,25 @@
"""Rétablir le rattachement des tâches planifiées aux contrats.
Revision ID: 8a95b1c2d3e4
Revises: 5784fc27cb2a
"""
from alembic import op
import sqlalchemy as sa
revision = "8a95b1c2d3e4"
down_revision = "5784fc27cb2a"
branch_labels = None
depends_on = None
def upgrade():
inspector = sa.inspect(op.get_bind())
columns = {c["name"] for c in inspector.get_columns("scheduled_tasks")}
if "contract_id" not in columns:
op.add_column("scheduled_tasks", sa.Column("contract_id", sa.Integer(), nullable=True))
names = {fk.get("name") for fk in inspector.get_foreign_keys("scheduled_tasks")}
if "fk_scheduled_tasks_contract_id" not in names:
op.create_foreign_key("fk_scheduled_tasks_contract_id", "scheduled_tasks", "contracts", ["contract_id"], ["id"])
def downgrade():
inspector = sa.inspect(op.get_bind())
names = {fk.get("name") for fk in inspector.get_foreign_keys("scheduled_tasks")}
if "fk_scheduled_tasks_contract_id" in names:
op.drop_constraint("fk_scheduled_tasks_contract_id", "scheduled_tasks", type_="foreignkey")
if "contract_id" in {c["name"] for c in inspector.get_columns("scheduled_tasks")}:
op.drop_column("scheduled_tasks", "contract_id")

View file

@ -0,0 +1,94 @@
from datetime import date, timedelta
from decimal import Decimal
from uuid import uuid4
import pytest
from app_new.extensions import db
from app_new.core.models import (
ProductGeneric, CommercialProduct, ProductPackaging, StockLocation,
Staff, ReusableContainer, CleaningForecastConfig, CollegeClosure, ClosureWorkDay,
)
from app_new.cleaning.services.stock import (
StockError, receive_stock, issue_stock, transfer_stock, dilute,
)
from app_new.cleaning.services.forecast import forecast_product
def setup_stock():
product = ProductGeneric(name=f"Nettoyant test {uuid4().hex[:8]}", reference_unit="L", product_type="liquide",
forecast_daily_quantity=Decimal("1.2"), stock_security=Decimal("2"))
ref = CommercialProduct(generic_product=product, commercial_name="Référence test")
packaging = ProductPackaging(generic_product=product, commercial_product=ref, name="Bidon 5 L",
purchase_unit="bidon", units_per_package=1,
reference_quantity_per_package=Decimal("5"))
suffix = uuid4().hex[:8]
loc_a = StockLocation(name=f"Stock test A {suffix}")
loc_b = StockLocation(name=f"Stock test B {suffix}")
db.session.add_all([product, ref, packaging, loc_a, loc_b])
db.session.flush()
return product, ref, packaging, loc_a, loc_b
def test_reception_conversion_and_fefo(app):
with app.app_context():
product, ref, packaging, loc_a, loc_b = setup_stock()
receive_stock(generic_product_id=product.id, commercial_product_id=ref.id, packaging_id=packaging.id,
packages=2, lot_number="ANCIEN", received_at=date.today(),
expiry_date=date.today() + timedelta(days=3), location_id=loc_a.id)
receive_stock(generic_product_id=product.id, commercial_product_id=ref.id, packaging_id=packaging.id,
packages=2, lot_number="RECENT", received_at=date.today(),
expiry_date=date.today() + timedelta(days=90), location_id=loc_a.id)
db.session.commit()
issue_stock(generic_product_id=product.id, quantity=Decimal("6"), source_location_id=loc_a.id)
db.session.commit()
lots = {lot.lot_number: sum((b.quantity for b in lot.balances), Decimal("0"))
for lot in product.lots}
assert lots["ANCIEN"] == Decimal("4.000000")
assert lots["RECENT"] == Decimal("10.000000")
def test_expired_lot_rejected_and_no_negative_stock(app):
with app.app_context():
product, ref, packaging, loc_a, _ = setup_stock()
receive_stock(generic_product_id=product.id, commercial_product_id=ref.id, packaging_id=packaging.id,
packages=1, lot_number="EXPIRE", received_at=date.today(),
expiry_date=date.today() - timedelta(days=1), location_id=loc_a.id)
db.session.commit()
with pytest.raises(StockError):
issue_stock(generic_product_id=product.id, quantity=1, source_location_id=loc_a.id)
with pytest.raises(StockError):
issue_stock(generic_product_id=product.id, quantity=100, source_location_id=loc_a.id)
db.session.rollback()
def test_transfer_and_dilution(app):
with app.app_context():
product, ref, packaging, loc_a, loc_b = setup_stock()
receive_stock(generic_product_id=product.id, commercial_product_id=ref.id, packaging_id=packaging.id,
packages=1, lot_number="TRANS", received_at=date.today(), location_id=loc_a.id)
container = ReusableContainer(identifier=f"FLACON-TEST-{uuid4().hex[:8]}", container_type="750 ml", volume_liters=Decimal("0.750"))
db.session.add(container); db.session.flush()
transfer_stock(generic_product_id=product.id, quantity=Decimal("1"), source_location_id=loc_a.id, destination_location_id=loc_b.id)
db.session.commit()
assert sum((b.quantity for b in product.lots[0].balances if b.location_id == loc_b.id), Decimal("0")) == Decimal("1.000000")
dilute(generic_product_id=product.id, total_liters=Decimal("0.750"), ratio_percent=Decimal("2"),
source_location_id=loc_b.id, container_id=container.id)
db.session.commit()
assert container.current_product_id == product.id
assert sum((b.quantity for b in product.lots[0].balances if b.location_id == loc_b.id), Decimal("0")) == Decimal("0.985000")
def test_forecast_distinguishes_school_days_permanence_and_closure(app):
with app.app_context():
product, ref, packaging, loc_a, _ = setup_stock()
start = date(2026, 9, 7)
closure = CollegeClosure(name="Vacances test", start_date=start + timedelta(days=5),
end_date=start + timedelta(days=9), closure_type="vacances_scolaires")
db.session.add(closure); db.session.flush()
db.session.add(ClosureWorkDay(closure_id=closure.id, work_date=start + timedelta(days=6),
start_time=__import__("datetime").time(8), end_time=__import__("datetime").time(12)))
db.session.commit()
result = forecast_product(product, start + timedelta(days=9), start_date=start)
assert result["school_days_consumption"] > 0
assert result["permanence_consumption"] == Decimal("0.480000")
assert result["closed_consumption"] == 0