Valogeo/loko
Valogeo/loko
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: add custom notification recipient fields and helpers to Symptom model

Browse source
This commit is contained in:
Karl Determe 2026-09-15 16:10:31 +02:00
parent 9ae7b0631b
commit 836b5c8edb
6 changed files with 444 additions and 48 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

63
loko/interventions/admin.py
View file

@ -1,6 +1,6 @@
from django.contrib.gis import admin
from django.contrib.admin.widgets import FilteredSelectMultiple
from django.utils.translation import get_language
from django.utils.translation import get_language, gettext_lazy as _
from django.utils.html import format_html
from django import forms
from django.urls import path
@ -353,16 +353,65 @@ class SymptomAdminForm(forms.ModelForm):
)
return cats
def _clean_emails(self, field_name):
emails_raw = self.cleaned_data.get(field_name, '')
if not emails_raw:
return ''
import re
from django.core.validators import validate_email
from django.core.exceptions import ValidationError
cleaned_emails = []
for email in re.split(r'[;,]', emails_raw):
email = email.strip()
if email:
try:
validate_email(email)
cleaned_emails.append(email)
except ValidationError:
raise forms.ValidationError(
_("L'adresse e-mail '%(email)s' n'est pas valide.") % {'email': email}
)
return ', '.join(cleaned_emails)
def clean_notification_emails(self):
return self._clean_emails('notification_emails')
def clean_cc_notification_emails(self):
return self._clean_emails('cc_notification_emails')
@admin.register(Symptom)
class SymptomAdmin(admin.ModelAdmin):
form = SymptomAdminForm
change_list_template = "admin/interventions/symptom_changelist.html"
list_display = ("id", "code", "thematic", "name_fr", "name_nl", "is_active", "auto_create_intervention", "auto_create_intervention_status", "require_equipment_selection", "asset_categories_list", "contract", "teams_list")
list_filter = ("thematic", "asset_categories", "priority", "is_active", "auto_create_intervention", "require_equipment_selection", "teams")
list_display = ("id", "code", "thematic", "name_fr", "name_nl", "is_active", "notify_contract_recipients", "auto_create_intervention", "auto_create_intervention_status", "require_equipment_selection", "asset_categories_list", "contract", "teams_list")
list_filter = ("thematic", "asset_categories", "priority", "is_active", "notify_contract_recipients", "auto_create_intervention", "require_equipment_selection", "teams")
search_fields = ("name_fr", "name_nl", "code")
filter_horizontal = ("teams",)
inlines = [SymptomRecommendationInline, SymptomContractAutoCloseInline]
fieldsets = (
(None, {
'fields': ('code', 'name_fr', 'name_nl', 'order_number', 'is_active'),
}),
(_("Classification"), {
'fields': ('thematic', 'asset_categories', 'priority', 'check_priority', 'origin_type', 'maintenance_type', 'require_equipment_selection'),
}),
(_("Attribution / Contrat"), {
'fields': ('provider', 'contract', 'teams'),
}),
(_("Notifications email"), {
'fields': ('notify_contract_recipients', 'notification_emails', 'cc_notification_emails'),
'description': _(
"Configuration des destinataires pour ce symptôme. "
"Par défaut, les destinataires configurés sur le contrat associé sont prévenus. "
"Vous pouvez désactiver cette notification du contrat et/ou spécifier des adresses email supplémentaires (principales et en copie, séparées par des virgules ou points-virgules)."
),
}),
(_("Workflow & Création automatique"), {
'fields': ('recommended_next_status', 'auto_create_intervention', 'auto_create_intervention_status'),
}),
)
def get_urls(self):
urls = super().get_urls()
@ -444,6 +493,9 @@ class SymptomAdmin(admin.ModelAdmin):
"Actif",
"Statut recommandé",
"Équipes",
"Notifier contrat",
"Emails supplémentaires symptôme",
"Emails CC supplémentaires symptôme",
"Emails notifications contrat",
"Emails CC contrat",
"Société (prestataire)",
@ -506,6 +558,9 @@ class SymptomAdmin(admin.ModelAdmin):
"Oui" if symptom.is_active else "Non",
recommended_status,
teams,
"Oui" if symptom.notify_contract_recipients else "Non",
symptom.notification_emails or "—",
symptom.cc_notification_emails or "—",
contract_emails,
contract_cc_emails,
provider_name,
@ -515,7 +570,7 @@ class SymptomAdmin(admin.ModelAdmin):
])
# Largeurs de colonnes
col_widths = [6, 14, 40, 40, 20, 30, 12, 16, 18, 25, 14, 8, 20, 40, 40, 40, 30, 20, 12, 50]
col_widths = [6, 14, 40, 40, 20, 30, 12, 16, 18, 25, 14, 8, 20, 40, 18, 40, 40, 40, 40, 30, 20, 12, 50]
for i, width in enumerate(col_widths, start=1):
ws.column_dimensions[ws.cell(row=1, column=i).column_letter].width = width

