3267 lines
119 KiB
Python
3267 lines
119 KiB
Python
from django.contrib.gis.db import models
|
||
from django.conf import settings
|
||
from django.db import transaction
|
||
from django.db.models import Sum, F, Q
|
||
from django.utils import timezone
|
||
from django.utils.translation import gettext_lazy as _
|
||
import uuid
|
||
import qrcode
|
||
from io import BytesIO
|
||
import base64
|
||
|
||
from common.storages import PrivateMediaStorage
|
||
|
||
private_storage = PrivateMediaStorage()
|
||
|
||
|
||
class Warehouse(models.Model):
|
||
code = models.CharField(max_length=100, unique=True)
|
||
name = models.CharField(max_length=100, unique=True)
|
||
location = models.TextField(blank=True, null=True)
|
||
description = models.TextField(blank=True, null=True)
|
||
manager = models.CharField(max_length=200)
|
||
lon = models.FloatField(null=True, blank=True)
|
||
lat = models.FloatField(null=True, blank=True)
|
||
geom = models.PointField(srid=3812, null=True, blank=True)
|
||
|
||
def __str__(self):
|
||
return self.name
|
||
|
||
class Meta:
|
||
ordering = ("name",)
|
||
verbose_name = _("Entrepôt")
|
||
verbose_name_plural = _("Entrepôts")
|
||
|
||
def total_products(self):
|
||
return (
|
||
Stock.objects.filter(location__warehouse=self)
|
||
.aggregate(total=Sum("quantity"))["total"]
|
||
or 0
|
||
)
|
||
|
||
def total_by_category(self):
|
||
return (
|
||
Stock.objects.filter(location__warehouse=self)
|
||
.values(category_code=F("product__category__code"))
|
||
.annotate(total=Sum("quantity"))
|
||
.order_by("-total")
|
||
)
|
||
|
||
def total_by_product(self):
|
||
return (
|
||
Stock.objects.filter(location__warehouse=self)
|
||
.values(product_name=F("product__name"))
|
||
.annotate(total=Sum("quantity"))
|
||
.order_by("-total")
|
||
)
|
||
|
||
def stock_by_product(self, product):
|
||
return (
|
||
Stock.objects.filter(location__warehouse=self, product=product)
|
||
.aggregate(total=Sum("quantity"))["total"]
|
||
or 0
|
||
)
|
||
|
||
def is_below_threshold(self, product):
|
||
total = self.stock_by_product(product)
|
||
return total < product.threshold
|
||
|
||
def product_status(self, product):
|
||
stock = self.stock_by_product(product)
|
||
is_below = stock < product.threshold
|
||
pending_order = product.has_pending_order()
|
||
|
||
if is_below and pending_order:
|
||
return _("Sous seuil (Commande en cours)")
|
||
if is_below:
|
||
return _("Sous seuil (Pas de commande)")
|
||
return _("Stock suffisant")
|
||
|
||
|
||
class WarehouseLocation(models.Model):
|
||
warehouse = models.ForeignKey(
|
||
Warehouse, on_delete=models.CASCADE, related_name="locations"
|
||
)
|
||
code = models.CharField(max_length=100, unique=True)
|
||
name = models.CharField(max_length=100)
|
||
description = models.TextField(blank=True, null=True)
|
||
is_default = models.BooleanField(
|
||
default=False,
|
||
verbose_name=_("Emplacement par défaut"),
|
||
help_text=_("Si activé, cet emplacement sera proposé par défaut dans les formulaires pour cet entrepôt."),
|
||
)
|
||
qr_code = models.ImageField(
|
||
upload_to='stock/qrcodes/locations/',
|
||
blank=True,
|
||
null=True,
|
||
verbose_name=_('QR Code'),
|
||
)
|
||
|
||
class Meta:
|
||
unique_together = ("warehouse", "name")
|
||
|
||
def __str__(self):
|
||
return f"{self.name} - {self.warehouse.name}"
|
||
|
||
def generate_qr_code(self):
|
||
"""Génère un QR code basé sur le code d'emplacement et le sauvegarde."""
|
||
qr = qrcode.QRCode(
|
||
version=1,
|
||
error_correction=qrcode.constants.ERROR_CORRECT_M,
|
||
box_size=10,
|
||
border=4,
|
||
)
|
||
qr.add_data(f"LOKO:stocklocation:{self.code}")
|
||
qr.make(fit=True)
|
||
img = qr.make_image(fill_color="black", back_color="white")
|
||
buffer = BytesIO()
|
||
img.save(buffer, format='PNG')
|
||
buffer.seek(0)
|
||
from django.core.files.base import ContentFile
|
||
filename = f"stocklocation_{self.code}.png"
|
||
self.qr_code.save(filename, ContentFile(buffer.read()), save=False)
|
||
|
||
def save(self, *args, **kwargs):
|
||
if self.is_default:
|
||
# Ensure only one default location per warehouse
|
||
WarehouseLocation.objects.filter(
|
||
warehouse=self.warehouse, is_default=True
|
||
).exclude(pk=self.pk).update(is_default=False)
|
||
generate = False
|
||
if self.pk:
|
||
try:
|
||
old = WarehouseLocation.objects.get(pk=self.pk)
|
||
if old.code != self.code or not self.qr_code:
|
||
generate = True
|
||
except WarehouseLocation.DoesNotExist:
|
||
generate = True
|
||
else:
|
||
generate = True
|
||
if generate:
|
||
self.generate_qr_code()
|
||
super().save(*args, **kwargs)
|
||
|
||
|
||
class ProductCategory(models.Model):
|
||
code = models.CharField(max_length=100, unique=True)
|
||
name_fr = models.CharField(max_length=100, unique=True)
|
||
name_en = models.CharField(max_length=100, unique=True, blank=True, null=True)
|
||
name_nl = models.CharField(max_length=100, unique=True, blank=True, null=True)
|
||
description = models.TextField(blank=True, null=True)
|
||
|
||
def __str__(self):
|
||
return self.name_fr
|
||
|
||
def name(self, lang="fr"):
|
||
if lang == "en" and self.name_en:
|
||
return self.name_en
|
||
if lang == "nl" and self.name_nl:
|
||
return self.name_nl
|
||
return self.name_fr
|
||
|
||
|
||
class Product(models.Model):
|
||
UNIT_PC = 'pc'
|
||
UNIT_KG = 'kg'
|
||
UNIT_G = 'g'
|
||
UNIT_T = 't'
|
||
UNIT_L = 'L'
|
||
UNIT_M = 'm'
|
||
UNIT_M2 = 'm2'
|
||
UNIT_M3 = 'm3'
|
||
|
||
UNIT_CHOICES = [
|
||
(UNIT_PC, _('Pièce(s)')),
|
||
(UNIT_KG, _('Kilogramme(s)')),
|
||
(UNIT_G, _('Gramme(s)')),
|
||
(UNIT_T, _('Tonne(s)')),
|
||
(UNIT_L, _('Litre(s)')),
|
||
(UNIT_M, _('Mètre(s)')),
|
||
(UNIT_M2, _('Mètre(s) carré(s)')),
|
||
(UNIT_M3, _('Mètre(s) cube(s)')),
|
||
]
|
||
|
||
code = models.CharField(max_length=100, unique=True)
|
||
name = models.CharField(max_length=200)
|
||
category = models.ForeignKey(
|
||
ProductCategory, on_delete=models.SET_NULL, null=True, related_name="products"
|
||
)
|
||
sku = models.CharField(max_length=100, unique=True)
|
||
description = models.TextField(blank=True, null=True)
|
||
price = models.DecimalField(max_digits=10, decimal_places=2, blank=True, null=True)
|
||
unit = models.CharField(
|
||
max_length=10,
|
||
choices=UNIT_CHOICES,
|
||
default=UNIT_PC,
|
||
verbose_name=_('Unité'),
|
||
)
|
||
threshold = models.IntegerField(default=10)
|
||
total_stock = models.IntegerField(default=0)
|
||
reserved_quantity = models.IntegerField(default=0)
|
||
qr_code = models.ImageField(upload_to='stock/qrcodes/products/', blank=True, null=True, verbose_name=_('QR Code'))
|
||
is_serialized = models.BooleanField(
|
||
default=False,
|
||
verbose_name=_("Articles tracés (numéro de série)"),
|
||
help_text=_("Si activé, chaque article de ce produit devra avoir un numéro de série unique."),
|
||
)
|
||
is_active = models.BooleanField(
|
||
default=True,
|
||
verbose_name=_("Actif"),
|
||
help_text=_("Si désactivé, le produit ne sera plus disponible pour les opérations (nouvelles entrées, sorties, transferts, interventions)."),
|
||
)
|
||
authorized_locations = models.ManyToManyField(
|
||
"WarehouseLocation",
|
||
blank=True,
|
||
related_name="authorized_products",
|
||
verbose_name=_("Emplacements autorisés"),
|
||
help_text=_("Emplacements dans lesquels ce produit est autorisé à être stocké (vide = tous autorisés)."),
|
||
)
|
||
|
||
STATUS_VALIDATED = 'validated'
|
||
STATUS_PENDING_VALIDATION = 'pending_validation'
|
||
STATUS_REJECTED = 'rejected'
|
||
|
||
STATUS_CHOICES = [
|
||
(STATUS_VALIDATED, _('Validé')),
|
||
(STATUS_PENDING_VALIDATION, _('En attente de validation superviseur')),
|
||
(STATUS_REJECTED, _('Rejeté')),
|
||
]
|
||
|
||
status = models.CharField(
|
||
max_length=30,
|
||
choices=STATUS_CHOICES,
|
||
default=STATUS_VALIDATED,
|
||
verbose_name=_("Statut"),
|
||
help_text=_("Statut de validation par le superviseur"),
|
||
)
|
||
supplier_reference = models.CharField(
|
||
max_length=100,
|
||
blank=True,
|
||
null=True,
|
||
verbose_name=_("Référence fournisseur"),
|
||
help_text=_("Référence de l'article chez le fournisseur"),
|
||
)
|
||
created_by = models.ForeignKey(
|
||
"auth.User",
|
||
on_delete=models.SET_NULL,
|
||
null=True,
|
||
blank=True,
|
||
related_name="created_products",
|
||
verbose_name=_("Créé par"),
|
||
)
|
||
validated_by = models.ForeignKey(
|
||
"auth.User",
|
||
on_delete=models.SET_NULL,
|
||
null=True,
|
||
blank=True,
|
||
related_name="validated_products",
|
||
verbose_name=_("Validé par"),
|
||
)
|
||
validated_at = models.DateTimeField(
|
||
null=True,
|
||
blank=True,
|
||
verbose_name=_("Validé le"),
|
||
)
|
||
validation_comment = models.TextField(
|
||
blank=True,
|
||
null=True,
|
||
verbose_name=_("Commentaire de validation"),
|
||
)
|
||
|
||
@property
|
||
def is_pending_validation(self):
|
||
return self.status == self.STATUS_PENDING_VALIDATION
|
||
|
||
def validate_product(self, user, comment=""):
|
||
"""Valide le produit par un superviseur."""
|
||
self.status = self.STATUS_VALIDATED
|
||
self.validated_by = user
|
||
self.validated_at = timezone.now()
|
||
if comment:
|
||
self.validation_comment = comment
|
||
self.save(update_fields=["status", "validated_by", "validated_at", "validation_comment"])
|
||
|
||
def reject_product(self, user, comment=""):
|
||
"""Rejette le produit par un superviseur."""
|
||
self.status = self.STATUS_REJECTED
|
||
self.is_active = False
|
||
self.validated_by = user
|
||
self.validated_at = timezone.now()
|
||
if comment:
|
||
self.validation_comment = comment
|
||
self.save(update_fields=["status", "is_active", "validated_by", "validated_at", "validation_comment"])
|
||
|
||
def __str__(self):
|
||
return self.name
|
||
|
||
def generate_qr_code(self):
|
||
"""Génère un QR code basé sur le code produit et le sauvegarde dans le champ qr_code."""
|
||
qr = qrcode.QRCode(
|
||
version=1,
|
||
error_correction=qrcode.constants.ERROR_CORRECT_M,
|
||
box_size=10,
|
||
border=4,
|
||
)
|
||
qr.add_data(f"LOKO:product:{self.code}")
|
||
qr.make(fit=True)
|
||
|
||
img = qr.make_image(fill_color="black", back_color="white")
|
||
buffer = BytesIO()
|
||
img.save(buffer, format='PNG')
|
||
buffer.seek(0)
|
||
|
||
from django.core.files.base import ContentFile
|
||
filename = f"product_{self.code}.png"
|
||
self.qr_code.save(filename, ContentFile(buffer.read()), save=False)
|
||
|
||
def save(self, *args, **kwargs):
|
||
# Ne pas re-générer le QR code lors des saves partiels (update_fields)
|
||
update_fields = kwargs.get('update_fields')
|
||
if update_fields is not None:
|
||
super().save(*args, **kwargs)
|
||
return
|
||
# Générer le QR code si absent ou si le code produit a changé
|
||
generate = False
|
||
if self.pk:
|
||
try:
|
||
old = Product.objects.get(pk=self.pk)
|
||
if old.code != self.code or not self.qr_code:
|
||
generate = True
|
||
except Product.DoesNotExist:
|
||
generate = True
|
||
else:
|
||
generate = True
|
||
if generate:
|
||
self.generate_qr_code()
|
||
super().save(*args, **kwargs)
|
||
|
||
def update_total_stock(self):
|
||
"""Recalcule et met à jour la quantité totale disponible"""
|
||
total = (
|
||
Stock.objects.filter(product=self)
|
||
.aggregate(total=Sum("quantity"))["total"]
|
||
or 0
|
||
)
|
||
self.total_stock = total
|
||
self.save(update_fields=["total_stock"])
|
||
|
||
def get_available_stock(self):
|
||
"""Retourne le stock disponible (stock total - quantité réservée)"""
|
||
return max(0, self.total_stock - self.reserved_quantity)
|
||
|
||
def update_reserved_quantity(self):
|
||
"""Recalcule et met à jour la quantité réservée depuis les bons de préparation non traités"""
|
||
reserved = (
|
||
PreparationOrderItem.objects.filter(
|
||
product=self,
|
||
preparation_order__status__in=["to_process", "pending"]
|
||
)
|
||
.aggregate(total=Sum("quantity"))["total"]
|
||
or 0
|
||
)
|
||
self.reserved_quantity = reserved
|
||
self.save(update_fields=["reserved_quantity"])
|
||
|
||
def has_pending_order(self):
|
||
return PurchaseOrderItem.objects.filter(
|
||
product=self, purchase_order__status__in=["pending", "ordered"]
|
||
).exists()
|
||
|
||
def get_default_location(self, warehouse):
|
||
"""
|
||
Retourne l'emplacement de stockage par défaut pour ce produit dans un entrepôt donné.
|
||
"""
|
||
if not warehouse or not self.pk:
|
||
return None
|
||
loc_map = get_product_default_locations_map(warehouse, product_ids=[self.pk])
|
||
loc_id = loc_map.get(self.pk)
|
||
if loc_id:
|
||
return warehouse.locations.filter(pk=loc_id).first()
|
||
return None
|
||
|
||
|
||
def notify_supervisors_pending_product(product, created_by=None):
|
||
"""Notifie les superviseurs / admins qu'un nouveau produit attend validation."""
|
||
from django.contrib.auth.models import User
|
||
from django.urls import reverse
|
||
from django.utils.translation import gettext as _
|
||
from notifications.services import notify_users
|
||
from stock.permissions import user_can_manage_products
|
||
|
||
supervisors = [
|
||
u for u in User.objects.filter(is_active=True).select_related('config')
|
||
if user_can_manage_products(u) and u != created_by
|
||
]
|
||
if not supervisors:
|
||
return
|
||
|
||
action_url = reverse("stock:product_detail", args=[product.pk])
|
||
title = _("Nouveau produit à valider : %(name)s") % {"name": product.name}
|
||
creator_name = (
|
||
created_by.get_full_name() or created_by.username
|
||
if created_by
|
||
else _("Magasinier")
|
||
)
|
||
content = _(
|
||
"Le magasinier %(creator)s a créé le produit '%(name)s' (Réf: %(ref)s) "
|
||
"via un bon de commande. Ce produit nécessite la validation d'un superviseur."
|
||
) % {
|
||
"creator": creator_name,
|
||
"name": product.name,
|
||
"ref": product.supplier_reference or product.code,
|
||
}
|
||
try:
|
||
notify_users(
|
||
recipients=supervisors,
|
||
title=title,
|
||
content=content,
|
||
notification_type="task",
|
||
action_url=action_url,
|
||
action_label=_("Examiner le produit"),
|
||
source_app="stock",
|
||
sender=created_by,
|
||
is_system=True,
|
||
is_single_claim=True,
|
||
)
|
||
except Exception:
|
||
import logging
|
||
logging.getLogger(__name__).exception("Failed to notify supervisors of pending product")
|
||
|
||
|
||
class Stock(models.Model):
|
||
product = models.ForeignKey(Product, on_delete=models.CASCADE, related_name="stocks")
|
||
location = models.ForeignKey(
|
||
WarehouseLocation, on_delete=models.CASCADE, related_name="stocks"
|
||
)
|
||
quantity = models.IntegerField(default=0)
|
||
|
||
class Meta:
|
||
unique_together = ("product", "location")
|
||
|
||
def __str__(self):
|
||
return _("{quantity} de {product} à {location} ({warehouse})").format(
|
||
quantity=self.quantity,
|
||
product=self.product.name,
|
||
location=self.location.name,
|
||
warehouse=self.location.warehouse.name,
|
||
)
|
||
|
||
def save(self, *args, **kwargs):
|
||
super().save(*args, **kwargs)
|
||
update_stock_alerts(self.product, self.location.warehouse)
|
||
# Mettre à jour le stock total du produit
|
||
self.product.update_total_stock()
|
||
|
||
|
||
def get_product_default_locations_map(warehouse, product_ids=None):
|
||
"""
|
||
Retourne un dictionnaire {product_id: location_id} pour les produits donnés
|
||
(ou tous les produits actifs si non spécifié) dans l'entrepôt spécifié.
|
||
|
||
Règles de détermination de l'emplacement par défaut :
|
||
1. Si le produit possède des emplacements autorisés dans cet entrepôt :
|
||
a. L'emplacement autorisé marqué is_default=True pour l'entrepôt s'il existe.
|
||
b. L'emplacement autorisé ayant le plus de stock (> 0).
|
||
c. Le premier emplacement autorisé (ordre alphabétique).
|
||
2. Si le produit n'a pas d'emplacement autorisé restreint dans cet entrepôt :
|
||
a. L'emplacement de cet entrepôt où le produit a déjà du stock (> 0, quantité max).
|
||
b. L'emplacement par défaut de l'entrepôt (is_default=True).
|
||
c. Le premier emplacement de l'entrepôt.
|
||
"""
|
||
if not warehouse:
|
||
return {}
|
||
|
||
locations = list(warehouse.locations.all().order_by('name'))
|
||
if not locations:
|
||
return {}
|
||
|
||
wh_default_loc = next((l for l in locations if l.is_default), locations[0])
|
||
wh_default_loc_id = wh_default_loc.id if wh_default_loc else None
|
||
wh_loc_ids = {l.id for l in locations}
|
||
loc_is_default = {l.id: l.is_default for l in locations}
|
||
|
||
# 1. Emplacements autorisés par produit pour cet entrepôt
|
||
auth_m2m = Product.authorized_locations.through.objects.filter(
|
||
warehouselocation_id__in=wh_loc_ids
|
||
)
|
||
if product_ids is not None:
|
||
auth_m2m = auth_m2m.filter(product_id__in=product_ids)
|
||
|
||
product_auth_locs = {}
|
||
for entry in auth_m2m.values('product_id', 'warehouselocation_id'):
|
||
p_id = entry['product_id']
|
||
l_id = entry['warehouselocation_id']
|
||
product_auth_locs.setdefault(p_id, []).append(l_id)
|
||
|
||
# 2. Stocks existants dans cet entrepôt
|
||
stock_qs = Stock.objects.filter(
|
||
location__warehouse=warehouse,
|
||
quantity__gt=0
|
||
)
|
||
if product_ids is not None:
|
||
stock_qs = stock_qs.filter(product_id__in=product_ids)
|
||
|
||
# Organiser les stocks par produit triés par quantité décroissante
|
||
stock_by_product = {}
|
||
for s in stock_qs.values('product_id', 'location_id', 'quantity').order_by('-quantity'):
|
||
stock_by_product.setdefault(s['product_id'], []).append((s['location_id'], s['quantity']))
|
||
|
||
# 3. Liste des identifiants produits cibles
|
||
if product_ids is not None:
|
||
target_pids = list(product_ids)
|
||
else:
|
||
target_pids = list(Product.objects.filter(is_active=True).values_list('id', flat=True))
|
||
|
||
result_map = {}
|
||
for p_id in target_pids:
|
||
auth_ids = product_auth_locs.get(p_id)
|
||
if auth_ids:
|
||
# 1a. Emplacement autorisé avec is_default
|
||
def_auth = next((l_id for l_id in auth_ids if loc_is_default.get(l_id)), None)
|
||
if def_auth:
|
||
result_map[p_id] = def_auth
|
||
continue
|
||
|
||
# 1b. Emplacement autorisé avec stock > 0
|
||
stocks = stock_by_product.get(p_id, [])
|
||
stock_auth = next((l_id for (l_id, qty) in stocks if l_id in auth_ids), None)
|
||
if stock_auth:
|
||
result_map[p_id] = stock_auth
|
||
continue
|
||
|
||
# 1c. Premier emplacement autorisé
|
||
result_map[p_id] = auth_ids[0]
|
||
else:
|
||
# 2a. Emplacement de cet entrepôt avec stock > 0
|
||
stocks = stock_by_product.get(p_id, [])
|
||
if stocks:
|
||
result_map[p_id] = stocks[0][0]
|
||
continue
|
||
|
||
# 2b / 2c. Emplacement par défaut de l'entrepôt
|
||
result_map[p_id] = wh_default_loc_id
|
||
|
||
return result_map
|
||
|
||
|
||
class StockMovementBatch(models.Model):
|
||
"""Regroupe un ensemble de mouvements de stock créés simultanément."""
|
||
TYPE_IN = "in"
|
||
TYPE_OUT = "out"
|
||
TYPE_TRANSFER = "transfer"
|
||
TYPE_CHOICES = [
|
||
(TYPE_IN, _("Entrée")),
|
||
(TYPE_OUT, _("Sortie")),
|
||
(TYPE_TRANSFER, _("Transfert")),
|
||
]
|
||
|
||
SUBTYPE_CHOICES = [
|
||
("purchase", _("Achat")),
|
||
("return", _("Retour")),
|
||
("inventory", _("Inventaire")),
|
||
("consumption", _("Consommation")),
|
||
("loss", _("Perte")),
|
||
("donation", _("Don")),
|
||
("internal", _("Interne")),
|
||
("external", _("Externe")),
|
||
("correction", _("Correction")),
|
||
("confection", _("Confection")),
|
||
("preparation_order", _("Bon de préparation")),
|
||
("epi", _("EPI")),
|
||
("loan", _("Emprunt")),
|
||
("worksite", _("Chantier")),
|
||
("other", _("Autre")),
|
||
]
|
||
|
||
movement_type = models.CharField(max_length=20, choices=TYPE_CHOICES, verbose_name=_("Type"))
|
||
movement_subtype = models.CharField(
|
||
max_length=30,
|
||
choices=SUBTYPE_CHOICES,
|
||
blank=True,
|
||
null=True,
|
||
verbose_name=_("Sous-type"),
|
||
)
|
||
date = models.DateTimeField(auto_now_add=True, verbose_name=_("Date"))
|
||
reason = models.TextField(blank=True, null=True, verbose_name=_("Raison"))
|
||
person = models.ForeignKey(
|
||
"contracts.CompanyMember",
|
||
on_delete=models.SET_NULL,
|
||
blank=True,
|
||
null=True,
|
||
related_name="stock_batches",
|
||
verbose_name=_("Personne"),
|
||
help_text=_("Personne à qui le matériel est attribué ou qui le restitue"),
|
||
)
|
||
company = models.ForeignKey(
|
||
"contracts.Company",
|
||
on_delete=models.SET_NULL,
|
||
blank=True,
|
||
null=True,
|
||
related_name="stock_batches",
|
||
verbose_name=_("Entreprise / Organisation"),
|
||
)
|
||
contract = models.ForeignKey(
|
||
"contracts.Contract",
|
||
on_delete=models.SET_NULL,
|
||
blank=True,
|
||
null=True,
|
||
related_name="stock_batches",
|
||
verbose_name=_("Contrat"),
|
||
)
|
||
created_by = models.ForeignKey(
|
||
"auth.User",
|
||
on_delete=models.SET_NULL,
|
||
blank=True,
|
||
null=True,
|
||
related_name="created_stock_batches",
|
||
verbose_name=_("Créé par"),
|
||
)
|
||
preparation_order = models.ForeignKey(
|
||
"PreparationOrder",
|
||
on_delete=models.SET_NULL,
|
||
blank=True,
|
||
null=True,
|
||
related_name="stock_batches",
|
||
verbose_name=_("Bon de préparation"),
|
||
)
|
||
purchase_order = models.ForeignKey(
|
||
"PurchaseOrder",
|
||
on_delete=models.SET_NULL,
|
||
blank=True,
|
||
null=True,
|
||
related_name="stock_batches",
|
||
verbose_name=_("Commande d'achat"),
|
||
)
|
||
production_order = models.ForeignKey(
|
||
"ProductionOrder",
|
||
on_delete=models.SET_NULL,
|
||
blank=True,
|
||
null=True,
|
||
related_name="stock_batches",
|
||
verbose_name=_("Bon de confection"),
|
||
)
|
||
requester = models.CharField(
|
||
max_length=255,
|
||
blank=True,
|
||
null=True,
|
||
verbose_name=_("Demandeur"),
|
||
help_text=_("Personne ou service à l'origine de la demande"),
|
||
)
|
||
signature = models.ImageField(
|
||
upload_to="stock_movements/signatures/",
|
||
blank=True,
|
||
null=True,
|
||
verbose_name=_("Signature"),
|
||
help_text=_("Signature du réceptionnaire (image depuis tablette)"),
|
||
)
|
||
signature_name = models.CharField(
|
||
max_length=255,
|
||
blank=True,
|
||
null=True,
|
||
verbose_name=_("Nom du signataire"),
|
||
)
|
||
is_loan = models.BooleanField(default=False, verbose_name=_("Prêt de matériel"))
|
||
estimated_return_date = models.DateField(null=True, blank=True, verbose_name=_("Date de retour attendue"))
|
||
loan_returned = models.BooleanField(default=False, verbose_name=_("Prêt retourné"))
|
||
|
||
class Meta:
|
||
ordering = ("-date",)
|
||
verbose_name = _("Lot de mouvements de stock")
|
||
verbose_name_plural = _("Lots de mouvements de stock")
|
||
|
||
def __str__(self):
|
||
return _("Lot #{id} — {type} — {date}").format(
|
||
id=self.pk,
|
||
type=self.get_movement_type_display(),
|
||
date=self.date.strftime("%d/%m/%Y %H:%M") if self.date else "",
|
||
)
|
||
|
||
@property
|
||
def is_overdue(self):
|
||
from django.utils import timezone
|
||
if self.is_loan and not self.loan_returned and self.estimated_return_date and self.movement_type == 'out':
|
||
return self.estimated_return_date < timezone.now().date()
|
||
return False
|
||
|
||
def movement_count(self):
|
||
return self.movements.count()
|
||
|
||
|
||
def stock_movement_document_upload_path(instance, filename):
|
||
return f"stock_movements/batch_{instance.batch_id}/{filename}"
|
||
|
||
|
||
class StockMovementDocument(models.Model):
|
||
"""Document (photo, pièce justificative…) attaché à un lot de mouvements."""
|
||
batch = models.ForeignKey(
|
||
StockMovementBatch,
|
||
on_delete=models.CASCADE,
|
||
related_name="documents",
|
||
verbose_name=_("Lot"),
|
||
)
|
||
file = models.FileField(upload_to=stock_movement_document_upload_path, storage=private_storage, verbose_name=_("Fichier"))
|
||
name = models.CharField(max_length=255, blank=True, verbose_name=_("Nom"))
|
||
uploaded_at = models.DateTimeField(auto_now_add=True)
|
||
|
||
def __str__(self):
|
||
return self.name or self.file.name
|
||
|
||
def filename(self):
|
||
import os
|
||
return os.path.basename(self.file.name)
|
||
|
||
def is_image(self):
|
||
name = self.file.name.lower()
|
||
return any(name.endswith(ext) for ext in (".jpg", ".jpeg", ".png", ".gif", ".webp", ".bmp"))
|
||
|
||
|
||
def stock_batch_signature_upload_path(instance, filename):
|
||
return f"stock_movements/signatures/batch_{instance.batch_id}/{filename}"
|
||
|
||
|
||
class StockBatchSignature(models.Model):
|
||
"""Signature immuable attachée à un lot de mouvements de stock."""
|
||
batch = models.ForeignKey(
|
||
StockMovementBatch,
|
||
on_delete=models.CASCADE,
|
||
related_name="signatures",
|
||
verbose_name=_("Lot"),
|
||
)
|
||
image = models.ImageField(
|
||
upload_to=stock_batch_signature_upload_path,
|
||
verbose_name=_("Image de signature"),
|
||
)
|
||
signed_by = models.ForeignKey(
|
||
"auth.User",
|
||
on_delete=models.SET_NULL,
|
||
null=True,
|
||
blank=True,
|
||
related_name="batch_signatures",
|
||
verbose_name=_("Signé par"),
|
||
)
|
||
signed_at = models.DateTimeField(auto_now_add=True, verbose_name=_("Signé le"))
|
||
signature_name = models.CharField(
|
||
max_length=255,
|
||
blank=True,
|
||
null=True,
|
||
verbose_name=_("Nom du signataire"),
|
||
)
|
||
|
||
class Meta:
|
||
ordering = ("signed_at",)
|
||
verbose_name = _("Signature de lot")
|
||
verbose_name_plural = _("Signatures de lot")
|
||
|
||
def __str__(self):
|
||
return _("Signature lot #{batch} — {date}").format(
|
||
batch=self.batch_id,
|
||
date=self.signed_at.strftime("%d/%m/%Y %H:%M") if self.signed_at else "",
|
||
)
|
||
|
||
|
||
class StockMovement(models.Model):
|
||
product = models.ForeignKey(Product, on_delete=models.CASCADE, related_name="stock_movements")
|
||
source_location = models.ForeignKey(
|
||
WarehouseLocation,
|
||
on_delete=models.SET_NULL,
|
||
blank=True,
|
||
null=True,
|
||
related_name="outgoing_movements",
|
||
)
|
||
destination_location = models.ForeignKey(
|
||
WarehouseLocation,
|
||
on_delete=models.SET_NULL,
|
||
blank=True,
|
||
null=True,
|
||
related_name="incoming_movements",
|
||
)
|
||
quantity = models.IntegerField()
|
||
date = models.DateTimeField(auto_now_add=True)
|
||
reason = models.TextField(blank=True, null=True)
|
||
batch = models.ForeignKey(
|
||
StockMovementBatch,
|
||
on_delete=models.SET_NULL,
|
||
blank=True,
|
||
null=True,
|
||
related_name="movements",
|
||
verbose_name=_("Lot"),
|
||
)
|
||
serialized_items = models.ManyToManyField(
|
||
'SerializedItem',
|
||
blank=True,
|
||
related_name='stock_movements',
|
||
verbose_name=_('Articles tracés'),
|
||
)
|
||
|
||
def movement_type(self):
|
||
if self.source_location and self.destination_location:
|
||
return _("Transfert")
|
||
elif self.destination_location:
|
||
return _("Entrée")
|
||
else:
|
||
return _("Sortie")
|
||
|
||
def __str__(self):
|
||
movement_type = self.movement_type()
|
||
return _("{movement} de {quantity} {product}").format(
|
||
movement=movement_type,
|
||
quantity=abs(self.quantity),
|
||
product=self.product.name,
|
||
)
|
||
|
||
|
||
class Inventory(models.Model):
|
||
STATUS_DRAFT = "draft"
|
||
STATUS_IN_PROGRESS = "in_progress"
|
||
STATUS_DONE = "done"
|
||
STATUS_VALIDATED = "validated"
|
||
|
||
|
||
STATUS_CHOICES = (
|
||
(STATUS_DRAFT, _("Brouillon")),
|
||
(STATUS_IN_PROGRESS, _("En cours")),
|
||
(STATUS_DONE, _("Terminé")),
|
||
(STATUS_VALIDATED, _("Validé")),
|
||
)
|
||
|
||
warehouse_location = models.ForeignKey(WarehouseLocation, on_delete=models.CASCADE, related_name="inventories")
|
||
performed_by = models.CharField(max_length=255, blank=True)
|
||
status = models.CharField(max_length=20, choices=STATUS_CHOICES, default=STATUS_DRAFT)
|
||
created_at = models.DateTimeField(auto_now_add=True)
|
||
updated_at = models.DateTimeField(auto_now=True)
|
||
validated_at = models.DateTimeField(blank=True, null=True)
|
||
validated_by = models.ForeignKey(
|
||
"auth.User",
|
||
on_delete=models.SET_NULL,
|
||
null=True,
|
||
blank=True,
|
||
related_name="validated_inventories",
|
||
verbose_name=_("Validé par"),
|
||
)
|
||
validator_comment = models.TextField(
|
||
blank=True,
|
||
default="",
|
||
verbose_name=_("Commentaire du validateur"),
|
||
)
|
||
qr_code = models.ImageField(
|
||
upload_to='stock/qrcodes/inventories/',
|
||
blank=True,
|
||
null=True,
|
||
verbose_name=_('QR Code'),
|
||
)
|
||
|
||
class Meta:
|
||
ordering = ("-created_at", "-id")
|
||
|
||
def __str__(self):
|
||
return _("Inventaire #{id} - {warehouse_location}").format(
|
||
id=self.id,
|
||
warehouse_location=self.warehouse_location.name,
|
||
)
|
||
|
||
@property
|
||
def is_validated(self):
|
||
return self.status == self.STATUS_VALIDATED
|
||
|
||
@property
|
||
def is_editable(self):
|
||
"""L'inventaire est éditable tant qu'il n'est pas validé."""
|
||
return self.status != self.STATUS_VALIDATED
|
||
|
||
@property
|
||
def is_quantity_editable(self):
|
||
"""Les quantités comptées ne peuvent être modifiées qu'en cours d'inventaire (draft/in_progress)."""
|
||
return self.status in (self.STATUS_DRAFT, self.STATUS_IN_PROGRESS)
|
||
|
||
@property
|
||
def is_worker_comment_editable(self):
|
||
"""Le commentaire magasinier est modifiable jusqu'à la validation (donc aussi en statut 'terminé')."""
|
||
return self.status != self.STATUS_VALIDATED
|
||
|
||
def total_items(self):
|
||
return self.items.count()
|
||
|
||
def total_adjustment(self):
|
||
return sum(item.difference for item in self.items.all())
|
||
|
||
def generate_qr_code(self):
|
||
"""Génère un QR code basé sur l'ID de l'inventaire et le sauvegarde."""
|
||
qr = qrcode.QRCode(
|
||
version=1,
|
||
error_correction=qrcode.constants.ERROR_CORRECT_M,
|
||
box_size=10,
|
||
border=4,
|
||
)
|
||
qr.add_data(f"LOKO:inventory:{self.pk}")
|
||
qr.make(fit=True)
|
||
img = qr.make_image(fill_color="black", back_color="white")
|
||
buffer = BytesIO()
|
||
img.save(buffer, format='PNG')
|
||
buffer.seek(0)
|
||
from django.core.files.base import ContentFile
|
||
filename = f"inventory_{self.pk}.png"
|
||
self.qr_code.save(filename, ContentFile(buffer.read()), save=False)
|
||
|
||
def get_validators(self):
|
||
"""
|
||
Retourne la liste des utilisateurs actifs ayant la permission de valider
|
||
un inventaire pour l'entrepôt concerné (ou cet emplacement spécifique).
|
||
"""
|
||
if not self.warehouse_location_id:
|
||
return []
|
||
from django.contrib.auth.models import User
|
||
from stock.permissions import user_can_validate_inventory, user_can_validate_inventory_on
|
||
|
||
warehouse = self.warehouse_location.warehouse
|
||
active_users = User.objects.filter(is_active=True).select_related("config")
|
||
validators = []
|
||
for user in active_users:
|
||
if user_can_validate_inventory(user, warehouse) or user_can_validate_inventory_on(user, self.warehouse_location):
|
||
validators.append(user)
|
||
return validators
|
||
|
||
def notify_validators(self, sender=None):
|
||
"""
|
||
Crée une notification avec prise en charge unique pour tous les utilisateurs
|
||
autorisés à valider cet inventaire lorsqu'il doit être validé.
|
||
"""
|
||
if not self.pk or not self.warehouse_location_id:
|
||
return None
|
||
|
||
from django.urls import reverse
|
||
from django.utils.translation import gettext as _
|
||
from notifications.models import NotificationRecipient
|
||
from notifications.services import notify_users
|
||
|
||
validators = self.get_validators()
|
||
if not validators:
|
||
return None
|
||
|
||
action_url = reverse("stock:inventory_detail", args=[self.pk])
|
||
|
||
# Éviter de renvoyer une notification s'il en existe déjà une non lue/active pour cet inventaire
|
||
has_active_notif = NotificationRecipient.objects.filter(
|
||
notification__source_app="stock",
|
||
notification__action_url=action_url,
|
||
read_at__isnull=True,
|
||
archived_at__isnull=True,
|
||
).exists()
|
||
if has_active_notif:
|
||
return None
|
||
|
||
location_name = self.warehouse_location.name
|
||
warehouse_name = self.warehouse_location.warehouse.name
|
||
title = _("Inventaire à valider : %(location)s (%(warehouse)s)") % {
|
||
"location": location_name,
|
||
"warehouse": warehouse_name,
|
||
}
|
||
performed = f" ({self.performed_by})" if self.performed_by else ""
|
||
content = _(
|
||
"L'inventaire #%(id)s pour %(location)s (entrepôt %(warehouse)s) "
|
||
"a été finalisé%(performed)s et doit être validé."
|
||
) % {
|
||
"id": self.id,
|
||
"location": location_name,
|
||
"warehouse": warehouse_name,
|
||
"performed": performed,
|
||
}
|
||
|
||
try:
|
||
return notify_users(
|
||
recipients=validators,
|
||
title=title,
|
||
content=content,
|
||
notification_type="task",
|
||
action_url=action_url,
|
||
action_label=_("Valider l'inventaire"),
|
||
source_app="stock",
|
||
sender=sender or getattr(self, "_current_user", None),
|
||
is_system=True,
|
||
is_single_claim=True,
|
||
)
|
||
except Exception:
|
||
import logging
|
||
logging.getLogger(__name__).exception(
|
||
f"Error sending validation notification for inventory #{self.pk}"
|
||
)
|
||
return None
|
||
|
||
def dismiss_validation_notifications(self):
|
||
"""
|
||
Marque comme traitée (lue et archivée) toute notification de validation
|
||
en cours pour cet inventaire pour tous ses destinataires.
|
||
"""
|
||
if not self.pk:
|
||
return
|
||
from django.urls import reverse
|
||
from django.utils import timezone
|
||
from notifications.models import NotificationRecipient
|
||
|
||
action_url = reverse("stock:inventory_detail", args=[self.pk])
|
||
now = timezone.now()
|
||
NotificationRecipient.objects.filter(
|
||
notification__source_app="stock",
|
||
notification__action_url=action_url,
|
||
read_at__isnull=True,
|
||
).update(read_at=now, archived_at=now)
|
||
|
||
def save(self, *args, **kwargs):
|
||
is_new = self._state.adding
|
||
old_status = None
|
||
if not is_new and self.pk:
|
||
old_status = Inventory.objects.filter(pk=self.pk).values_list("status", flat=True).first()
|
||
|
||
update_fields = kwargs.get('update_fields')
|
||
if update_fields is not None:
|
||
super().save(*args, **kwargs)
|
||
else:
|
||
needs_qr = self.pk is None or not self.qr_code
|
||
super().save(*args, **kwargs)
|
||
if needs_qr:
|
||
self.generate_qr_code()
|
||
super().save(update_fields=['qr_code'])
|
||
|
||
status_became_done = (
|
||
(is_new and self.status == self.STATUS_DONE)
|
||
or (old_status and old_status != self.STATUS_DONE and self.status == self.STATUS_DONE)
|
||
)
|
||
if status_became_done:
|
||
try:
|
||
self.notify_validators()
|
||
except Exception:
|
||
import logging
|
||
logging.getLogger(__name__).exception(
|
||
f"Error in notify_validators for inventory #{self.pk}"
|
||
)
|
||
|
||
if (old_status == self.STATUS_DONE and self.status != self.STATUS_DONE) or self.status == self.STATUS_VALIDATED:
|
||
try:
|
||
self.dismiss_validation_notifications()
|
||
except Exception:
|
||
import logging
|
||
logging.getLogger(__name__).exception(
|
||
f"Error in dismiss_validation_notifications for inventory #{self.pk}"
|
||
)
|
||
|
||
def validate(self, user=None):
|
||
if self.is_validated and self.validated_at:
|
||
return
|
||
self.status = self.STATUS_VALIDATED
|
||
if not self.validated_at:
|
||
self.validated_at = timezone.now()
|
||
if user and not self.validated_by:
|
||
self.validated_by = user
|
||
self.save(update_fields=["status", "validated_at", "validated_by", "updated_at"])
|
||
self.apply_adjustments()
|
||
try:
|
||
self.dismiss_validation_notifications()
|
||
except Exception:
|
||
pass
|
||
|
||
def apply_adjustments(self):
|
||
for item in self.items.select_related("product"):
|
||
stock, created = Stock.objects.get_or_create(
|
||
product=item.product,
|
||
location=item.inventory.warehouse_location,
|
||
defaults={"quantity": 0},
|
||
)
|
||
difference = item.counted_quantity - stock.quantity
|
||
if difference == 0:
|
||
continue
|
||
StockMovement.objects.create(
|
||
product=item.product,
|
||
source_location=item.inventory.warehouse_location if difference < 0 else None,
|
||
destination_location=item.inventory.warehouse_location if difference > 0 else None,
|
||
quantity=difference,
|
||
reason=_("Ajustement inventaire #{inventory_id}").format(
|
||
inventory_id=self.id
|
||
),
|
||
)
|
||
stock.quantity = item.counted_quantity
|
||
stock.save()
|
||
|
||
|
||
class InventoryItem(models.Model):
|
||
inventory = models.ForeignKey(Inventory, on_delete=models.CASCADE, related_name="items")
|
||
product = models.ForeignKey(Product, on_delete=models.CASCADE)
|
||
expected_quantity = models.IntegerField(blank=True, null=True)
|
||
counted_quantity = models.IntegerField()
|
||
is_validated = models.BooleanField(default=False, verbose_name=_("Validé"))
|
||
validated_at = models.DateTimeField(blank=True, null=True, verbose_name=_("Validé le"))
|
||
validated_by = models.ForeignKey(
|
||
"auth.User",
|
||
on_delete=models.SET_NULL,
|
||
null=True,
|
||
blank=True,
|
||
related_name="validated_inventory_items",
|
||
verbose_name=_("Validé par"),
|
||
)
|
||
comment = models.TextField(blank=True, default="", verbose_name=_("Commentaire"))
|
||
comment_by = models.ForeignKey(
|
||
"auth.User",
|
||
on_delete=models.SET_NULL,
|
||
null=True,
|
||
blank=True,
|
||
related_name="inventory_item_comments",
|
||
verbose_name=_("Commentaire par"),
|
||
)
|
||
validator_comment = models.TextField(blank=True, default="", verbose_name=_("Commentaire du validateur"))
|
||
validator_comment_by = models.ForeignKey(
|
||
"auth.User",
|
||
on_delete=models.SET_NULL,
|
||
null=True,
|
||
blank=True,
|
||
related_name="inventory_item_validator_comments",
|
||
verbose_name=_("Commentaire validateur par"),
|
||
)
|
||
|
||
class Meta:
|
||
unique_together = ("inventory", "product")
|
||
|
||
def __str__(self):
|
||
return _("Inventaire {product} ({location})").format(
|
||
product=self.product.name,
|
||
location=self.inventory.warehouse_location.name,
|
||
)
|
||
|
||
@property
|
||
def difference(self):
|
||
return self.counted_quantity - (self.expected_quantity or 0)
|
||
|
||
def clean(self):
|
||
super().clean()
|
||
if self.product and hasattr(self, 'inventory') and self.inventory:
|
||
try:
|
||
location = self.inventory.warehouse_location
|
||
except AttributeError:
|
||
location = None
|
||
if location:
|
||
if self.product.authorized_locations.exists() and location not in self.product.authorized_locations.all():
|
||
from django.core.exceptions import ValidationError
|
||
raise ValidationError(
|
||
_("Le produit \"%(prod)s\" n'est pas autorisé dans l'emplacement de cet inventaire (\"%(loc)s\").") % {
|
||
'prod': self.product.name,
|
||
'loc': location.name
|
||
}
|
||
)
|
||
|
||
def save(self, *args, **kwargs):
|
||
self.full_clean()
|
||
if self.expected_quantity is None:
|
||
try:
|
||
location = self.inventory.warehouse_location
|
||
except AttributeError:
|
||
location = None
|
||
if location:
|
||
expected = (
|
||
Stock.objects.filter(
|
||
product=self.product,
|
||
location=location,
|
||
)
|
||
.aggregate(total=Sum("quantity"))["total"]
|
||
)
|
||
self.expected_quantity = expected or 0
|
||
else:
|
||
self.expected_quantity = 0
|
||
super().save(*args, **kwargs)
|
||
|
||
|
||
class Supplier(models.Model):
|
||
name = models.CharField(max_length=200)
|
||
contact_email = models.EmailField(blank=True, null=True)
|
||
phone_number = models.CharField(max_length=20, blank=True, null=True)
|
||
address = models.TextField(blank=True, null=True)
|
||
is_active = models.BooleanField(default=True)
|
||
enterprise_number = models.CharField(max_length=100, blank=True, null=True)
|
||
|
||
def __str__(self):
|
||
return self.name
|
||
|
||
class Meta:
|
||
ordering = ("name",)
|
||
verbose_name = _("Fournisseur")
|
||
verbose_name_plural = _("Fournisseurs")
|
||
|
||
|
||
class RequestForQuote(models.Model):
|
||
STATUS_DRAFT = "draft"
|
||
STATUS_SENT = "sent"
|
||
STATUS_COMPLETED = "completed"
|
||
STATUS_CANCELLED = "cancelled"
|
||
|
||
STATUS_CHOICES = [
|
||
(STATUS_DRAFT, _("Brouillon")),
|
||
(STATUS_SENT, _("Envoyée")),
|
||
(STATUS_COMPLETED, _("Convertie en commande")),
|
||
(STATUS_CANCELLED, _("Annulée")),
|
||
]
|
||
|
||
code = models.CharField(max_length=50, unique=True, blank=True)
|
||
uuid = models.UUIDField(default=uuid.uuid4, editable=False)
|
||
warehouse = models.ForeignKey(
|
||
Warehouse, on_delete=models.CASCADE, related_name="request_for_quotes",
|
||
verbose_name=_("Entrepôt"),
|
||
)
|
||
date_created = models.DateField(auto_now_add=True)
|
||
status = models.CharField(
|
||
_("Statut"),
|
||
max_length=50,
|
||
choices=STATUS_CHOICES,
|
||
default=STATUS_DRAFT,
|
||
)
|
||
created_by = models.ForeignKey(
|
||
"auth.User",
|
||
on_delete=models.SET_NULL,
|
||
blank=True,
|
||
null=True,
|
||
related_name="created_request_for_quotes",
|
||
verbose_name=_("Créé par"),
|
||
)
|
||
suppliers = models.ManyToManyField(
|
||
Supplier,
|
||
blank=True,
|
||
related_name="request_for_quotes",
|
||
verbose_name=_("Fournisseurs sollicités"),
|
||
)
|
||
notes = models.TextField(blank=True, null=True, verbose_name=_("Notes"))
|
||
|
||
class Meta:
|
||
ordering = ("-date_created", "-id")
|
||
verbose_name = _("Demande d'offre")
|
||
verbose_name_plural = _("Demandes d'offres")
|
||
|
||
def __str__(self):
|
||
return _("Demande d'offre {code}").format(
|
||
code=self.code or f"#{self.id}"
|
||
)
|
||
|
||
def save(self, *args, **kwargs):
|
||
if not self.code:
|
||
if self.pk:
|
||
self.code = f"RFQ-{self.pk:06d}"
|
||
else:
|
||
super().save(*args, **kwargs)
|
||
self.code = f"RFQ-{self.pk:06d}"
|
||
kwargs['force_insert'] = False
|
||
super().save(*args, **kwargs)
|
||
|
||
|
||
class RequestForQuoteItem(models.Model):
|
||
request_for_quote = models.ForeignKey(
|
||
RequestForQuote, on_delete=models.CASCADE, related_name="items"
|
||
)
|
||
product = models.ForeignKey(Product, on_delete=models.CASCADE, verbose_name=_("Produit"))
|
||
quantity = models.IntegerField(
|
||
verbose_name=_("Quantité totale"),
|
||
help_text=_("Quantité totale souhaitée"),
|
||
)
|
||
num_articles = models.PositiveIntegerField(
|
||
null=True, blank=True,
|
||
verbose_name=_("Nombre d'articles"),
|
||
)
|
||
quantity_per_article = models.DecimalField(
|
||
max_digits=10, decimal_places=3,
|
||
null=True, blank=True,
|
||
verbose_name=_("Quantité par article"),
|
||
)
|
||
|
||
def save(self, *args, **kwargs):
|
||
if self.num_articles and self.quantity_per_article:
|
||
self.quantity = int(self.num_articles * self.quantity_per_article)
|
||
super().save(*args, **kwargs)
|
||
|
||
@property
|
||
def total_quantity_label(self):
|
||
unit = self.product.get_unit_display()
|
||
if self.num_articles and self.quantity_per_article:
|
||
return _("{num} article(s) × {qpa} {unit} = {total} {unit}").format(
|
||
num=self.num_articles,
|
||
qpa=self.quantity_per_article,
|
||
total=self.quantity,
|
||
unit=unit,
|
||
)
|
||
return _("{total} {unit}").format(total=self.quantity, unit=unit)
|
||
|
||
def __str__(self):
|
||
return f"{self.quantity} x {self.product.name} (RFQ {self.request_for_quote.id})"
|
||
|
||
|
||
class PurchaseOrder(models.Model):
|
||
code = models.CharField(max_length=50, unique=True, blank=True)
|
||
uuid = models.UUIDField(default=uuid.uuid4, editable=False)
|
||
supplier = models.ForeignKey(
|
||
Supplier, on_delete=models.CASCADE, related_name="purchase_orders",
|
||
verbose_name=_("Fournisseur"),
|
||
)
|
||
warehouse = models.ForeignKey(
|
||
Warehouse, on_delete=models.CASCADE, related_name="purchase_orders",
|
||
verbose_name=_("Entrepôt"),
|
||
)
|
||
request_for_quote = models.ForeignKey(
|
||
RequestForQuote,
|
||
on_delete=models.SET_NULL,
|
||
null=True,
|
||
blank=True,
|
||
related_name="purchase_orders",
|
||
verbose_name=_("Demande d'offre d'origine"),
|
||
)
|
||
date_ordered = models.DateField(auto_now_add=True)
|
||
status = models.CharField(
|
||
_("Statut"),
|
||
max_length=50,
|
||
choices=[
|
||
("pending", _("En attente")),
|
||
("ordered", _("Envoyée")),
|
||
("partial", _("Partiellement reçue")),
|
||
("received", _("Reçue")),
|
||
("cancelled", _("Annulée")),
|
||
],
|
||
default="pending",
|
||
)
|
||
received_at = models.DateTimeField(blank=True, null=True)
|
||
|
||
@property
|
||
def total_net_amount(self):
|
||
return sum((item.net_amount for item in self.items.all()), 0)
|
||
|
||
@property
|
||
def total_vat_amount(self):
|
||
return sum((item.vat_amount for item in self.items.all()), 0)
|
||
|
||
@property
|
||
def total_amount(self):
|
||
return sum((item.total_amount for item in self.items.all()), 0)
|
||
|
||
class Meta:
|
||
ordering = ("-date_ordered", "-id")
|
||
|
||
def __str__(self):
|
||
return _("Commande {code} - {supplier}").format(
|
||
code=self.code or f"#{self.id}",
|
||
supplier=self.supplier.name,
|
||
)
|
||
|
||
def save(self, *args, **kwargs):
|
||
# Générer le code automatiquement si non défini
|
||
if not self.code:
|
||
# Si l'objet existe déjà (mise à jour), on peut utiliser l'ID
|
||
if self.pk:
|
||
self.code = f"PO-{self.pk:06d}"
|
||
else:
|
||
# Pour une nouvelle instance, sauvegarder d'abord pour obtenir l'ID
|
||
super().save(*args, **kwargs)
|
||
self.code = f"PO-{self.pk:06d}"
|
||
kwargs['force_insert'] = False # Empêcher une erreur lors du second save
|
||
super().save(*args, **kwargs)
|
||
|
||
def generate_qr_code(self):
|
||
"""Génère un QR code basé sur le code de commande et retourne l'image en base64"""
|
||
qr = qrcode.QRCode(
|
||
version=1,
|
||
error_correction=qrcode.constants.ERROR_CORRECT_L,
|
||
box_size=10,
|
||
border=4,
|
||
)
|
||
qr.add_data(f"LOKO:order:{self.uuid}")
|
||
qr.make(fit=True)
|
||
|
||
img = qr.make_image(fill_color="black", back_color="white")
|
||
buffer = BytesIO()
|
||
img.save(buffer, format='PNG')
|
||
buffer.seek(0)
|
||
|
||
# Convertir en base64 pour l'affichage HTML
|
||
img_base64 = base64.b64encode(buffer.getvalue()).decode()
|
||
return f"data:image/png;base64,{img_base64}"
|
||
|
||
def receive(self):
|
||
"""Réceptionne tous les articles restants de la commande."""
|
||
if self.received_at:
|
||
return
|
||
with transaction.atomic():
|
||
now = timezone.now()
|
||
PurchaseOrder.objects.filter(pk=self.pk).update(
|
||
status="received", received_at=now
|
||
)
|
||
self.status = "received"
|
||
self.received_at = now
|
||
batch = StockMovementBatch.objects.create(
|
||
movement_type=StockMovementBatch.TYPE_IN,
|
||
movement_subtype='purchase',
|
||
reason=_("Réception commande #{order_id}").format(order_id=self.id),
|
||
purchase_order=self,
|
||
)
|
||
for item in self.items.select_related("product", "location"):
|
||
remaining = item.quantity - item.received_quantity
|
||
if remaining <= 0:
|
||
continue
|
||
stock, _created = Stock.objects.get_or_create(
|
||
product=item.product,
|
||
location=item.location,
|
||
defaults={"quantity": 0},
|
||
)
|
||
stock.quantity += remaining
|
||
stock.save()
|
||
StockMovement.objects.create(
|
||
product=item.product,
|
||
destination_location=item.location,
|
||
quantity=remaining,
|
||
reason=_("Réception commande #{order_id}").format(
|
||
order_id=self.id
|
||
),
|
||
batch=batch,
|
||
)
|
||
item.received_quantity = item.quantity
|
||
item.save(update_fields=["received_quantity"])
|
||
|
||
def receive_items(self, item_quantities):
|
||
"""
|
||
Réceptionne partiellement des articles.
|
||
item_quantities : dict {item_id: quantity_to_receive}
|
||
"""
|
||
with transaction.atomic():
|
||
now = timezone.now()
|
||
batch = StockMovementBatch.objects.create(
|
||
movement_type=StockMovementBatch.TYPE_IN,
|
||
movement_subtype='purchase',
|
||
reason=_("Réception partielle commande #{order_id}").format(order_id=self.id),
|
||
purchase_order=self,
|
||
)
|
||
for item in self.items.select_related("product", "location"):
|
||
qty = item_quantities.get(item.pk, 0)
|
||
if qty <= 0:
|
||
continue
|
||
# Ne pas dépasser le restant
|
||
remaining = item.quantity - item.received_quantity
|
||
qty = min(qty, remaining)
|
||
if qty <= 0:
|
||
continue
|
||
stock, _created = Stock.objects.get_or_create(
|
||
product=item.product,
|
||
location=item.location,
|
||
defaults={"quantity": 0},
|
||
)
|
||
stock.quantity += qty
|
||
stock.save()
|
||
StockMovement.objects.create(
|
||
product=item.product,
|
||
destination_location=item.location,
|
||
quantity=qty,
|
||
reason=_("Réception partielle commande #{order_id}").format(
|
||
order_id=self.id
|
||
),
|
||
batch=batch,
|
||
)
|
||
item.received_quantity += qty
|
||
item.save(update_fields=["received_quantity"])
|
||
|
||
# Mettre à jour le statut de la commande
|
||
all_items = list(self.items.all())
|
||
if all(i.received_quantity >= i.quantity for i in all_items):
|
||
PurchaseOrder.objects.filter(pk=self.pk).update(
|
||
status="received", received_at=now
|
||
)
|
||
self.status = "received"
|
||
self.received_at = now
|
||
else:
|
||
any_received = any(i.received_quantity > 0 for i in all_items)
|
||
if any_received and self.status not in ("partial", "received"):
|
||
PurchaseOrder.objects.filter(pk=self.pk).update(status="partial")
|
||
self.status = "partial"
|
||
elif any_received and self.status == "ordered":
|
||
PurchaseOrder.objects.filter(pk=self.pk).update(status="partial")
|
||
self.status = "partial"
|
||
|
||
|
||
class PurchaseOrderItem(models.Model):
|
||
purchase_order = models.ForeignKey(
|
||
PurchaseOrder, on_delete=models.CASCADE, related_name="items"
|
||
)
|
||
product = models.ForeignKey(Product, on_delete=models.CASCADE)
|
||
location = models.ForeignKey(WarehouseLocation, on_delete=models.CASCADE)
|
||
quantity = models.IntegerField(
|
||
verbose_name=_('Quantité totale'),
|
||
help_text=_('Quantité totale dans l\'unité du produit à ajouter au stock'),
|
||
)
|
||
num_articles = models.PositiveIntegerField(
|
||
null=True, blank=True,
|
||
verbose_name=_('Nombre d\'articles'),
|
||
help_text=_('Nombre d\'articles commandés (ex: nombre de sacs)'),
|
||
)
|
||
quantity_per_article = models.DecimalField(
|
||
max_digits=10, decimal_places=3,
|
||
null=True, blank=True,
|
||
verbose_name=_('Quantité par article'),
|
||
help_text=_('Quantité dans l\'unité du produit par article (ex: 10 kg par sac)'),
|
||
)
|
||
received_quantity = models.IntegerField(
|
||
default=0,
|
||
verbose_name=_('Quantité reçue'),
|
||
help_text=_('Quantité déjà réceptionnée pour cet article'),
|
||
)
|
||
unit_price = models.DecimalField(
|
||
max_digits=10, decimal_places=2,
|
||
null=True, blank=True,
|
||
verbose_name=_("Prix unitaire (HTVA)"),
|
||
)
|
||
vat_rate = models.DecimalField(
|
||
max_digits=5, decimal_places=2,
|
||
default=21.00,
|
||
verbose_name=_("Taux de TVA (%)"),
|
||
)
|
||
|
||
@property
|
||
def net_amount(self):
|
||
if self.unit_price is not None:
|
||
from decimal import Decimal
|
||
return Decimal(self.quantity) * self.unit_price
|
||
return 0
|
||
|
||
@property
|
||
def vat_amount(self):
|
||
from decimal import Decimal
|
||
return self.net_amount * (self.vat_rate / Decimal('100.00'))
|
||
|
||
@property
|
||
def total_amount(self):
|
||
return self.net_amount + self.vat_amount
|
||
|
||
@property
|
||
def remaining_quantity(self):
|
||
"""Quantité restant à réceptionner."""
|
||
return max(0, self.quantity - self.received_quantity)
|
||
|
||
@property
|
||
def is_fully_received(self):
|
||
return self.received_quantity >= self.quantity
|
||
|
||
@property
|
||
def total_quantity_label(self):
|
||
"""Retourne un label lisible de la quantité totale avec l'unité."""
|
||
unit = self.product.get_unit_display()
|
||
if self.num_articles and self.quantity_per_article:
|
||
return _("{num} article(s) × {qpa} {unit} = {total} {unit}").format(
|
||
num=self.num_articles,
|
||
qpa=self.quantity_per_article,
|
||
total=self.quantity,
|
||
unit=unit,
|
||
)
|
||
return _("{total} {unit}").format(total=self.quantity, unit=unit)
|
||
|
||
def save(self, *args, **kwargs):
|
||
# Auto-compute quantity from articles if both fields are provided
|
||
if self.num_articles and self.quantity_per_article:
|
||
from decimal import Decimal
|
||
self.quantity = int(self.num_articles * self.quantity_per_article)
|
||
super().save(*args, **kwargs)
|
||
|
||
def __str__(self):
|
||
return _("{quantity} x {product} dans la commande {order}").format(
|
||
quantity=self.quantity,
|
||
product=self.product.name,
|
||
order=self.purchase_order.id,
|
||
)
|
||
|
||
|
||
class StockAlert(models.Model):
|
||
product = models.ForeignKey(Product, on_delete=models.CASCADE, related_name="alerts")
|
||
warehouse = models.ForeignKey(Warehouse, on_delete=models.CASCADE, related_name="alerts")
|
||
current_quantity = models.IntegerField(default=0)
|
||
triggered_at = models.DateTimeField(auto_now_add=True)
|
||
resolved_at = models.DateTimeField(blank=True, null=True)
|
||
is_active = models.BooleanField(default=True)
|
||
|
||
class Meta:
|
||
unique_together = ("product", "warehouse", "is_active")
|
||
|
||
def __str__(self):
|
||
return _("Alerte stock {product} ({warehouse})").format(
|
||
product=self.product.name,
|
||
warehouse=self.warehouse.name,
|
||
)
|
||
|
||
|
||
def update_stock_alerts(product, warehouse):
|
||
total = (
|
||
Stock.objects.filter(product=product, location__warehouse=warehouse)
|
||
.aggregate(total=Sum("quantity"))["total"]
|
||
or 0
|
||
)
|
||
alert = StockAlert.objects.filter(
|
||
product=product, warehouse=warehouse, is_active=True
|
||
).first()
|
||
if total < product.threshold:
|
||
if alert:
|
||
alert.current_quantity = total
|
||
alert.save(update_fields=["current_quantity"])
|
||
else:
|
||
StockAlert.objects.create(
|
||
product=product,
|
||
warehouse=warehouse,
|
||
current_quantity=total,
|
||
)
|
||
elif alert:
|
||
# Supprimer les anciennes alertes inactives pour ce produit/entrepôt afin d'éviter la violation d'unicité
|
||
StockAlert.objects.filter(
|
||
product=product, warehouse=warehouse, is_active=False
|
||
).exclude(pk=alert.pk).delete()
|
||
alert.current_quantity = total
|
||
alert.is_active = False
|
||
alert.resolved_at = timezone.now()
|
||
alert.save(update_fields=["current_quantity", "is_active", "resolved_at"])
|
||
|
||
|
||
class PreparationOrder(models.Model):
|
||
order_number = models.CharField(max_length=100, unique=True, blank=True)
|
||
date = models.DateTimeField(auto_now_add=True)
|
||
status = models.CharField(
|
||
max_length=50,
|
||
choices=[
|
||
("to_process", _("A traiter")),
|
||
("pending", _("En attente")),
|
||
("processed", _("Traité")),
|
||
("cancelled", _("Annulé")),
|
||
],
|
||
default="to_process",
|
||
)
|
||
priority = models.CharField(
|
||
max_length=50,
|
||
choices=[
|
||
('1', _('Urgent')),
|
||
('2', _('Elevé')),
|
||
('3', _('Normal')),
|
||
('4', _('Bas')),
|
||
],
|
||
default='3',
|
||
verbose_name=_("Priorité"),
|
||
)
|
||
intervention = models.ForeignKey(
|
||
"interventions.Intervention",
|
||
on_delete=models.SET_NULL,
|
||
null=True,
|
||
blank=True,
|
||
related_name="preparation_orders",
|
||
verbose_name=_("Intervention")
|
||
)
|
||
requester = models.CharField(
|
||
max_length=255,
|
||
blank=True,
|
||
null=True,
|
||
verbose_name=_("Demandeur"),
|
||
help_text=_("Personne ou service à l'origine de la demande"),
|
||
)
|
||
is_loan = models.BooleanField(default=False, verbose_name=_("Prêt de matériel"))
|
||
estimated_return_date = models.DateField(null=True, blank=True, verbose_name=_("Date de retour attendue"))
|
||
loan_returned = models.BooleanField(default=False, verbose_name=_("Prêt retourné"))
|
||
items = models.ManyToManyField(Product, through="PreparationOrderItem")
|
||
qr_code = models.ImageField(upload_to='stock/qrcodes/preparation_orders/', blank=True, null=True, verbose_name=_('QR Code'))
|
||
|
||
STATUS_BADGE_CLASSES = {
|
||
'to_process': 'bg-warning text-dark',
|
||
'pending': 'bg-info',
|
||
'processed': 'bg-success',
|
||
'cancelled': 'bg-secondary',
|
||
}
|
||
STATUS_ICONS = {
|
||
'to_process': 'bi-hourglass-split',
|
||
'pending': 'bi-clock',
|
||
'processed': 'bi-check-circle',
|
||
'cancelled': 'bi-x-circle',
|
||
}
|
||
PRIORITY_BADGE_CLASSES = {
|
||
'1': 'bg-danger',
|
||
'2': 'bg-warning text-dark',
|
||
'3': 'bg-primary',
|
||
'4': 'bg-secondary',
|
||
}
|
||
|
||
class Meta:
|
||
ordering = ("-date",)
|
||
verbose_name = _("Bon de préparation")
|
||
verbose_name_plural = _("Bons de préparation")
|
||
|
||
def __str__(self):
|
||
return f"Preparation Order #{self.order_number}"
|
||
|
||
@property
|
||
def priority_badge_class(self):
|
||
return self.PRIORITY_BADGE_CLASSES.get(self.priority, 'bg-secondary')
|
||
|
||
@property
|
||
def status_badge_class(self):
|
||
return self.STATUS_BADGE_CLASSES.get(self.status, 'bg-secondary')
|
||
|
||
@property
|
||
def status_icon(self):
|
||
return self.STATUS_ICONS.get(self.status, 'bi-circle')
|
||
|
||
@property
|
||
def is_already_prepared(self):
|
||
"""Retourne True si le bon est traité, ou si des sorties de stock ont été exécutées, ou si des articles sont marqués préparés."""
|
||
if self.status == 'processed':
|
||
return True
|
||
if self.stock_batches.filter(movement_type='out').exists():
|
||
return True
|
||
if self.preparationorderitem_set.filter(is_prepared=True).exists():
|
||
return True
|
||
if self.other_items.filter(is_prepared=True).exists():
|
||
return True
|
||
return False
|
||
|
||
def generate_qr_code(self):
|
||
"""Génère un QR code basé sur le numéro de bon de préparation et le sauvegarde dans le champ qr_code."""
|
||
qr = qrcode.QRCode(
|
||
version=1,
|
||
error_correction=qrcode.constants.ERROR_CORRECT_M,
|
||
box_size=10,
|
||
border=4,
|
||
)
|
||
qr.add_data(f"LOKO:preparation:{self.order_number}")
|
||
qr.make(fit=True)
|
||
|
||
img = qr.make_image(fill_color="black", back_color="white")
|
||
buffer = BytesIO()
|
||
img.save(buffer, format='PNG')
|
||
buffer.seek(0)
|
||
|
||
from django.core.files.base import ContentFile
|
||
filename = f"preparation_{self.order_number}.png"
|
||
self.qr_code.save(filename, ContentFile(buffer.read()), save=False)
|
||
|
||
@property
|
||
def is_overdue(self):
|
||
from django.utils import timezone
|
||
if self.is_loan and not self.loan_returned and self.estimated_return_date and self.status == 'processed':
|
||
return self.estimated_return_date < timezone.now().date()
|
||
return False
|
||
|
||
@classmethod
|
||
def generate_order_number(cls, intervention=None):
|
||
"""
|
||
Génère un numéro de bon de préparation unique.
|
||
- Si rattaché à une intervention : BP-<code_intervention>-<index:02d>
|
||
Ex: BP-I113392-01, BP-I113392-02, etc.
|
||
- Si autonome (sans intervention) : BP-YYYYMMDD-<seq:03d>
|
||
Ex: BP-20260930-001, BP-20260930-002, etc.
|
||
"""
|
||
if intervention:
|
||
interv_code = (getattr(intervention, 'code', None) or '').strip()
|
||
if not interv_code:
|
||
interv_id = getattr(intervention, 'id', intervention)
|
||
interv_code = f"I{interv_id}"
|
||
|
||
prefix = f"BP-{interv_code}-"
|
||
|
||
interv_obj = intervention if hasattr(intervention, 'pk') else None
|
||
q_filter = Q(order_number__startswith=prefix)
|
||
if interv_obj:
|
||
q_filter |= Q(intervention=interv_obj)
|
||
|
||
existing_orders = list(cls.objects.filter(q_filter).values_list('order_number', flat=True))
|
||
|
||
existing_indexes = []
|
||
for num in existing_orders:
|
||
if num and num.startswith(prefix):
|
||
suffix = num[len(prefix):]
|
||
if suffix.isdigit():
|
||
existing_indexes.append(int(suffix))
|
||
|
||
linked_count = 0
|
||
if interv_obj:
|
||
linked_count = cls.objects.filter(intervention=interv_obj).count()
|
||
|
||
if existing_indexes:
|
||
next_index = max(max(existing_indexes), linked_count) + 1
|
||
elif linked_count > 0:
|
||
next_index = linked_count + 1
|
||
else:
|
||
next_index = 1
|
||
|
||
candidate = f"{prefix}{next_index:02d}"
|
||
while cls.objects.filter(order_number=candidate).exists():
|
||
next_index += 1
|
||
candidate = f"{prefix}{next_index:02d}"
|
||
return candidate
|
||
else:
|
||
import datetime
|
||
today = datetime.date.today()
|
||
prefix = f"BP-{today.strftime('%Y%m%d')}-"
|
||
existing = cls.objects.filter(order_number__startswith=prefix).values_list('order_number', flat=True)
|
||
existing_nums = []
|
||
for num in existing:
|
||
suffix = num[len(prefix):]
|
||
if suffix.isdigit():
|
||
existing_nums.append(int(suffix))
|
||
next_num = max(existing_nums, default=0) + 1
|
||
candidate = f"{prefix}{next_num:03d}"
|
||
while cls.objects.filter(order_number=candidate).exists():
|
||
next_num += 1
|
||
candidate = f"{prefix}{next_num:03d}"
|
||
return candidate
|
||
|
||
def save(self, *args, **kwargs):
|
||
is_new = self.pk is None
|
||
old_status = None
|
||
if not is_new:
|
||
old_status = PreparationOrder.objects.filter(pk=self.pk).values_list('status', flat=True).first()
|
||
|
||
if not self.order_number:
|
||
self.order_number = self.generate_order_number(intervention=self.intervention)
|
||
|
||
# Ne pas re-générer le QR code lors des saves partiels (update_fields)
|
||
update_fields = kwargs.get('update_fields')
|
||
if update_fields is None and self.order_number:
|
||
if is_new:
|
||
# Pour un nouvel objet, on sauvegarde d'abord pour obtenir le pk,
|
||
# puis on génère le QR code et on sauvegarde à nouveau.
|
||
super().save(*args, **kwargs)
|
||
self.generate_qr_code()
|
||
super().save(update_fields=['qr_code'])
|
||
else:
|
||
old = None
|
||
try:
|
||
old = PreparationOrder.objects.get(pk=self.pk)
|
||
except PreparationOrder.DoesNotExist:
|
||
pass
|
||
if old is None or old.order_number != self.order_number or not self.qr_code:
|
||
self.generate_qr_code()
|
||
super().save(*args, **kwargs)
|
||
else:
|
||
super().save(*args, **kwargs)
|
||
|
||
# Mettre à jour les quantités réservées si le statut change
|
||
if not is_new and old_status and old_status != self.status:
|
||
self.update_reserved_quantities()
|
||
|
||
def update_reserved_quantities(self):
|
||
"""Met à jour les quantités réservées pour tous les produits de ce bon"""
|
||
for item in self.items.all():
|
||
item.update_reserved_quantity()
|
||
|
||
def mark_as_processed(self):
|
||
"""Marque le bon comme traité, marque ses articles comme préparés et libère les quantités réservées"""
|
||
self.status = "processed"
|
||
self.save()
|
||
self.preparationorderitem_set.all().update(is_prepared=True)
|
||
self.other_items.all().update(is_prepared=True)
|
||
self.update_reserved_quantities()
|
||
|
||
def reopen(self):
|
||
"""Réouvre un bon de préparation traité pour modification (statut : to_process)."""
|
||
if self.status == "processed":
|
||
self.status = "to_process"
|
||
self.save()
|
||
self.update_reserved_quantities()
|
||
|
||
def cancel(self):
|
||
"""Annule le bon et libère les quantités réservées"""
|
||
self.status = "cancelled"
|
||
self.save()
|
||
self.update_reserved_quantities()
|
||
|
||
def get_executed_outputs_by_product_and_location(self):
|
||
"""
|
||
Retourne un dictionnaire des mouvements de sortie nets déjà exécutés pour ce bon.
|
||
Structure : { product_id: { location_id: net_quantity } }
|
||
Calcul : sorties (TYPE_OUT) moins entrées/retours (TYPE_IN).
|
||
"""
|
||
executed = {}
|
||
for batch in self.stock_batches.all():
|
||
for move in batch.movements.all():
|
||
prod_id = move.product_id
|
||
if prod_id not in executed:
|
||
executed[prod_id] = {}
|
||
|
||
if batch.movement_type == "out" and move.source_location_id:
|
||
loc_id = move.source_location_id
|
||
executed[prod_id][loc_id] = executed[prod_id].get(loc_id, 0) + move.quantity
|
||
elif batch.movement_type == "in" and move.destination_location_id:
|
||
loc_id = move.destination_location_id
|
||
executed[prod_id][loc_id] = executed[prod_id].get(loc_id, 0) - move.quantity
|
||
|
||
res = {}
|
||
for p_id, loc_dict in executed.items():
|
||
filtered_locs = {l_id: qty for l_id, qty in loc_dict.items() if qty > 0}
|
||
if filtered_locs:
|
||
res[p_id] = filtered_locs
|
||
return res
|
||
|
||
def get_executed_output_quantities(self):
|
||
"""
|
||
Retourne la quantité totale nette sortie par produit pour ce bon.
|
||
Structure : { product_id: net_total_quantity }
|
||
"""
|
||
outputs_by_loc = self.get_executed_outputs_by_product_and_location()
|
||
return {p_id: sum(loc_dict.values()) for p_id, loc_dict in outputs_by_loc.items()}
|
||
|
||
def get_executed_output_summary(self):
|
||
"""
|
||
Retourne un dictionnaire récapitulatif par produit des mouvements d'entrée et sortie liés.
|
||
Structure : { product_id: { 'out': gross_out, 'in': gross_in, 'net': net_qty } }
|
||
"""
|
||
summary = {}
|
||
for batch in self.stock_batches.all():
|
||
for move in batch.movements.all():
|
||
p_id = move.product_id
|
||
if p_id not in summary:
|
||
summary[p_id] = {'out': 0, 'in': 0, 'net': 0}
|
||
if batch.movement_type == 'out':
|
||
summary[p_id]['out'] += move.quantity
|
||
summary[p_id]['net'] += move.quantity
|
||
elif batch.movement_type == 'in':
|
||
summary[p_id]['in'] += move.quantity
|
||
summary[p_id]['net'] -= move.quantity
|
||
return summary
|
||
|
||
def get_default_return_location_id(self):
|
||
"""
|
||
Si tous les produits sortis pour ce bon proviennent du même emplacement unique,
|
||
retourne l'ID de cet emplacement (pour préremplir l'emplacement de destination lors d'un retour).
|
||
"""
|
||
executed_by_loc = self.get_executed_outputs_by_product_and_location()
|
||
loc_set = set()
|
||
for p_id, loc_dict in executed_by_loc.items():
|
||
for loc_id, qty in loc_dict.items():
|
||
if qty > 0:
|
||
loc_set.add(loc_id)
|
||
if len(loc_set) == 1:
|
||
return list(loc_set)[0]
|
||
|
||
gross_locs = set()
|
||
for batch in self.stock_batches.filter(movement_type='out'):
|
||
for move in batch.movements.all():
|
||
if move.source_location_id:
|
||
gross_locs.add(move.source_location_id)
|
||
if len(gross_locs) == 1:
|
||
return list(gross_locs)[0]
|
||
return None
|
||
|
||
def get_all_items(self):
|
||
"""Retourne la liste combinée des articles catalogue et autres articles avec leurs informations d'utilisation."""
|
||
items = []
|
||
for it in self.preparationorderitem_set.select_related('product').all():
|
||
items.append({
|
||
'id': it.id,
|
||
'is_other': False,
|
||
'name': it.product.name,
|
||
'code': it.product.code,
|
||
'quantity': it.quantity,
|
||
'unit': it.product.get_unit_display() if hasattr(it.product, 'get_unit_display') else '',
|
||
'is_prepared': it.is_prepared,
|
||
'is_used': it.is_used,
|
||
'item_obj': it,
|
||
})
|
||
for oi in self.other_items.all():
|
||
items.append({
|
||
'id': oi.id,
|
||
'is_other': True,
|
||
'name': oi.description,
|
||
'code': '',
|
||
'quantity': oi.quantity,
|
||
'unit': '',
|
||
'is_prepared': oi.is_prepared,
|
||
'is_used': oi.is_used,
|
||
'item_obj': oi,
|
||
})
|
||
return items
|
||
|
||
@property
|
||
def all_items_used(self):
|
||
items = self.get_all_items()
|
||
return bool(items) and all(it['is_used'] for it in items)
|
||
|
||
@property
|
||
def used_items_count(self):
|
||
items = self.get_all_items()
|
||
return sum(1 for it in items if it['is_used'])
|
||
|
||
@property
|
||
def total_items_count(self):
|
||
return self.preparationorderitem_set.count() + self.other_items.count()
|
||
|
||
def log_history(self, action, description, user=None, details=None):
|
||
"""Enregistre un événement dans l'historique du bon de préparation."""
|
||
from common.utils import get_short_name
|
||
user_short = ""
|
||
if user and getattr(user, "is_authenticated", False):
|
||
user_short = get_short_name(user)
|
||
return self.history_entries.create(
|
||
user=user if (user and getattr(user, "is_authenticated", False)) else None,
|
||
user_short_name=user_short,
|
||
action=action,
|
||
description=description,
|
||
details=details or {},
|
||
)
|
||
|
||
|
||
|
||
|
||
class PreparationOrderItem(models.Model):
|
||
preparation_order = models.ForeignKey(PreparationOrder, on_delete=models.CASCADE)
|
||
product = models.ForeignKey(Product, on_delete=models.CASCADE)
|
||
quantity = models.IntegerField() # Quantité demandée initialement (0 si ajouté pendant le traitement)
|
||
is_prepared = models.BooleanField(default=False)
|
||
is_used = models.BooleanField(
|
||
default=False,
|
||
verbose_name=_("Utilisé"),
|
||
help_text=_("Indique si le matériel a été utilisé lors de l'intervention."),
|
||
)
|
||
quantity_modified = models.BooleanField(
|
||
default=False,
|
||
verbose_name=_("Quantité modifiée"),
|
||
help_text=_("Indique que la quantité a été modifiée après que l'article ait été marqué comme préparé."),
|
||
)
|
||
previous_quantity = models.IntegerField(
|
||
null=True, blank=True,
|
||
verbose_name=_("Quantité précédente"),
|
||
help_text=_("Ancienne quantité avant modification (pour information du magasinier)."),
|
||
)
|
||
|
||
def __str__(self):
|
||
return _("{quantity} x {product} dans la commande {order}").format(
|
||
quantity=self.quantity,
|
||
product=self.product.name,
|
||
order=self.preparation_order.order_number,
|
||
)
|
||
|
||
def save(self, *args, **kwargs):
|
||
super().save(*args, **kwargs)
|
||
# Mettre à jour la quantité réservée du produit
|
||
self.product.update_reserved_quantity()
|
||
|
||
def delete(self, *args, **kwargs):
|
||
product = self.product
|
||
super().delete(*args, **kwargs)
|
||
# Mettre à jour la quantité réservée du produit après suppression
|
||
product.update_reserved_quantity()
|
||
|
||
|
||
# class PreparationOrderDesign(models.Model):
|
||
# """
|
||
# Lie une conception (ProductDesign) à un bon de préparation.
|
||
# Permet d'associer plusieurs conceptions avec quantité à un même bon.
|
||
# Un bon de production est créé automatiquement pour chaque entrée.
|
||
# """
|
||
# preparation_order = models.ForeignKey(
|
||
# PreparationOrder, on_delete=models.CASCADE, related_name="design_orders"
|
||
# )
|
||
# product_design = models.ForeignKey(
|
||
# "ProductDesign", on_delete=models.CASCADE,
|
||
# verbose_name=_("Conception")
|
||
# )
|
||
# quantity = models.IntegerField(default=1, verbose_name=_("Quantité"))
|
||
#
|
||
# def __str__(self):
|
||
# return _("{quantity} x {design} dans le bon {order}").format(
|
||
# quantity=self.quantity,
|
||
# design=self.product_design.name,
|
||
# order=self.preparation_order.order_number,
|
||
# )
|
||
|
||
|
||
def purchase_order_document_upload_path(instance, filename):
|
||
return f"purchase_orders/order_{instance.purchase_order_id}/{filename}"
|
||
|
||
|
||
class PurchaseOrderDocument(models.Model):
|
||
"""Document (photo, pièce justificative…) attaché à une commande fournisseur."""
|
||
purchase_order = models.ForeignKey(
|
||
PurchaseOrder,
|
||
on_delete=models.CASCADE,
|
||
related_name="documents",
|
||
verbose_name=_("Commande fournisseur"),
|
||
)
|
||
file = models.FileField(upload_to=purchase_order_document_upload_path, storage=private_storage, verbose_name=_("Fichier"))
|
||
name = models.CharField(max_length=255, blank=True, verbose_name=_("Nom"))
|
||
uploaded_at = models.DateTimeField(auto_now_add=True)
|
||
|
||
def __str__(self):
|
||
return self.name or self.file.name
|
||
|
||
def filename(self):
|
||
import os
|
||
return os.path.basename(self.file.name)
|
||
|
||
def is_image(self):
|
||
name = self.file.name.lower()
|
||
return any(name.endswith(ext) for ext in (".jpg", ".jpeg", ".png", ".gif", ".webp", ".bmp"))
|
||
|
||
|
||
def preparation_order_document_upload_path(instance, filename):
|
||
return f"preparation_orders/order_{instance.preparation_order_id}/{filename}"
|
||
|
||
|
||
class PreparationOrderDocument(models.Model):
|
||
"""Document (photo, pièce justificative…) attaché à un bon de préparation."""
|
||
preparation_order = models.ForeignKey(
|
||
PreparationOrder,
|
||
on_delete=models.CASCADE,
|
||
related_name="documents",
|
||
verbose_name=_("Bon de préparation"),
|
||
)
|
||
file = models.FileField(upload_to=preparation_order_document_upload_path, storage=private_storage, verbose_name=_("Fichier"))
|
||
name = models.CharField(max_length=255, blank=True, verbose_name=_("Nom"))
|
||
uploaded_at = models.DateTimeField(auto_now_add=True)
|
||
|
||
def __str__(self):
|
||
return self.name or self.file.name
|
||
|
||
def filename(self):
|
||
import os
|
||
return os.path.basename(self.file.name)
|
||
|
||
def is_image(self):
|
||
name = self.file.name.lower()
|
||
return any(name.endswith(ext) for ext in (".jpg", ".jpeg", ".png", ".gif", ".webp", ".bmp"))
|
||
|
||
|
||
def production_order_document_upload_path(instance, filename):
|
||
return f"production_orders/order_{instance.production_order_id}/{filename}"
|
||
|
||
|
||
class ProductionOrderDocument(models.Model):
|
||
"""Document (photo, pièce justificative…) attaché à un bon de confection."""
|
||
production_order = models.ForeignKey(
|
||
'ProductionOrder',
|
||
on_delete=models.CASCADE,
|
||
related_name="documents",
|
||
verbose_name=_("Bon de confection"),
|
||
)
|
||
file = models.FileField(upload_to=production_order_document_upload_path, storage=private_storage, verbose_name=_("Fichier"))
|
||
name = models.CharField(max_length=255, blank=True, verbose_name=_("Nom"))
|
||
uploaded_at = models.DateTimeField(auto_now_add=True)
|
||
|
||
def __str__(self):
|
||
return self.name or self.file.name
|
||
|
||
def filename(self):
|
||
import os
|
||
return os.path.basename(self.file.name)
|
||
|
||
def is_image(self):
|
||
name = self.file.name.lower()
|
||
return any(name.endswith(ext) for ext in (".jpg", ".jpeg", ".png", ".gif", ".webp", ".bmp"))
|
||
|
||
|
||
class PreparationOrderOtherItem(models.Model):
|
||
"""
|
||
Ligne "autre produit" dans un bon de préparation.
|
||
L'utilisateur décrit librement ce qu'il souhaite, choisit une catégorie
|
||
et, si la catégorie est Signalisation, peut préciser un type de panneau.
|
||
"""
|
||
preparation_order = models.ForeignKey(
|
||
PreparationOrder, on_delete=models.CASCADE, related_name="other_items"
|
||
)
|
||
description = models.TextField(verbose_name=_("Description"))
|
||
quantity = models.IntegerField(default=1, verbose_name=_("Quantité"))
|
||
category = models.ForeignKey(
|
||
ProductCategory,
|
||
on_delete=models.SET_NULL,
|
||
null=True,
|
||
blank=True,
|
||
verbose_name=_("Catégorie"),
|
||
)
|
||
sign_panel_type = models.ForeignKey(
|
||
"sign.SignPanelType",
|
||
on_delete=models.SET_NULL,
|
||
null=True,
|
||
blank=True,
|
||
verbose_name=_("Type de panneau"),
|
||
)
|
||
is_prepared = models.BooleanField(default=False, verbose_name=_("Préparé"))
|
||
is_used = models.BooleanField(
|
||
default=False,
|
||
verbose_name=_("Utilisé"),
|
||
help_text=_("Indique si le matériel a été utilisé lors de l'intervention."),
|
||
)
|
||
|
||
@property
|
||
def name(self):
|
||
return self.description
|
||
|
||
def __str__(self):
|
||
return _("{quantity} x {description} dans le bon {order}").format(
|
||
quantity=self.quantity,
|
||
description=self.description[:50],
|
||
order=self.preparation_order.order_number,
|
||
)
|
||
|
||
|
||
class PreparationOrderHistory(models.Model):
|
||
"""
|
||
Historique des modifications apportées à un bon de préparation.
|
||
Trace les ajouts, modifications de quantité, suppressions et changements de statut avec utilisateur (shortname) et date.
|
||
"""
|
||
ACTION_CHOICES = [
|
||
("created", _("Création")),
|
||
("item_added", _("Ajout de produit")),
|
||
("quantity_changed", _("Modification de quantité")),
|
||
("item_removed", _("Suppression de produit")),
|
||
("other_item_added", _("Ajout d'article personnalisé")),
|
||
("other_item_updated", _("Modification d'article personnalisé")),
|
||
("other_item_removed", _("Suppression d'article personnalisé")),
|
||
("status_changed", _("Changement de statut")),
|
||
("reopened", _("Réouverture")),
|
||
("processed", _("Traitement")),
|
||
("cancelled", _("Annulation")),
|
||
("loan_updated", _("Modification du prêt")),
|
||
("document_added", _("Ajout de document")),
|
||
("document_removed", _("Suppression de document")),
|
||
("other", _("Autre modification")),
|
||
]
|
||
|
||
preparation_order = models.ForeignKey(
|
||
PreparationOrder,
|
||
on_delete=models.CASCADE,
|
||
related_name="history_entries",
|
||
verbose_name=_("Bon de préparation"),
|
||
)
|
||
user = models.ForeignKey(
|
||
settings.AUTH_USER_MODEL,
|
||
on_delete=models.SET_NULL,
|
||
null=True,
|
||
blank=True,
|
||
related_name="preparation_order_history_entries",
|
||
verbose_name=_("Utilisateur"),
|
||
)
|
||
user_short_name = models.CharField(
|
||
max_length=150,
|
||
blank=True,
|
||
verbose_name=_("Nom court"),
|
||
)
|
||
created_at = models.DateTimeField(
|
||
auto_now_add=True,
|
||
verbose_name=_("Date et heure"),
|
||
)
|
||
action = models.CharField(
|
||
max_length=50,
|
||
choices=ACTION_CHOICES,
|
||
default="other",
|
||
verbose_name=_("Action"),
|
||
)
|
||
description = models.TextField(
|
||
verbose_name=_("Description"),
|
||
)
|
||
details = models.JSONField(
|
||
default=dict,
|
||
blank=True,
|
||
verbose_name=_("Détails"),
|
||
)
|
||
|
||
class Meta:
|
||
verbose_name = _("Historique de bon de préparation")
|
||
verbose_name_plural = _("Historiques de bons de préparation")
|
||
ordering = ["-created_at"]
|
||
|
||
def __str__(self):
|
||
return f"[{self.created_at.strftime('%d/%m/%Y %H:%M')}] {self.author_short_name}: {self.description}"
|
||
|
||
def save(self, *args, **kwargs):
|
||
if not self.user_short_name and self.user:
|
||
from common.utils import get_short_name
|
||
self.user_short_name = get_short_name(self.user)
|
||
super().save(*args, **kwargs)
|
||
|
||
@property
|
||
def author_short_name(self):
|
||
if self.user:
|
||
from common.utils import get_short_name
|
||
name = get_short_name(self.user)
|
||
if name:
|
||
return name
|
||
return self.user_short_name or _("Système")
|
||
|
||
@property
|
||
def badge_class(self):
|
||
badges = {
|
||
"created": "bg-primary",
|
||
"item_added": "bg-success",
|
||
"quantity_changed": "bg-warning text-dark",
|
||
"item_removed": "bg-danger",
|
||
"other_item_added": "bg-info text-dark",
|
||
"other_item_updated": "bg-warning text-dark",
|
||
"other_item_removed": "bg-danger",
|
||
"status_changed": "bg-info text-dark",
|
||
"reopened": "bg-warning text-dark",
|
||
"processed": "bg-success",
|
||
"cancelled": "bg-secondary",
|
||
"loan_updated": "bg-primary",
|
||
"document_added": "bg-secondary",
|
||
"document_removed": "bg-secondary",
|
||
}
|
||
return badges.get(self.action, "bg-secondary")
|
||
|
||
@property
|
||
def icon_class(self):
|
||
icons = {
|
||
"created": "bi-plus-circle-fill",
|
||
"item_added": "bi-box-seam-fill",
|
||
"quantity_changed": "bi-pencil-fill",
|
||
"item_removed": "bi-trash-fill",
|
||
"other_item_added": "bi-tag-fill",
|
||
"other_item_updated": "bi-pencil-square",
|
||
"other_item_removed": "bi-trash-fill",
|
||
"status_changed": "bi-arrow-repeat",
|
||
"reopened": "bi-arrow-counterclockwise",
|
||
"processed": "bi-check2-circle",
|
||
"cancelled": "bi-x-circle-fill",
|
||
"loan_updated": "bi-calendar-event-fill",
|
||
"document_added": "bi-paperclip",
|
||
"document_removed": "bi-trash",
|
||
}
|
||
return icons.get(self.action, "bi-clock-history")
|
||
|
||
|
||
|
||
|
||
|
||
# class ProductDesign(models.Model):
|
||
# """
|
||
# Conception de produit : une fiche/recette décrivant comment assembler
|
||
# un produit fini à partir de pièces du stock.
|
||
# Ex: un panneau de signalisation spécifique composé de plusieurs pièces.
|
||
# """
|
||
# STATUS_DRAFT = "draft"
|
||
# STATUS_ACTIVE = "active"
|
||
# STATUS_ARCHIVED = "archived"
|
||
|
||
# STATUS_CHOICES = [
|
||
# (STATUS_DRAFT, _("Brouillon")),
|
||
# (STATUS_ACTIVE, _("Actif")),
|
||
# (STATUS_ARCHIVED, _("Archivé")),
|
||
# ]
|
||
|
||
# code = models.CharField(max_length=100, unique=True)
|
||
# name = models.CharField(max_length=300)
|
||
# description = models.TextField(blank=True, null=True)
|
||
# image = models.ImageField(upload_to="product_designs/", blank=True, null=True)
|
||
# status = models.CharField(max_length=20, choices=STATUS_CHOICES, default=STATUS_DRAFT)
|
||
# category = models.ForeignKey(
|
||
# ProductCategory, on_delete=models.SET_NULL, null=True, blank=True,
|
||
# related_name="product_designs"
|
||
# )
|
||
# warehouse = models.ForeignKey(
|
||
# Warehouse, on_delete=models.SET_NULL, null=True, blank=True,
|
||
# related_name="product_designs",
|
||
# verbose_name=_("Entrepôt de production")
|
||
# )
|
||
# default_location = models.ForeignKey(
|
||
# WarehouseLocation, on_delete=models.SET_NULL, null=True, blank=True,
|
||
# related_name="product_designs",
|
||
# verbose_name=_("Emplacement par défaut")
|
||
# )
|
||
# created_at = models.DateTimeField(auto_now_add=True)
|
||
# updated_at = models.DateTimeField(auto_now=True)
|
||
# output_product = models.ForeignKey(
|
||
# "Product",
|
||
# on_delete=models.SET_NULL,
|
||
# null=True, blank=True,
|
||
# related_name="design_output_for",
|
||
# verbose_name=_("Produit fini par défaut"),
|
||
# help_text=_("Produit existant à augmenter en stock quand un bon de confection est terminé. "
|
||
# "Peut être surchargé par le bon de confection."),
|
||
# )
|
||
# components = models.ManyToManyField(
|
||
# Product, through="ProductDesignComponent", related_name="used_in_designs"
|
||
# )
|
||
|
||
# class Meta:
|
||
# ordering = ("-created_at",)
|
||
|
||
# def __str__(self):
|
||
# return f"{self.code} - {self.name}"
|
||
|
||
# def total_components(self):
|
||
# """Nombre total de pièces nécessaires"""
|
||
# return (
|
||
# self.design_components.aggregate(total=Sum("quantity"))["total"] or 0
|
||
# )
|
||
|
||
# def estimated_cost(self):
|
||
# """Coût estimé basé sur le prix des composants"""
|
||
# total = 0
|
||
# for comp in self.design_components.select_related("product"):
|
||
# if comp.product.price:
|
||
# total += comp.product.price * comp.quantity
|
||
# return total
|
||
|
||
# def can_produce(self, warehouse=None):
|
||
# """
|
||
# Vérifie si le stock est suffisant pour produire une unité.
|
||
# Si un entrepôt est spécifié, ne vérifie que dans celui-ci.
|
||
# """
|
||
# for comp in self.design_components.select_related("product"):
|
||
# if warehouse:
|
||
# available = (
|
||
# Stock.objects.filter(
|
||
# product=comp.product, location__warehouse=warehouse
|
||
# ).aggregate(total=Sum("quantity"))["total"] or 0
|
||
# )
|
||
# else:
|
||
# available = comp.product.total_stock
|
||
# if available < comp.quantity:
|
||
# return False
|
||
# return True
|
||
|
||
|
||
# class ProductDesignComponent(models.Model):
|
||
# """
|
||
# Un composant (pièce) faisant partie d'une conception de produit.
|
||
# """
|
||
# product_design = models.ForeignKey(
|
||
# ProductDesign, on_delete=models.CASCADE, related_name="design_components"
|
||
# )
|
||
# product = models.ForeignKey(
|
||
# Product, on_delete=models.CASCADE, related_name="design_usages"
|
||
# )
|
||
# quantity = models.IntegerField(default=1)
|
||
# notes = models.CharField(max_length=500, blank=True, null=True)
|
||
|
||
# class Meta:
|
||
# unique_together = ("product_design", "product")
|
||
|
||
# def __str__(self):
|
||
# return _("{quantity} x {product} pour {design}").format(
|
||
# quantity=self.quantity,
|
||
# product=self.product.name,
|
||
# design=self.product_design.name,
|
||
# )
|
||
|
||
|
||
class ProductionOrder(models.Model):
|
||
"""
|
||
Bon de confection : assemblage d'un produit conçu à partir des composants du stock.
|
||
Peut être lié à un bon de préparation.
|
||
Quand il est terminé, le produit fini est automatiquement ajouté au stock.
|
||
"""
|
||
STATUS_DRAFT = "draft"
|
||
STATUS_TO_PROCESS = "to_process"
|
||
STATUS_IN_PRODUCTION = "in_production"
|
||
STATUS_COMPLETED = "completed"
|
||
STATUS_CANCELLED = "cancelled"
|
||
|
||
STATUS_CHOICES = [
|
||
(STATUS_DRAFT, _("Brouillon")),
|
||
(STATUS_TO_PROCESS, _("À traiter")),
|
||
(STATUS_IN_PRODUCTION, _("En confection")),
|
||
(STATUS_COMPLETED, _("Terminé")),
|
||
(STATUS_CANCELLED, _("Annulé")),
|
||
]
|
||
|
||
DESTINATION_INTERNAL = "internal"
|
||
DESTINATION_EXTERNAL = "external"
|
||
DESTINATION_CHOICES = [
|
||
(DESTINATION_INTERNAL, _("Interne")),
|
||
(DESTINATION_EXTERNAL, _("Externe")),
|
||
]
|
||
|
||
STATUS_BADGE_CLASSES = {
|
||
'draft': 'bg-secondary',
|
||
'to_process': 'bg-danger',
|
||
'in_production': 'bg-primary',
|
||
'completed': 'bg-success',
|
||
'cancelled': 'bg-secondary',
|
||
}
|
||
STATUS_ICONS = {
|
||
'draft': 'bi-pencil',
|
||
'to_process': 'bi-hourglass',
|
||
'in_production': 'bi-gear-wide-connected',
|
||
'completed': 'bi-check-circle',
|
||
'cancelled': 'bi-x-circle',
|
||
}
|
||
|
||
@property
|
||
def status_badge_class(self):
|
||
return self.STATUS_BADGE_CLASSES.get(self.status, 'bg-secondary')
|
||
|
||
@property
|
||
def status_icon(self):
|
||
return self.STATUS_ICONS.get(self.status, 'bi-circle')
|
||
|
||
code = models.CharField(max_length=100, unique=True, blank=True)
|
||
# product_design = models.ForeignKey(
|
||
# ProductDesign, on_delete=models.SET_NULL,
|
||
# null=True, blank=True,
|
||
# related_name="production_orders"
|
||
# )
|
||
preparation_order = models.ForeignKey(
|
||
PreparationOrder, on_delete=models.SET_NULL,
|
||
null=True, blank=True, related_name="production_orders"
|
||
)
|
||
preparation_order_item = models.ForeignKey(
|
||
PreparationOrderItem, on_delete=models.SET_NULL,
|
||
null=True, blank=True, related_name="production_orders",
|
||
verbose_name=_("Article du bon de préparation"),
|
||
)
|
||
preparation_order_other_item = models.ForeignKey(
|
||
PreparationOrderOtherItem, on_delete=models.SET_NULL,
|
||
null=True, blank=True, related_name="production_orders",
|
||
verbose_name=_("Autre produit du bon de préparation"),
|
||
)
|
||
warehouse = models.ForeignKey(
|
||
Warehouse, on_delete=models.CASCADE, related_name="production_orders"
|
||
)
|
||
quantity = models.IntegerField(default=1, help_text=_("Nombre d'unités à concevoir"))
|
||
status = models.CharField(max_length=20, choices=STATUS_CHOICES, default=STATUS_TO_PROCESS)
|
||
destination = models.CharField(
|
||
max_length=20, choices=DESTINATION_CHOICES, default=DESTINATION_INTERNAL,
|
||
verbose_name=_("Destination"),
|
||
help_text=_("Indiquer si la confection est destinée à l'usage interne ou externe."),
|
||
)
|
||
notes = models.TextField(blank=True, null=True)
|
||
created_at = models.DateTimeField(auto_now_add=True)
|
||
updated_at = models.DateTimeField(auto_now=True)
|
||
completed_at = models.DateTimeField(blank=True, null=True)
|
||
qr_code = models.ImageField(upload_to='stock/qrcodes/conception_orders/', blank=True, null=True, verbose_name=_('QR Code'))
|
||
|
||
# Produit fini à ajouter au stock lors de la complétion du bon de confection
|
||
output_product = models.ForeignKey(
|
||
Product,
|
||
on_delete=models.SET_NULL,
|
||
null=True, blank=True,
|
||
related_name="conception_orders_output",
|
||
verbose_name=_("Produit fini à stocker"),
|
||
help_text=_("Produit existant dont la quantité sera augmentée à la fin de la confection. "
|
||
"Si vide et qu'un nom est fourni, un nouveau produit sera créé."),
|
||
)
|
||
output_product_name = models.CharField(
|
||
max_length=300, blank=True, null=True,
|
||
verbose_name=_("Nom du produit fini (nouveau)"),
|
||
help_text=_("Nom du nouveau produit à créer si 'Produit fini à stocker' n'est pas sélectionné."),
|
||
)
|
||
output_location = models.ForeignKey(
|
||
WarehouseLocation,
|
||
on_delete=models.SET_NULL,
|
||
null=True, blank=True,
|
||
related_name="conception_orders_output",
|
||
verbose_name=_("Emplacement de stockage du produit fini"),
|
||
help_text=_("Emplacement où le produit fini sera ajouté au stock. "
|
||
"Si vide, utilise l'emplacement par défaut de la conception."),
|
||
)
|
||
|
||
class Meta:
|
||
ordering = ("-created_at",)
|
||
verbose_name = _("Bon de confection")
|
||
verbose_name_plural = _("Bons de confection")
|
||
|
||
@property
|
||
def is_external(self):
|
||
"""Retourne True si le bon de confection est destiné à l'usage externe."""
|
||
return self.destination == self.DESTINATION_EXTERNAL
|
||
|
||
@property
|
||
def is_reopened(self):
|
||
"""Retourne True si le bon a été réouvert et est actuellement en cours ou à traiter."""
|
||
if self.status in [self.STATUS_TO_PROCESS, self.STATUS_IN_PRODUCTION]:
|
||
return self.history_entries.filter(action="reopened").exists()
|
||
return False
|
||
|
||
@property
|
||
def display_output_product_name(self):
|
||
"""Retourne le nom d'affichage du produit fini / confectionné."""
|
||
if self.output_product:
|
||
return self.output_product.name
|
||
if self.output_product_name:
|
||
return self.output_product_name
|
||
if self.preparation_order_other_item and self.preparation_order_other_item.description:
|
||
return self.preparation_order_other_item.description
|
||
if self.notes and ': ' in self.notes:
|
||
for line in self.notes.splitlines():
|
||
if ': ' in line:
|
||
return line.split(': ', 1)[1].strip()
|
||
return ""
|
||
|
||
def __str__(self):
|
||
return _("Confection {code} x{qty}").format(
|
||
code=self.code or f"#{self.pk}",
|
||
qty=self.quantity,
|
||
)
|
||
|
||
def generate_qr_code(self):
|
||
"""Génère un QR code basé sur le code du bon de confection et le sauvegarde dans le champ qr_code."""
|
||
qr = qrcode.QRCode(
|
||
version=1,
|
||
error_correction=qrcode.constants.ERROR_CORRECT_M,
|
||
box_size=10,
|
||
border=4,
|
||
)
|
||
qr.add_data(f"LOKO:conception:{self.code}")
|
||
qr.make(fit=True)
|
||
|
||
img = qr.make_image(fill_color="black", back_color="white")
|
||
buffer = BytesIO()
|
||
img.save(buffer, format='PNG')
|
||
buffer.seek(0)
|
||
|
||
from django.core.files.base import ContentFile
|
||
filename = f"conception_{self.code}.png"
|
||
self.qr_code.save(filename, ContentFile(buffer.read()), save=False)
|
||
|
||
def save(self, *args, **kwargs):
|
||
# Ne pas re-générer le QR code lors des saves partiels (update_fields)
|
||
update_fields = kwargs.get('update_fields')
|
||
|
||
if not self.code:
|
||
if self.pk:
|
||
self.code = f"CONC-{self.pk:06d}"
|
||
else:
|
||
super().save(*args, **kwargs)
|
||
self.code = f"CONC-{self.pk:06d}"
|
||
kwargs['force_insert'] = False
|
||
|
||
if update_fields is None:
|
||
generate = False
|
||
if self.pk:
|
||
try:
|
||
old = ProductionOrder.objects.get(pk=self.pk)
|
||
if old.code != self.code or not self.qr_code:
|
||
generate = True
|
||
except ProductionOrder.DoesNotExist:
|
||
generate = True
|
||
else:
|
||
generate = True
|
||
if generate:
|
||
self.generate_qr_code()
|
||
|
||
super().save(*args, **kwargs)
|
||
|
||
def _get_active_components(self):
|
||
"""
|
||
Retourne les composants propres au bon de confection.
|
||
"""
|
||
if self.pk and self.order_components.exists():
|
||
return list(self.order_components.select_related("product"))
|
||
return []
|
||
|
||
def can_start(self):
|
||
"""Vérifie si tous les composants sont disponibles dans l'entrepôt"""
|
||
if self.status not in [self.STATUS_TO_PROCESS, self.STATUS_DRAFT, self.STATUS_IN_PRODUCTION]:
|
||
return False
|
||
for comp in self._get_active_components():
|
||
available = (
|
||
Stock.objects.filter(
|
||
product=comp.product, location__warehouse=self.warehouse
|
||
).aggregate(total=Sum("quantity"))["total"] or 0
|
||
)
|
||
if available < comp.quantity * self.quantity:
|
||
return False
|
||
return True
|
||
|
||
def get_components_status(self):
|
||
"""Retourne le statut de chaque composant par rapport au stock"""
|
||
components_status = []
|
||
for comp in self._get_active_components():
|
||
available = (
|
||
Stock.objects.filter(
|
||
product=comp.product, location__warehouse=self.warehouse
|
||
).aggregate(total=Sum("quantity"))["total"] or 0
|
||
)
|
||
required = comp.quantity * self.quantity
|
||
components_status.append({
|
||
'component': comp,
|
||
'required': required,
|
||
'available': available,
|
||
'sufficient': available >= required,
|
||
})
|
||
return components_status
|
||
|
||
def get_output_product(self):
|
||
"""
|
||
Retourne le produit fini à stocker, ou en crée un nouveau si nécessaire.
|
||
Retourne None si aucun produit de sortie n'est spécifié.
|
||
"""
|
||
# Produit explicitement lié au bon
|
||
if self.output_product_id:
|
||
return self.output_product
|
||
# Créer un nouveau produit si un nom est fourni
|
||
if self.output_product_name:
|
||
import uuid as _uuid
|
||
code = f"CUSTOM-{_uuid.uuid4().hex[:8].upper()}"
|
||
product = Product.objects.create(
|
||
code=code,
|
||
name=self.output_product_name,
|
||
sku=code,
|
||
unit=Product.UNIT_PC,
|
||
threshold=0,
|
||
total_stock=0,
|
||
)
|
||
# Lier le produit créé pour les prochaines fois
|
||
self.output_product = product
|
||
self.save(update_fields=["output_product", "updated_at"])
|
||
return product
|
||
return None
|
||
|
||
def get_output_location(self):
|
||
"""
|
||
Retourne l'emplacement de stockage du produit fini.
|
||
Priorité : output_location > première location de l'entrepôt
|
||
"""
|
||
if self.output_location_id:
|
||
return self.output_location
|
||
return self.warehouse.locations.first()
|
||
|
||
def log_history(self, action, description, user=None, details=None):
|
||
"""Enregistre un événement dans l'historique du bon de confection."""
|
||
from common.utils import get_short_name
|
||
user_short = ""
|
||
if user and getattr(user, "is_authenticated", False):
|
||
user_short = get_short_name(user)
|
||
return self.history_entries.create(
|
||
user=user if (user and getattr(user, "is_authenticated", False)) else None,
|
||
user_short_name=user_short,
|
||
action=action,
|
||
description=description,
|
||
details=details or {},
|
||
)
|
||
|
||
def complete(self, user=None):
|
||
"""
|
||
Marque le bon de confection comme terminé :
|
||
- Consomme le stock des composants
|
||
- Ajoute le produit fini au stock
|
||
- Enregistre les mouvements dans des lots associés
|
||
- Trace l'historique
|
||
"""
|
||
if self.status == self.STATUS_COMPLETED:
|
||
return
|
||
with transaction.atomic():
|
||
now = timezone.now()
|
||
|
||
# 1. Consommer les composants
|
||
active_components = self._get_active_components()
|
||
if active_components:
|
||
batch_out = StockMovementBatch.objects.create(
|
||
movement_type=StockMovementBatch.TYPE_OUT,
|
||
movement_subtype="confection",
|
||
production_order=self,
|
||
reason=_("Bon de confection {code}").format(code=self.code),
|
||
created_by=user,
|
||
)
|
||
for comp in active_components:
|
||
needed = comp.quantity * self.quantity
|
||
# Prendre le stock depuis les emplacements de l'entrepôt
|
||
stocks = Stock.objects.filter(
|
||
product=comp.product,
|
||
location__warehouse=self.warehouse,
|
||
quantity__gt=0,
|
||
).order_by("-quantity")
|
||
remaining = needed
|
||
for stock_entry in stocks:
|
||
if remaining <= 0:
|
||
break
|
||
take = min(stock_entry.quantity, remaining)
|
||
stock_entry.quantity -= take
|
||
stock_entry.save()
|
||
remaining -= take
|
||
StockMovement.objects.create(
|
||
product=comp.product,
|
||
source_location=stock_entry.location,
|
||
quantity=-take,
|
||
reason=_("Bon de confection {code}").format(
|
||
code=self.code,
|
||
),
|
||
batch=batch_out,
|
||
)
|
||
|
||
# 2. Ajouter le produit fini au stock
|
||
finished_product = self.get_output_product()
|
||
if finished_product:
|
||
output_loc = self.get_output_location()
|
||
if output_loc:
|
||
stock_entry, _created = Stock.objects.get_or_create(
|
||
product=finished_product,
|
||
location=output_loc,
|
||
defaults={"quantity": 0},
|
||
)
|
||
stock_entry.quantity += self.quantity
|
||
stock_entry.save()
|
||
batch_in = StockMovementBatch.objects.create(
|
||
movement_type=StockMovementBatch.TYPE_IN,
|
||
movement_subtype="confection",
|
||
production_order=self,
|
||
reason=_("Bon de confection {code} terminé").format(
|
||
code=self.code,
|
||
),
|
||
created_by=user,
|
||
)
|
||
StockMovement.objects.create(
|
||
product=finished_product,
|
||
destination_location=output_loc,
|
||
quantity=self.quantity,
|
||
reason=_("Bon de confection {code} terminé").format(
|
||
code=self.code,
|
||
),
|
||
batch=batch_in,
|
||
)
|
||
|
||
self.status = self.STATUS_COMPLETED
|
||
self.completed_at = now
|
||
self.save(update_fields=["status", "completed_at", "updated_at"])
|
||
|
||
self.log_history(
|
||
action="completed",
|
||
description=_("Confection terminée. Composants consommés et produit fini stocké."),
|
||
user=user,
|
||
details={'status': self.STATUS_COMPLETED},
|
||
)
|
||
|
||
def reopen(self, user=None):
|
||
"""
|
||
Réouvre un bon de confection terminé pour modification (statut : in_production) :
|
||
- Restitue les composants consommés au stock
|
||
- Annule l'ajout du produit fini au stock
|
||
- Enregistre l'événement dans l'historique
|
||
"""
|
||
if self.status != self.STATUS_COMPLETED:
|
||
return False
|
||
with transaction.atomic():
|
||
now = timezone.now()
|
||
# 1. Annuler l'ajout du produit fini au stock
|
||
finished_product = self.get_output_product()
|
||
if finished_product:
|
||
output_loc = self.get_output_location()
|
||
if output_loc:
|
||
stock_entry, _created = Stock.objects.select_for_update().get_or_create(
|
||
product=finished_product,
|
||
location=output_loc,
|
||
defaults={"quantity": 0},
|
||
)
|
||
stock_entry.quantity -= self.quantity
|
||
stock_entry.save()
|
||
batch_rev_in = StockMovementBatch.objects.create(
|
||
movement_type=StockMovementBatch.TYPE_OUT,
|
||
movement_subtype="correction",
|
||
production_order=self,
|
||
reason=_("Réouverture bon de confection {code} - Annulation produit fini").format(code=self.code),
|
||
created_by=user,
|
||
)
|
||
StockMovement.objects.create(
|
||
product=finished_product,
|
||
source_location=output_loc,
|
||
destination_location=None,
|
||
quantity=-self.quantity,
|
||
reason=_("Réouverture bon de confection {code} - Annulation produit fini").format(code=self.code),
|
||
batch=batch_rev_in,
|
||
)
|
||
|
||
# 2. Restituer les composants au stock
|
||
component_moves = StockMovement.objects.filter(
|
||
models.Q(batch__production_order=self, batch__movement_type=StockMovementBatch.TYPE_OUT, batch__movement_subtype="confection") |
|
||
models.Q(batch__production_order=self, batch__movement_type=StockMovementBatch.TYPE_OUT, batch__movement_subtype="") |
|
||
models.Q(reason=f"Bon de confection {self.code}", quantity__lt=0)
|
||
).select_related('product', 'source_location')
|
||
if finished_product:
|
||
component_moves = component_moves.exclude(product=finished_product)
|
||
if 'batch_rev_in' in locals() and batch_rev_in:
|
||
component_moves = component_moves.exclude(batch=batch_rev_in)
|
||
|
||
batch_rev_out = StockMovementBatch.objects.create(
|
||
movement_type=StockMovementBatch.TYPE_IN,
|
||
movement_subtype="return",
|
||
production_order=self,
|
||
reason=_("Réouverture bon de confection {code} - Restitution composants").format(code=self.code),
|
||
created_by=user,
|
||
)
|
||
|
||
if component_moves.exists():
|
||
for move in component_moves:
|
||
restore_qty = abs(move.quantity)
|
||
loc = move.source_location or self.warehouse.locations.first()
|
||
if loc:
|
||
stock_entry, _created = Stock.objects.select_for_update().get_or_create(
|
||
product=move.product,
|
||
location=loc,
|
||
defaults={"quantity": 0},
|
||
)
|
||
stock_entry.quantity += restore_qty
|
||
stock_entry.save()
|
||
StockMovement.objects.create(
|
||
product=move.product,
|
||
source_location=None,
|
||
destination_location=loc,
|
||
quantity=restore_qty,
|
||
reason=_("Réouverture bon de confection {code} - Restitution composants").format(code=self.code),
|
||
batch=batch_rev_out,
|
||
)
|
||
else:
|
||
for comp in self._get_active_components():
|
||
restore_qty = comp.quantity * self.quantity
|
||
loc = self.warehouse.locations.first()
|
||
if loc:
|
||
stock_entry, _created = Stock.objects.select_for_update().get_or_create(
|
||
product=comp.product,
|
||
location=loc,
|
||
defaults={"quantity": 0},
|
||
)
|
||
stock_entry.quantity += restore_qty
|
||
stock_entry.save()
|
||
StockMovement.objects.create(
|
||
product=comp.product,
|
||
source_location=None,
|
||
destination_location=loc,
|
||
quantity=restore_qty,
|
||
reason=_("Réouverture bon de confection {code} - Restitution composants").format(code=self.code),
|
||
batch=batch_rev_out,
|
||
)
|
||
|
||
self.status = self.STATUS_IN_PRODUCTION
|
||
self.completed_at = None
|
||
self.save(update_fields=["status", "completed_at", "updated_at"])
|
||
|
||
self.log_history(
|
||
action="reopened",
|
||
description=_("Réouverture du bon de confection (statut : En confection). Restitution des composants en stock et annulation de l'entrée du produit fini."),
|
||
user=user,
|
||
details={'status': self.STATUS_IN_PRODUCTION},
|
||
)
|
||
return True
|
||
|
||
def cancel(self):
|
||
"""Annule le bon de confection"""
|
||
if self.status in [self.STATUS_TO_PROCESS, self.STATUS_DRAFT, self.STATUS_IN_PRODUCTION]:
|
||
self.status = self.STATUS_CANCELLED
|
||
self.save(update_fields=["status", "updated_at"])
|
||
|
||
|
||
class SerializedItem(models.Model):
|
||
"""Article tracé individuellement par numéro de série."""
|
||
|
||
STATUS_AVAILABLE = "available"
|
||
STATUS_ASSIGNED = "assigned"
|
||
STATUS_LOST = "lost"
|
||
STATUS_RETIRED = "retired"
|
||
|
||
STATUS_CHOICES = [
|
||
(STATUS_AVAILABLE, _("Disponible")),
|
||
(STATUS_ASSIGNED, _("Attribué")),
|
||
(STATUS_LOST, _("Perdu")),
|
||
(STATUS_RETIRED, _("Retiré du service")),
|
||
]
|
||
|
||
product = models.ForeignKey(
|
||
Product,
|
||
on_delete=models.CASCADE,
|
||
related_name="serialized_items",
|
||
limit_choices_to={"is_serialized": True},
|
||
verbose_name=_("Produit"),
|
||
)
|
||
serial_number = models.CharField(
|
||
max_length=200,
|
||
unique=True,
|
||
verbose_name=_("Numéro de série"),
|
||
)
|
||
status = models.CharField(
|
||
max_length=20,
|
||
choices=STATUS_CHOICES,
|
||
default=STATUS_AVAILABLE,
|
||
verbose_name=_("Statut"),
|
||
)
|
||
current_location = models.ForeignKey(
|
||
WarehouseLocation,
|
||
on_delete=models.SET_NULL,
|
||
null=True,
|
||
blank=True,
|
||
related_name="serialized_items",
|
||
verbose_name=_("Emplacement actuel"),
|
||
)
|
||
current_holder = models.ForeignKey(
|
||
"contracts.CompanyMember",
|
||
on_delete=models.SET_NULL,
|
||
null=True,
|
||
blank=True,
|
||
related_name="held_serialized_items",
|
||
verbose_name=_("Détenteur actuel"),
|
||
)
|
||
notes = models.TextField(blank=True, default="", verbose_name=_("Notes"))
|
||
qr_code = models.ImageField(
|
||
upload_to='stock/qrcodes/serialized_items/',
|
||
blank=True,
|
||
null=True,
|
||
verbose_name=_("QR Code"),
|
||
)
|
||
created_at = models.DateTimeField(auto_now_add=True, verbose_name=_("Créé le"))
|
||
created_by = models.ForeignKey(
|
||
"auth.User",
|
||
on_delete=models.SET_NULL,
|
||
null=True,
|
||
blank=True,
|
||
related_name="created_serialized_items",
|
||
verbose_name=_("Créé par"),
|
||
)
|
||
is_active = models.BooleanField(
|
||
default=True,
|
||
verbose_name=_("Actif"),
|
||
)
|
||
|
||
class Meta:
|
||
ordering = ("product__name", "serial_number")
|
||
verbose_name = _("Article tracé")
|
||
verbose_name_plural = _("Articles tracés")
|
||
|
||
def __str__(self):
|
||
return f"{self.product.name} — {self.serial_number}"
|
||
|
||
def generate_qr_code(self):
|
||
"""Génère un QR code basé sur le numéro de série et le sauvegarde dans le champ qr_code."""
|
||
qr = qrcode.QRCode(
|
||
version=1,
|
||
error_correction=qrcode.constants.ERROR_CORRECT_M,
|
||
box_size=10,
|
||
border=4,
|
||
)
|
||
qr.add_data(f"LOKO:serializeditem:{self.serial_number}")
|
||
qr.make(fit=True)
|
||
|
||
img = qr.make_image(fill_color="black", back_color="white")
|
||
buffer = BytesIO()
|
||
img.save(buffer, format='PNG')
|
||
buffer.seek(0)
|
||
|
||
from django.core.files.base import ContentFile
|
||
safe_sn = self.serial_number.replace('/', '_').replace('\\', '_')
|
||
filename = f"serialized_{safe_sn}.png"
|
||
self.qr_code.save(filename, ContentFile(buffer.read()), save=False)
|
||
|
||
def save(self, *args, **kwargs):
|
||
update_fields = kwargs.get('update_fields')
|
||
if update_fields is not None:
|
||
super().save(*args, **kwargs)
|
||
return
|
||
generate = False
|
||
if self.pk:
|
||
try:
|
||
old = SerializedItem.objects.get(pk=self.pk)
|
||
if old.serial_number != self.serial_number or not self.qr_code:
|
||
generate = True
|
||
except SerializedItem.DoesNotExist:
|
||
generate = True
|
||
else:
|
||
generate = True
|
||
if generate:
|
||
self.generate_qr_code()
|
||
super().save(*args, **kwargs)
|
||
|
||
|
||
class SerializedItemEvent(models.Model):
|
||
"""Evénement de l'historique d'un article tracé."""
|
||
|
||
TYPE_ENTRY = "entry"
|
||
TYPE_ASSIGNED = "assigned"
|
||
TYPE_RETURNED = "returned"
|
||
TYPE_TRANSFERRED = "transferred"
|
||
TYPE_LOST = "lost"
|
||
TYPE_RETIRED = "retired"
|
||
TYPE_NOTE = "note"
|
||
|
||
TYPE_CHOICES = [
|
||
(TYPE_ENTRY, _("Entrée en stock")),
|
||
(TYPE_ASSIGNED, _("Attribué")),
|
||
(TYPE_RETURNED, _("Rendu")),
|
||
(TYPE_TRANSFERRED, _("Transféré")),
|
||
(TYPE_LOST, _("Perdu")),
|
||
(TYPE_RETIRED, _("Retiré du service")),
|
||
(TYPE_NOTE, _("Note")),
|
||
]
|
||
|
||
item = models.ForeignKey(
|
||
SerializedItem,
|
||
on_delete=models.CASCADE,
|
||
related_name="events",
|
||
verbose_name=_("Article"),
|
||
)
|
||
event_type = models.CharField(
|
||
max_length=20,
|
||
choices=TYPE_CHOICES,
|
||
verbose_name=_("Type d'événement"),
|
||
)
|
||
date = models.DateTimeField(default=timezone.now, verbose_name=_("Date"))
|
||
person = models.ForeignKey(
|
||
"contracts.CompanyMember",
|
||
on_delete=models.SET_NULL,
|
||
null=True,
|
||
blank=True,
|
||
related_name="serialized_item_events",
|
||
verbose_name=_("Personne"),
|
||
)
|
||
from_location = models.ForeignKey(
|
||
WarehouseLocation,
|
||
on_delete=models.SET_NULL,
|
||
null=True,
|
||
blank=True,
|
||
related_name="serialized_events_from",
|
||
verbose_name=_("Depuis"),
|
||
)
|
||
to_location = models.ForeignKey(
|
||
WarehouseLocation,
|
||
on_delete=models.SET_NULL,
|
||
null=True,
|
||
blank=True,
|
||
related_name="serialized_events_to",
|
||
verbose_name=_("Vers"),
|
||
)
|
||
notes = models.TextField(blank=True, default="", verbose_name=_("Remarques"))
|
||
created_by = models.ForeignKey(
|
||
"auth.User",
|
||
on_delete=models.SET_NULL,
|
||
null=True,
|
||
blank=True,
|
||
related_name="serialized_item_events",
|
||
verbose_name=_("Enregistré par"),
|
||
)
|
||
created_at = models.DateTimeField(auto_now_add=True)
|
||
|
||
class Meta:
|
||
ordering = ("-date", "-created_at")
|
||
verbose_name = _("Événement article tracé")
|
||
verbose_name_plural = _("Événements articles tracés")
|
||
|
||
def __str__(self):
|
||
return f"{self.get_event_type_display()} — {self.item} — {self.date:%d/%m/%Y}"
|
||
|
||
|
||
class ProductionOrderComponent(models.Model):
|
||
"""
|
||
Composant (article du stock) directement associé à un bon de confection.
|
||
Si des composants sont définis ici, ils ont priorité sur ceux du design lié.
|
||
"""
|
||
production_order = models.ForeignKey(
|
||
ProductionOrder, on_delete=models.CASCADE,
|
||
related_name="order_components"
|
||
)
|
||
product = models.ForeignKey(
|
||
Product, on_delete=models.CASCADE,
|
||
related_name="production_order_usages"
|
||
)
|
||
quantity = models.IntegerField(default=1)
|
||
notes = models.CharField(max_length=500, blank=True, null=True)
|
||
|
||
class Meta:
|
||
unique_together = ("production_order", "product")
|
||
ordering = ("product__name",)
|
||
|
||
def __str__(self):
|
||
return _("{quantity} x {product} pour {order}").format(
|
||
quantity=self.quantity,
|
||
product=self.product.name,
|
||
order=self.production_order.code,
|
||
)
|
||
|
||
|
||
class ProductionOrderHistory(models.Model):
|
||
"""
|
||
Historique des modifications apportées à un bon de confection.
|
||
Trace les créations, changements de statut, réouvertures, modifications de composants/quantité,
|
||
documents et complétions avec utilisateur (shortname) et date.
|
||
"""
|
||
ACTION_CHOICES = [
|
||
("created", _("Création")),
|
||
("status_changed", _("Changement de statut")),
|
||
("reopened", _("Réouverture")),
|
||
("completed", _("Terminé")),
|
||
("cancelled", _("Annulation")),
|
||
("component_added", _("Ajout de composant")),
|
||
("component_updated", _("Modification de composant")),
|
||
("component_removed", _("Suppression de composant")),
|
||
("quantity_changed", _("Modification de quantité")),
|
||
("document_added", _("Ajout de document")),
|
||
("document_removed", _("Suppression de document")),
|
||
("other", _("Autre modification")),
|
||
]
|
||
|
||
production_order = models.ForeignKey(
|
||
ProductionOrder,
|
||
on_delete=models.CASCADE,
|
||
related_name="history_entries",
|
||
verbose_name=_("Bon de confection"),
|
||
)
|
||
user = models.ForeignKey(
|
||
settings.AUTH_USER_MODEL,
|
||
on_delete=models.SET_NULL,
|
||
null=True,
|
||
blank=True,
|
||
related_name="production_order_history_entries",
|
||
verbose_name=_("Utilisateur"),
|
||
)
|
||
user_short_name = models.CharField(
|
||
max_length=150,
|
||
blank=True,
|
||
verbose_name=_("Nom court"),
|
||
)
|
||
created_at = models.DateTimeField(
|
||
auto_now_add=True,
|
||
verbose_name=_("Date et heure"),
|
||
)
|
||
action = models.CharField(
|
||
max_length=50,
|
||
choices=ACTION_CHOICES,
|
||
default="other",
|
||
verbose_name=_("Action"),
|
||
)
|
||
description = models.TextField(
|
||
verbose_name=_("Description"),
|
||
)
|
||
details = models.JSONField(
|
||
default=dict,
|
||
blank=True,
|
||
verbose_name=_("Détails"),
|
||
)
|
||
|
||
class Meta:
|
||
verbose_name = _("Historique de bon de confection")
|
||
verbose_name_plural = _("Historiques de bons de confection")
|
||
ordering = ["-created_at"]
|
||
|
||
def __str__(self):
|
||
return f"[{self.created_at.strftime('%d/%m/%Y %H:%M')}] {self.author_short_name}: {self.description}"
|
||
|
||
def save(self, *args, **kwargs):
|
||
if not self.user_short_name and self.user:
|
||
from common.utils import get_short_name
|
||
self.user_short_name = get_short_name(self.user)
|
||
super().save(*args, **kwargs)
|
||
|
||
@property
|
||
def author_short_name(self):
|
||
if self.user:
|
||
from common.utils import get_short_name
|
||
name = get_short_name(self.user)
|
||
if name:
|
||
return name
|
||
return self.user_short_name or _("Système")
|
||
|
||
@property
|
||
def badge_class(self):
|
||
badges = {
|
||
"created": "bg-primary",
|
||
"status_changed": "bg-info text-dark",
|
||
"reopened": "bg-warning text-dark",
|
||
"completed": "bg-success",
|
||
"cancelled": "bg-secondary",
|
||
"component_added": "bg-success",
|
||
"component_updated": "bg-warning text-dark",
|
||
"component_removed": "bg-danger",
|
||
"quantity_changed": "bg-warning text-dark",
|
||
"document_added": "bg-secondary",
|
||
"document_removed": "bg-secondary",
|
||
"other": "bg-secondary",
|
||
}
|
||
return badges.get(self.action, "bg-secondary")
|
||
|
||
@property
|
||
def icon_class(self):
|
||
icons = {
|
||
"created": "bi-plus-circle-fill",
|
||
"status_changed": "bi-arrow-repeat",
|
||
"reopened": "bi-arrow-counterclockwise",
|
||
"completed": "bi-check2-circle",
|
||
"cancelled": "bi-x-circle-fill",
|
||
"component_added": "bi-box-seam-fill",
|
||
"component_updated": "bi-pencil-fill",
|
||
"component_removed": "bi-trash-fill",
|
||
"quantity_changed": "bi-pencil-square",
|
||
"document_added": "bi-paperclip",
|
||
"document_removed": "bi-trash",
|
||
"other": "bi-gear-fill",
|
||
}
|
||
return icons.get(self.action, "bi-clock-history")
|
||
|