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

refactor: support multiple intervention types in subtypes and update non-compliance notifications for company members

Browse source
This commit is contained in:
kdeterme 2026-09-17 16:05:36 +02:00
parent 7ca27197bf
commit 396d26414c
8 changed files with 231 additions and 35 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

56
loko/inspections/models.py
View file

@ -304,7 +304,19 @@ class Inspection(models.Model):
if not teams:
return
from notifications.services import notify_team
from contracts.models import CompanyMember
from notifications.models import Notification, NotificationRecipient
from notifications.webpush import send_webpush_to_users
# Collecter tous les membres actifs avec un compte utilisateur
recipient_user_ids = set(
CompanyMember.objects.filter(
teams__in=teams, user__isnull=False, user__is_active=True
).values_list('user_id', flat=True)
)
if not recipient_user_ids:
return
status_label = self.get_result_status_display()
title = _("Non-conformité durabilité (%(status)s) : %(code)s") % {
'status': status_label,
@ -325,22 +337,32 @@ class Inspection(models.Model):
}
action_url = f"/interventions/{self.checked_intervention.id}/"
for team in teams:
try:
notify_team(
team=team,
title=str(title),
content=str(content),
notification_type='alert',
action_url=action_url,
action_label=str(_("Consulter l'intervention")),
source_app='inspections',
)
except Exception:
import logging
logging.getLogger(__name__).exception(
f"Error sending non-compliance notification for inspection {self.pk} to team {team.pk}"
)
try:
notification = Notification.objects.create(
title=str(title),
content=str(content),
notification_type='alert',
action_url=action_url,
action_label=str(_("Consulter l'intervention")),
source_app='inspections',
is_system=True,
)
NotificationRecipient.objects.bulk_create([
NotificationRecipient(notification=notification, user_id=uid)
for uid in recipient_user_ids
], ignore_conflicts=True)
send_webpush_to_users(
users=list(recipient_user_ids),
title=str(title),
body=str(content),
action_url=action_url,
)
except Exception:
import logging
logging.getLogger(__name__).exception(
f"Error sending non-compliance notification for inspection {self.pk}"
)
self.non_compliance_notified = True
Inspection.objects.filter(pk=self.pk).update(non_compliance_notified=True)

6
loko/inspections/services.py
View file

@ -53,10 +53,12 @@ def schedule_post_intervention_inspection(intervention):
# 2. Résolution administrative par commune associée
if not zone:
from common.models import Municipality
for loc in intervention.locations.all():
loc_obj = loc.location
if loc_obj and hasattr(loc_obj, 'municipality') and loc_obj.municipality:
zone = zones.filter(municipalities=loc_obj.municipality).first()
muni_to_check = loc_obj if isinstance(loc_obj, Municipality) else getattr(loc_obj, 'municipality', None)
if muni_to_check:
zone = zones.filter(municipalities=muni_to_check).first()
if zone:
break

18
loko/inspections/tests.py
View file

