2140 lines
88 KiB
Python
2140 lines
88 KiB
Python
from django.contrib.gis.db import models
|
||
from django.contrib.contenttypes.fields import GenericForeignKey
|
||
from django.contrib.contenttypes.models import ContentType
|
||
from django.conf import settings
|
||
from django.core.exceptions import ValidationError
|
||
from django.utils.translation import get_language, gettext_lazy as _
|
||
from django.utils import timezone
|
||
from django.db.models import Sum, F, Q
|
||
from django.db.models.functions import Coalesce
|
||
|
||
import os
|
||
from datetime import datetime, timedelta
|
||
from decimal import Decimal
|
||
|
||
from common.models import Thematic, Document
|
||
from common.storages import PrivateMediaStorage
|
||
|
||
private_storage = PrivateMediaStorage()
|
||
from assets.models import AssetCategory
|
||
from contracts.models import Contract, ContractPost, ContractOrder, Company, CompanyTeam, CompanyMember
|
||
|
||
STATUS_CHOICES = {
|
||
'in_preparation': _('En préparation'), # status_order 1
|
||
'to_be_approved': _('A approuver'), # status_order 2
|
||
'to_be_planned': _('A planifier'), # status_order 5
|
||
'to_be_processed': _('A traiter'), # status_order 11
|
||
'assigned': _('Pris en charge'), # status_order 15
|
||
'in_progress': _('Démarré'), # status_order 21
|
||
'on_pause': _('En pause'), # status_order 22
|
||
'finished': _('Terminé'), # status_order 25
|
||
'processed': _('Traité'), # status_order 31
|
||
'to_be_corrected': _('A corriger'), # status_order 41
|
||
'corrected': _('Corrigé'), # status_order 45
|
||
'validated': _('Validé'), # status_order 51
|
||
'invoiced': _('Comptabilisé'), # status_order 61
|
||
'closed': _('Finalisé'), # status_order 71
|
||
# 'archived': _('Archivé'), # status_order 100
|
||
'canceled': _('Annulé'), # status_order 200
|
||
}
|
||
|
||
STATUS_ORDERS = {
|
||
'in_preparation': 1,
|
||
'to_be_approved': 2,
|
||
'to_be_planned': 5,
|
||
'to_be_processed': 11,
|
||
'assigned': 15,
|
||
'in_progress': 21,
|
||
'on_pause': 22,
|
||
'finished': 25,
|
||
'processed': 31,
|
||
'to_be_corrected': 41,
|
||
'corrected': 45,
|
||
'validated': 51,
|
||
'invoiced': 61,
|
||
'closed': 71,
|
||
#'archived': 100,
|
||
'canceled': 200,
|
||
}
|
||
|
||
# Couleurs canoniques des statuts (hex). Source de vérité partagée avec status_colors.js.
|
||
STATUS_COLORS = {
|
||
'in_preparation': '#757575',
|
||
'to_be_approved': '#ffb74d',
|
||
'to_be_planned': '#ff8a80',
|
||
'to_be_processed': '#f4511e',
|
||
'assigned': '#63b3ed',
|
||
'in_progress': '#3182ce',
|
||
'on_pause': '#6b226e',
|
||
'finished': '#26a69a',
|
||
'processed': '#9ae9e2',
|
||
'to_be_corrected': '#9e87f6',
|
||
'corrected': '#81c784',
|
||
'validated': '#388e3c',
|
||
'invoiced': '#738f7a',
|
||
'closed': '#09425f',
|
||
'canceled': '#cfd8dc',
|
||
}
|
||
|
||
# Statuts dont le texte doit être sombre (fond clair).
|
||
STATUS_COLORS_DARK_TEXT = {'to_be_approved', 'to_be_planned', 'assigned', 'processed', 'to_be_corrected', 'corrected', 'canceled'}
|
||
|
||
|
||
PRIORITY_CHOICES = {
|
||
'1': _('Urgent'),
|
||
'2': _('Elevé'),
|
||
'3': _('Normal'),
|
||
'4': _('Bas'),
|
||
}
|
||
|
||
MAINTAIN_CHOICES = [
|
||
('corrective', _('Maintenance corrective')),
|
||
('preventive', _('Maintenance préventive')),
|
||
('ameliorative', _('Maintenance améliorative')),
|
||
]
|
||
|
||
TYPE_CHOICES = [
|
||
('fixing', _('Dépannage')),
|
||
('repair', _('Réparation')),
|
||
('repair_post_preventive', _('Réparation post préventif')),
|
||
('replace', _('Remplacement')),
|
||
('care', _('Entretien')),
|
||
('control', _('Contrôle')),
|
||
('inspection', _('Inspection')),
|
||
('build', _('Construction')),
|
||
('collection', _('Ramassage')),
|
||
('reprogramming', _('Reprogrammation')),
|
||
]
|
||
|
||
|
||
def get_type_choices(thematic=None):
|
||
"""
|
||
Renvoie la liste des choix de type d'intervention selon la thématique.
|
||
Le type 'reprogramming' (Reprogrammation) n'est disponible que pour la thématique 'trafficlights'.
|
||
"""
|
||
thematic_code = None
|
||
if thematic:
|
||
if isinstance(thematic, str):
|
||
thematic_code = thematic
|
||
elif hasattr(thematic, 'code'):
|
||
thematic_code = thematic.code
|
||
|
||
if thematic_code == 'trafficlights':
|
||
return list(TYPE_CHOICES)
|
||
return [c for c in TYPE_CHOICES if c[0] != 'reprogramming']
|
||
|
||
ORIGIN_TYPE_CHOICES = [
|
||
('damage', _('Avarie')),
|
||
('defect', _('Défaut/Panne')),
|
||
('demand', _('Demande'))
|
||
]
|
||
|
||
PAUSE_REASON_CHOICES = (
|
||
("reschedule", _("Intervention à replanifier")),
|
||
("order_material", _("Matériel à commander")),
|
||
("treated_in_next_maintenance", _("Traité dans le prochain entretien")),
|
||
("waiting_for_material_delivery", _("En attente livraison de matériel")),
|
||
("waiting_for_supplier", _("En attente sous-traitant")),
|
||
("offer_by_provider", _("Offre à faire par le prestataire")),
|
||
("waiting_for_validation", _("En attente de validation")),
|
||
("under_framework_contract", _("Sous garantie marché transversal")),
|
||
("other", _("Autre")),
|
||
)
|
||
|
||
CANCELLATION_TYPE_CHOICES = (
|
||
("duplicate", _("Doublon")),
|
||
("to_redirect", _("A rediriger")),
|
||
("not_our_assets", _("Pas sous notre gestion")),
|
||
("non_relevant", _("Non-relevant")),
|
||
("other", _("Autre")),
|
||
)
|
||
|
||
|
||
class SourceCategory(models.Model):
|
||
name_fr = models.CharField(max_length=200)
|
||
name_nl = models.CharField(max_length=200)
|
||
description = models.TextField(null=True,blank=True)
|
||
has_reference = models.BooleanField(default=False)
|
||
is_active = models.BooleanField(default=True)
|
||
|
||
def getName(self):
|
||
lang = get_language()
|
||
if lang == 'nl':
|
||
return self.name_nl or self.name_fr
|
||
return self.name_fr
|
||
|
||
def __str__(self):
|
||
lang = get_language()
|
||
if lang == 'nl':
|
||
return self.name_nl or self.name_fr
|
||
return self.name_fr
|
||
|
||
class Meta:
|
||
ordering = ["name_fr"]
|
||
|
||
|
||
SYMPTOM_MAINTENANCE_TYPE_CHOICES = {
|
||
'infra': _('TL - Infrastructure'),
|
||
'controller': _('TL - Contrôleur de traffic'),
|
||
'electromeca': _('TU - Electromécanique'),
|
||
'civil_engineering': _('TU - Génie civil'),
|
||
'equipments': _('ITS - Équipement'),
|
||
'clean': _('Clean - Nettoyage'),
|
||
'collection': _('Clean - Ramassage'),
|
||
'other': _('Autre'),
|
||
}
|
||
|
||
|
||
class Symptom(models.Model):
|
||
code = models.CharField(max_length=50, unique=True, null=True, blank=True, help_text=_("Code unique optionnel pour identifier le symptôme (ex: pour les raccourcis)."))
|
||
name_fr = models.CharField(max_length=255)
|
||
name_nl = models.CharField(max_length=255)
|
||
thematic = models.ForeignKey(Thematic, on_delete=models.PROTECT)
|
||
asset_categories = models.ManyToManyField(
|
||
AssetCategory,
|
||
blank=True,
|
||
related_name='symptoms',
|
||
)
|
||
priority = models.CharField(choices=PRIORITY_CHOICES, default='3')
|
||
check_priority = models.BooleanField(default=False, help_text=_("Indique si la priorité doit être vérifiée."))
|
||
origin_type = models.CharField(choices=ORIGIN_TYPE_CHOICES, default='damage', null=True, blank=True)
|
||
provider = models.ForeignKey(Company, on_delete=models.SET_NULL, null=True, blank=True)
|
||
contract = models.ForeignKey(Contract, on_delete=models.SET_NULL, null=True, blank=True)
|
||
maintenance_type = models.CharField(max_length=50, choices=SYMPTOM_MAINTENANCE_TYPE_CHOICES, null=True, blank=True, help_text=_("Type de maintenance concerné (ex: 'controller' pour les contrôleurs de traffic)."))
|
||
require_equipment_selection = models.BooleanField(
|
||
default=False,
|
||
verbose_name=_("Équipement obligatoire"),
|
||
help_text=_("Si coché, une intervention avec ce symptôme doit obligatoirement être liée à un équipement."),
|
||
)
|
||
teams = models.ManyToManyField(
|
||
CompanyTeam,
|
||
blank=True,
|
||
related_name='symptoms',
|
||
verbose_name=_("Équipes désignées"),
|
||
help_text=_("Équipes désignées pour prendre en charge les observations liées à ce symptôme."),
|
||
)
|
||
order_number = models.IntegerField(null=True, blank=True)
|
||
is_active = models.BooleanField(default=True)
|
||
recommended_next_status = models.CharField(choices=STATUS_CHOICES, null=True, blank=True, max_length=50, default='to_be_processed')
|
||
auto_create_intervention = models.BooleanField(
|
||
default=False,
|
||
verbose_name=_("Création automatique d'interventions"),
|
||
help_text=_("Si coché, une intervention est automatiquement créée lorsque l'observation passe au statut 'À traiter'.")
|
||
)
|
||
auto_create_intervention_status = models.CharField(
|
||
max_length=50,
|
||
choices=STATUS_CHOICES,
|
||
default='to_be_processed',
|
||
null=True,
|
||
blank=True,
|
||
verbose_name=_("Statut de l'intervention auto-créée"),
|
||
help_text=_("Statut dans lequel l'intervention automatique doit être créée.")
|
||
)
|
||
internal_notification_emails = models.CharField(blank=True, null=True, max_length=1000, help_text=_("Liste d'emails pour les notifications internes."))
|
||
|
||
def __str__(self):
|
||
lang = get_language()
|
||
if lang == 'nl':
|
||
return self.name_nl or self.name_fr
|
||
return self.name_fr
|
||
|
||
def get_name(self):
|
||
lang = get_language()
|
||
if lang == 'nl':
|
||
return self.name_nl or self.name_fr
|
||
return self.name_fr
|
||
|
||
class Meta:
|
||
ordering = ["thematic", "order_number"]
|
||
|
||
|
||
class SymptomRecommendation(models.Model):
|
||
"""
|
||
Modèle pour les recommandations associées aux symptômes.
|
||
|
||
Les textes peuvent contenir des variables entre doubles crochets qui seront
|
||
remplacées par les valeurs du contrat associé à l'intervention :
|
||
- [[phone_number]] : Numéro de téléphone de contact du contrat
|
||
- [[contract_number]] : Numéro du contrat
|
||
- [[company_name]] : Nom de la société/prestataire
|
||
"""
|
||
symptom = models.ForeignKey(Symptom, on_delete=models.CASCADE, related_name='recommendations')
|
||
icon = models.CharField(max_length=255, null=True, blank=True)
|
||
text_fr = models.TextField()
|
||
text_nl = models.TextField()
|
||
order_number = models.PositiveIntegerField(null=True, blank=True)
|
||
document = models.FileField(upload_to='symptom_recommendations/', storage=private_storage, null=True, blank=True)
|
||
confirm_call = models.BooleanField(default=False)
|
||
|
||
# Variables disponibles pour le remplacement dans les textes
|
||
TEMPLATE_VARIABLES = {
|
||
'phone_number': lambda contract: contract.contact_phone if contract else None,
|
||
'contract_number': lambda contract: contract.contract_number if contract else None,
|
||
'company_name': lambda contract: contract.company.name if contract and contract.company else None,
|
||
'company_phone': lambda contract: contract.company.phone_number if contract and contract.company else None,
|
||
}
|
||
|
||
def get_rendered_text(self, contract=None, lang=None):
|
||
"""
|
||
Retourne le texte de la recommandation avec les variables remplacées.
|
||
|
||
Les variables entre doubles crochets [[variable_name]] sont remplacées
|
||
par les valeurs correspondantes du contrat.
|
||
|
||
Args:
|
||
contract: Instance du contrat pour récupérer les valeurs
|
||
lang: Code langue ('fr' ou 'nl'). Si None, utilise la langue courante.
|
||
|
||
Returns:
|
||
str: Texte avec les variables remplacées
|
||
"""
|
||
import re
|
||
|
||
if lang is None:
|
||
lang = get_language()
|
||
|
||
text = self.text_nl if lang == 'nl' else self.text_fr
|
||
|
||
def replace_variable(match):
|
||
var_name = match.group(1)
|
||
if var_name in self.TEMPLATE_VARIABLES:
|
||
value = self.TEMPLATE_VARIABLES[var_name](contract)
|
||
return value if value else f'[[{var_name}]]'
|
||
return match.group(0)
|
||
|
||
# Remplace les variables [[variable_name]] par leurs valeurs
|
||
rendered_text = re.sub(r'\[\[([\w_]+)\]\]', replace_variable, text)
|
||
|
||
return rendered_text
|
||
|
||
def __str__(self):
|
||
lang = get_language()
|
||
return self.text_nl if lang == 'nl' else self.text_fr
|
||
|
||
class Meta:
|
||
ordering = ['order_number']
|
||
|
||
|
||
class ReflexSheet(models.Model):
|
||
title_fr = models.CharField(max_length=255)
|
||
title_nl = models.CharField(max_length=255)
|
||
symptoms = models.ManyToManyField(Symptom, related_name='reflex_sheets')
|
||
contact_persons = models.ManyToManyField(settings.AUTH_USER_MODEL, blank=True, related_name='reflex_sheets')
|
||
comment_fr = models.TextField(blank=True, null=True)
|
||
comment_nl = models.TextField(blank=True, null=True)
|
||
asset_category = models.ForeignKey(AssetCategory, on_delete=models.SET_NULL, null=True, blank=True)
|
||
created_at = models.DateTimeField(auto_now_add=True)
|
||
updated_at = models.DateTimeField(auto_now=True)
|
||
is_active = models.BooleanField(default=True)
|
||
thematic = models.ForeignKey(Thematic, on_delete=models.PROTECT)
|
||
|
||
def __str__(self):
|
||
lang = get_language()
|
||
return self.title_nl if lang == 'nl' else self.title_fr
|
||
|
||
class Meta:
|
||
ordering = ['thematic', 'title_fr']
|
||
|
||
|
||
def intervention_sheet_upload_path(instance, filename):
|
||
# On récupère l'année et le mois actuels
|
||
now = timezone.now()
|
||
year = now.strftime("%Y")
|
||
month = now.strftime("%m")
|
||
|
||
# On construit le chemin avec l'ID de l'intervention (attention : doit être déjà enregistrée)
|
||
intervention_id = instance.id
|
||
return f"intervention_documents/{year}/{month}/{intervention_id}/{filename}"
|
||
|
||
class Intervention(models.Model):
|
||
code = models.CharField(max_length=50, unique=True, blank=True, db_index=True)
|
||
title = models.CharField(max_length=200)
|
||
lon = models.FloatField(null=True,blank=True)
|
||
lat = models.FloatField(null=True,blank=True)
|
||
address = models.CharField(max_length=500, null=True, blank=True)
|
||
location_code = models.CharField(max_length=500, null=True, blank=True) # Location code
|
||
status = models.CharField(choices=STATUS_CHOICES, default='in_preparation')
|
||
status_order = models.IntegerField(default=1) # Order of the status in the workflow
|
||
pause_reason = models.CharField(choices=PAUSE_REASON_CHOICES, max_length=50, null=True, blank=True)
|
||
pause_reason_other = models.CharField(max_length=255, null=True, blank=True)
|
||
cancellation_type = models.CharField(choices=CANCELLATION_TYPE_CHOICES, max_length=50, null=True, blank=True)
|
||
priority = models.CharField(choices=PRIORITY_CHOICES, default='3')
|
||
description = models.TextField(null=True,blank=True)
|
||
maintain_type = models.CharField(choices=MAINTAIN_CHOICES, default='corrective')
|
||
type = models.CharField(choices=TYPE_CHOICES,null=True,blank=True)
|
||
origin_type = models.CharField(choices=ORIGIN_TYPE_CHOICES, default='damage')
|
||
begin_time = models.DateTimeField(null=True,blank=True)
|
||
end_time = models.DateTimeField(null=True,blank=True)
|
||
planned_begin_time = models.DateTimeField(null=True, blank=True)
|
||
planned_end_time = models.DateTimeField(null=True, blank=True)
|
||
expected_begin_time = models.DateTimeField(null=True, blank=True)
|
||
expected_end_time = models.DateTimeField(null=True, blank=True)
|
||
source_category = models.ForeignKey(SourceCategory, on_delete=models.SET_NULL, null=True, blank=True)
|
||
source_detail = models.CharField(null=True,blank=True)
|
||
source_ref = models.CharField(null=True,blank=True)
|
||
pmo_number = models.IntegerField(null=True, blank=True, verbose_name=_("Numéro PMO"))
|
||
asset_category = models.ForeignKey(AssetCategory, on_delete=models.SET_NULL, null=True, blank=True)
|
||
symptom = models.ForeignKey(Symptom, on_delete=models.SET_NULL, null=True, blank=True)
|
||
created_by = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.PROTECT, related_name="created_interventions", null=True, blank=True)
|
||
creation_time = models.DateTimeField(auto_now_add=True)
|
||
last_modified_by = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.PROTECT, related_name="modified_interventions", null=True, blank=True)
|
||
last_modified_time = models.DateTimeField(auto_now=True, null=True, blank=True)
|
||
transmission_time = models.DateTimeField(null=True,blank=True)
|
||
init_call_confirmed_time = models.DateTimeField(null=True,blank=True)
|
||
init_call_received = models.BooleanField(default=False)
|
||
init_description = models.TextField(null=True,blank=True)
|
||
contract = models.ForeignKey(Contract, on_delete=models.SET_NULL, null=True, blank=True, related_name="interventions")
|
||
intervention_manager = models.ForeignKey(
|
||
settings.AUTH_USER_MODEL,
|
||
on_delete=models.SET_NULL,
|
||
null=True,
|
||
blank=True,
|
||
related_name="managed_interventions",
|
||
verbose_name=_("Gestionnaire d'intervention"),
|
||
help_text=_("Utilisateur qui approuve cette intervention (spécifique aux contrats concernés)."),
|
||
)
|
||
order = models.ForeignKey(ContractOrder, on_delete=models.SET_NULL, null=True, blank=True, related_name="interventions")
|
||
assigned_provider = models.ForeignKey(Company, on_delete=models.SET_NULL, null=True, blank=True, related_name="assigned_interventions")
|
||
assigned_team = models.ForeignKey(CompanyTeam, on_delete=models.SET_NULL, null=True, blank=True, related_name="assigned_interventions")
|
||
assigned_member = models.ForeignKey(CompanyMember, on_delete=models.SET_NULL, null=True, blank=True, related_name="assigned_interventions")
|
||
osiris = models.IntegerField(null=True, blank=True)
|
||
linked_folder = models.CharField(max_length=1024, null=True, blank=True)
|
||
intervention_sheet = models.FileField(upload_to=intervention_sheet_upload_path, storage=private_storage, null=True, blank=True)
|
||
thematic = models.ForeignKey(Thematic, on_delete=models.SET_NULL, null=True, blank=True)
|
||
geom = models.MultiPolygonField(srid=3812, null=True,blank=True)
|
||
# Validation par périodes
|
||
period_validation_enabled = models.BooleanField(default=False)
|
||
period_number = models.PositiveIntegerField(default=1)
|
||
period_is_final = models.BooleanField(default=False)
|
||
equipment_operation_templates = models.ManyToManyField(
|
||
'OperationTemplate',
|
||
blank=True,
|
||
related_name='interventions_equipment',
|
||
)
|
||
guided_operation_template = models.ForeignKey(
|
||
'GuidedOperationTemplate',
|
||
on_delete=models.SET_NULL,
|
||
null=True,
|
||
blank=True,
|
||
related_name='interventions',
|
||
verbose_name=_("Template d'opérations guidées"),
|
||
help_text=_("Assigné automatiquement lors du choix d'un template d'opérations avec un template guidé associé.")
|
||
)
|
||
# Indique si l'intervention a été créée par un prestataire (external_manager)
|
||
created_by_provider = models.BooleanField(
|
||
default=False,
|
||
verbose_name="Créée par le prestataire",
|
||
help_text="Indique que cette intervention a été créée par un prestataire externe."
|
||
)
|
||
# Indique si le traitement est réalisé via une autre intervention
|
||
processed_via_other_intervention = models.BooleanField(
|
||
default=False,
|
||
verbose_name="Traitement via une autre intervention",
|
||
help_text="Indique que le traitement de cette intervention est réalisé via une autre intervention."
|
||
)
|
||
# Raison de la correction demandée (quand le statut passe à 'à corriger')
|
||
reason_for_correction = models.TextField(
|
||
null=True,
|
||
blank=True,
|
||
verbose_name="Raison de la correction",
|
||
help_text="Raison pour laquelle une correction est demandée."
|
||
)
|
||
# Détails de ce qui a été corrigé (quand le statut passe à 'corrigé')
|
||
correction_details = models.TextField(
|
||
null=True,
|
||
blank=True,
|
||
verbose_name="Détails de la correction",
|
||
help_text="Détails de ce qui a été corrigé."
|
||
)
|
||
# Vérification : l'intervention a été vérifiée par un utilisateur ayant le droit can_check_interventions
|
||
is_checked = models.BooleanField(
|
||
default=False,
|
||
verbose_name=_("Vérifiée"),
|
||
help_text=_("Indique si l'intervention a été vérifiée par un utilisateur habilité.")
|
||
)
|
||
# Blocage contractuel / budgétaire
|
||
is_blocked_due_to_contract = models.BooleanField(
|
||
default=False,
|
||
db_index=True,
|
||
verbose_name=_("En attente (marché bloqué)"),
|
||
help_text=_("Indique que cette intervention est en attente car le marché associé est temporairement bloqué pour des raisons budgétaires.")
|
||
)
|
||
# Résultat du contrôle de suivi
|
||
FOLLOW_UP_RESULT_CHOICES = [
|
||
('ok', _('OK')),
|
||
('attention', _('Points d’attention')),
|
||
('not_ok', _('Not OK')),
|
||
]
|
||
follow_up_performed = models.CharField(
|
||
max_length=20,
|
||
choices=FOLLOW_UP_RESULT_CHOICES,
|
||
null=True,
|
||
blank=True,
|
||
default=None,
|
||
verbose_name=_("Résultat du contrôle"),
|
||
help_text=_("Résultat du contrôle / rapport de suivi.")
|
||
)
|
||
|
||
def save(self, *args, **kwargs):
|
||
is_new = self._state.adding # True uniquement lors de la création
|
||
status_changed_to_closed = False
|
||
status_changed_to_validated = False
|
||
old_status = None
|
||
if not is_new:
|
||
old_status = Intervention.objects.filter(pk=self.pk).values_list('status', flat=True).first()
|
||
if old_status and old_status != 'closed' and self.status == 'closed':
|
||
status_changed_to_closed = True
|
||
if old_status and old_status != 'validated' and self.status == 'validated':
|
||
status_changed_to_validated = True
|
||
else:
|
||
if self.status == 'validated':
|
||
status_changed_to_validated = True
|
||
|
||
if not self.code: # Generate code only if not defined
|
||
next_number = InterventionSequence.get_next_number()
|
||
self.code = f"I{str(next_number).zfill(6)}" # Prefix "I" followed by the number
|
||
|
||
# Gestion de la sécurité sur contrat bloqué
|
||
if self.contract and getattr(self.contract, 'is_blocked', False):
|
||
if is_new and self.status in ('to_be_processed', 'assigned', 'in_progress'):
|
||
self.status = 'in_preparation'
|
||
if self.status in self.BLOCKED_CONTRACT_DISPLAY_STATUSES:
|
||
self.is_blocked_due_to_contract = True
|
||
else:
|
||
self.is_blocked_due_to_contract = False
|
||
else:
|
||
self.is_blocked_due_to_contract = False
|
||
|
||
if self.status in STATUS_ORDERS:
|
||
self.status_order = STATUS_ORDERS[self.status]
|
||
if is_new and self.created_by_id:
|
||
from common.models import UserConfig # import local pour éviter les imports circulaires
|
||
user_config = UserConfig.objects.filter(user_id=self.created_by_id).first()
|
||
if user_config and not user_config.is_intern:
|
||
self.created_by_provider = True
|
||
super().save(*args, **kwargs)
|
||
|
||
# Sync daily assignments history
|
||
self.sync_daily_assignments()
|
||
|
||
if status_changed_to_validated:
|
||
from interventions.utils import create_preventive_intervention_report
|
||
import logging
|
||
try:
|
||
create_preventive_intervention_report(self)
|
||
except Exception:
|
||
logging.getLogger(__name__).exception(
|
||
f"Error automatically generating preventive report for intervention {self.code}"
|
||
)
|
||
|
||
if status_changed_to_closed:
|
||
from observations.models import Observation, ObservationComment
|
||
linked_obs = self.observations.exclude(status='closed')
|
||
for obs in linked_obs:
|
||
obs.status = 'closed'
|
||
obs.save(update_fields=['status'])
|
||
try:
|
||
ObservationComment.objects.create(
|
||
observation=obs,
|
||
comment=f"Observation clôturée automatiquement suite à la clôture de l'intervention {self.code}",
|
||
)
|
||
except Exception:
|
||
pass
|
||
|
||
if old_status and old_status != self.status:
|
||
self.notify_subscribers(old_status, self.status)
|
||
if self.status == 'on_pause':
|
||
self.notify_manager_on_pause()
|
||
|
||
def notify_subscribers(self, old_status, new_status):
|
||
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
|
||
|
||
subscribers = User.objects.filter(intervention_subscriptions__intervention=self)
|
||
if self.last_modified_by:
|
||
subscribers = subscribers.exclude(pk=self.last_modified_by.pk)
|
||
if self.status == 'on_pause' and self.intervention_manager_id:
|
||
subscribers = subscribers.exclude(pk=self.intervention_manager_id)
|
||
|
||
if subscribers.exists():
|
||
old_label = dict(STATUS_CHOICES).get(old_status, old_status)
|
||
new_label = dict(STATUS_CHOICES).get(self.status, self.status)
|
||
|
||
title = _("Mise à jour de l'intervention %(code)s") % {'code': self.code}
|
||
content = _("L'intervention « %(title)s » est passée du statut « %(old)s » à « %(new)s ».") % {
|
||
'title': self.title,
|
||
'old': old_label,
|
||
'new': new_label
|
||
}
|
||
|
||
action_url = reverse('interventions:intervention_detail', args=[self.id])
|
||
|
||
try:
|
||
notify_users(
|
||
recipients=subscribers,
|
||
title=title,
|
||
content=content,
|
||
notification_type='info',
|
||
action_url=action_url,
|
||
action_label=_("Voir l'intervention"),
|
||
source_app='interventions',
|
||
sender=self.last_modified_by,
|
||
is_system=True,
|
||
)
|
||
except Exception:
|
||
import logging
|
||
logging.getLogger(__name__).exception(
|
||
f"Error sending subscription notification for intervention {self.code}"
|
||
)
|
||
|
||
def notify_manager_on_pause(self):
|
||
"""Notifie le gestionnaire de l'intervention lorsqu'elle passe en pause."""
|
||
if not self.intervention_manager_id:
|
||
return
|
||
|
||
# Ne pas notifier le gestionnaire s'il est lui-même l'auteur de la mise en pause
|
||
if self.last_modified_by_id and self.last_modified_by_id == self.intervention_manager_id:
|
||
return
|
||
|
||
from django.urls import reverse
|
||
from django.utils.translation import gettext as _
|
||
from notifications.services import notify_user
|
||
|
||
pause_reason_label = dict(PAUSE_REASON_CHOICES).get(self.pause_reason, self.pause_reason)
|
||
if self.pause_reason == 'other' and self.pause_reason_other:
|
||
reason_text = f"{pause_reason_label} : {self.pause_reason_other}"
|
||
elif pause_reason_label:
|
||
reason_text = str(pause_reason_label)
|
||
else:
|
||
reason_text = None
|
||
|
||
title = _("Intervention %(code)s mise en pause") % {'code': self.code}
|
||
if reason_text:
|
||
content = _("L'intervention « %(title)s » a été mise en pause.\nMotif : %(reason)s") % {
|
||
'title': self.title,
|
||
'reason': reason_text,
|
||
}
|
||
else:
|
||
content = _("L'intervention « %(title)s » a été mise en pause.") % {
|
||
'title': self.title,
|
||
}
|
||
|
||
action_url = reverse('interventions:intervention_detail', args=[self.id])
|
||
|
||
try:
|
||
notify_user(
|
||
recipient=self.intervention_manager,
|
||
title=title,
|
||
content=content,
|
||
notification_type='alert',
|
||
action_url=action_url,
|
||
action_label=_("Voir l'intervention"),
|
||
source_app='interventions',
|
||
sender=self.last_modified_by,
|
||
is_system=True,
|
||
)
|
||
except Exception:
|
||
import logging
|
||
logging.getLogger(__name__).exception(
|
||
f"Error sending pause notification to manager for intervention {self.code}"
|
||
)
|
||
|
||
def sync_daily_assignments(self):
|
||
"""Synchronise les assignations quotidiennes de l'intervention."""
|
||
start_dt = self.planned_begin_time or self.begin_time or self.expected_begin_time or self.creation_time
|
||
end_dt = self.planned_end_time or self.end_time or self.expected_end_time or start_dt
|
||
|
||
if not start_dt:
|
||
return
|
||
|
||
start_date = start_dt.date()
|
||
end_date = end_dt.date()
|
||
|
||
from datetime import timedelta
|
||
active_dates = set()
|
||
curr = start_date
|
||
while curr <= end_date:
|
||
active_dates.add(curr)
|
||
curr += timedelta(days=1)
|
||
if self.begin_time:
|
||
active_dates.add(self.begin_time.date())
|
||
if self.end_time:
|
||
active_dates.add(self.end_time.date())
|
||
|
||
today = timezone.now().date()
|
||
|
||
if self.assigned_team:
|
||
for d in active_dates:
|
||
if d < today:
|
||
# Pour les dates passées, on conserve l'historique : get_or_create pour ne pas écraser
|
||
InterventionAssignment.objects.get_or_create(
|
||
intervention=self,
|
||
date=d,
|
||
defaults={
|
||
'team': self.assigned_team,
|
||
'member': self.assigned_member,
|
||
'status': self.status,
|
||
}
|
||
)
|
||
else:
|
||
# Pour aujourd'hui ou le futur, on met à jour en cas d'erreur ou réassignation
|
||
InterventionAssignment.objects.update_or_create(
|
||
intervention=self,
|
||
date=d,
|
||
defaults={
|
||
'team': self.assigned_team,
|
||
'member': self.assigned_member,
|
||
'status': self.status,
|
||
}
|
||
)
|
||
# Nettoyage des assignations futures obsolètes uniquement
|
||
qs_to_delete = InterventionAssignment.objects.filter(intervention=self).exclude(date__in=active_dates)
|
||
qs_to_delete.filter(date__gte=today).delete()
|
||
else:
|
||
# Si l'intervention n'a plus d'équipe assignée, supprimer toutes les assignations pour aujourd'hui et le futur
|
||
InterventionAssignment.objects.filter(intervention=self, date__gte=today).delete()
|
||
|
||
def __str__(self):
|
||
return f"{self.code} - {self.title}"
|
||
|
||
class Meta:
|
||
ordering = ["-creation_time"]
|
||
indexes = [
|
||
models.Index(fields=['status', 'thematic']),
|
||
models.Index(fields=['status', 'creation_time']),
|
||
models.Index(fields=['thematic', 'creation_time']),
|
||
# Accélère l'agrégat par (thématique, statut, type de maintenance)
|
||
models.Index(fields=['thematic', 'status', 'maintain_type'], name='intv_thematic_status_mtype_idx'),
|
||
# Accélère la sous-requête de permissions par contrat
|
||
models.Index(fields=['contract', 'status'], name='intv_contract_status_idx'),
|
||
# Accélère la requête du graphique : GROUP BY date_trunc(COALESCE(begin_time, ...))
|
||
models.Index(
|
||
fields=['begin_time', 'planned_begin_time', 'expected_begin_time', 'maintain_type'],
|
||
name='intv_chart_begin_maintain_idx',
|
||
),
|
||
]
|
||
|
||
@property
|
||
def intervention_sheet_filename(self):
|
||
if self.intervention_sheet:
|
||
return os.path.basename(self.intervention_sheet.name)
|
||
return None
|
||
|
||
@property
|
||
def status_color_hex(self):
|
||
return STATUS_COLORS.get(self.status, '#757575')
|
||
|
||
@property
|
||
def status_text_color_hex(self):
|
||
return '#000000' if self.status in STATUS_COLORS_DARK_TEXT else '#ffffff'
|
||
|
||
@property
|
||
def provider_or_company_name(self):
|
||
"""Retourne le nom de l'organisme / prestataire associé à cette intervention."""
|
||
if self.assigned_provider:
|
||
return self.assigned_provider.name
|
||
if self.contract and self.contract.company:
|
||
return self.contract.company.name
|
||
if self.assigned_team and self.assigned_team.company:
|
||
return self.assigned_team.company.name
|
||
return ""
|
||
|
||
BLOCKED_CONTRACT_DISPLAY_STATUSES = ('in_preparation', 'to_be_approved', 'to_be_planned')
|
||
|
||
@property
|
||
def is_contract_blocked(self):
|
||
"""Retourne True si l'intervention est marquée bloquée ou si son contrat est actuellement bloqué."""
|
||
return bool(self.is_blocked_due_to_contract or (self.contract and getattr(self.contract, 'is_blocked', False)))
|
||
|
||
@property
|
||
def show_contract_blocked_warning(self):
|
||
"""
|
||
Indique si l'avertissement / badge de marché bloqué doit être affiché.
|
||
Uniquement affiché pour les interventions en préparation, à approuver ou à planifier.
|
||
"""
|
||
return bool(self.is_contract_blocked and self.status in self.BLOCKED_CONTRACT_DISPLAY_STATUSES)
|
||
|
||
def get_status_color(self):
|
||
return {
|
||
"in_preparation": "secondary",
|
||
"to_be_processed": "danger",
|
||
"to_be_approved": "warning",
|
||
"to_be_planned": "warning",
|
||
"assigned": "warning",
|
||
"in_progress": "primary",
|
||
"on_pause": "warning",
|
||
"finished": "info",
|
||
"processed": "info",
|
||
"to_be_corrected": "danger",
|
||
"corrected": "info",
|
||
"validated": "success",
|
||
"invoiced": "dark",
|
||
"closed": "dark",
|
||
"canceled": "secondary"
|
||
}.get(self.status, "secondary")
|
||
|
||
def get_total_occupation_minutes(self):
|
||
return (self.occupations
|
||
.aggregate(total=Coalesce(Sum('minutes'), 0))
|
||
.get('total', 0))
|
||
|
||
def get_preparation_orders_status(self):
|
||
"""
|
||
Agrège les statuts des bons de préparation liés.
|
||
Retourne: 'processed' si tous les bons sont traités/annulés,
|
||
'not_processed' si au moins un bon est à traiter/en attente,
|
||
'none' si aucun bon n'existe.
|
||
"""
|
||
prep_orders = self.preparation_orders.all().values_list('status', flat=True)
|
||
if not prep_orders:
|
||
return 'none'
|
||
|
||
# Statuts considérés comme 'traité'
|
||
treated_statuses = {'processed', 'cancelled'}
|
||
# Statuts considérés comme 'non traité'
|
||
untreated_statuses = {'to_process', 'pending'}
|
||
|
||
statuses = set(prep_orders)
|
||
has_untreated = bool(statuses & untreated_statuses)
|
||
|
||
if has_untreated:
|
||
return 'not_processed'
|
||
return 'processed'
|
||
|
||
@property
|
||
def has_pending_preparation(self):
|
||
"""Retourne True si au moins un bon de préparation est à traiter ou en attente."""
|
||
return self.preparation_orders.filter(
|
||
status__in=['to_process', 'pending']
|
||
).exists()
|
||
|
||
@property
|
||
def has_material(self):
|
||
"""Indique si l'intervention dispose de matériel mentionné dans ses bons de préparation."""
|
||
if not hasattr(self, 'preparation_orders'):
|
||
return False
|
||
return self.preparation_orders.filter(
|
||
Q(preparationorderitem__isnull=False) | Q(other_items__isnull=False)
|
||
).distinct().exists()
|
||
|
||
def get_material_stats(self):
|
||
"""
|
||
Retourne le statut d'utilisation du matériel pour cette intervention :
|
||
- total : nombre total d'articles (catalogue + autres)
|
||
- used : nombre d'articles marqués comme utilisés
|
||
- all_used : True si tous les articles sont utilisés (et total > 0)
|
||
- has_material : True si au moins un article existe
|
||
"""
|
||
if not hasattr(self, 'preparation_orders'):
|
||
return {'total': 0, 'used': 0, 'all_used': False, 'has_material': False}
|
||
from stock.models import PreparationOrderItem, PreparationOrderOtherItem
|
||
order_ids = list(self.preparation_orders.values_list('id', flat=True))
|
||
if not order_ids:
|
||
return {'total': 0, 'used': 0, 'all_used': False, 'has_material': False}
|
||
std_items = PreparationOrderItem.objects.filter(preparation_order_id__in=order_ids)
|
||
other_items = PreparationOrderOtherItem.objects.filter(preparation_order_id__in=order_ids)
|
||
total = std_items.count() + other_items.count()
|
||
used = std_items.filter(is_used=True).count() + other_items.filter(is_used=True).count()
|
||
return {
|
||
'total': total,
|
||
'used': used,
|
||
'all_used': (total > 0 and used == total),
|
||
'has_material': total > 0,
|
||
}
|
||
|
||
def get_associated_observations(self):
|
||
"""
|
||
Retourne le QuerySet de toutes les observations associées à cette intervention :
|
||
- Celles dont le champ intervention pointe directement vers cette intervention
|
||
- Celles correspondant au source_ref de cette intervention
|
||
- Celles associées aux interventions liées via InterventionLink (origine, duplication, réparation)
|
||
"""
|
||
if not self.pk:
|
||
from observations.models import Observation
|
||
return Observation.objects.none()
|
||
|
||
from observations.models import Observation
|
||
|
||
possible_obs_ids = set()
|
||
for obs_id in self.observations.values_list('id', flat=True):
|
||
possible_obs_ids.add(obs_id)
|
||
|
||
if self.source_category and self.source_ref:
|
||
sc_names = [getattr(self.source_category, a, '') or '' for a in ('name_fr', 'name_nl')]
|
||
if any(n.strip().lower() == 'observation' for n in sc_names):
|
||
try:
|
||
possible_obs_ids.add(int(self.source_ref))
|
||
except (ValueError, TypeError):
|
||
pass
|
||
if self.source_ref.startswith('OBS') and self.source_ref[3:].isdigit():
|
||
try:
|
||
possible_obs_ids.add(int(self.source_ref[3:]))
|
||
except (ValueError, TypeError):
|
||
pass
|
||
|
||
# Interventions liées (duplication, réparation, lien manuel)
|
||
linked_itv_ids = set()
|
||
for f, s in InterventionLink.objects.filter(
|
||
Q(first=self) | Q(second=self)
|
||
).values_list('first_id', 'second_id'):
|
||
linked_itv_ids.add(f if f != self.id else s)
|
||
|
||
if linked_itv_ids:
|
||
for obs_id in Observation.objects.filter(intervention_id__in=linked_itv_ids).values_list('id', flat=True):
|
||
possible_obs_ids.add(obs_id)
|
||
for itv_other in Intervention.objects.filter(id__in=linked_itv_ids, source_category__isnull=False).exclude(source_ref__isnull=True).exclude(source_ref=''):
|
||
sc_names = [getattr(itv_other.source_category, a, '') or '' for a in ('name_fr', 'name_nl')]
|
||
if any(n.strip().lower() == 'observation' for n in sc_names):
|
||
try:
|
||
possible_obs_ids.add(int(itv_other.source_ref))
|
||
except (ValueError, TypeError):
|
||
pass
|
||
if itv_other.source_ref.startswith('OBS') and itv_other.source_ref[3:].isdigit():
|
||
try:
|
||
possible_obs_ids.add(int(itv_other.source_ref[3:]))
|
||
except (ValueError, TypeError):
|
||
pass
|
||
|
||
return Observation.objects.filter(id__in=possible_obs_ids).order_by('-created_at')
|
||
|
||
|
||
class InterventionLink(models.Model):
|
||
class LinkType(models.TextChoices):
|
||
RELATED = "related", _("Lien manuel")
|
||
REPAIR = "repair", _("Réparation")
|
||
|
||
first = models.ForeignKey(
|
||
"Intervention",
|
||
on_delete=models.CASCADE,
|
||
related_name="links_as_first",
|
||
)
|
||
second = models.ForeignKey(
|
||
"Intervention",
|
||
on_delete=models.CASCADE,
|
||
related_name="links_as_second",
|
||
)
|
||
link_type = models.CharField(max_length=32, choices=LinkType.choices, default=LinkType.RELATED)
|
||
created_at = models.DateTimeField(auto_now_add=True)
|
||
|
||
class Meta:
|
||
constraints = [
|
||
models.UniqueConstraint(
|
||
fields=["first", "second", "link_type"],
|
||
name="uniq_intervention_link",
|
||
),
|
||
models.CheckConstraint(
|
||
condition=~Q(first=F("second")),
|
||
name="prevent_self_link",
|
||
),
|
||
]
|
||
ordering = ["-created_at"]
|
||
|
||
def save(self, *args, **kwargs):
|
||
if self.first_id and self.second_id:
|
||
if self.first_id == self.second_id:
|
||
raise ValueError("An intervention link cannot reference the same intervention twice.")
|
||
if self.first_id > self.second_id:
|
||
self.first, self.second = self.second, self.first
|
||
super().save(*args, **kwargs)
|
||
|
||
def __str__(self):
|
||
return f"{self.first} ↔ {self.second} ({self.get_link_type_display()})"
|
||
|
||
@classmethod
|
||
def normalize_pair(cls, a, b):
|
||
if a.pk == b.pk:
|
||
raise ValueError("Cannot create a link between the same intervention")
|
||
return (a, b) if a.pk < b.pk else (b, a)
|
||
|
||
@classmethod
|
||
def link(cls, a, b, link_type: str):
|
||
first, second = cls.normalize_pair(a, b)
|
||
return cls.objects.get_or_create(first=first, second=second, link_type=link_type)
|
||
|
||
def other(self, intervention):
|
||
if intervention.pk == self.first_id:
|
||
return self.second
|
||
if intervention.pk == self.second_id:
|
||
return self.first
|
||
raise ValueError("Intervention is not part of this link")
|
||
|
||
|
||
class SymptomContractAutoClose(models.Model):
|
||
"""
|
||
Règle de clôture automatique spécifique à une combinaison (symptôme × contrat).
|
||
L'intervention sera clôturée automatiquement si elle porte ce symptôme ET est assignée
|
||
à ce contrat, après le délai indiqué.
|
||
"""
|
||
symptom = models.ForeignKey(
|
||
'Symptom',
|
||
on_delete=models.CASCADE,
|
||
related_name='auto_close_rules',
|
||
verbose_name=_("Symptôme"),
|
||
)
|
||
contract = models.ForeignKey(
|
||
Contract,
|
||
on_delete=models.CASCADE,
|
||
related_name='auto_close_rules',
|
||
verbose_name=_("Contrat"),
|
||
)
|
||
delay_days = models.PositiveIntegerField(
|
||
default=30,
|
||
verbose_name=_("Délai (jours)"),
|
||
help_text=_("Délai en jours après lequel l'intervention est clôturée automatiquement."),
|
||
)
|
||
|
||
class Meta:
|
||
unique_together = [('symptom', 'contract')]
|
||
verbose_name = _("Règle de clôture automatique (symptôme × contrat)")
|
||
verbose_name_plural = _("Règles de clôture automatique (symptôme × contrat)")
|
||
ordering = ['symptom', 'contract']
|
||
|
||
def __str__(self):
|
||
return f"{self.symptom} × {self.contract.contract_number} ({self.delay_days}j)"
|
||
|
||
|
||
class ContractPauseReasonAssignment(models.Model):
|
||
"""
|
||
Permet de configurer, pour un contrat donné, le membre à assigner automatiquement
|
||
lorsqu'une intervention est mise en pause avec un motif particulier.
|
||
"""
|
||
contract = models.ForeignKey(
|
||
Contract,
|
||
on_delete=models.CASCADE,
|
||
related_name='pause_reason_assignments',
|
||
verbose_name=_("Contrat"),
|
||
)
|
||
pause_reason = models.CharField(
|
||
max_length=50,
|
||
choices=PAUSE_REASON_CHOICES,
|
||
verbose_name=_("Motif de pause"),
|
||
)
|
||
assigned_member = models.ForeignKey(
|
||
CompanyMember,
|
||
on_delete=models.SET_NULL,
|
||
null=True,
|
||
blank=True,
|
||
related_name='pause_reason_assignments',
|
||
verbose_name=_("Membre assigné"),
|
||
help_text=_("Membre à qui réassigner automatiquement l'intervention lors d'une mise en pause avec ce motif. Laisser vide pour désassigner."),
|
||
)
|
||
|
||
class Meta:
|
||
verbose_name = _("Réassignation automatique sur pause")
|
||
verbose_name_plural = _("Réassignations automatiques sur pause")
|
||
unique_together = [('contract', 'pause_reason')]
|
||
|
||
def __str__(self):
|
||
member_label = str(self.assigned_member) if self.assigned_member else _("(désassigner)")
|
||
return f"{self.contract} – {self.get_pause_reason_display()} → {member_label}"
|
||
|
||
|
||
class InterventionSequence(models.Model):
|
||
current_number = models.PositiveIntegerField(default=0)
|
||
|
||
@staticmethod
|
||
def get_next_number():
|
||
# Atomically retrieve and increment the number
|
||
from django.db import transaction
|
||
with transaction.atomic():
|
||
(sequence, created) = InterventionSequence.objects.get_or_create(id=1) # Fixed ID for a single sequence
|
||
sequence.current_number += 1
|
||
sequence.save()
|
||
return sequence.current_number
|
||
|
||
|
||
|
||
ACTION_TYPE_CHOICES = {
|
||
'creation': _('Création'),
|
||
'status_change': _('Changement de statut'),
|
||
'transfer': _('Transfert'),
|
||
}
|
||
|
||
class InterventionTimeLine(models.Model):
|
||
intervention = models.ForeignKey(Intervention, related_name='events', on_delete=models.CASCADE)
|
||
event_user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.PROTECT, related_name="actions", null=True, blank=True)
|
||
event_time = models.DateTimeField()
|
||
event_type = models.CharField(choices=ACTION_TYPE_CHOICES)
|
||
from_status = models.CharField(choices=STATUS_CHOICES,null=True,blank=True)
|
||
to_status = models.CharField(choices=STATUS_CHOICES,null=True,blank=True)
|
||
|
||
class Meta:
|
||
ordering = ["event_time"]
|
||
|
||
|
||
PLAN_ACTION_TYPE_CHOICES = {
|
||
'expected_begin': _('Début prévu'),
|
||
'expected_end': _('Fin prévue'),
|
||
'planned_begin': _('Début planifié'),
|
||
'planned_end': _('Fin planifiée'),
|
||
'plan_other': _('Autre'),
|
||
}
|
||
|
||
class InterventionPlanificationTimeLine(models.Model):
|
||
intervention = models.ForeignKey(Intervention, related_name='planned_events', on_delete=models.CASCADE)
|
||
event_user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.PROTECT, related_name="planned_actions", null=True, blank=True)
|
||
event_time = models.DateTimeField()
|
||
event_type = models.CharField(choices=PLAN_ACTION_TYPE_CHOICES, default='plan_other')
|
||
event_description = models.CharField(max_length=1024)
|
||
|
||
class Meta:
|
||
ordering = ["event_time"]
|
||
|
||
|
||
REAL_ACTION_TYPE_CHOICES = {
|
||
'real_begin': _("Démarrage de l'intervention"),
|
||
'real_end': _("Fin de l'intervention"),
|
||
'real_pause': _("Pause de l'intervention"),
|
||
'real_resume': _("Reprise de l'intervention"),
|
||
'real_other': _('Autre'),
|
||
}
|
||
|
||
class InterventionRealizationTimeLine(models.Model):
|
||
intervention = models.ForeignKey(Intervention, related_name='realized_events', on_delete=models.CASCADE)
|
||
event_user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.PROTECT, related_name="realized_actions", null=True, blank=True)
|
||
event_time = models.DateTimeField()
|
||
event_type = models.CharField(choices=REAL_ACTION_TYPE_CHOICES, default='real_other')
|
||
event_description = models.CharField(max_length=1024)
|
||
|
||
class Meta:
|
||
ordering = ["event_time"]
|
||
|
||
|
||
class InterventionOccupation(models.Model):
|
||
intervention = models.ForeignKey(
|
||
Intervention,
|
||
related_name='occupations',
|
||
on_delete=models.CASCADE,
|
||
)
|
||
member = models.ForeignKey(
|
||
CompanyMember,
|
||
on_delete=models.CASCADE,
|
||
related_name='occupations',
|
||
)
|
||
minutes = models.PositiveIntegerField()
|
||
created_at = models.DateTimeField(auto_now_add=True)
|
||
updated_at = models.DateTimeField(auto_now=True)
|
||
created_by = models.ForeignKey(
|
||
settings.AUTH_USER_MODEL,
|
||
on_delete=models.SET_NULL,
|
||
null=True,
|
||
blank=True,
|
||
related_name='created_intervention_occupations',
|
||
)
|
||
updated_by = models.ForeignKey(
|
||
settings.AUTH_USER_MODEL,
|
||
on_delete=models.SET_NULL,
|
||
null=True,
|
||
blank=True,
|
||
related_name='updated_intervention_occupations',
|
||
)
|
||
|
||
class Meta:
|
||
ordering = ['created_at']
|
||
|
||
def __str__(self):
|
||
member_name = getattr(self.member, 'name', '')
|
||
return f"{member_name} - {self.minutes} min"
|
||
|
||
|
||
class InterventionAssignment(models.Model):
|
||
"""Historique des assignations quotidiennes par intervention et par équipe."""
|
||
intervention = models.ForeignKey(
|
||
Intervention,
|
||
on_delete=models.CASCADE,
|
||
related_name='daily_assignments',
|
||
verbose_name=_("Intervention"),
|
||
)
|
||
team = models.ForeignKey(
|
||
CompanyTeam,
|
||
on_delete=models.CASCADE,
|
||
related_name='daily_assignments',
|
||
verbose_name=_("Équipe"),
|
||
)
|
||
member = models.ForeignKey(
|
||
CompanyMember,
|
||
on_delete=models.SET_NULL,
|
||
null=True,
|
||
blank=True,
|
||
related_name='daily_assignments',
|
||
verbose_name=_("Membre"),
|
||
)
|
||
status = models.CharField(
|
||
choices=STATUS_CHOICES,
|
||
max_length=50,
|
||
null=True,
|
||
blank=True,
|
||
verbose_name=_("Statut"),
|
||
)
|
||
date = models.DateField(verbose_name=_("Date d'assignation"))
|
||
created_at = models.DateTimeField(auto_now_add=True)
|
||
|
||
class Meta:
|
||
unique_together = ('intervention', 'date')
|
||
verbose_name = _("Assignation quotidienne d'intervention")
|
||
verbose_name_plural = _("Assignations quotidiennes d'interventions")
|
||
ordering = ['date', 'team']
|
||
|
||
def __str__(self):
|
||
return f"{self.intervention.code} - {self.team.name} ({self.date})"
|
||
|
||
|
||
POST_CONTROL_STATUS_CHOICES = {
|
||
'validated': _('Approuvé'),
|
||
'refused': _('Refusé'),
|
||
'pending': _('A justifier'),
|
||
}
|
||
|
||
PERIOD_STATUS_CHOICES = {
|
||
'pending': _('En attente'),
|
||
'validated': _('Validé'),
|
||
'invoiced': _('Facturé'),
|
||
}
|
||
|
||
class InterventionContractPost(models.Model):
|
||
"""Poste de contrat associé à une intervention."""
|
||
intervention = models.ForeignKey(Intervention, on_delete=models.CASCADE, related_name="interv_contract_posts")
|
||
contract_post = models.ForeignKey(ContractPost, on_delete=models.CASCADE)
|
||
quantity = models.DecimalField(verbose_name=_("Quantité"), null=True, blank=True, max_digits=10, decimal_places=2)
|
||
comment = models.CharField(max_length=1024, verbose_name=_("Commentaire"), null=True, blank=True)
|
||
control_status = models.CharField(choices=POST_CONTROL_STATUS_CHOICES, null=True, blank=True)
|
||
control_comment = models.CharField(max_length=1024, verbose_name=_("Commentaire de contrôle"), null=True, blank=True)
|
||
|
||
def __str__(self):
|
||
return f"{self.intervention} - {self.contract_post.get_description_for_language()} ({self.quantity})"
|
||
|
||
@property
|
||
def total_price(self):
|
||
"""Calcule le total pour ce poste dans l'intervention.
|
||
Si is_coefficient_multipliable est True, force la quantité à 1 (sauf si 0 ou vide)
|
||
et multiplie le prix unitaire par la somme des coûts totaux des postes du même post_type."""
|
||
if not self.contract_post.unit_price:
|
||
return Decimal("0.00")
|
||
|
||
qty = self.quantity or 0
|
||
|
||
# Si le poste est un coefficient multiplicateur
|
||
if self.contract_post.is_coefficient_multipliable and self.contract_post.post_type:
|
||
# Pour les postes avec coefficient, la quantité doit être 1 (sauf si 0 ou vide)
|
||
# Si l'utilisateur a saisi une quantité > 0, on la ramène à 1 pour le calcul
|
||
effective_qty = Decimal('1') if qty > 0 else Decimal('0')
|
||
|
||
# Calculer la somme des coûts totaux (quantité × prix unitaire) de tous les postes du même type
|
||
same_type_posts = InterventionContractPost.objects.filter(
|
||
intervention=self.intervention,
|
||
contract_post__post_type=self.contract_post.post_type
|
||
).exclude(
|
||
contract_post__is_coefficient_multipliable=True # Exclure les postes multiplicateurs
|
||
).select_related('contract_post')
|
||
|
||
total_same_type_cost = sum(
|
||
(post.quantity or Decimal('0')) * (post.contract_post.unit_price or Decimal('0'))
|
||
for post in same_type_posts
|
||
)
|
||
|
||
# Si c'est une validation par période, ajouter les quantités des périodes
|
||
if self.intervention.period_validation_enabled:
|
||
period_quantities = InterventionPostPeriodQuantity.objects.filter(
|
||
intervention=self.intervention,
|
||
contract_post=self.contract_post
|
||
)
|
||
total_same_type_cost += sum(
|
||
(pq.quantity or Decimal('0')) * (self.contract_post.unit_price or Decimal('0'))
|
||
for pq in period_quantities
|
||
)
|
||
|
||
# Multiplier 1 × prix unitaire × somme totale des coûts
|
||
return effective_qty * self.contract_post.unit_price * total_same_type_cost
|
||
|
||
# Calcul standard
|
||
base = Decimal(qty) * self.contract_post.unit_price
|
||
if self.contract_post.is_amount_to_be_justified and self.contract_post.coeff_amount_to_be_justified:
|
||
return base * self.contract_post.coeff_amount_to_be_justified
|
||
return base
|
||
|
||
class Meta:
|
||
ordering = ["contract_post__order_number"]
|
||
|
||
|
||
class InterventionPostPeriodQuantity(models.Model):
|
||
"""
|
||
Historique des quantités validées par période pour une intervention.
|
||
Une ligne par poste de contrat et par période.
|
||
"""
|
||
intervention = models.ForeignKey(Intervention, on_delete=models.CASCADE, related_name="period_quantities")
|
||
contract_post = models.ForeignKey(ContractPost, on_delete=models.CASCADE, related_name="period_quantities")
|
||
period_number = models.PositiveIntegerField()
|
||
quantity = models.DecimalField(max_digits=10, decimal_places=2)
|
||
validated_at = models.DateTimeField()
|
||
validated_by = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True)
|
||
status = models.CharField(max_length=20, choices=PERIOD_STATUS_CHOICES, default='pending', verbose_name=_("Statut"))
|
||
|
||
class Meta:
|
||
unique_together = ("intervention", "contract_post", "period_number")
|
||
ordering = ["intervention", "period_number", "contract_post__order_number"]
|
||
|
||
def __str__(self):
|
||
return f"{self.intervention.code} - P{self.period_number} - {self.contract_post.code}: {self.quantity}"
|
||
|
||
|
||
class InterventionAsset(models.Model):
|
||
intervention = models.ForeignKey(Intervention, on_delete=models.CASCADE)
|
||
content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE)
|
||
object_id = models.PositiveIntegerField()
|
||
asset = GenericForeignKey('content_type', 'object_id')
|
||
|
||
|
||
class InterventionNote(models.Model):
|
||
class NoteType(models.TextChoices):
|
||
WORK_DONE = "work_done", _("Travaux réalisés")
|
||
MATERIAL = "material", _("Matériel à prévoir")
|
||
EQUIPMENT = "equipment", _("Matériel utilisé")
|
||
IMPORTANT_INFO = "important_info", _("Information importante")
|
||
ORIGIN = "origin", _("Origine")
|
||
COMMENT = "comment", _("Commentaire")
|
||
OTHER = "other", _("Autre")
|
||
|
||
NOTE_TYPE_STYLES = {
|
||
NoteType.WORK_DONE: {"badge_class": "bg-transparent text-secondary border border-success", "icon_class": "bi-clipboard-check"},
|
||
NoteType.MATERIAL: {"badge_class": "bg-transparent text-secondary border border-warning", "icon_class": "bi-box-seam"},
|
||
NoteType.EQUIPMENT: {"badge_class": "bg-transparent text-secondary border border-warning", "icon_class": "bi-tools"},
|
||
NoteType.IMPORTANT_INFO: {"badge_class": "bg-transparent text-secondary border border-danger", "icon_class": "bi-exclamation-octagon-fill"},
|
||
NoteType.ORIGIN: {"badge_class": "bg-transparent text-secondary border border-secondary", "icon_class": "bi-flag-fill"},
|
||
NoteType.COMMENT: {"badge_class": "bg-transparent text-secondary border border-secondary", "icon_class": "bi-chat-dots"},
|
||
NoteType.OTHER: {"badge_class": "bg-transparent text-secondary border border-secondary", "icon_class": "bi-sticky"},
|
||
}
|
||
|
||
NOTE_TYPE_PRIORITIES = {
|
||
NoteType.EQUIPMENT: 0,
|
||
NoteType.IMPORTANT_INFO: 1,
|
||
NoteType.ORIGIN: 2,
|
||
NoteType.WORK_DONE: 3,
|
||
NoteType.COMMENT: 3,
|
||
NoteType.OTHER: 3,
|
||
}
|
||
|
||
intervention = models.ForeignKey(Intervention, on_delete=models.CASCADE, related_name="notes")
|
||
content = models.TextField()
|
||
note_time = models.DateTimeField()
|
||
note_type = models.CharField(
|
||
max_length=32,
|
||
choices=NoteType.choices,
|
||
default=NoteType.WORK_DONE,
|
||
verbose_name=_("Type de note"),
|
||
)
|
||
note_author = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, related_name="interventions_notes", null=True, blank=True)
|
||
updated_at = models.DateTimeField(null=True, blank=True)
|
||
updated_by = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, related_name="updated_interventions_notes", null=True, blank=True)
|
||
|
||
def save(self, *args, **kwargs):
|
||
# Ne définir updated_at que lors des mises à jour (pas à la création)
|
||
if self.pk:
|
||
self.updated_at = timezone.now()
|
||
super().save(*args, **kwargs)
|
||
|
||
@property
|
||
def type_priority_value(self):
|
||
return self.NOTE_TYPE_PRIORITIES.get(self.note_type, 99)
|
||
|
||
@property
|
||
def type_badge_class(self):
|
||
return self.NOTE_TYPE_STYLES.get(self.note_type, {"badge_class": "bg-transparent text-secondary border border-secondary"}).get("badge_class", "bg-transparent text-secondary border border-secondary")
|
||
|
||
@property
|
||
def type_icon_class(self):
|
||
return self.NOTE_TYPE_STYLES.get(self.note_type, {"icon_class": "bi-card-text"}).get("icon_class", "bi-card-text")
|
||
|
||
|
||
|
||
def intervention_document_upload_path(instance, filename):
|
||
# On récupère l'année et le mois actuels
|
||
now = timezone.now()
|
||
year = now.strftime("%Y")
|
||
month = now.strftime("%m")
|
||
|
||
# On construit le chemin avec l'ID de l'intervention (attention : doit être déjà enregistrée)
|
||
intervention_id = instance.intervention.id if instance.intervention else "unknown"
|
||
return f"intervention_documents/{year}/{month}/{intervention_id}/{filename}"
|
||
|
||
class InterventionDocument(Document):
|
||
intervention = models.ForeignKey(Intervention, on_delete=models.CASCADE, related_name="documents", verbose_name="Intervention associée")
|
||
file = models.FileField(upload_to=intervention_document_upload_path, storage=private_storage, verbose_name="Fichier", max_length=255)
|
||
thumbnail = models.ImageField(upload_to=intervention_document_upload_path, storage=private_storage, null=True, blank=True, max_length=255)
|
||
upload_date = models.DateTimeField(auto_now_add=True, verbose_name="Date d'upload", null=True, blank=True)
|
||
uploaded_by = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.PROTECT, related_name="uploaded_intervention_documents", null=True, blank=True)
|
||
|
||
|
||
|
||
class InterventionLocation(models.Model):
|
||
intervention = models.ForeignKey(
|
||
"interventions.Intervention",
|
||
on_delete=models.CASCADE,
|
||
related_name="locations",
|
||
db_index=True,
|
||
)
|
||
content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE, db_index=True)
|
||
object_id = models.PositiveIntegerField(db_index=True)
|
||
location = GenericForeignKey("content_type", "object_id")
|
||
|
||
class Meta:
|
||
constraints = [
|
||
models.UniqueConstraint(
|
||
fields=["intervention", "content_type", "object_id"],
|
||
name="uniq_intervention_location"
|
||
),
|
||
]
|
||
indexes = [
|
||
models.Index(fields=["content_type", "object_id"], name="ix_ct_obj"),
|
||
models.Index(fields=["intervention", "content_type"], name="ix_intv_ct"),
|
||
]
|
||
|
||
|
||
### Traffic Lights ###
|
||
|
||
# class InterventionTrafficLightIntersection(models.Model):
|
||
# intervention = models.ForeignKey(Intervention, on_delete=models.CASCADE, related_name="intervention_intersections")
|
||
# intersection = models.ForeignKey(TrafficLightIntersection, on_delete=models.CASCADE, related_name="intervention_intersections")
|
||
|
||
|
||
### Public Lighting ###
|
||
|
||
# class InterventionPublicLightingStreet(models.Model):
|
||
# intervention = models.ForeignKey(Intervention, on_delete=models.CASCADE, related_name="intervention_streets")
|
||
# street = models.ForeignKey(PublicLightingStreet, on_delete=models.CASCADE, related_name="intervention_streets")
|
||
|
||
|
||
### Green ###
|
||
|
||
# class InterventionGreenSection(models.Model):
|
||
# intervention = models.ForeignKey(Intervention, on_delete=models.CASCADE, related_name="intervention_green_sections")
|
||
# section = models.ForeignKey(GreenSection, on_delete=models.CASCADE, related_name="intervention_sections")
|
||
|
||
|
||
### Structures ###
|
||
|
||
# class InterventionStructureLocation(models.Model):
|
||
# intervention = models.ForeignKey(Intervention, on_delete=models.CASCADE, related_name="intervention_structures_locations")
|
||
# location = models.ForeignKey(StructureLocation, on_delete=models.CASCADE, related_name="intervention_locations")
|
||
|
||
|
||
|
||
### Operations ###
|
||
|
||
class OperationStatusSet(models.Model):
|
||
name = models.CharField(max_length=255, unique=True)
|
||
|
||
def __str__(self):
|
||
return self.name
|
||
|
||
|
||
BOOTSTRAP_COLOR_CHOICES = [
|
||
('primary', 'Primary'),
|
||
('secondary', 'Secondary'),
|
||
('success', 'Success'),
|
||
('danger', 'Danger'),
|
||
('warning', 'Warning'),
|
||
('info', 'Info'),
|
||
('light', 'Light'),
|
||
('dark', 'Dark'),
|
||
]
|
||
|
||
|
||
class OperationStatus(models.Model):
|
||
status_set = models.ForeignKey(OperationStatusSet, related_name='statuses', on_delete=models.CASCADE)
|
||
label_fr = models.CharField(max_length=50)
|
||
label_nl = models.CharField(max_length=50)
|
||
value = models.CharField(max_length=50)
|
||
order = models.PositiveIntegerField(default=0)
|
||
bootstrap_icon = models.CharField(max_length=50, blank=True, null=True, help_text=_("Icône Bootstrap (ex: bi-check-circle)"))
|
||
bootstrap_color = models.CharField(max_length=20, choices=BOOTSTRAP_COLOR_CHOICES, blank=True, null=True, help_text=_("Couleur Bootstrap"))
|
||
requires_comment = models.BooleanField(default=False, verbose_name=_("Commentaire obligatoire"), help_text=_("Si coché, un commentaire sera requis lorsque ce statut est sélectionné"))
|
||
|
||
class Meta:
|
||
unique_together = ('status_set', 'value')
|
||
ordering = ['order']
|
||
|
||
def __str__(self):
|
||
lang = get_language()
|
||
if lang == 'nl':
|
||
return self.label_nl or self.label_fr
|
||
return self.label_fr
|
||
|
||
|
||
VALUE_TYPE_CHOICES = [
|
||
('numeric', _('Numérique')),
|
||
('duration', _('Durée (hh:mm:ss)')),
|
||
('datetime', _('Date et heure')),
|
||
]
|
||
|
||
|
||
class MeasurementDefinition(models.Model):
|
||
name = models.CharField(max_length=255)
|
||
value_type = models.CharField(
|
||
max_length=20,
|
||
choices=VALUE_TYPE_CHOICES,
|
||
default='numeric',
|
||
verbose_name=_('Type de valeur'),
|
||
)
|
||
unit = models.CharField(max_length=50, blank=True)
|
||
# Bornes pour type numérique
|
||
min_value = models.DecimalField(max_digits=8, decimal_places=2, null=True, blank=True)
|
||
max_value = models.DecimalField(max_digits=8, decimal_places=2, null=True, blank=True)
|
||
# Bornes pour type durée
|
||
min_duration = models.DurationField(null=True, blank=True, verbose_name=_('Durée min'))
|
||
max_duration = models.DurationField(null=True, blank=True, verbose_name=_('Durée max'))
|
||
# Bornes pour type date/heure
|
||
min_datetime = models.DateTimeField(null=True, blank=True, verbose_name=_('Date/heure min'))
|
||
max_datetime = models.DateTimeField(null=True, blank=True, verbose_name=_('Date/heure max'))
|
||
|
||
def __str__(self):
|
||
return f"{self.name} ({self.unit})" if self.unit else self.name
|
||
|
||
|
||
class GuidedOperationTemplate(models.Model):
|
||
"""
|
||
Représente un template d'opérations guidées avec une interface HTML personnalisée.
|
||
Le slug identifie le fichier template: interventions/guided_operations/{slug}.html
|
||
"""
|
||
slug = models.SlugField(
|
||
max_length=100,
|
||
unique=True,
|
||
verbose_name=_("Slug"),
|
||
help_text=_("Identifiant unique. Doit correspondre au nom du fichier HTML: guided_operations/{slug}.html")
|
||
)
|
||
name_fr = models.CharField(max_length=255, verbose_name=_("Nom (FR)"))
|
||
name_nl = models.CharField(max_length=255, blank=True, verbose_name=_("Nom (NL)"))
|
||
description = models.TextField(blank=True, verbose_name=_("Description"))
|
||
|
||
class Meta:
|
||
verbose_name = _("Template d'opérations guidées")
|
||
verbose_name_plural = _("Templates d'opérations guidées")
|
||
ordering = ["name_fr"]
|
||
|
||
def getName(self):
|
||
lang = get_language()
|
||
if lang == 'nl':
|
||
return self.name_nl or self.name_fr
|
||
return self.name_fr
|
||
|
||
def __str__(self):
|
||
lang = get_language()
|
||
if lang == 'nl':
|
||
return self.name_nl or self.name_fr
|
||
return self.name_fr
|
||
|
||
|
||
class OperationTemplate(models.Model):
|
||
code = models.CharField(max_length=100, null=True, blank=True)
|
||
name = models.CharField(max_length=255, unique=True)
|
||
description = models.TextField(blank=True)
|
||
thematic = models.ForeignKey(Thematic, on_delete=models.SET_NULL, null=True, blank=True)
|
||
team = models.ForeignKey(
|
||
CompanyTeam,
|
||
on_delete=models.SET_NULL,
|
||
null=True,
|
||
blank=True,
|
||
related_name='operation_templates',
|
||
verbose_name=_("Équipe associée"),
|
||
help_text=_("Si définie, cette équipe sera automatiquement assignée lors de la création d'interventions préventives utilisant cette gamme.")
|
||
)
|
||
guided_template = models.ForeignKey(
|
||
GuidedOperationTemplate,
|
||
on_delete=models.SET_NULL,
|
||
null=True,
|
||
blank=True,
|
||
related_name='operation_templates',
|
||
verbose_name=_("Template guidé associé"),
|
||
help_text=_("Si défini, ce template d'opérations déclenchera automatiquement l'assignation du template guidé à l'intervention préventive.")
|
||
)
|
||
|
||
# Optional: filter equipment operations to specific categories
|
||
categories = models.ManyToManyField(
|
||
'assets.AssetCategory',
|
||
blank=True,
|
||
related_name='operation_templates',
|
||
verbose_name=_("Catégories"),
|
||
help_text=_("Si défini, les opérations équipement ne seront créées que pour les assets de ces catégories.")
|
||
)
|
||
|
||
def __str__(self):
|
||
return self.name
|
||
|
||
def get_category_ids(self):
|
||
"""Return set of category IDs this template applies to, or None if no filter."""
|
||
ids = set(self.categories.values_list('id', flat=True))
|
||
return ids if ids else None
|
||
|
||
def get_asset_model_filters(self):
|
||
"""
|
||
Return dict of {content_type_model: set(object_ids)} for asset model filtering.
|
||
Returns None if no asset model filters are defined.
|
||
"""
|
||
filters = {}
|
||
for entry in self.asset_model_filters.all():
|
||
ct = entry.asset_model_content_type
|
||
if ct:
|
||
model_name = ct.model.lower()
|
||
if model_name not in filters:
|
||
filters[model_name] = set()
|
||
filters[model_name].add(entry.asset_model_object_id)
|
||
return filters if filters else None
|
||
|
||
|
||
class InspectionConfiguration(models.Model):
|
||
name_fr = models.CharField(max_length=255, verbose_name=_("Nom (FR)"))
|
||
name_nl = models.CharField(max_length=255, blank=True, verbose_name=_("Nom (NL)"))
|
||
description = models.TextField(blank=True, verbose_name=_("Description"))
|
||
thematic = models.ForeignKey(Thematic, on_delete=models.PROTECT, verbose_name=_("Thématique"))
|
||
maintain_type = models.CharField(
|
||
max_length=50,
|
||
choices=MAINTAIN_CHOICES,
|
||
default='preventive',
|
||
verbose_name=_("Type de maintenance")
|
||
)
|
||
operation_templates = models.ManyToManyField(
|
||
OperationTemplate,
|
||
blank=True,
|
||
verbose_name=_("Templates d'opérations")
|
||
)
|
||
default_category = models.ForeignKey(
|
||
'assets.AssetCategory',
|
||
on_delete=models.SET_NULL,
|
||
null=True,
|
||
blank=True,
|
||
verbose_name=_("Catégorie par défaut")
|
||
)
|
||
displayed_location_types = models.ManyToManyField(
|
||
ContentType,
|
||
blank=True,
|
||
related_name='inspection_configs_as_location',
|
||
limit_choices_to=Q(app_label='assets'),
|
||
verbose_name=_("Types de localisations à afficher sur la carte")
|
||
)
|
||
displayed_asset_types = models.ManyToManyField(
|
||
ContentType,
|
||
blank=True,
|
||
related_name='inspection_configs_as_asset',
|
||
limit_choices_to=Q(app_label='assets'),
|
||
verbose_name=_("Types d'équipements à afficher sur la carte")
|
||
)
|
||
concerned_categories = models.ManyToManyField(
|
||
'assets.AssetCategory',
|
||
blank=True,
|
||
related_name='inspection_configs',
|
||
verbose_name=_("Catégories concernées"),
|
||
help_text=_("Filtre les localisations et équipements affichés sur la carte par catégorie. Si vide, tous sont affichés.")
|
||
)
|
||
show_equipment = models.BooleanField(
|
||
default=True,
|
||
verbose_name=_("Afficher le champ équipement"),
|
||
help_text=_("Si décoché, le champ équipement sera masqué lors de l'encodage de cette inspection.")
|
||
)
|
||
|
||
|
||
class Meta:
|
||
verbose_name = _("Configuration d'inspection")
|
||
verbose_name_plural = _("Configurations d'inspections")
|
||
ordering = ['name_fr']
|
||
|
||
@property
|
||
def name(self):
|
||
lang = get_language()
|
||
if lang == 'nl':
|
||
return self.name_nl or self.name_fr
|
||
return self.name_fr
|
||
|
||
def __str__(self):
|
||
return self.name
|
||
|
||
|
||
class InspectionConfigurationContract(models.Model):
|
||
inspection_configuration = models.ForeignKey(
|
||
InspectionConfiguration,
|
||
on_delete=models.CASCADE,
|
||
related_name='contracts_providers',
|
||
verbose_name=_("Configuration d'inspection")
|
||
)
|
||
provider = models.ForeignKey(
|
||
Company,
|
||
on_delete=models.CASCADE,
|
||
related_name='inspection_configuration_contracts',
|
||
verbose_name=_("Prestataire")
|
||
)
|
||
contract = models.ForeignKey(
|
||
Contract,
|
||
on_delete=models.CASCADE,
|
||
related_name='inspection_configuration_contracts',
|
||
verbose_name=_("Contrat")
|
||
)
|
||
|
||
class Meta:
|
||
verbose_name = _("Prestataire et Contrat d'inspection")
|
||
verbose_name_plural = _("Prestataires et Contrats d'inspections")
|
||
unique_together = ('inspection_configuration', 'provider', 'contract')
|
||
|
||
def __str__(self):
|
||
return f"{self.provider.name} - {self.contract.contract_number}"
|
||
|
||
|
||
|
||
|
||
class OperationTemplateAssetModelFilter(models.Model):
|
||
"""
|
||
Stores a reference to a specific asset model (e.g., StructureAssetModel, TrafficLightPoleModel).
|
||
Uses GenericForeignKey to allow referencing any asset model type.
|
||
"""
|
||
template = models.ForeignKey(
|
||
OperationTemplate,
|
||
on_delete=models.CASCADE,
|
||
related_name='asset_model_filters',
|
||
verbose_name=_("Template")
|
||
)
|
||
asset_model_content_type = models.ForeignKey(
|
||
ContentType,
|
||
on_delete=models.CASCADE,
|
||
verbose_name=_("Type de modèle d'asset"),
|
||
limit_choices_to={'app_label': 'assets'}
|
||
)
|
||
asset_model_object_id = models.PositiveIntegerField(verbose_name=_("ID du modèle"))
|
||
asset_model = GenericForeignKey('asset_model_content_type', 'asset_model_object_id')
|
||
|
||
class Meta:
|
||
verbose_name = _("Filtre de modèle d'asset")
|
||
verbose_name_plural = _("Filtres de modèles d'assets")
|
||
unique_together = ('template', 'asset_model_content_type', 'asset_model_object_id')
|
||
|
||
def __str__(self):
|
||
if self.asset_model:
|
||
return str(self.asset_model)
|
||
return f"Model {self.asset_model_content_type_id}:{self.asset_model_object_id}"
|
||
|
||
|
||
class OperationType(models.Model):
|
||
code = models.CharField(max_length=100, unique=True)
|
||
name_fr = models.CharField(max_length=255)
|
||
name_nl = models.CharField(max_length=255)
|
||
|
||
def __str__(self):
|
||
lang = get_language()
|
||
if lang == 'nl':
|
||
return self.name_nl
|
||
return self.name_fr
|
||
|
||
|
||
class OperationTemplateItem(models.Model):
|
||
code = models.CharField(max_length=100, null=True, blank=True)
|
||
template = models.ForeignKey(OperationTemplate, related_name='items', on_delete=models.CASCADE)
|
||
name_fr = models.CharField(max_length=255)
|
||
name_nl = models.CharField(max_length=255)
|
||
operation_type = models.ForeignKey(OperationType, on_delete=models.SET_NULL, null=True, blank=True)
|
||
|
||
status_set = models.ForeignKey(OperationStatusSet, on_delete=models.SET_NULL, null=True, blank=True)
|
||
measurement_definition = models.ForeignKey(MeasurementDefinition, on_delete=models.SET_NULL, null=True, blank=True)
|
||
|
||
order = models.PositiveIntegerField(default=0)
|
||
|
||
# Indique si cette opération est générique (une fois par intervention) ou concerne un équipement (répétée par équipement)
|
||
is_generic = models.BooleanField(default=False, verbose_name=_("Opération générique"),
|
||
help_text=_("Si coché, cette opération sera ajoutée une seule fois par intervention. Sinon, elle sera répétée pour chaque équipement."))
|
||
|
||
# Indique si cette opération permet/requiert l'ajout de photos
|
||
requires_photo = models.BooleanField(default=False, verbose_name=_("Requiert une photo"))
|
||
|
||
class Meta:
|
||
ordering = ['order']
|
||
|
||
def __str__(self):
|
||
lang = get_language()
|
||
if lang == 'nl':
|
||
return f"{self.template.name} - {self.name_nl}"
|
||
return f"{self.template.name} - {self.name_fr}"
|
||
|
||
|
||
class ContractOperationTemplatePost(models.Model):
|
||
"""
|
||
Associe un poste de contrat à un template d'opérations avec une quantité par défaut.
|
||
Utilisé lors de la création d'interventions préventives pour pré-remplir les postes de contrat.
|
||
"""
|
||
contract = models.ForeignKey(
|
||
Contract,
|
||
on_delete=models.CASCADE,
|
||
related_name='operation_template_posts',
|
||
verbose_name=_("Contrat"),
|
||
)
|
||
operation_template = models.ForeignKey(
|
||
OperationTemplate,
|
||
on_delete=models.CASCADE,
|
||
related_name='contract_posts',
|
||
verbose_name=_("Template d'opérations"),
|
||
)
|
||
contract_post = models.ForeignKey(
|
||
ContractPost,
|
||
on_delete=models.CASCADE,
|
||
related_name='operation_template_posts',
|
||
verbose_name=_("Poste de contrat"),
|
||
)
|
||
default_quantity = models.DecimalField(
|
||
max_digits=10,
|
||
decimal_places=2,
|
||
default=1,
|
||
verbose_name=_("Quantité par défaut"),
|
||
)
|
||
comment = models.CharField(
|
||
max_length=1024,
|
||
blank=True,
|
||
null=True,
|
||
verbose_name=_("Commentaire par défaut"),
|
||
)
|
||
|
||
class Meta:
|
||
unique_together = [('contract', 'operation_template', 'contract_post')]
|
||
ordering = ['contract', 'operation_template', 'contract_post__order_number']
|
||
verbose_name = _("Poste par défaut (Contrat × Template d'opérations)")
|
||
verbose_name_plural = _("Postes par défaut (Contrats × Templates d'opérations)")
|
||
|
||
def __str__(self):
|
||
return f"{self.contract.contract_number} × {self.operation_template.name} → {self.contract_post.code} ({self.default_quantity})"
|
||
|
||
def clean(self):
|
||
super().clean()
|
||
if self.contract_post_id and self.contract_id:
|
||
if self.contract_post.contract_id != self.contract_id:
|
||
raise ValidationError({
|
||
'contract_post': _("Ce poste n'appartient pas au contrat sélectionné.")
|
||
})
|
||
|
||
def save(self, *args, **kwargs):
|
||
if self.contract_post_id and not self.contract_id:
|
||
self.contract = self.contract_post.contract
|
||
super().save(*args, **kwargs)
|
||
|
||
|
||
EMAIL_LOG_TYPE_CHOICES = {
|
||
'transmission': _('Envoi de la fiche'),
|
||
'reminder': _('Rappel'),
|
||
'modification': _('Modification'),
|
||
'cancellation': _('Annulation'),
|
||
'daily_summary': _('Résumé journalier'),
|
||
}
|
||
|
||
|
||
class InterventionEmailLog(models.Model):
|
||
"""Historique des envois d'emails pour une intervention."""
|
||
intervention = models.ForeignKey(Intervention, related_name='email_logs', on_delete=models.CASCADE)
|
||
email_type = models.CharField(choices=EMAIL_LOG_TYPE_CHOICES, max_length=50)
|
||
sent_at = models.DateTimeField(auto_now_add=True)
|
||
sent_by = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name='intervention_emails_sent')
|
||
recipients = models.TextField(help_text=_("Liste des destinataires (séparés par des virgules)"))
|
||
subject = models.CharField(max_length=500, null=True, blank=True)
|
||
detail = models.TextField(null=True, blank=True, help_text=_("Détails supplémentaires (ex: motif de modification)"))
|
||
success = models.BooleanField(default=True)
|
||
error_message = models.TextField(null=True, blank=True)
|
||
|
||
class Meta:
|
||
ordering = ['-sent_at']
|
||
verbose_name = _("Historique d'envoi d'email")
|
||
verbose_name_plural = _("Historiques d'envois d'emails")
|
||
|
||
def __str__(self):
|
||
return f"{self.intervention.code} - {self.get_email_type_display()} - {self.sent_at.strftime('%d/%m/%Y %H:%M')}"
|
||
|
||
@property
|
||
def recipients_list(self):
|
||
"""Retourne la liste des destinataires."""
|
||
if not self.recipients:
|
||
return []
|
||
return [r.strip() for r in self.recipients.split(',') if r.strip()]
|
||
|
||
|
||
class Operation(models.Model):
|
||
intervention = models.ForeignKey(Intervention, related_name='operations', on_delete=models.CASCADE)
|
||
name_fr = models.CharField(max_length=255)
|
||
name_nl = models.CharField(max_length=255)
|
||
comment = models.TextField(blank=True)
|
||
|
||
status_set = models.ForeignKey(OperationStatusSet, on_delete=models.SET_NULL, null=True, blank=True)
|
||
status = models.ForeignKey(OperationStatus, on_delete=models.SET_NULL, null=True, blank=True)
|
||
|
||
# Pour les opérations de mesure
|
||
measurement_definition = models.ForeignKey(MeasurementDefinition, on_delete=models.SET_NULL, null=True, blank=True)
|
||
measured_value = models.DecimalField(max_digits=8, decimal_places=2, null=True, blank=True)
|
||
measured_duration = models.DurationField(null=True, blank=True, verbose_name=_('Durée mesurée'))
|
||
measured_datetime = models.DateTimeField(null=True, blank=True, verbose_name=_('Date/heure mesurée'))
|
||
|
||
# Option "Non applicable" pour marquer une opération comme non pertinente
|
||
is_not_applicable = models.BooleanField(default=False, verbose_name=_("Non applicable"))
|
||
|
||
order = models.PositiveIntegerField(default=0)
|
||
|
||
# Pour grouper les opérations par template
|
||
template_group = models.CharField(max_length=255, blank=True, null=True, verbose_name=_("Groupe de template"))
|
||
# Pour lier à l'asset si nécessaire (pour templates équipement)
|
||
asset_content_type = models.ForeignKey(ContentType, on_delete=models.SET_NULL, null=True, blank=True, related_name='operations')
|
||
asset_object_id = models.PositiveIntegerField(null=True, blank=True)
|
||
asset = GenericForeignKey('asset_content_type', 'asset_object_id')
|
||
|
||
# Marquer les opérations ajoutées manuellement par l'utilisateur
|
||
is_custom = models.BooleanField(default=False, verbose_name=_("Opération personnalisée"))
|
||
created_by = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name='created_operations')
|
||
created_at = models.DateTimeField(auto_now_add=True, null=True, blank=True)
|
||
|
||
# Indique si cette opération est générique (une fois par intervention) ou concerne un équipement (répétée par équipement)
|
||
is_generic = models.BooleanField(default=False, verbose_name=_("Opération générique"))
|
||
|
||
# Indique si cette opération permet/requiert l'ajout de photos
|
||
requires_photo = models.BooleanField(default=False, verbose_name=_("Requiert une photo"))
|
||
|
||
class Meta:
|
||
ordering = ['template_group', 'order']
|
||
|
||
def __str__(self):
|
||
lang = get_language()
|
||
if lang == 'nl':
|
||
return f"{self.name_nl} - {self.status.label_nl if self.status else '/'}"
|
||
return f"{self.name_fr} - {self.status.label_fr if self.status else '/'}"
|
||
|
||
def is_measurement(self):
|
||
return self.measurement_definition is not None
|
||
|
||
def is_measurement_out_of_range(self):
|
||
"""Check if measured value is outside the allowed range"""
|
||
if not self.is_measurement():
|
||
return False
|
||
md = self.measurement_definition
|
||
if md.value_type == 'duration':
|
||
if self.measured_duration is None:
|
||
return False
|
||
if md.min_duration and self.measured_duration < md.min_duration:
|
||
return True
|
||
if md.max_duration and self.measured_duration > md.max_duration:
|
||
return True
|
||
return False
|
||
elif md.value_type == 'datetime':
|
||
if self.measured_datetime is None:
|
||
return False
|
||
if md.min_datetime and self.measured_datetime < md.min_datetime:
|
||
return True
|
||
if md.max_datetime and self.measured_datetime > md.max_datetime:
|
||
return True
|
||
return False
|
||
else: # numeric
|
||
if self.measured_value is None:
|
||
return False
|
||
if md.min_value is None and md.max_value is None:
|
||
return False
|
||
if md.min_value is not None and self.measured_value < md.min_value:
|
||
return True
|
||
if md.max_value is not None and self.measured_value > md.max_value:
|
||
return True
|
||
return False
|
||
|
||
@property
|
||
def measured_duration_display(self):
|
||
"""Format measured_duration as HH:MM:SS string"""
|
||
if self.measured_duration is None:
|
||
return ''
|
||
total_seconds = int(self.measured_duration.total_seconds())
|
||
hours, remainder = divmod(abs(total_seconds), 3600)
|
||
minutes, seconds = divmod(remainder, 60)
|
||
return f"{hours:02d}:{minutes:02d}:{seconds:02d}"
|
||
|
||
|
||
def operation_photo_upload_path(instance, filename):
|
||
"""Generate upload path for operation photos"""
|
||
from django.utils.text import get_valid_filename
|
||
|
||
now = timezone.now()
|
||
year = now.strftime("%Y")
|
||
month = now.strftime("%m")
|
||
|
||
intervention_id = instance.operation.intervention.id if instance.operation and instance.operation.intervention else "unknown"
|
||
operation_id = instance.operation.id if instance.operation else "unknown"
|
||
|
||
# Sanitize filename to prevent path traversal attacks
|
||
safe_filename = get_valid_filename(filename)
|
||
|
||
return f"operation_photos/{year}/{month}/{intervention_id}/{operation_id}/{safe_filename}"
|
||
|
||
|
||
class OperationPhoto(models.Model):
|
||
"""Photos attached to operations"""
|
||
operation = models.ForeignKey(Operation, related_name='photos', on_delete=models.CASCADE)
|
||
photo = models.ImageField(upload_to=operation_photo_upload_path, storage=private_storage, verbose_name=_("Photo"))
|
||
thumbnail = models.ImageField(upload_to=operation_photo_upload_path, storage=private_storage, null=True, blank=True, max_length=255)
|
||
uploaded_at = models.DateTimeField(auto_now_add=True, verbose_name=_("Date d'upload"))
|
||
uploaded_by = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, blank=True, related_name='uploaded_operation_photos')
|
||
description = models.CharField(max_length=255, blank=True, null=True, verbose_name=_("Description"))
|
||
|
||
class Meta:
|
||
ordering = ['-uploaded_at']
|
||
verbose_name = _("Photo d'opération")
|
||
verbose_name_plural = _("Photos d'opérations")
|
||
|
||
def __str__(self):
|
||
return f"Photo for {self.operation} - {self.uploaded_at}"
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Guided Operation Templates — données saisies & médias
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
class GuidedOperationData(models.Model):
|
||
"""Stocke les réponses saisies via un template d'opérations guidées."""
|
||
intervention = models.OneToOneField(
|
||
Intervention,
|
||
on_delete=models.CASCADE,
|
||
related_name='guided_operation_data',
|
||
verbose_name=_("Intervention")
|
||
)
|
||
guided_template = models.ForeignKey(
|
||
GuidedOperationTemplate,
|
||
on_delete=models.CASCADE,
|
||
related_name='data_entries',
|
||
verbose_name=_("Template guidé")
|
||
)
|
||
data = models.JSONField(default=dict, verbose_name=_("Données"))
|
||
created_at = models.DateTimeField(auto_now_add=True)
|
||
updated_at = models.DateTimeField(auto_now=True)
|
||
|
||
class Meta:
|
||
verbose_name = _("Données d'opérations guidées")
|
||
verbose_name_plural = _("Données d'opérations guidées")
|
||
|
||
def __str__(self):
|
||
return f"{self.intervention.code} – {self.guided_template}"
|
||
|
||
|
||
def guided_operation_media_upload_path(instance, filename):
|
||
"""Generate upload path for guided operation media files."""
|
||
from django.utils.text import get_valid_filename
|
||
now = timezone.now()
|
||
year = now.strftime("%Y")
|
||
month = now.strftime("%m")
|
||
intervention_id = instance.guided_data.intervention_id if instance.guided_data_id else "unknown"
|
||
field_key = instance.field_key or "general"
|
||
safe_filename = get_valid_filename(filename)
|
||
return f"guided_operation_media/{year}/{month}/{intervention_id}/{field_key}/{safe_filename}"
|
||
|
||
|
||
GUIDED_MEDIA_TYPE_CHOICES = [
|
||
('photo', _('Photo')),
|
||
('document', _('Document')),
|
||
]
|
||
|
||
|
||
class GuidedOperationMedia(models.Model):
|
||
"""Photos ou documents attachés à une réponse d'opérations guidées."""
|
||
guided_data = models.ForeignKey(
|
||
GuidedOperationData,
|
||
on_delete=models.CASCADE,
|
||
related_name='media',
|
||
verbose_name=_("Données guidées")
|
||
)
|
||
field_key = models.CharField(
|
||
max_length=100,
|
||
blank=True,
|
||
default='',
|
||
verbose_name=_("Clé de question"),
|
||
help_text=_("Identifie la question à laquelle ce média est rattaché (vide = général).")
|
||
)
|
||
file = models.FileField(
|
||
upload_to=guided_operation_media_upload_path,
|
||
storage=private_storage,
|
||
verbose_name=_("Fichier")
|
||
)
|
||
media_type = models.CharField(
|
||
max_length=20,
|
||
choices=GUIDED_MEDIA_TYPE_CHOICES,
|
||
default='photo',
|
||
verbose_name=_("Type de média")
|
||
)
|
||
description = models.CharField(max_length=255, blank=True, verbose_name=_("Description"))
|
||
uploaded_at = models.DateTimeField(auto_now_add=True)
|
||
uploaded_by = models.ForeignKey(
|
||
settings.AUTH_USER_MODEL,
|
||
on_delete=models.SET_NULL,
|
||
null=True,
|
||
blank=True,
|
||
related_name='uploaded_guided_media'
|
||
)
|
||
|
||
class Meta:
|
||
ordering = ['field_key', '-uploaded_at']
|
||
verbose_name = _("Média d'opérations guidées")
|
||
verbose_name_plural = _("Médias d'opérations guidées")
|
||
|
||
def __str__(self):
|
||
return f"{self.guided_data} – {self.field_key or 'general'} – {self.uploaded_at.strftime('%d/%m/%Y')}"
|
||
|
||
|
||
class InterventionSubscription(models.Model):
|
||
user = models.ForeignKey(
|
||
settings.AUTH_USER_MODEL,
|
||
on_delete=models.CASCADE,
|
||
related_name='intervention_subscriptions',
|
||
verbose_name=_("Utilisateur")
|
||
)
|
||
intervention = models.ForeignKey(
|
||
'Intervention',
|
||
on_delete=models.CASCADE,
|
||
related_name='subscriptions',
|
||
verbose_name=_("Intervention")
|
||
)
|
||
created_at = models.DateTimeField(
|
||
auto_now_add=True,
|
||
verbose_name=_("Abonné le")
|
||
)
|
||
|
||
class Meta:
|
||
unique_together = ('user', 'intervention')
|
||
verbose_name = _("Abonnement à une intervention")
|
||
verbose_name_plural = _("Abonnements aux interventions")
|
||
|
||
def __str__(self):
|
||
return f"{self.user.username} → {self.intervention.code}"
|
||
|
||
|
||
class DirectContractRouting(models.Model):
|
||
THEMATIC_CHOICES = [
|
||
('structures', _("Structures")),
|
||
('trafficlights', _("Traffic Lights")),
|
||
('clean', _("Clean")),
|
||
('its', _("ITS")),
|
||
('nature', _("Nature")),
|
||
('roads', _("Roads")),
|
||
('publiclighting', _("Public Lighting")),
|
||
]
|
||
thematic = models.CharField(
|
||
max_length=50,
|
||
choices=THEMATIC_CHOICES,
|
||
verbose_name=_("Thématique"),
|
||
)
|
||
asset_code = models.CharField(
|
||
max_length=100,
|
||
verbose_name=_("Code de l'asset (ex: OA054)"),
|
||
help_text=_("Le code de l'ouvrage d'art ou de l'équipement concerné."),
|
||
)
|
||
symptom = models.ForeignKey(
|
||
"interventions.Symptom",
|
||
on_delete=models.CASCADE,
|
||
null=True,
|
||
blank=True,
|
||
verbose_name=_("Symptôme spécifique"),
|
||
help_text=_("Si renseigné, cette règle s'applique uniquement à ce symptôme. Sinon, elle s'applique par défaut."),
|
||
)
|
||
contract = models.ForeignKey(
|
||
"contracts.Contract",
|
||
on_delete=models.CASCADE,
|
||
verbose_name=_("Contrat à forcer"),
|
||
)
|
||
|
||
class Meta:
|
||
verbose_name = _("Routage direct de contrat")
|
||
verbose_name_plural = _("Routages directs de contrats")
|
||
unique_together = ('thematic', 'asset_code', 'symptom')
|
||
|
||
def __str__(self):
|
||
symptom_str = f" [{self.symptom.code}]" if self.symptom else " [default]"
|
||
return f"{self.thematic} × {self.asset_code}{symptom_str} → {self.contract.contract_number}"
|