from django.contrib.gis.db import models from datetime import datetime from django.conf import settings from django.utils.translation import gettext_lazy as _, get_language from django.utils import timezone from common.models import Thematic from common.storages import PrivateMediaStorage from assets.models import AssetCategory private_storage = PrivateMediaStorage() OBSERVATION_TYPE_CHOICES = [ ('incident', 'Incident'), ('inventory', 'Inventaire'), ('remark', 'Remarque'), ] BOOTSTRAP_COLOR_CHOICES = [ ('primary', 'Primary'), ('secondary', 'Secondary'), ('success', 'Success'), ('danger', 'Danger'), ('warning', 'Warning'), ('info', 'Info'), ('light', 'Light'), ('dark', 'Dark'), ] # ============================================================================ # Attribute Choice Sets (Reusable choice sets for observation attributes) # ============================================================================ def attribute_choice_image_upload_path(instance, filename): """Generate upload path for attribute choice images""" now = timezone.now() year = now.strftime("%Y") month = now.strftime("%m") return f"observations/attribute_choices/{year}/{month}/{filename}" class ObservationAttributeChoiceSet(models.Model): """ Ensemble de choix réutilisable pour les attributs d'observation. Exemple: "État du marquage" avec les options Bon/Dégradé/Absent """ name = models.CharField(max_length=255, unique=True, verbose_name=_("Nom")) description = models.TextField(blank=True, null=True, verbose_name=_("Description")) is_active = models.BooleanField(default=True, verbose_name=_("Actif")) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta: verbose_name = _("Ensemble de choix d'attribut") verbose_name_plural = _("Ensembles de choix d'attributs") ordering = ['name'] def __str__(self): return self.name class ObservationAttributeChoice(models.Model): """ Option de choix individuelle dans un ensemble. Avec support pour les icônes, couleurs et images. """ choice_set = models.ForeignKey( ObservationAttributeChoiceSet, on_delete=models.CASCADE, related_name='choices', verbose_name=_("Ensemble de choix") ) value = models.CharField(max_length=100, verbose_name=_("Valeur (clé)"), help_text=_("Valeur technique stockée (ne pas modifier après création)")) label_fr = models.CharField(max_length=255, verbose_name=_("Label (FR)")) label_nl = models.CharField(max_length=255, blank=True, null=True, verbose_name=_("Label (NL)")) bootstrap_icon = models.CharField( max_length=50, blank=True, null=True, verbose_name=_("Icône Bootstrap"), help_text=_("Ex: bi-check-circle, bi-x-circle, bi-dash-circle") ) bootstrap_color = models.CharField( max_length=20, choices=BOOTSTRAP_COLOR_CHOICES, default='primary', verbose_name=_("Couleur Bootstrap") ) image = models.ImageField( upload_to=attribute_choice_image_upload_path, blank=True, null=True, verbose_name=_("Image"), help_text=_("Image optionnelle pour représenter ce choix (pour usage mobile)") ) order = models.PositiveIntegerField(default=0, verbose_name=_("Ordre")) is_active = models.BooleanField(default=True, verbose_name=_("Actif")) class Meta: verbose_name = _("Choix d'attribut") verbose_name_plural = _("Choix d'attributs") ordering = ['choice_set', 'order'] unique_together = ['choice_set', 'value'] def get_label(self): lang = get_language() if lang == 'nl' and self.label_nl: return self.label_nl return self.label_fr def __str__(self): return f"{self.choice_set.name} - {self.get_label()}" # ============================================================================ # Checklist Models (Generic checklist system for observations) # ============================================================================ class ChecklistTemplate(models.Model): """ Template de checklist associé à une catégorie d'asset et un type d'observation. Exemple: "Checklist Inventaire Emplacements Taxi" """ name_fr = models.CharField(max_length=255, verbose_name=_("Nom (FR)")) name_nl = models.CharField(max_length=255, verbose_name=_("Nom (NL)"), blank=True, null=True) description = models.TextField(blank=True, null=True, verbose_name=_("Description")) category = models.ForeignKey( AssetCategory, on_delete=models.CASCADE, related_name='checklist_templates', verbose_name=_("Catégorie d'asset") ) observation_type = models.CharField( max_length=20, choices=OBSERVATION_TYPE_CHOICES, default='inventory', verbose_name=_("Type d'observation") ) is_active = models.BooleanField(default=True, verbose_name=_("Actif")) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) def get_name(self): lang = get_language() name = getattr(self, f"name_{lang}", self.name_fr) return name or self.name_fr def __str__(self): return f"{self.get_name()} ({self.category})" class Meta: verbose_name = _("Template de checklist") verbose_name_plural = _("Templates de checklist") ordering = ['category', 'name_fr'] class ChecklistItem(models.Model): """ Élément individuel d'un template de checklist. Exemple: "Marquage au sol", "Poteau avec signalisation", etc. """ ITEM_TYPE_CHOICES = [ ('boolean', _('Oui/Non')), ('text', _('Texte libre')), ('number', _('Nombre')), ('choice', _('Choix simple')), ('choice_multiple', _('Choix multiple')), ] template = models.ForeignKey( ChecklistTemplate, on_delete=models.CASCADE, related_name='items', verbose_name=_("Template") ) name_fr = models.CharField(max_length=255, verbose_name=_("Nom (FR)")) name_nl = models.CharField(max_length=255, verbose_name=_("Nom (NL)"), blank=True, null=True) description_fr = models.TextField(blank=True, null=True, verbose_name=_("Description (FR)")) description_nl = models.TextField(blank=True, null=True, verbose_name=_("Description (NL)")) item_type = models.CharField( max_length=20, choices=ITEM_TYPE_CHOICES, default='boolean', verbose_name=_("Type de réponse") ) # Option 1: Référence à un ensemble de choix réutilisable (recommandé) choice_set = models.ForeignKey( ObservationAttributeChoiceSet, on_delete=models.SET_NULL, null=True, blank=True, related_name='checklist_items', verbose_name=_("Ensemble de choix"), help_text=_("Sélectionnez un ensemble de choix prédéfini (recommandé) ou utilisez les options textuelles ci-dessous.") ) # Option 2: Choix définis directement en texte (ancien système, conservé pour rétrocompatibilité) choices_fr = models.TextField( blank=True, null=True, verbose_name=_("Options FR (séparées par des virgules)"), help_text=_("Alternative aux ensembles de choix. Ex: Bon,Moyen,Mauvais") ) choices_nl = models.TextField( blank=True, null=True, verbose_name=_("Options NL (séparées par des virgules)"), help_text=_("Alternative aux ensembles de choix. Ex: Goed,Gemiddeld,Slecht") ) choices_icons = models.TextField( blank=True, null=True, verbose_name=_("Icônes Bootstrap (séparées par des virgules)"), help_text=_("Icônes pour chaque option. Ex: bi-check-circle,bi-dash-circle,bi-x-circle") ) choices_colors = models.TextField( blank=True, null=True, verbose_name=_("Couleurs Bootstrap (séparées par des virgules)"), help_text=_("Couleurs pour chaque option. Ex: success,warning,danger (défaut: primary)") ) is_required = models.BooleanField(default=False, verbose_name=_("Obligatoire")) order = models.PositiveIntegerField(default=0, verbose_name=_("Ordre d'affichage")) def get_name(self): lang = get_language() name = getattr(self, f"name_{lang}", self.name_fr) return name or self.name_fr def get_description(self): lang = get_language() desc = getattr(self, f"description_{lang}", self.description_fr) return desc or self.description_fr or "" def is_multiple_choice(self): """Retourne True si l'item permet la sélection multiple.""" return self.item_type == 'choice_multiple' def get_choices_list(self): """Retourne la liste des choix possibles pour les items de type 'choice' selon la langue.""" # Priorité à l'ensemble de choix s'il existe if self.choice_set: return [choice.get_label() for choice in self.choice_set.choices.filter(is_active=True)] # Fallback sur les choix textuels lang = get_language() choices_field = getattr(self, f"choices_{lang}", self.choices_fr) choices = choices_field or self.choices_fr if choices: return [c.strip() for c in choices.split(',') if c.strip()] return [] def get_choices_list_with_keys(self): """ Retourne la liste des choix avec clés, valeurs traduites, icônes, couleurs et images. Utile pour stocker la clé mais afficher la traduction avec style. """ # Priorité à l'ensemble de choix s'il existe if self.choice_set: result = [] for choice in self.choice_set.choices.filter(is_active=True).order_by('order'): result.append({ 'key': choice.value, 'label': choice.get_label(), 'icon': choice.bootstrap_icon or '', 'color': choice.bootstrap_color or 'primary', 'image': choice.image.url if choice.image else None, 'image_id': choice.id if choice.image else None, }) return result # Fallback sur les choix textuels (ancien système) choices_fr = self.choices_fr or '' choices_nl = self.choices_nl or '' choices_icons = self.choices_icons or '' choices_colors = self.choices_colors or '' fr_list = [c.strip() for c in choices_fr.split(',') if c.strip()] nl_list = [c.strip() for c in choices_nl.split(',') if c.strip()] icons_list = [c.strip() for c in choices_icons.split(',')] colors_list = [c.strip() for c in choices_colors.split(',')] lang = get_language() result = [] for i, fr_choice in enumerate(fr_list): label = nl_list[i] if (lang == 'nl' and i < len(nl_list)) else fr_choice icon = icons_list[i] if i < len(icons_list) and icons_list[i] else '' color = colors_list[i] if i < len(colors_list) and colors_list[i] else 'primary' result.append({ 'key': fr_choice, 'label': label, 'icon': icon, 'color': color, 'image': None, 'image_id': None, }) return result def get_choice_metadata(self, choice_key): """ Retourne les métadonnées (label, icon, color, image) pour un choix spécifique. """ choices = self.get_choices_list_with_keys() for choice in choices: if choice['key'] == choice_key: return choice # Si non trouvé, retourner le choix brut avec défauts return {'key': choice_key, 'label': choice_key, 'icon': '', 'color': 'primary', 'image': None, 'image_id': None} def __str__(self): return f"{self.get_name()} ({self.template.get_name()})" class Meta: verbose_name = _("Élément de checklist") verbose_name_plural = _("Éléments de checklist") ordering = ['template', 'order', 'name_fr'] class ObservationChecklistResponse(models.Model): """ Réponse à un élément de checklist pour une observation spécifique. """ observation = models.ForeignKey( 'Observation', on_delete=models.CASCADE, related_name='checklist_responses', verbose_name=_("Observation") ) checklist_item = models.ForeignKey( ChecklistItem, on_delete=models.CASCADE, related_name='responses', verbose_name=_("Élément de checklist") ) # Réponses selon le type d'item value_boolean = models.BooleanField(null=True, blank=True, verbose_name=_("Valeur (Oui/Non)")) value_text = models.TextField(blank=True, null=True, verbose_name=_("Valeur (Texte)")) value_number = models.DecimalField( max_digits=10, decimal_places=2, null=True, blank=True, verbose_name=_("Valeur (Nombre)") ) value_choice = models.CharField(max_length=255, blank=True, null=True, verbose_name=_("Valeur (Choix simple)")) # Pour les choix multiples, stocker les valeurs sélectionnées en JSON value_choices = models.JSONField( blank=True, null=True, verbose_name=_("Valeurs (Choix multiples)"), help_text=_("Liste des clés de choix sélectionnées") ) comment = models.TextField(blank=True, null=True, verbose_name=_("Commentaire")) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) def get_value(self): """Retourne la valeur selon le type d'item.""" item_type = self.checklist_item.item_type if item_type == 'boolean': if self.value_boolean is True: return _("Oui") elif self.value_boolean is False: return _("Non") return "-" elif item_type == 'text': return self.value_text or "-" elif item_type == 'number': return str(self.value_number) if self.value_number is not None else "-" elif item_type == 'choice': return self.value_choice or "-" elif item_type == 'choice_multiple': if self.value_choices: return ", ".join(self.value_choices) return "-" return "-" def get_value_with_metadata(self): """ Retourne la valeur avec les métadonnées (label, icon, color) pour l'affichage. Utilisé pour les types 'choice' et 'choice_multiple'. """ item_type = self.checklist_item.item_type if item_type == 'choice': if self.value_choice: return [self.checklist_item.get_choice_metadata(self.value_choice)] return [] elif item_type == 'choice_multiple': if self.value_choices: return [self.checklist_item.get_choice_metadata(key) for key in self.value_choices] return [] return [] def __str__(self): return f"{self.observation} - {self.checklist_item.get_name()}: {self.get_value()}" class Meta: verbose_name = _("Réponse de checklist") verbose_name_plural = _("Réponses de checklist") unique_together = ['observation', 'checklist_item'] ordering = ['observation', 'checklist_item__order'] class Observation(models.Model): code = models.CharField(max_length=20, null=True, blank=True, unique=True, verbose_name=_("Code")) description = models.TextField(null=True, blank=True) latitude = models.FloatField() longitude = models.FloatField() address = models.CharField(max_length=1024, null=True, blank=True) created_at = models.DateTimeField(auto_now_add=True) created_by = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, related_name='observations', null=True, blank=True) thematic = models.ForeignKey(Thematic, on_delete=models.SET_NULL, related_name='observations', null=True, blank=True) category = models.ForeignKey(AssetCategory, on_delete=models.SET_NULL, related_name='observations', null=True, blank=True) observation_type = models.CharField(max_length=20, choices=OBSERVATION_TYPE_CHOICES, default='incident') symptom = models.ForeignKey('interventions.Symptom', on_delete=models.SET_NULL, related_name='observations', null=True, blank=True) geom = models.PointField(null=True, blank=True, srid=3812) STATUS_CHOICES = [ ('in_preparation', _('En préparation')), ('to_process', _('À traiter')), ('in_progress', _('En cours')), ('processed', _('Traité')), ('not_relevant', _('Non relevant')), ('to_redirect', _('À rediriger')), ('duplicate', _('Doublon')), ('closed', _('Clôturé')), ('archived', _('Archivé')), ] status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='in_preparation') intervention = models.ForeignKey('interventions.Intervention', on_delete=models.SET_NULL, null=True, blank=True, related_name='observations') project = models.ForeignKey('projects.Project', on_delete=models.SET_NULL, null=True, blank=True, related_name='observations', verbose_name=_("Projet lié")) def save(self, *args, **kwargs): is_new = self.pk is None old_status = None old_symptom_id = None if not is_new: try: old_obj = Observation.objects.only('status', 'symptom_id').get(pk=self.pk) old_status = old_obj.status old_symptom_id = old_obj.symptom_id except Observation.DoesNotExist: pass super().save(*args, **kwargs) if is_new and not self.code: self.code = f'OBS{self.pk:05d}' Observation.objects.filter(pk=self.pk).update(code=self.code) if (self.symptom_id and self.symptom.auto_create_intervention and self.status == 'to_process' and not self.intervention_id): status_changed_to_to_process = (old_status != 'to_process' and self.status == 'to_process') symptom_changed_to_auto = (old_symptom_id != self.symptom_id and self.symptom.auto_create_intervention) if is_new or status_changed_to_to_process or symptom_changed_to_auto: self.create_auto_intervention() def create_auto_intervention(self): from django.contrib.gis.geos import Point, MultiPolygon from interventions.models import Intervention, SourceCategory, InterventionTimeLine, InterventionPlanificationTimeLine, InterventionDocument from django.core.files.base import ContentFile from os.path import basename source_cat = SourceCategory.objects.filter(name_fr__iexact='OBSERVATION').first() status = self.symptom.auto_create_intervention_status or 'to_be_processed' # Geometry buffer geom = None if self.geom: geom = MultiPolygon(self.geom.buffer(1)) elif self.longitude is not None and self.latitude is not None: try: point = Point(self.longitude, self.latitude, srid=4326) point.transform(3812) geom = MultiPolygon(point.buffer(1)) except Exception: pass # Create the intervention title = self.symptom.get_name() if self.address: title = f"{title} - {self.address}" intervention = Intervention.objects.create( title=title[:200], lon=self.longitude, lat=self.latitude, address=self.address, status=status, priority=self.symptom.priority or '3', description=self.description, maintain_type='corrective', type='fixing', origin_type=self.symptom.origin_type or 'damage', source_category=source_cat, source_ref=self.code or f"OBS{self.pk:05d}", source_detail=self.description, asset_category=self.category, symptom=self.symptom, created_by=self.created_by, thematic=self.thematic, geom=geom, contract=self.symptom.contract, assigned_provider=self.symptom.provider, ) # Link to observation and set observation status to 'in_progress' self.intervention = intervention self.status = 'in_progress' Observation.objects.filter(pk=self.pk).update(intervention=intervention, status='in_progress') # Create timeline events InterventionTimeLine.objects.create( intervention=intervention, event_user=self.created_by, event_time=timezone.now(), event_type='creation', from_status=None, to_status=status ) InterventionPlanificationTimeLine.objects.create( intervention=intervention, event_user=self.created_by, event_time=timezone.now() + timezone.timedelta(hours=2), event_type='expected_begin', event_description="Début prévu initial." ) # Copy existing photos (if any) for photo_obj in self.photos.all(): if photo_obj.photo: try: photo_obj.photo.open('rb') InterventionDocument.objects.create( intervention=intervention, file=ContentFile(photo_obj.photo.read(), name=basename(photo_obj.photo.name)), uploaded_by=self.created_by, ) except Exception: pass finally: try: photo_obj.photo.close() except Exception: pass # Add comment on observation from observations.models import ObservationComment ObservationComment.objects.create( observation=self, author=None, comment=f"Observation mise en cours suite à la création automatique de l'intervention {intervention.code}" ) def __str__(self): return self.code or f'OBS#{self.pk}' class Meta: indexes = [ # Accélère les agrégats par type et statut (widget home + listes) models.Index(fields=['thematic', 'observation_type', 'status'], name='obs_thm_type_status_idx'), # Accélère le filtre créateur pour les utilisateurs externes models.Index(fields=['thematic', 'created_by', 'status'], name='obs_thm_creator_status_idx'), # Tri par date de création sur les incidents models.Index(fields=['observation_type', 'status', 'created_at'], name='obs_type_status_created_idx'), ] def observation_photo_upload_path(instance, filename): now = timezone.now() year = now.strftime("%Y") month = now.strftime("%m") return f"observations/photos/{year}/{month}/{filename}" class ObservationPhoto(models.Model): observation = models.ForeignKey(Observation, related_name='photos', on_delete=models.CASCADE) photo = models.ImageField(upload_to=observation_photo_upload_path, storage=private_storage) thumbnail = models.ImageField( upload_to=observation_photo_upload_path, storage=private_storage, null=True, blank=True, max_length=255, verbose_name="Miniature", ) def save(self, *args, **kwargs): is_new = self.pk is None super().save(*args, **kwargs) if is_new and self.observation_id and self.photo: # Check if observation has an intervention obs = Observation.objects.filter(pk=self.observation_id).only('intervention_id', 'created_by_id').first() if obs and obs.intervention_id: from interventions.models import InterventionDocument from django.core.files.base import ContentFile from os.path import basename try: self.photo.open('rb') InterventionDocument.objects.create( intervention_id=obs.intervention_id, file=ContentFile(self.photo.read(), name=basename(self.photo.name)), uploaded_by_id=obs.created_by_id, ) except Exception: pass finally: try: self.photo.close() except Exception: pass def generate_thumbnail(self): """Génère une miniature 300×300 (max) de la photo originale.""" if not self.photo or not self.photo.name: return try: from PIL import Image, ExifTags from io import BytesIO from django.core.files.base import ContentFile with Image.open(self.photo.path) as img: # Corriger l'orientation EXIF (photos mobiles) try: exif = img._getexif() if exif: for tag, value in exif.items(): if tag in ExifTags.TAGS and ExifTags.TAGS[tag] == "Orientation": if value == 3: img = img.rotate(180, expand=True) elif value == 6: img = img.rotate(270, expand=True) elif value == 8: img = img.rotate(90, expand=True) except (AttributeError, Exception): pass # Conversion RGB si nécessaire if img.mode in ("RGBA", "P"): img = img.convert("RGB") img.thumbnail((300, 300), Image.Resampling.LANCZOS) output = BytesIO() img.save(output, format="JPEG", quality=85) output.seek(0) thumb_name = f"{self.photo.name}_thumb.jpg" self.thumbnail.save(thumb_name, ContentFile(output.read()), save=False) ObservationPhoto.objects.filter(pk=self.pk).update(thumbnail=self.thumbnail.name) except Exception as e: import logging logging.getLogger(__name__).warning( "Impossible de générer la miniature pour ObservationPhoto pk=%s : %s", self.pk, e ) class ObservationComment(models.Model): observation = models.ForeignKey(Observation, related_name='comments', on_delete=models.CASCADE) author = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True) comment = models.TextField() created_at = models.DateTimeField(auto_now_add=True)