@ -1507,9 +1507,11 @@ class PostInterventionDurabilityInspectionTestCase(TestCase):
can_inspect=True
)
from notifications.models import Team
self.team = Team.objects.create(name='Équipe Contrôle Enrobés')
self.team.members.add(self.user)
from contracts.models import Company, CompanyTeam, CompanyMember
self.company = Company.objects.create(name='Test Company Asphalt')
self.team = CompanyTeam.objects.create(name='Équipe Contrôle Enrobés', company=self.company)
self.member = CompanyMember.objects.create(name='Test Inspector', user=self.user)
self.member.teams.add(self.team)
from interventions.models import InterventionSubType
self.hotbox_subtype = InterventionSubType.objects.create(
@ -1517,7 +1519,7 @@ class PostInterventionDurabilityInspectionTestCase(TestCase):
name_fr='Hot Box (asphalte à chaud)',
name_nl='Hot Box (warm asfalt)',
thematic=self.roads_thematic,
intervention_type='repair',
intervention_types=['repair'],
requires_post_inspection=True,
post_inspection_delay_min_days=30,
post_inspection_delay_max_days=60,
@ -1529,7 +1531,7 @@ class PostInterventionDurabilityInspectionTestCase(TestCase):
code='test_roads_repair_standard',
name_fr='Réparation standard',
thematic=self.roads_thematic,
intervention_type='repair',
intervention_types=['repair'],
requires_post_inspection=False
)
@ -1570,7 +1572,7 @@ class PostInterventionDurabilityInspectionTestCase(TestCase):
thematic=self.roads_thematic
)
zone.inspectors.add(self.user)
muni = Municipality.objects.create(name_fr='Schaerbeek Test', code='21099')
muni = Municipality.objects.create(name_fr='Schaerbeek Test', name_nl='Schaarbeek Test', niscode='21099', fid='21099')
zone.municipalities.add(muni)
intervention = Intervention.objects.create(
@ -1596,7 +1598,7 @@ class PostInterventionDurabilityInspectionTestCase(TestCase):
self.assertTrue(insp.can_edit(self.user))
def test_non_compliance_triggers_team_notification(self):
from notifications.models import TeamNotificationTarget
from notifications.models import NotificationRecipient
intervention = Intervention.objects.create(
thematic=self.roads_thematic,
type='repair',
@ -1615,7 +1617,7 @@ class PostInterventionDurabilityInspectionTestCase(TestCase):
insp.refresh_from_db()
self.assertTrue(insp.non_compliance_notified)
self.assertTrue(TeamNotificationTarget.objects.filter(team=self.team).exists())
self.assertTrue(NotificationRecipient.objects.filter(user=self.user, notification__source_app='inspections').exists())
def test_subtype_without_post_inspection_does_not_trigger(self):
intervention = Intervention.objects.create(

28
loko/interventions/admin.py
View file

@ -42,20 +42,42 @@ from .services import generate_interventions_excel
from common.models import Thematic
from assets.models import AssetCategory
from interventions.views.preventive_import import process_worksheet as _process_worksheet
from interventions.models import STATUS_CHOICES, DirectContractRouting
from interventions.models import STATUS_CHOICES, DirectContractRouting, TYPE_CHOICES
class InterventionSubTypeAdminForm(forms.ModelForm):
intervention_types = forms.MultipleChoiceField(
choices=TYPE_CHOICES,
required=False,
widget=FilteredSelectMultiple(_("Types d'intervention"), is_stacked=False),
label=_("Types d'intervention"),
help_text=_("Types d'intervention auxquels ce sous-type s'applique. Si aucun n'est sélectionné, s'applique à tous les types."),
)
class Meta:
model = InterventionSubType
fields = '__all__'
@admin.register(InterventionSubType)
class InterventionSubTypeAdmin(admin.ModelAdmin):
form = InterventionSubTypeAdminForm
list_display = (
"code", "name_fr", "thematic", "intervention_type",
"code", "name_fr", "thematic", "types_display",
"requires_post_inspection", "post_inspection_delay_min_days",
"post_inspection_delay_max_days", "is_active"
)
list_filter = ("thematic", "intervention_type", "requires_post_inspection", "is_active")
list_filter = ("thematic", "requires_post_inspection", "is_active")
search_fields = ("code", "name_fr", "name_nl")
filter_horizontal = ("notification_teams",)
def types_display(self, obj):
if not obj.intervention_types:
return _("Tous")
type_dict = dict(TYPE_CHOICES)
return ", ".join(str(type_dict.get(t, t)) for t in obj.intervention_types)
types_display.short_description = _("Types d'intervention")
@admin.register(Intervention)
class InterventionAdmin(admin.ModelAdmin):

54
loko/interventions/migrations/0011_alter_interventionsubtype_options_and_more.py Normal file
View file

@ -0,0 +1,54 @@
# Generated by Django 6.0.7 on 2026-09-17 13:59
import django.contrib.postgres.fields
from django.db import migrations, models
def migrate_single_type_to_multiple(apps, schema_editor):
InterventionSubType = apps.get_model('interventions', 'InterventionSubType')
for subtype in InterventionSubType.objects.all():
old_val = getattr(subtype, 'intervention_type', None)
if old_val:
subtype.intervention_types = [old_val]
subtype.save(update_fields=['intervention_types'])
elif subtype.code == 'roads_repair_hotbox':
subtype.intervention_types = ['repair']
subtype.save(update_fields=['intervention_types'])
def reverse_multiple_to_single(apps, schema_editor):
InterventionSubType = apps.get_model('interventions', 'InterventionSubType')
for subtype in InterventionSubType.objects.all():
if subtype.intervention_types:
subtype.intervention_type = subtype.intervention_types[0]
subtype.save(update_fields=['intervention_type'])
class Migration(migrations.Migration):
dependencies = [
('contracts', '0007_contract_approval_except_urgent'),
('interventions', '0010_seed_roads_hotbox_subtype'),
]
operations = [
migrations.AlterModelOptions(
name='interventionsubtype',
options={'ordering': ['thematic', 'order_number', 'name_fr'], 'verbose_name': "Sous-type d'intervention", 'verbose_name_plural': "Sous-types d'interventions"},
),
migrations.AddField(
model_name='interventionsubtype',
name='intervention_types',
field=django.contrib.postgres.fields.ArrayField(base_field=models.CharField(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')], max_length=50), blank=True, default=list, help_text="Types d'intervention auxquels ce sous-type s'applique. Laisser vide pour s'appliquer à tous les types.", verbose_name="Types d'intervention"),
),
migrations.RunPython(migrate_single_type_to_multiple, reverse_code=reverse_multiple_to_single),
migrations.RemoveField(
model_name='interventionsubtype',
name='intervention_type',
),
migrations.AlterField(
model_name='interventionsubtype',
name='notification_teams',
field=models.ManyToManyField(blank=True, help_text="Équipes notifiées si l'inspection de durabilité révèle un constat non-conforme.", related_name='intervention_subtypes', to='contracts.companyteam', verbose_name='Équipes à notifier'),
),
]

13
loko/interventions/models.py
View file

@ -5,6 +5,7 @@ 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.contrib.postgres.fields import ArrayField
from django.db.models import Sum, F, Q
from django.db.models.functions import Coalesce
@ -201,7 +202,13 @@ class InterventionSubType(models.Model):
name_fr = models.CharField(max_length=200, verbose_name=_("Nom (FR)"))
name_nl = models.CharField(max_length=200, blank=True, verbose_name=_("Nom (NL)"))
thematic = models.ForeignKey(Thematic, on_delete=models.PROTECT, related_name='intervention_subtypes', verbose_name=_("Thématique"))
intervention_type = models.CharField(max_length=50, choices=TYPE_CHOICES, verbose_name=_("Type d'intervention"))
intervention_types = ArrayField(
models.CharField(max_length=50, choices=TYPE_CHOICES),
default=list,
blank=True,
verbose_name=_("Types d'intervention"),
help_text=_("Types d'intervention auxquels ce sous-type s'applique. Laisser vide pour s'appliquer à tous les types.")
)
# Paramètres de durabilité / inspection post-travaux
requires_post_inspection = models.BooleanField(
@ -226,7 +233,7 @@ class InterventionSubType(models.Model):
verbose_name=_("Nature de l'inspection")
)
notification_teams = models.ManyToManyField(
'notifications.Team',
CompanyTeam,
blank=True,
related_name='intervention_subtypes',
verbose_name=_("Équipes à notifier"),
@ -238,7 +245,7 @@ class InterventionSubType(models.Model):
class Meta:
verbose_name = _("Sous-type d'intervention")
verbose_name_plural = _("Sous-types d'interventions")
ordering = ['thematic', 'intervention_type', 'order_number', 'name_fr']
ordering = ['thematic', 'order_number', 'name_fr']
def __str__(self):
return f"{self.get_name()} ({self.thematic.name})"

81
loko/interventions/tests/test_inspections.py
View file

@ -566,5 +566,86 @@ class InspectionTests(TestCase):
self.assertIn('member_assignment', allowed_fields)
class InterventionSubTypeTests(TestCase):
def setUp(self):
User = get_user_model()
self.user = User.objects.create_user(username='subtype_test_user', password='pwd')
self.thematic = Thematic.objects.create(code='roads_test', name_fr='Voirie Test', name_nl='Wegen Test')
self.company = Company.objects.create(name='Test Contractor')
self.team1 = CompanyTeam.objects.create(name='Équipe Régie 1', company=self.company)
self.team2 = CompanyTeam.objects.create(name='Équipe Prestataire 2', company=self.company)
from interventions.models import InterventionSubType
self.subtype_multi = InterventionSubType.objects.create(
code='sub_multi',
name_fr='Hot Box Multi',
name_nl='Hot Box NL',
thematic=self.thematic,
intervention_types=['repair', 'fixing'],
requires_post_inspection=True,
order_number=1,
is_active=True,
)
self.subtype_multi.notification_teams.add(self.team1, self.team2)
self.subtype_universal = InterventionSubType.objects.create(
code='sub_univ',
name_fr='Universel',
name_nl='Universeel',
thematic=self.thematic,
intervention_types=[],
requires_post_inspection=False,
order_number=2,
is_active=True,
)
self.subtype_build_only = InterventionSubType.objects.create(
code='sub_build',
name_fr='Construction Seule',
thematic=self.thematic,
intervention_types=['build'],
order_number=3,
is_active=True,
)
def test_subtype_multiple_types_and_teams(self):
from interventions.models import InterventionSubType
st = InterventionSubType.objects.get(code='sub_multi')
self.assertEqual(st.intervention_types, ['repair', 'fixing'])
self.assertEqual(st.notification_teams.count(), 2)
self.assertIn(self.team1, st.notification_teams.all())
self.assertIn(self.team2, st.notification_teams.all())
def test_form_filters_subtypes_by_type(self):
from interventions.views.forms import InterventionForm
# Form for 'repair' should include sub_multi and sub_univ, but NOT sub_build
form_repair = InterventionForm(thematic=self.thematic, user=self.user, initial={'type': 'repair'})
qs_repair = form_repair.fields['sub_type'].queryset
self.assertIn(self.subtype_multi, qs_repair)
self.assertIn(self.subtype_universal, qs_repair)
self.assertNotIn(self.subtype_build_only, qs_repair)
# Form for 'build' should include sub_build and sub_univ, but NOT sub_multi
form_build = InterventionForm(thematic=self.thematic, user=self.user, initial={'type': 'build'})
qs_build = form_build.fields['sub_type'].queryset
self.assertIn(self.subtype_build_only, qs_build)
self.assertIn(self.subtype_universal, qs_build)
self.assertNotIn(self.subtype_multi, qs_build)
def test_admin_types_display(self):
from interventions.admin import InterventionSubTypeAdmin
from django.contrib.admin.sites import AdminSite
from django.utils import translation
admin_instance = InterventionSubTypeAdmin(self.subtype_multi.__class__, AdminSite())
with translation.override('fr'):
display_text = admin_instance.types_display(self.subtype_multi)
self.assertIn("Réparation", display_text)
self.assertIn("Dépannage", display_text)
display_univ = admin_instance.types_display(self.subtype_universal)
self.assertEqual(str(display_univ), "Tous")

10
loko/interventions/views/forms.py
View file

@ -98,16 +98,22 @@ class InterventionForm(forms.ModelForm):
if 'type' in self.fields:
self.fields['type'].choices = get_type_choices(thematic_obj)
type_value = self.initial.get('type') or self.data.get('type') or (getattr(self.instance, 'type', None) if self.instance else None)
maintain_type_value = self.initial.get('maintain_type') or self.data.get('maintain_type') or (getattr(self.instance, 'maintain_type', None) if self.instance else None)
if 'sub_type' in self.fields:
from interventions.models import InterventionSubType
sub_type_qs = InterventionSubType.objects.filter(is_active=True)
if thematic_obj:
sub_type_qs = sub_type_qs.filter(thematic=thematic_obj)
if type_value:
from django.db.models import Q
sub_type_qs = sub_type_qs.filter(
Q(intervention_types__contains=[type_value]) | Q(intervention_types=[])
)
self.fields['sub_type'].queryset = sub_type_qs
self.fields['sub_type'].required = False
type_value = self.initial.get('type') or self.data.get('type')
maintain_type_value = self.initial.get('maintain_type') or self.data.get('maintain_type')
if type_value == 'inspection':
if not self.initial.get('status') and not self.data.get('status'):
self.fields['status'].initial = 'to_be_processed'