55
loko/interventions/migrations/0008_symptom_notification_recipients.py Normal file
View file

@ -0,0 +1,55 @@
from django.db import migrations, models
def migrate_internal_emails_forward(apps, schema_editor):
Symptom = apps.get_model('interventions', 'Symptom')
for symptom in Symptom.objects.exclude(internal_notification_emails__isnull=True).exclude(internal_notification_emails=''):
if not symptom.notification_emails:
symptom.notification_emails = symptom.internal_notification_emails
symptom.save(update_fields=['notification_emails'])
def migrate_internal_emails_reverse(apps, schema_editor):
pass
class Migration(migrations.Migration):
dependencies = [
('interventions', '0007_cleanup_blocked_status_interventions'),
]
operations = [
migrations.AddField(
model_name='symptom',
name='notify_contract_recipients',
field=models.BooleanField(
default=True,
help_text="Indique si les destinataires définis sur le contrat associé doivent être prévenus (oui par défaut).",
verbose_name="Notifier les destinataires du contrat",
),
),
migrations.AddField(
model_name='symptom',
name='notification_emails',
field=models.CharField(
blank=True,
help_text="Adresses email supplémentaires à notifier pour ce symptôme (séparées par des virgules ou points-virgules).",
max_length=1000,
null=True,
verbose_name="Emails supplémentaires",
),
),
migrations.AddField(
model_name='symptom',
name='cc_notification_emails',
field=models.CharField(
blank=True,
help_text="Adresses email supplémentaires à ajouter en copie pour ce symptôme (séparées par des virgules ou points-virgules).",
max_length=1000,
null=True,
verbose_name="Emails en copie (CC) supplémentaires",
),
),
migrations.RunPython(migrate_internal_emails_forward, migrate_internal_emails_reverse),
]

19
loko/interventions/models.py
View file

@ -231,6 +231,25 @@ class Symptom(models.Model):
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."))
notify_contract_recipients = models.BooleanField(
default=True,
verbose_name=_("Notifier les destinataires du contrat"),
help_text=_("Indique si les destinataires définis sur le contrat associé doivent être prévenus (oui par défaut)."),
)
notification_emails = models.CharField(
max_length=1000,
blank=True,
null=True,
verbose_name=_("Emails supplémentaires"),
help_text=_("Adresses email supplémentaires à notifier pour ce symptôme (séparées par des virgules ou points-virgules)."),
)
cc_notification_emails = models.CharField(
max_length=1000,
blank=True,
null=True,
verbose_name=_("Emails en copie (CC) supplémentaires"),
help_text=_("Adresses email supplémentaires à ajouter en copie pour ce symptôme (séparées par des virgules ou points-virgules)."),
)
def __str__(self):
lang = get_language()

246
loko/interventions/tests/test_symptom_notifications.py Normal file
View file

