1163 lines
45 KiB
Python
1163 lines
45 KiB
Python
from django.contrib.gis.db import models
|
|
from django.conf import settings
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.contrib.contenttypes.fields import GenericForeignKey
|
|
from django.utils import timezone
|
|
from django.utils.translation import gettext_lazy as _
|
|
from common.storages import PrivateMediaStorage
|
|
|
|
private_storage = PrivateMediaStorage()
|
|
|
|
INSPECTION_RESULT_CONFIG = {
|
|
'compliant': {
|
|
'label': _('Conforme'),
|
|
'badge_class': 'bg-success',
|
|
'btn_class': 'btn-outline-success',
|
|
'icon': 'bi-check-circle-fill',
|
|
'color_hex': '#198754',
|
|
},
|
|
'non_compliant': {
|
|
'label': _('Non conforme'),
|
|
'badge_class': 'bg-warning text-dark',
|
|
'btn_class': 'btn-outline-warning',
|
|
'icon': 'bi-exclamation-triangle-fill',
|
|
'color_hex': '#ffc107',
|
|
},
|
|
'damaged': {
|
|
'label': _('Endommagé'),
|
|
'badge_class': 'bg-danger',
|
|
'btn_class': 'btn-outline-danger',
|
|
'icon': 'bi-x-circle-fill',
|
|
'color_hex': '#dc3545',
|
|
},
|
|
'missing': {
|
|
'label': _('Absent / Retiré'),
|
|
'badge_class': 'bg-dark',
|
|
'btn_class': 'btn-outline-dark',
|
|
'icon': 'bi-dash-circle-fill',
|
|
'color_hex': '#212529',
|
|
},
|
|
'cancelled': {
|
|
'label': _('Annulé'),
|
|
'badge_class': 'bg-secondary',
|
|
'btn_class': 'btn-outline-secondary',
|
|
'icon': 'bi-slash-circle',
|
|
'color_hex': '#6c757d',
|
|
},
|
|
'pending': {
|
|
'label': _('En attente'),
|
|
'badge_class': 'bg-light text-dark border',
|
|
'btn_class': 'btn-outline-secondary',
|
|
'icon': 'bi-clock',
|
|
'color_hex': '#6c757d',
|
|
},
|
|
}
|
|
|
|
INSPECTION_RESULT_CHOICES = tuple(
|
|
(code, cfg['label']) for code, cfg in INSPECTION_RESULT_CONFIG.items()
|
|
)
|
|
|
|
INSPECTION_STATUS_CHOICES = (
|
|
('scheduled', _('Planifiée / À réaliser')),
|
|
('in_progress', _('En cours')),
|
|
('completed', _('Réalisée')),
|
|
('cancelled', _('Annulée')),
|
|
)
|
|
|
|
INSPECTION_NATURE_CHOICES = (
|
|
('visual', _('Inspection visuelle')),
|
|
('technical', _('Inspection technique')),
|
|
('post_intervention', _('Suivi / Qualité post-intervention')),
|
|
)
|
|
|
|
INSPECTOR_ROLE_CHOICES = (
|
|
('observer', _('Observer / Agent de terrain')),
|
|
('provider', _('Prestataire technique externe')),
|
|
('manager', _('Gestionnaire / Contrôleur interne')),
|
|
)
|
|
|
|
|
|
class Inspection(models.Model):
|
|
"""
|
|
Modèle d'enregistrement générique d'une vérification/inspection sur le terrain d'une cible (Asset, Localisation, Intervention).
|
|
"""
|
|
asset_content_type = models.ForeignKey(
|
|
ContentType,
|
|
on_delete=models.CASCADE,
|
|
verbose_name=_("Type d'asset / Cible"),
|
|
limit_choices_to={'app_label__in': ['assets', 'interventions']}
|
|
)
|
|
asset_object_id = models.PositiveIntegerField(verbose_name=_("ID de la cible"))
|
|
asset = GenericForeignKey('asset_content_type', 'asset_object_id')
|
|
|
|
# Liaisons optionnelles avec les Interventions
|
|
mission_intervention = models.ForeignKey(
|
|
'interventions.Intervention',
|
|
on_delete=models.SET_NULL,
|
|
null=True,
|
|
blank=True,
|
|
related_name='generated_inspections',
|
|
verbose_name=_("Intervention / Mission d'inspection")
|
|
)
|
|
checked_intervention = models.ForeignKey(
|
|
'interventions.Intervention',
|
|
on_delete=models.SET_NULL,
|
|
null=True,
|
|
blank=True,
|
|
related_name='quality_inspections',
|
|
verbose_name=_("Intervention de travaux contrôlée")
|
|
)
|
|
|
|
nature = models.CharField(
|
|
max_length=50,
|
|
choices=INSPECTION_NATURE_CHOICES,
|
|
default='visual',
|
|
verbose_name=_("Nature de l'inspection")
|
|
)
|
|
inspector_role = models.CharField(
|
|
max_length=50,
|
|
choices=INSPECTOR_ROLE_CHOICES,
|
|
default='observer',
|
|
verbose_name=_("Rôle de l'inspecteur")
|
|
)
|
|
|
|
inspector = models.ForeignKey(
|
|
settings.AUTH_USER_MODEL,
|
|
on_delete=models.SET_NULL,
|
|
null=True,
|
|
blank=True,
|
|
related_name='inspections',
|
|
verbose_name=_("Inspecteur / Vérificateur")
|
|
)
|
|
inspection_date = models.DateTimeField(default=timezone.now, verbose_name=_("Date d'inspection"))
|
|
|
|
result_status = models.CharField(
|
|
max_length=50,
|
|
choices=INSPECTION_RESULT_CHOICES,
|
|
default='compliant',
|
|
verbose_name=_("Résultat du contrôle")
|
|
)
|
|
|
|
status = models.CharField(
|
|
max_length=50,
|
|
choices=INSPECTION_STATUS_CHOICES,
|
|
default='completed',
|
|
verbose_name=_("Statut d'avancement")
|
|
)
|
|
scheduled_start_date = models.DateField(
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("Début de fenêtre d'inspection")
|
|
)
|
|
scheduled_due_date = models.DateField(
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("Échéance limite d'inspection")
|
|
)
|
|
assigned_team = models.ForeignKey(
|
|
'contracts.CompanyTeam',
|
|
on_delete=models.SET_NULL,
|
|
null=True,
|
|
blank=True,
|
|
related_name='assigned_inspections',
|
|
verbose_name=_("Équipe assignée")
|
|
)
|
|
assigned_member = models.ForeignKey(
|
|
'contracts.CompanyMember',
|
|
on_delete=models.SET_NULL,
|
|
null=True,
|
|
blank=True,
|
|
related_name='assigned_inspections',
|
|
verbose_name=_("Membre assigné")
|
|
)
|
|
non_compliance_notified = models.BooleanField(
|
|
default=False,
|
|
verbose_name=_("Notification de non-conformité envoyée")
|
|
)
|
|
|
|
verified_obsolescence = models.CharField(
|
|
max_length=50,
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("État d'obsolescence constaté")
|
|
)
|
|
|
|
notes = models.TextField(null=True, blank=True, verbose_name=_("Remarques / Constats"))
|
|
apply_changes_to_asset = models.BooleanField(
|
|
default=False,
|
|
verbose_name=_("Appliquer directement les modifications à la cible")
|
|
)
|
|
|
|
|
|
created_at = models.DateTimeField(auto_now_add=True)
|
|
updated_at = models.DateTimeField(auto_now=True)
|
|
|
|
class Meta:
|
|
verbose_name = _("Inspection (Générique)")
|
|
verbose_name_plural = _("Inspections (Génériques)")
|
|
ordering = ['-inspection_date']
|
|
|
|
@property
|
|
def is_mission_inspection(self):
|
|
return self.mission_intervention_id is not None
|
|
|
|
@property
|
|
def is_cross_check(self):
|
|
return self.checked_intervention_id is not None
|
|
|
|
@property
|
|
def related_intervention(self):
|
|
"""
|
|
Renvoie l'intervention liée à cette inspection si elle existe :
|
|
- checked_intervention (intervention de travaux contrôlée, ex: durabilité)
|
|
- mission_intervention (mission d'inspection)
|
|
- asset si la cible de l'inspection est une intervention
|
|
"""
|
|
if self.checked_intervention_id:
|
|
return self.checked_intervention
|
|
if self.mission_intervention_id:
|
|
return self.mission_intervention
|
|
if self.asset_content_type and self.asset_content_type.model == 'intervention':
|
|
return self.asset
|
|
return None
|
|
|
|
@property
|
|
def related_intervention_url(self):
|
|
interv = self.related_intervention
|
|
if interv:
|
|
from django.urls import reverse
|
|
return reverse('interventions:intervention_detail', kwargs={'intervention_id': interv.id})
|
|
return ''
|
|
|
|
@property
|
|
def mobile_related_intervention_url(self):
|
|
interv = self.related_intervention
|
|
if interv:
|
|
from django.urls import reverse
|
|
return reverse('interventions:intervention_detail_mobile', kwargs={'intervention_id': interv.id})
|
|
return ''
|
|
|
|
@property
|
|
def is_cancelled(self):
|
|
return self.result_status == 'cancelled' or self.status == 'cancelled'
|
|
|
|
@property
|
|
def is_scheduled(self):
|
|
return self.status == 'scheduled'
|
|
|
|
@property
|
|
def timeliness(self):
|
|
"""
|
|
Niveau d'urgence temporelle pour une inspection planifiée:
|
|
'future', 'current', 'urgent', 'overdue', ou None.
|
|
"""
|
|
if self.status != 'scheduled' or not self.scheduled_due_date:
|
|
return None
|
|
today = timezone.now().date()
|
|
if self.scheduled_start_date and today < self.scheduled_start_date:
|
|
return 'future'
|
|
if today > self.scheduled_due_date:
|
|
return 'overdue'
|
|
days_left = (self.scheduled_due_date - today).days
|
|
if days_left <= 7:
|
|
return 'urgent'
|
|
return 'current'
|
|
|
|
@property
|
|
def timeliness_badge_class(self):
|
|
mapping = {
|
|
'future': 'bg-secondary text-white',
|
|
'current': 'bg-success text-white',
|
|
'urgent': 'bg-warning text-dark',
|
|
'overdue': 'bg-danger text-white',
|
|
}
|
|
return mapping.get(self.timeliness, 'bg-light text-dark')
|
|
|
|
@property
|
|
def timeliness_label(self):
|
|
mapping = {
|
|
'future': _('À venir'),
|
|
'current': _('À réaliser'),
|
|
'urgent': _('Échéance proche'),
|
|
'overdue': _('En retard'),
|
|
}
|
|
return mapping.get(self.timeliness, '')
|
|
|
|
def notify_sub_type_teams_if_needed(self):
|
|
"""
|
|
Notifie les équipes configurées sur le sous-type d'intervention
|
|
si le résultat du contrôle de durabilité est non-conforme.
|
|
"""
|
|
if not self.checked_intervention or not getattr(self.checked_intervention, 'sub_type_id', None):
|
|
return
|
|
sub_type = self.checked_intervention.sub_type
|
|
if not sub_type or not sub_type.requires_post_inspection:
|
|
return
|
|
teams = list(sub_type.notification_teams.all())
|
|
if not teams:
|
|
return
|
|
|
|
from contracts.models import CompanyMember
|
|
from notifications.models import Notification, NotificationRecipient
|
|
from notifications.webpush import send_webpush_to_users
|
|
|
|
# Collecter tous les membres actifs avec un compte utilisateur
|
|
recipient_user_ids = set(
|
|
CompanyMember.objects.filter(
|
|
teams__in=teams, user__isnull=False, user__is_active=True
|
|
).values_list('user_id', flat=True)
|
|
)
|
|
if not recipient_user_ids:
|
|
return
|
|
|
|
status_label = self.get_result_status_display()
|
|
title = _("Non-conformité durabilité (%(status)s) : %(code)s") % {
|
|
'status': status_label,
|
|
'code': self.checked_intervention.code,
|
|
}
|
|
inspector_name = self.inspector.get_full_name() if self.inspector else str(_("Non assigné"))
|
|
content = _(
|
|
"L'inspection de durabilité sur l'intervention %(code)s (%(subtype)s) "
|
|
"a révélé un constat non conforme : %(status)s.\n"
|
|
"Inspecteur : %(inspector)s\n"
|
|
"Remarques : %(notes)s"
|
|
) % {
|
|
'code': self.checked_intervention.code,
|
|
'subtype': sub_type.get_name(),
|
|
'status': status_label,
|
|
'inspector': inspector_name,
|
|
'notes': self.notes or '-',
|
|
}
|
|
action_url = f"/interventions/{self.checked_intervention.id}/"
|
|
|
|
try:
|
|
notification = Notification.objects.create(
|
|
title=str(title),
|
|
content=str(content),
|
|
notification_type='alert',
|
|
action_url=action_url,
|
|
action_label=str(_("Consulter l'intervention")),
|
|
source_app='inspections',
|
|
is_system=True,
|
|
)
|
|
NotificationRecipient.objects.bulk_create([
|
|
NotificationRecipient(notification=notification, user_id=uid)
|
|
for uid in recipient_user_ids
|
|
], ignore_conflicts=True)
|
|
|
|
send_webpush_to_users(
|
|
users=list(recipient_user_ids),
|
|
title=str(title),
|
|
body=str(content),
|
|
action_url=action_url,
|
|
)
|
|
except Exception:
|
|
import logging
|
|
logging.getLogger(__name__).exception(
|
|
f"Error sending non-compliance notification for inspection {self.pk}"
|
|
)
|
|
|
|
self.non_compliance_notified = True
|
|
Inspection.objects.filter(pk=self.pk).update(non_compliance_notified=True)
|
|
|
|
def can_edit(self, user):
|
|
from .permissions import can_edit_inspection
|
|
return can_edit_inspection(user, self)
|
|
|
|
def save(self, *args, **kwargs):
|
|
super().save(*args, **kwargs)
|
|
|
|
# Notification automatique en cas de non-conformité post-travaux
|
|
if self.status == 'completed' and self.result_status in ('non_compliant', 'damaged', 'missing') and not self.non_compliance_notified:
|
|
self.notify_sub_type_teams_if_needed()
|
|
|
|
if self.status == 'completed' and self.apply_changes_to_asset and self.asset:
|
|
target_asset = self.asset
|
|
modified = False
|
|
|
|
if self.is_cancelled:
|
|
# Recalculer l'état de l'asset à partir de la dernière inspection active restante
|
|
latest_active = Inspection.objects.filter(
|
|
asset_content_type=self.asset_content_type,
|
|
asset_object_id=self.asset_object_id
|
|
).exclude(result_status='cancelled').exclude(pk=self.pk).order_by('-inspection_date').first()
|
|
|
|
if latest_active:
|
|
if hasattr(target_asset, 'last_inspection_date'):
|
|
target_asset.last_inspection_date = latest_active.inspection_date
|
|
modified = True
|
|
|
|
if hasattr(target_asset, 'inspection_status'):
|
|
try:
|
|
setattr(target_asset, 'inspection_status', latest_active.result_status)
|
|
modified = True
|
|
except AttributeError:
|
|
pass
|
|
|
|
if hasattr(target_asset, 'validation_status'):
|
|
target_asset.validation_status = 'validated'
|
|
modified = True
|
|
|
|
if hasattr(target_asset, 'is_compliant'):
|
|
if latest_active.result_status in ('non_compliant', 'damaged', 'missing'):
|
|
target_asset.is_compliant = False
|
|
if hasattr(target_asset, 'non_compliance_reason'):
|
|
target_asset.non_compliance_reason = latest_active.notes or ''
|
|
else:
|
|
target_asset.is_compliant = True
|
|
if hasattr(target_asset, 'non_compliance_reason'):
|
|
target_asset.non_compliance_reason = None
|
|
modified = True
|
|
|
|
if latest_active.verified_obsolescence and hasattr(target_asset, 'obsolescence'):
|
|
target_asset.obsolescence = latest_active.verified_obsolescence
|
|
modified = True
|
|
else:
|
|
if hasattr(target_asset, 'last_inspection_date'):
|
|
target_asset.last_inspection_date = None
|
|
modified = True
|
|
|
|
if hasattr(target_asset, 'inspection_status'):
|
|
try:
|
|
setattr(target_asset, 'inspection_status', 'unknown')
|
|
modified = True
|
|
except AttributeError:
|
|
pass
|
|
|
|
if hasattr(target_asset, 'is_compliant'):
|
|
target_asset.is_compliant = True
|
|
if hasattr(target_asset, 'non_compliance_reason'):
|
|
target_asset.non_compliance_reason = None
|
|
modified = True
|
|
|
|
if modified:
|
|
target_asset.save()
|
|
else:
|
|
if hasattr(target_asset, 'last_inspection_date'):
|
|
target_asset.last_inspection_date = self.inspection_date
|
|
modified = True
|
|
|
|
if hasattr(target_asset, 'inspection_status'):
|
|
try:
|
|
setattr(target_asset, 'inspection_status', self.result_status)
|
|
modified = True
|
|
except AttributeError:
|
|
pass
|
|
|
|
if hasattr(target_asset, 'validation_status'):
|
|
target_asset.validation_status = 'validated'
|
|
modified = True
|
|
|
|
if hasattr(target_asset, 'is_compliant'):
|
|
if self.result_status in ('non_compliant', 'damaged', 'missing'):
|
|
target_asset.is_compliant = False
|
|
if hasattr(target_asset, 'non_compliance_reason') and self.notes:
|
|
target_asset.non_compliance_reason = self.notes
|
|
else:
|
|
target_asset.is_compliant = True
|
|
if hasattr(target_asset, 'non_compliance_reason'):
|
|
target_asset.non_compliance_reason = None
|
|
modified = True
|
|
|
|
if self.verified_obsolescence and hasattr(target_asset, 'obsolescence'):
|
|
target_asset.obsolescence = self.verified_obsolescence
|
|
modified = True
|
|
|
|
if modified:
|
|
target_asset.save()
|
|
|
|
@property
|
|
def status_badge_class(self):
|
|
return INSPECTION_RESULT_CONFIG.get(self.result_status, {}).get('badge_class', 'bg-secondary')
|
|
|
|
@property
|
|
def status_btn_class(self):
|
|
return INSPECTION_RESULT_CONFIG.get(self.result_status, {}).get('btn_class', 'btn-outline-secondary')
|
|
|
|
@property
|
|
def status_icon(self):
|
|
return INSPECTION_RESULT_CONFIG.get(self.result_status, {}).get('icon', 'bi-question-circle')
|
|
|
|
@property
|
|
def status_color_hex(self):
|
|
return INSPECTION_RESULT_CONFIG.get(self.result_status, {}).get('color_hex', '#6c757d')
|
|
|
|
@classmethod
|
|
def get_result_choices_config(cls):
|
|
"""Retourne la liste des configurations pour chaque statut de résultat (code, label, badge_class, btn_class, icon, color_hex)."""
|
|
return [
|
|
{'code': code, **cfg}
|
|
for code, cfg in INSPECTION_RESULT_CONFIG.items()
|
|
]
|
|
|
|
@property
|
|
def target_display(self):
|
|
if self.asset:
|
|
return str(self.asset)
|
|
return f"Cible #{self.asset_object_id}"
|
|
|
|
@property
|
|
def thematic(self):
|
|
from assets.permissions import _get_asset_thematic
|
|
if self.asset:
|
|
return _get_asset_thematic(self.asset)
|
|
return None
|
|
|
|
@property
|
|
def lon_lat(self):
|
|
"""Retourne les coordonnées (lon, lat) WGS84 du centroïde de la cible s'il est géolocalisé."""
|
|
if not self.asset:
|
|
return None
|
|
if hasattr(self.asset, 'lon') and hasattr(self.asset, 'lat') and self.asset.lon and self.asset.lat:
|
|
return (float(self.asset.lon), float(self.asset.lat))
|
|
if hasattr(self.asset, 'geom') and self.asset.geom:
|
|
g = self.asset.geom
|
|
if g.srid and g.srid != 4326:
|
|
try:
|
|
g = g.transform(4326, clone=True)
|
|
except Exception:
|
|
pass
|
|
centroid = g.centroid
|
|
return (float(centroid.x), float(centroid.y))
|
|
return None
|
|
|
|
@property
|
|
def thematic_icon(self):
|
|
th = self.thematic
|
|
if th and hasattr(th, 'icon') and th.icon:
|
|
return th.icon
|
|
return 'bi-geo-alt'
|
|
|
|
@property
|
|
def is_hotbox_related(self):
|
|
"""Détermine si l'inspection concerne une réparation Hot Box (asphalte à chaud)."""
|
|
if hasattr(self, 'hotboxinspection'):
|
|
return True
|
|
interv = self.related_intervention
|
|
if interv and getattr(interv, 'sub_type', None):
|
|
st = interv.sub_type
|
|
code = (st.code or '').lower()
|
|
name = (st.name_fr or '').lower()
|
|
if any(k in code or k in name for k in ('hotbox', 'hot_box', 'hot box')):
|
|
return True
|
|
if self.nature == 'post_intervention' and 'hot box' in (self.notes or '').lower():
|
|
return True
|
|
return False
|
|
|
|
@property
|
|
def detailed_inspection(self):
|
|
"""Downcast à la sous-classe spécialisée si disponible (ex: ParkingSpotInspection, GiepInspection, SignPanelInspection, HotBoxInspection)."""
|
|
if self.__class__ is not Inspection:
|
|
return self
|
|
if hasattr(self, 'parkingspotinspection'):
|
|
return self.parkingspotinspection
|
|
if hasattr(self, 'giepinspection'):
|
|
return self.giepinspection
|
|
if hasattr(self, 'signpanelinspection'):
|
|
return self.signpanelinspection
|
|
if hasattr(self, 'hotboxinspection'):
|
|
return self.hotboxinspection
|
|
if self.is_hotbox_related and self.pk:
|
|
try:
|
|
hb = HotBoxInspection(inspection_ptr=self)
|
|
for f in self._meta.fields:
|
|
setattr(hb, f.attname, getattr(self, f.attname))
|
|
hb.save_base(raw=True)
|
|
return hb
|
|
except Exception:
|
|
pass
|
|
return self
|
|
|
|
@property
|
|
def details_summary(self):
|
|
"""Résumé textuel concis des constatations spécifiques de l'inspection."""
|
|
leaf = self.detailed_inspection
|
|
parts = []
|
|
|
|
if hasattr(leaf, 'verified_standard_stand_count') and leaf.verified_standard_stand_count is not None:
|
|
parts.append(str(_("%(count)d arceaux classiques") % {'count': leaf.verified_standard_stand_count}))
|
|
if hasattr(leaf, 'verified_cargo_stand_count') and leaf.verified_cargo_stand_count is not None:
|
|
parts.append(str(_("%(count)d arceaux cargo") % {'count': leaf.verified_cargo_stand_count}))
|
|
if hasattr(leaf, 'verified_has_scooter_drop_zone') and leaf.verified_has_scooter_drop_zone is not None:
|
|
parts.append(str(_("Drop-zone trottinette") if leaf.verified_has_scooter_drop_zone else _("Sans drop-zone")))
|
|
if hasattr(leaf, 'verified_marking_quality') and leaf.verified_marking_quality:
|
|
marking_labels = {
|
|
'good': _("Marquage bon état"),
|
|
'medium': _("Marquage moyen"),
|
|
'damaged': _("Marquage dégradé"),
|
|
'missing': _("Marquage absent / effacé"),
|
|
'bad': _("Marquage mauvais"),
|
|
}
|
|
val = marking_labels.get(leaf.verified_marking_quality, leaf.verified_marking_quality)
|
|
parts.append(str(val))
|
|
|
|
if hasattr(leaf, 'verified_cleanliness') and leaf.verified_cleanliness:
|
|
parts.append(str(leaf.get_verified_cleanliness_display()))
|
|
if hasattr(leaf, 'verified_signpanel_type') and leaf.verified_signpanel_type:
|
|
parts.append(str(leaf.verified_signpanel_type.code))
|
|
if hasattr(leaf, 'verified_signpanel_text') and leaf.verified_signpanel_text:
|
|
parts.append(f"« {leaf.verified_signpanel_text[:30]} »")
|
|
|
|
if hasattr(leaf, 'data') and isinstance(leaf.data, dict) and leaf.data:
|
|
ouvrages = leaf.data.get('ouvrages', [])
|
|
if ouvrages:
|
|
parts.append(str(_("%(count)d ouvrage(s) GIEP") % {'count': len(ouvrages)}))
|
|
if getattr(leaf, 'date_derniere_pluie', None) and leaf.date_derniere_pluie != 'null':
|
|
parts.append(str(_("Pluie : %s") % leaf.get_date_derniere_pluie_display()))
|
|
|
|
# Hot Box summary
|
|
if hasattr(leaf, 'joint_level') and hasattr(leaf, 'joint_sealing'):
|
|
if leaf.joint_level:
|
|
parts.append(f"{_('Planéité')} : {leaf.get_joint_level_display()}")
|
|
if leaf.joint_sealing:
|
|
parts.append(f"{_('Joints')} : {leaf.get_joint_sealing_display()}")
|
|
if leaf.cracking:
|
|
parts.append(f"{_('Fissuration')} : {leaf.get_cracking_display()}")
|
|
if leaf.water_drainage:
|
|
parts.append(f"{_('Évacuation')} : {leaf.get_water_drainage_display()}")
|
|
|
|
if leaf.verified_obsolescence:
|
|
parts.append(str(_("Obsolescence : %s") % leaf.verified_obsolescence))
|
|
|
|
return ", ".join(parts) if parts else "-"
|
|
|
|
def get_asset_url(self, is_mobile=False):
|
|
"""Retourne l'URL de détail de l'asset cible (route mobile /mobile/... si is_mobile est True)."""
|
|
if not self.asset:
|
|
return None
|
|
if is_mobile:
|
|
from django.urls import reverse
|
|
from assets.permissions import _get_asset_thematic
|
|
th = _get_asset_thematic(self.asset)
|
|
thematic_code = th.code if th else 'parking'
|
|
asset_model = self.asset.__class__.__name__.lower()
|
|
try:
|
|
return reverse('mobile:mobile_asset_detail', kwargs={
|
|
'thematic_code': thematic_code,
|
|
'asset_model': asset_model,
|
|
'asset_id': self.asset.id
|
|
})
|
|
except Exception:
|
|
pass
|
|
if hasattr(self.asset, 'get_absolute_url'):
|
|
return self.asset.get_absolute_url()
|
|
return None
|
|
|
|
@property
|
|
def mobile_asset_url(self):
|
|
"""Propriété raccourcie pour l'URL mobile de l'asset."""
|
|
return self.get_asset_url(is_mobile=True)
|
|
|
|
def __str__(self):
|
|
asset_str = str(self.asset) if self.asset else f"Cible #{self.asset_object_id}"
|
|
date_str = self.inspection_date.strftime('%d/%m/%Y') if self.inspection_date else ''
|
|
return f"Inspection {asset_str} ({date_str}) - {self.get_result_status_display()}"
|
|
|
|
|
|
# Alias de compatibilité
|
|
AssetInspection = Inspection
|
|
|
|
|
|
class ParkingSpotInspection(Inspection):
|
|
"""
|
|
Spécialisation de l'inspection pour les emplacements de stationnement.
|
|
"""
|
|
verified_standard_stand_count = models.IntegerField(
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("Nombre d'arceaux classiques constaté")
|
|
)
|
|
verified_cargo_stand_count = models.IntegerField(
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("Nombre d'arceaux cargo constaté")
|
|
)
|
|
verified_drop_off_stand_count = models.IntegerField(
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("Nombre d'arceaux drop-off constaté")
|
|
)
|
|
verified_stand_type = models.CharField(
|
|
max_length=50,
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("Type d'arceau constaté")
|
|
)
|
|
verified_has_scooter_drop_zone = models.BooleanField(
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("Drop zone trottinette constatée")
|
|
)
|
|
MARKING_QUALITY_CHOICES = (
|
|
('good', _("Bon état")),
|
|
('medium', _("Moyen")),
|
|
('damaged', _("Dégradé")),
|
|
('missing', _("Absent / Effacé")),
|
|
)
|
|
verified_marking_quality = models.CharField(
|
|
max_length=50,
|
|
choices=MARKING_QUALITY_CHOICES,
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("Qualité du marquage constatée")
|
|
)
|
|
apply_physical_counts_to_asset = models.BooleanField(
|
|
default=False,
|
|
verbose_name=_("Mettre à jour la capacité / comptages de l'asset")
|
|
)
|
|
|
|
class Meta:
|
|
verbose_name = _("Inspection d'emplacement de stationnement")
|
|
verbose_name_plural = _("Inspections d'emplacements de stationnement")
|
|
|
|
def save(self, *args, **kwargs):
|
|
super().save(*args, **kwargs)
|
|
if self.apply_physical_counts_to_asset and self.asset:
|
|
target_asset = self.asset
|
|
modified = False
|
|
if not self.is_cancelled:
|
|
if self.verified_standard_stand_count is not None and hasattr(target_asset, 'standard_stand_count'):
|
|
target_asset.standard_stand_count = self.verified_standard_stand_count
|
|
modified = True
|
|
if self.verified_cargo_stand_count is not None and hasattr(target_asset, 'cargo_stand_count'):
|
|
target_asset.cargo_stand_count = self.verified_cargo_stand_count
|
|
modified = True
|
|
if self.verified_drop_off_stand_count is not None and hasattr(target_asset, 'drop_off_stand_count'):
|
|
target_asset.drop_off_stand_count = self.verified_drop_off_stand_count
|
|
modified = True
|
|
if self.verified_has_scooter_drop_zone is not None and hasattr(target_asset, 'has_scooter_drop_zone'):
|
|
target_asset.has_scooter_drop_zone = self.verified_has_scooter_drop_zone
|
|
modified = True
|
|
if self.verified_marking_quality and hasattr(target_asset, 'marking_quality'):
|
|
target_asset.marking_quality = self.verified_marking_quality
|
|
modified = True
|
|
if modified:
|
|
target_asset.save()
|
|
else:
|
|
latest_active_ps = ParkingSpotInspection.objects.filter(
|
|
asset_content_type=self.asset_content_type,
|
|
asset_object_id=self.asset_object_id
|
|
).exclude(result_status='cancelled').exclude(pk=self.pk).order_by('-inspection_date').first()
|
|
if latest_active_ps and latest_active_ps.apply_physical_counts_to_asset:
|
|
if latest_active_ps.verified_standard_stand_count is not None and hasattr(target_asset, 'standard_stand_count'):
|
|
target_asset.standard_stand_count = latest_active_ps.verified_standard_stand_count
|
|
modified = True
|
|
if latest_active_ps.verified_cargo_stand_count is not None and hasattr(target_asset, 'cargo_stand_count'):
|
|
target_asset.cargo_stand_count = latest_active_ps.verified_cargo_stand_count
|
|
modified = True
|
|
if latest_active_ps.verified_drop_off_stand_count is not None and hasattr(target_asset, 'drop_off_stand_count'):
|
|
target_asset.drop_off_stand_count = latest_active_ps.verified_drop_off_stand_count
|
|
modified = True
|
|
if latest_active_ps.verified_has_scooter_drop_zone is not None and hasattr(target_asset, 'has_scooter_drop_zone'):
|
|
target_asset.has_scooter_drop_zone = latest_active_ps.verified_has_scooter_drop_zone
|
|
modified = True
|
|
if latest_active_ps.verified_marking_quality and hasattr(target_asset, 'marking_quality'):
|
|
target_asset.marking_quality = latest_active_ps.verified_marking_quality
|
|
modified = True
|
|
if modified:
|
|
target_asset.save()
|
|
|
|
|
|
class GiepInspection(Inspection):
|
|
"""
|
|
Spécialisation de l'inspection pour les zones / ouvrages de gestion infiltrative des eaux pluviales (GIEP).
|
|
"""
|
|
DATE_DERNIERE_PLUIE_CHOICES = (
|
|
('aujourdhui', _("Aujourd'hui")),
|
|
('moins_24h', _("Moins de 24h")),
|
|
('plus_24h', _("Plus de 24h")),
|
|
('null', _("Non vérifié")),
|
|
)
|
|
date_derniere_pluie = models.CharField(
|
|
max_length=50,
|
|
choices=DATE_DERNIERE_PLUIE_CHOICES,
|
|
default='null',
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("Date de la dernière pluie")
|
|
)
|
|
orage_moins_une_heure = models.BooleanField(
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("Orage il y a moins d'une heure")
|
|
)
|
|
data = models.JSONField(
|
|
default=dict,
|
|
blank=True,
|
|
verbose_name=_("Données structurées d'inspection GIEP")
|
|
)
|
|
|
|
class Meta:
|
|
verbose_name = _("Inspection GIEP / Eaux pluviales")
|
|
verbose_name_plural = _("Inspections GIEP / Eaux pluviales")
|
|
|
|
@classmethod
|
|
def compute_status_from_data(cls, data):
|
|
"""
|
|
Calcule le résultat global ('damaged', 'non_compliant', 'compliant')
|
|
sur base des réponses aux questions de chaque ouvrage d'une inspection GIEP.
|
|
Priorité : damaged > non_compliant > compliant.
|
|
"""
|
|
if not data or not isinstance(data, dict):
|
|
return 'compliant'
|
|
ouvrages = data.get('ouvrages', [])
|
|
if not isinstance(ouvrages, list) or not ouvrages:
|
|
return 'compliant'
|
|
|
|
has_damaged = False
|
|
has_non_compliant = False
|
|
|
|
for ouv in ouvrages:
|
|
if not isinstance(ouv, dict):
|
|
continue
|
|
device_type = (ouv.get('device_type') or 'evc').lower()
|
|
q = ouv.get('questions') or {}
|
|
if not isinstance(q, dict):
|
|
continue
|
|
|
|
def is_val(key, *expected):
|
|
v = q.get(key)
|
|
if isinstance(v, bool):
|
|
v_str = 'true' if v else 'false'
|
|
elif v is None:
|
|
v_str = 'null'
|
|
else:
|
|
v_str = str(v).strip().lower()
|
|
return v_str in [str(e).strip().lower() for e in expected]
|
|
|
|
if device_type == 'evc':
|
|
# 1. Point d'entrée
|
|
if is_val('evc_pe_ok', 'false', 'non'):
|
|
has_non_compliant = True
|
|
|
|
# 2. Empierrement
|
|
if is_val('evc_empierrement_present', True, 'true'):
|
|
if is_val('evc_empierrement_ok', 'false', 'non'):
|
|
has_non_compliant = True
|
|
if is_val('evc_empierrement_emporte', True, 'true'):
|
|
has_damaged = True
|
|
|
|
# 3. Végétation
|
|
if is_val('evc_ev_veg_etat', 'mauvais'):
|
|
has_non_compliant = True
|
|
if is_val('evc_ev_plantes_mortes', 'plus_10', '+ de 10%'):
|
|
has_damaged = True
|
|
|
|
# 4. Chenaux d'érosion
|
|
if is_val('evc_ev_chenaux_erosion', 'true', 'oui', True):
|
|
has_non_compliant = True
|
|
|
|
# 5. Flaques d'eau
|
|
if is_val('evc_ev_flaques', 'true', 'oui', True):
|
|
has_non_compliant = True
|
|
|
|
# 6. Sédiments
|
|
if is_val('evc_ev_sediments', 'true', 'oui', True):
|
|
has_non_compliant = True
|
|
if is_val('evc_ev_croute_sediments', 'true', 'oui', True):
|
|
has_damaged = True
|
|
|
|
# 7. Espèces envahissantes
|
|
if is_val('evc_ev_envahissantes', 'true', 'oui', True):
|
|
has_non_compliant = True
|
|
|
|
# 8. Surverse
|
|
if is_val('evc_surverse_presente', True, 'true'):
|
|
if is_val('evc_sv_grille_propre', 'false', 'non'):
|
|
has_non_compliant = True
|
|
if is_val('evc_sv_decantation_vide', 'false', 'non'):
|
|
has_non_compliant = True
|
|
if is_val('evc_sv_haute', 'false', 'non'):
|
|
has_damaged = True
|
|
|
|
elif device_type == 'oed':
|
|
# 1. Revêtement général
|
|
if is_val('oed_revetement_ok', 'false', 'non'):
|
|
has_non_compliant = True
|
|
|
|
# 2. Bouche d'injection
|
|
if is_val('oed_injection_presente', True, 'true'):
|
|
if is_val('oed_bi_niveaux_ok', 'false', 'non'):
|
|
has_damaged = True
|
|
if is_val('oed_bi_grille_propre', 'false', 'non'):
|
|
has_non_compliant = True
|
|
if is_val('oed_bi_decantation_vide', 'false', 'non'):
|
|
has_damaged = True
|
|
if is_val('oed_bi_drain_colmate', 'true', 'oui', True):
|
|
has_damaged = True
|
|
|
|
# 3. Structure enherbée / minérale
|
|
if is_val('oed_rhm_presente', True, 'true'):
|
|
if is_val('oed_rhm_stagne', 'true', 'oui', True):
|
|
has_non_compliant = True
|
|
if is_val('oed_rhm_joints_ok', 'false', 'non'):
|
|
has_non_compliant = True
|
|
if is_val('oed_rhm_joints_fines', 'true', 'oui', True):
|
|
has_damaged = True
|
|
if is_val('oed_rhm_joints_remplissage', 'true', 'oui', True):
|
|
has_non_compliant = True
|
|
|
|
# 4. Modulaire poreux / continu poreux
|
|
if is_val('oed_rp_presente', True, 'true'):
|
|
if is_val('oed_rp_stagne', 'true', 'oui', True):
|
|
has_non_compliant = True
|
|
if is_val('oed_rp_depots', 'true', 'oui', True):
|
|
has_non_compliant = True
|
|
|
|
elif device_type == 'ta':
|
|
# 1. Point d'entrée
|
|
if is_val('ta_pe_ok', 'false', 'non'):
|
|
has_non_compliant = True
|
|
if is_val('ta_pe_conduits_ok', 'false', 'non'):
|
|
has_damaged = True
|
|
if is_val('ta_pe_decantation_vide', 'false', 'non'):
|
|
has_non_compliant = True
|
|
|
|
# 2. Bassin d'orage
|
|
if is_val('ta_bo_vegetation_etat', 'mauvais'):
|
|
has_non_compliant = True
|
|
if is_val('ta_bo_plantes_mortes', 'plus_10', '+ de 10%'):
|
|
has_damaged = True
|
|
if is_val('ta_bo_chenaux', 'true', 'oui', True):
|
|
has_non_compliant = True
|
|
if is_val('ta_sediments', 'true', 'oui', True):
|
|
has_non_compliant = True
|
|
if is_val('ta_sediments_rempli', 'true', 'oui', True):
|
|
has_damaged = True
|
|
if is_val('ta_bo_envahissantes', 'true', 'oui', True):
|
|
has_non_compliant = True
|
|
|
|
if has_damaged:
|
|
return 'damaged'
|
|
if has_non_compliant:
|
|
return 'non_compliant'
|
|
return 'compliant'
|
|
|
|
|
|
class SignPanelInspection(Inspection):
|
|
"""
|
|
Spécialisation de l'inspection pour les panneaux de signalisation routière.
|
|
"""
|
|
CLEANLINESS_CHOICES = (
|
|
('clean', _("Propre / Bon état")),
|
|
('dirty', _("Sale / Mousse / Nettoyage requis")),
|
|
('vegetation', _("Masqué par la végétation (Élagage requis)")),
|
|
('vandalized', _("Tagué / Vandalisé / Déformé")),
|
|
('missing', _("Panneau absent / Support vide")),
|
|
)
|
|
verified_signpanel_type = models.ForeignKey(
|
|
'assets.SignPanelType',
|
|
on_delete=models.SET_NULL,
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("Type de panneau constaté")
|
|
)
|
|
verified_signpanel_text = models.CharField(
|
|
max_length=1000,
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("Texte / Mention constatée")
|
|
)
|
|
verified_cleanliness = models.CharField(
|
|
max_length=50,
|
|
choices=CLEANLINESS_CHOICES,
|
|
default='clean',
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("Niveau de propreté / Visibilité")
|
|
)
|
|
ai_detected_code = models.CharField(
|
|
max_length=50,
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("Code détecté par IA")
|
|
)
|
|
ai_confidence = models.FloatField(
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("Confiance de détection IA")
|
|
)
|
|
apply_sign_updates_to_asset = models.BooleanField(
|
|
default=True,
|
|
verbose_name=_("Mettre à jour le type et le texte de l'asset")
|
|
)
|
|
|
|
class Meta:
|
|
verbose_name = _("Inspection de panneau de signalisation")
|
|
verbose_name_plural = _("Inspections de panneaux de signalisation")
|
|
|
|
def save(self, *args, **kwargs):
|
|
super().save(*args, **kwargs)
|
|
if self.apply_sign_updates_to_asset and self.asset:
|
|
target_asset = self.asset
|
|
modified = False
|
|
if not self.is_cancelled:
|
|
if self.verified_signpanel_type:
|
|
target_asset.signpanel_type = self.verified_signpanel_type
|
|
modified = True
|
|
if self.verified_signpanel_text is not None:
|
|
target_asset.signpanel_text = self.verified_signpanel_text
|
|
modified = True
|
|
if self.verified_cleanliness in ('dirty', 'vegetation', 'vandalized', 'missing') or self.result_status in ('non_compliant', 'damaged', 'missing'):
|
|
target_asset.is_compliant = False
|
|
modified = True
|
|
elif self.verified_cleanliness == 'clean' and self.result_status == 'compliant':
|
|
target_asset.is_compliant = True
|
|
modified = True
|
|
if self.verified_cleanliness == 'missing' or self.result_status == 'missing':
|
|
target_asset.status = 'inactive'
|
|
modified = True
|
|
if modified:
|
|
target_asset.save()
|
|
|
|
|
|
class HotBoxInspection(Inspection):
|
|
"""
|
|
Spécialisation de l'inspection pour les réparations d'enrobé à chaud (Hot Box).
|
|
Formulaire visuel simplifié adapté aux observateurs non experts :
|
|
- Planéité (Arasement)
|
|
- Joints (Raccordement)
|
|
- Fissuration
|
|
- Évacuation des eaux pluviales
|
|
"""
|
|
JOINT_LEVEL_CHOICES = (
|
|
('flush', _("Bien à niveau / Arasé")),
|
|
('slight_bump_hollow', _("Léger décalage / Creux ou bosse")),
|
|
('step', _("Ressaut important (> 1-2 cm)")),
|
|
)
|
|
joint_level = models.CharField(
|
|
max_length=50,
|
|
choices=JOINT_LEVEL_CHOICES,
|
|
default=None,
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("Planéité / Arasement")
|
|
)
|
|
|
|
JOINT_SEALING_CHOICES = (
|
|
('good', _("Joints bien fermés / scellés")),
|
|
('cracked', _("Joints fissurés ou abîmés")),
|
|
('open', _("Joints ouverts ou décollés")),
|
|
)
|
|
joint_sealing = models.CharField(
|
|
max_length=50,
|
|
choices=JOINT_SEALING_CHOICES,
|
|
default=None,
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("Joints de raccordement")
|
|
)
|
|
|
|
CRACKING_CHOICES = (
|
|
('none', _("Aucune fissure")),
|
|
('minor', _("Petites fissures / Faïençage")),
|
|
('major', _("Grosses fissures / Fissures ouvertes")),
|
|
)
|
|
cracking = models.CharField(
|
|
max_length=50,
|
|
choices=CRACKING_CHOICES,
|
|
default=None,
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("Fissuration")
|
|
)
|
|
|
|
WATER_DRAINAGE_CHOICES = (
|
|
('good', _("Bon écoulement / Pas de flaque")),
|
|
('stagnation', _("Eau stagnante / Flaque d'eau")),
|
|
('not_checked', _("Non vérifié")),
|
|
)
|
|
water_drainage = models.CharField(
|
|
max_length=50,
|
|
choices=WATER_DRAINAGE_CHOICES,
|
|
default=None,
|
|
null=True,
|
|
blank=True,
|
|
verbose_name=_("Évacuation des eaux pluviales")
|
|
)
|
|
|
|
class Meta:
|
|
verbose_name = _("Inspection de réparation Hot Box")
|
|
verbose_name_plural = _("Inspections de réparations Hot Box")
|
|
|
|
def compute_status(self):
|
|
"""
|
|
Calcule automatiquement le statut de conformité ('compliant', 'non_compliant', 'damaged')
|
|
selon la sévérité des constats visuels, ou None si l'évaluation n'est pas complète.
|
|
"""
|
|
if self.joint_level == 'step' or self.joint_sealing == 'open' or self.cracking == 'major':
|
|
return 'damaged'
|
|
if (
|
|
self.joint_level == 'slight_bump_hollow' or
|
|
self.joint_sealing == 'cracked' or
|
|
self.cracking == 'minor' or
|
|
self.water_drainage == 'stagnation'
|
|
):
|
|
return 'non_compliant'
|
|
if self.joint_level and self.joint_sealing and self.cracking and self.water_drainage:
|
|
return 'compliant'
|
|
return None
|
|
|
|
@property
|
|
def is_fully_evaluated(self):
|
|
"""Indique si les 4 questions de l'inspection ont été renseignées."""
|
|
return bool(self.joint_level and self.joint_sealing and self.cracking and self.water_drainage)
|
|
|
|
|
|
def inspection_document_upload_path(instance, filename):
|
|
now = timezone.now()
|
|
year = now.strftime("%Y")
|
|
month = now.strftime("%m")
|
|
return f"inspections/documents/{year}/{month}/{filename}"
|
|
|
|
|
|
class InspectionDocument(models.Model):
|
|
"""
|
|
Document / Photo / Pièce jointe associée à une inspection.
|
|
"""
|
|
inspection = models.ForeignKey(
|
|
Inspection,
|
|
related_name='documents',
|
|
on_delete=models.CASCADE,
|
|
verbose_name=_("Inspection")
|
|
)
|
|
field_key = models.CharField(
|
|
max_length=100,
|
|
blank=True,
|
|
default='',
|
|
verbose_name=_("Clé de question / Média"),
|
|
help_text=_("Identifie la question ou l'ouvrage auquel ce média est rattaché (vide = général).")
|
|
)
|
|
file = models.FileField(
|
|
upload_to=inspection_document_upload_path,
|
|
storage=private_storage,
|
|
verbose_name=_("Fichier / Photo / Document")
|
|
)
|
|
created_at = models.DateTimeField(auto_now_add=True)
|
|
|
|
class Meta:
|
|
verbose_name = _("Document / Photo d'inspection")
|
|
verbose_name_plural = _("Documents / Photos d'inspection")
|
|
|
|
@property
|
|
def is_image(self):
|
|
if not self.file or not self.file.name:
|
|
return False
|
|
ext = self.file.name.split('.')[-1].lower()
|
|
return ext in ['jpg', 'jpeg', 'png', 'gif', 'webp', 'heic', 'bmp']
|
|
|
|
@property
|
|
def filename(self):
|
|
import os
|
|
return os.path.basename(self.file.name) if self.file else ""
|
|
|
|
@property
|
|
def file_url(self):
|
|
if not self.file:
|
|
return ""
|
|
from common.private_files.helpers import build_private_url
|
|
return build_private_url('inspections', 'InspectionDocument', self.pk, 'file')
|
|
|
|
def __str__(self):
|
|
return f"Document {self.filename} (Inspection #{self.inspection_id})"
|
|
|