@ -0,0 +1,246 @@
from django.test import TestCase, override_settings
from django.utils import timezone
from django.core.exceptions import ValidationError
from common.models import Thematic
from contracts.models import Company, Contract
from assets.models import AssetCategory
from interventions.models import Intervention, Symptom
from interventions.admin import SymptomAdminForm
from interventions.views.helpers import get_contract_recipients, get_contract_cc_recipients
@override_settings(EMAIL_OVERRIDE=None)
class SymptomNotificationRecipientsTests(TestCase):
def setUp(self):
self.thematic, _ = Thematic.objects.get_or_create(code='trafficlights', defaults={'name': 'Feux de signalisation'})
self.category, _ = AssetCategory.objects.get_or_create(name_fr='Test Category', thematic=self.thematic)
self.company = Company.objects.create(name='Prestataire Feux')
today = timezone.now().date()
self.contract = Contract.objects.create(
contract_number='CTR-FEUX-001',
company=self.company,
start_date=today,
end_date=today + timezone.timedelta(days=365),
is_active=True,
is_blocked=False,
notification_emails='contract_to1@example.com; contract_to2@example.com',
cc_notification_emails='contract_cc1@example.com, contract_cc2@example.com',
send_email_notifications=True,
)
self.contract.thematics.add(self.thematic)
self.contract_blocked = Contract.objects.create(
contract_number='CTR-FEUX-BLOCKED',
company=self.company,
start_date=today,
end_date=today + timezone.timedelta(days=365),
is_active=True,
is_blocked=True,
blocked_reason='Gel budget',
notification_emails='blocked_contract@example.com',
cc_notification_emails='blocked_cc@example.com',
send_email_notifications=True,
)
self.contract_blocked.thematics.add(self.thematic)
def test_default_symptom_notifies_contract_recipients(self):
"""Par défaut (notify_contract_recipients=True, sans emails supplémentaires), seuls les emails du contrat sont retournés."""
symptom = Symptom.objects.create(
name_fr='Ampoule grillée',
thematic=self.thematic,
contract=self.contract,
notify_contract_recipients=True,
)
intervention = Intervention.objects.create(
title='Remplacement ampoule',
thematic=self.thematic,
contract=self.contract,
symptom=symptom,
)
recipients = get_contract_recipients(self.contract, intervention)
self.assertIn('contract_to1@example.com', recipients)
self.assertIn('contract_to2@example.com', recipients)
self.assertEqual(len(recipients), 2)
cc_recipients = get_contract_cc_recipients(self.contract, intervention)
self.assertIn('contract_cc1@example.com', cc_recipients)
self.assertIn('contract_cc2@example.com', cc_recipients)
self.assertIn('bm.perm.tech@sprb.brussels', cc_recipients)
def test_symptom_with_additional_recipients_and_cc(self):
"""Avec notify_contract_recipients=True et des emails supplémentaires, les emails du contrat et du symptôme sont cumulés."""
symptom = Symptom.objects.create(
name_fr='Panne contrôleur critique',
thematic=self.thematic,
contract=self.contract,
notify_contract_recipients=True,
notification_emails='extra_expert@sprb.brussels; contract_to1@example.com', # Doublon volontaire avec contrat
cc_notification_emails='extra_supervision@sprb.brussels',
)
intervention = Intervention.objects.create(
title='Contrôleur HS',
thematic=self.thematic,
contract=self.contract,
symptom=symptom,
)
recipients = get_contract_recipients(self.contract, intervention)
self.assertEqual(
set(recipients),
{'contract_to1@example.com', 'contract_to2@example.com', 'extra_expert@sprb.brussels'}
)
cc_recipients = get_contract_cc_recipients(self.contract, intervention)
self.assertEqual(
set(cc_recipients),
{'contract_cc1@example.com', 'contract_cc2@example.com', 'extra_supervision@sprb.brussels', 'bm.perm.tech@sprb.brussels'}
)
def test_symptom_disabling_contract_recipients(self):
"""Avec notify_contract_recipients=False, les emails du contrat sont exclus, seuls les emails du symptôme sont prévenus."""
symptom = Symptom.objects.create(
name_fr='Vandalisme interne',
thematic=self.thematic,
contract=self.contract,
notify_contract_recipients=False,
notification_emails='police_bruxelles@sprb.brussels, securite@sprb.brussels',
cc_notification_emails='responsable_securite@sprb.brussels',
)
intervention = Intervention.objects.create(
title='Dégradation armoire',
thematic=self.thematic,
contract=self.contract,
symptom=symptom,
)
recipients = get_contract_recipients(self.contract, intervention)
self.assertEqual(
set(recipients),
{'police_bruxelles@sprb.brussels', 'securite@sprb.brussels'}
)
self.assertNotIn('contract_to1@example.com', recipients)
self.assertNotIn('contract_to2@example.com', recipients)
cc_recipients = get_contract_cc_recipients(self.contract, intervention)
self.assertIn('responsable_securite@sprb.brussels', cc_recipients)
self.assertIn('bm.perm.tech@sprb.brussels', cc_recipients)
self.assertNotIn('contract_cc1@example.com', cc_recipients)
self.assertNotIn('contract_cc2@example.com', cc_recipients)
def test_symptom_disabling_contract_recipients_without_extra_emails(self):
"""Avec notify_contract_recipients=False et sans emails supplémentaires, aucun destinataire principal n'est retourné."""
symptom = Symptom.objects.create(
name_fr='Test sans notification',
thematic=self.thematic,
contract=self.contract,
notify_contract_recipients=False,
)
intervention = Intervention.objects.create(
title='Intervention silencieuse',
thematic=self.thematic,
contract=self.contract,
symptom=symptom,
)
recipients = get_contract_recipients(self.contract, intervention)
self.assertEqual(recipients, [])
def test_blocked_contract_returns_empty_even_with_symptom_emails(self):
"""Un marché bloqué ne doit envoyer aucun email, même si le symptôme configure des destinataires."""
symptom = Symptom.objects.create(
name_fr='Panne sur marché bloqué',
thematic=self.thematic,
contract=self.contract_blocked,
notify_contract_recipients=True,
notification_emails='extra@sprb.brussels',
cc_notification_emails='extra_cc@sprb.brussels',
)
intervention = Intervention.objects.create(
title='Intervention bloquée',
thematic=self.thematic,
contract=self.contract_blocked,
symptom=symptom,
)
recipients = get_contract_recipients(self.contract_blocked, intervention)
self.assertEqual(recipients, [])
cc_recipients = get_contract_cc_recipients(self.contract_blocked, intervention)
self.assertEqual(cc_recipients, [])
def test_intervention_without_contract_uses_symptom_recipients(self):
"""Une intervention sans contrat mais avec un symptôme configuré peut quand même notifier ses destinataires spécifiques."""
symptom = Symptom.objects.create(
name_fr='Alerte voirie spéciale',
thematic=self.thematic,
contract=None,
notify_contract_recipients=True,
notification_emails='service_voirie@sprb.brussels',
cc_notification_emails='cc_voirie@sprb.brussels',
)
intervention = Intervention.objects.create(
title='Intervention sans contrat',
thematic=self.thematic,
contract=None,
symptom=symptom,
)
recipients = get_contract_recipients(None, intervention)
self.assertEqual(recipients, ['service_voirie@sprb.brussels'])
cc_recipients = get_contract_cc_recipients(None, intervention)
self.assertIn('cc_voirie@sprb.brussels', cc_recipients)
self.assertIn('bm.perm.tech@sprb.brussels', cc_recipients)
def test_backward_compatibility_without_intervention(self):
"""L'appel sans intervention (intervention=None) conserve le comportement historique du contrat."""
recipients = get_contract_recipients(self.contract)
self.assertEqual(
set(recipients),
{'contract_to1@example.com', 'contract_to2@example.com'}
)
cc_recipients = get_contract_cc_recipients(self.contract)
self.assertIn('contract_cc1@example.com', cc_recipients)
self.assertIn('contract_cc2@example.com', cc_recipients)
self.assertIn('bm.perm.tech@sprb.brussels', cc_recipients)
def test_symptom_admin_form_email_validation(self):
"""Le formulaire d'administration valide et normalise les listes d'adresses email."""
# Emails valides avec mélange de virgules et points-virgules
form = SymptomAdminForm(data={
'name_fr': 'Test Form',
'name_nl': 'Test Form NL',
'thematic': self.thematic.pk,
'priority': '3',
'is_active': True,
'notify_contract_recipients': True,
'notification_emails': 'alpha@example.com; beta@example.com, gamma@example.com',
'cc_notification_emails': 'delta@example.com; epsilon@example.com',
})
self.assertTrue(form.is_valid(), form.errors)
self.assertEqual(
form.cleaned_data['notification_emails'],
'alpha@example.com, beta@example.com, gamma@example.com'
)
self.assertEqual(
form.cleaned_data['cc_notification_emails'],
'delta@example.com, epsilon@example.com'
)
# Email invalide
form_invalid = SymptomAdminForm(data={
'name_fr': 'Test Form Invalid',
'name_nl': 'Test Form Invalid NL',
'thematic': self.thematic.pk,
'priority': '3',
'is_active': True,
'notify_contract_recipients': True,
'notification_emails': 'adresse_pas_valide',
})
self.assertFalse(form_invalid.is_valid())
self.assertIn('notification_emails', form_invalid.errors)

71
loko/interventions/views/helpers.py
View file

@ -3681,77 +3681,98 @@ def is_user_same_as_provider(user, intervention) -> bool:
def get_contract_recipients(contract, intervention=None) -> List[str]:
"""Retourne une liste unique d'adresses email associées au contrat."""
"""Retourne une liste unique d'adresses email associées au contrat et/ou au symptôme."""
# En dev/test, redirige tous les emails vers l'adresse de test
email_override = getattr(settings, 'EMAIL_OVERRIDE', None)
if email_override:
return [email_override]
if not contract:
if not contract and intervention:
contract = getattr(intervention, 'contract', None)
if contract and getattr(contract, 'is_blocked', False):
return []
if getattr(contract, 'is_blocked', False):
return []
if hasattr(contract, 'send_email_notifications') and not contract.send_email_notifications:
return []
if getattr(contract, 'email_only_if_urgent', False) and intervention and getattr(intervention, 'priority', None) != '1':
return []
symptom = getattr(intervention, 'symptom', None) if intervention else None
notify_contract = getattr(symptom, 'notify_contract_recipients', True) if symptom else True
emails = set()
# 1) Champ texte sur le contrat (séparé par ; ou ,)
# 1) Destinataires du contrat (si notif contrat autorisée pour ce symptôme)
if notify_contract and contract:
contract_notifs_enabled = not (hasattr(contract, 'send_email_notifications') and not contract.send_email_notifications)
urgent_only_ok = not (getattr(contract, 'email_only_if_urgent', False) and intervention and getattr(intervention, 'priority', None) != '1')
if contract_notifs_enabled and urgent_only_ok:
# Champ texte sur le contrat (séparé par ; ou ,)
if hasattr(contract, 'notification_emails') and contract.notification_emails:
for e in re.split(r'[;,]', contract.notification_emails):
e = e.strip()
if e:
emails.add(e)
# 2) Relations de contacts (ex: contract.contacts)
# Relations de contacts (ex: contract.contacts)
if hasattr(contract, 'contacts'):
qs = contract.contacts.all()
# Filtre facultatif si vous avez un flag pour recevoir les rapports
if hasattr(qs.model, 'receive_intervention_reports'):
qs = qs.filter(receive_intervention_reports=True)
emails.update([c.email for c in qs if getattr(c, 'email', None)])
# 2) Destinataires supplémentaires configurés sur le symptôme
if symptom:
for raw in [getattr(symptom, 'notification_emails', None), getattr(symptom, 'internal_notification_emails', None)]:
if raw:
for e in re.split(r'[;,]', raw):
e = e.strip()
if e:
emails.add(e)
return [e for e in emails if e]
def get_contract_cc_recipients(contract, intervention=None) -> List[str]:
"""Retourne une liste unique d'adresses email en copie (CC) associées au contrat."""
"""Retourne une liste unique d'adresses email en copie (CC) associées au contrat et/ou au symptôme."""
# En dev/test, ne pas utiliser de CC pour éviter les envois externes
email_override = getattr(settings, 'EMAIL_OVERRIDE', None)
if email_override:
return []
if not contract:
if not contract and intervention:
contract = getattr(intervention, 'contract', None)
if contract and getattr(contract, 'is_blocked', False):
return []
if getattr(contract, 'is_blocked', False):
return []
if hasattr(contract, 'send_email_notifications') and not contract.send_email_notifications:
return []
if getattr(contract, 'email_only_if_urgent', False) and intervention and getattr(intervention, 'priority', None) != '1':
return []
symptom = getattr(intervention, 'symptom', None) if intervention else None
notify_contract = getattr(symptom, 'notify_contract_recipients', True) if symptom else True
emails = set()
# Champ texte CC sur le contrat (séparé par ; ou ,)
# 1) CC du contrat (si notif contrat autorisée pour ce symptôme)
if notify_contract and contract:
contract_notifs_enabled = not (hasattr(contract, 'send_email_notifications') and not contract.send_email_notifications)
urgent_only_ok = not (getattr(contract, 'email_only_if_urgent', False) and intervention and getattr(intervention, 'priority', None) != '1')
if contract_notifs_enabled and urgent_only_ok:
if hasattr(contract, 'cc_notification_emails') and contract.cc_notification_emails:
for e in re.split(r'[;,]', contract.cc_notification_emails):
e = e.strip()
if e:
emails.add(e)
# Toujours ajouter l'email par défaut en CC
# 2) CC supplémentaires configurés sur le symptôme
if symptom and hasattr(symptom, 'cc_notification_emails') and symptom.cc_notification_emails:
for e in re.split(r'[;,]', symptom.cc_notification_emails):
e = e.strip()
if e:
emails.add(e)
# Toujours ajouter l'email par défaut en CC si des destinataires ou un contrat/symptôme existent
if contract or symptom:
emails.add(getattr(settings, 'EMAIL_TECH_SUPPORT', 'bm.perm.tech@sprb.brussels'))
return [e for e in emails if e]

12
loko/interventions/views/workflow.py
View file

@ -807,8 +807,8 @@ def attempt_status_update(request, intervention, user, new_status, pause_reason=
reverse("interventions:intervention_detail", args=[intervention.id])
)
recipients = get_contract_recipients(contract, intervention) if contract else []
cc_recipients = get_contract_cc_recipients(contract, intervention) if contract else [getattr(settings, 'EMAIL_TECH_SUPPORT', 'bm.perm.tech@sprb.brussels')]
recipients = get_contract_recipients(contract, intervention)
cc_recipients = get_contract_cc_recipients(contract, intervention)
# Si le contrat utilise l'envoi groupé journalier, ne pas envoyer l'email individuel
if recipients and not (contract and getattr(contract, 'send_daily_summary', False)):
@ -1000,8 +1000,8 @@ def attempt_status_update(request, intervention, user, new_status, pause_reason=
intervention.save(update_fields=['intervention_sheet'])
contract = getattr(intervention, 'contract', None)
recipients = get_contract_recipients(contract, intervention) if contract else []
cc_recipients = get_contract_cc_recipients(contract, intervention) if contract else [getattr(settings, 'EMAIL_TECH_SUPPORT', 'bm.perm.tech@sprb.brussels')]
recipients = get_contract_recipients(contract, intervention)
cc_recipients = get_contract_cc_recipients(contract, intervention)
if recipients:
detail_url = request.build_absolute_uri(
@ -1241,7 +1241,7 @@ def send_intervention_reminder(request, intervention_id):
if not recipients:
return JsonResponse({
'success': False,
'error': _("Aucun destinataire n'a été configuré pour le contrat associé."),
'error': _("Aucun destinataire n'a été configuré pour le contrat ou le symptôme associé."),
}, status=400)
buffer = io.BytesIO()
@ -1381,7 +1381,7 @@ def send_intervention_modification(request, intervention_id):
if not recipients:
return JsonResponse({
'success': False,
'error': _("Aucun destinataire n'a été configuré pour le contrat associé."),
'error': _("Aucun destinataire n'a été configuré pour le contrat ou le symptôme associé."),
}, status=400)
buffer = io.BytesIO()