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

fix(interventions): assign controller only when intervention status changes to finished

Browse source
This commit is contained in:
kdeterme 2026-10-01 09:53:06 +02:00
parent 8df84704dc
commit 3920648000
3 changed files with 212 additions and 96 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

38
loko/interventions/models.py
View file

@ -658,21 +658,39 @@ class Intervention(models.Model):
if user_config and not user_config.is_intern:
self.created_by_provider = True
# Attribution automatique de l'équipe de contrôle
if (
self.contract and getattr(self.contract, 'auto_assign_control_team', False)
and not self.assigned_control_team_id
):
# Attribution automatique de l'équipe de contrôle et du contrôleur
if self.contract and getattr(self.contract, 'auto_assign_control_team', False):
try:
from interventions.services_control_assignment import resolve_control_team_and_member_for_intervention
ctrl_team, ctrl_member = resolve_control_team_and_member_for_intervention(self)
if ctrl_team:
self.assigned_control_team = ctrl_team
self.assigned_controller = ctrl_member
# 1. Équipe de contrôle : attribuée dès la création ou sauvegarde si non définie
if not self.assigned_control_team_id:
ctrl_team, _ = resolve_control_team_and_member_for_intervention(self)
if ctrl_team:
self.assigned_control_team = ctrl_team
if 'update_fields' in kwargs and kwargs['update_fields'] is not None:
kwargs['update_fields'] = set(kwargs['update_fields']) | {'assigned_control_team'}
# 2. Contrôleur : attribué UNIQUEMENT lorsque l'intervention passe au statut 'terminé' ('finished')
if status_changed_to_finished and not self.assigned_controller_id:
if self.assigned_control_team:
_, ctrl_member = resolve_control_team_and_member_for_intervention(
self, target_team=self.assigned_control_team
)
else:
ctrl_team, ctrl_member = resolve_control_team_and_member_for_intervention(self)
if ctrl_team and not self.assigned_control_team_id:
self.assigned_control_team = ctrl_team
if 'update_fields' in kwargs and kwargs['update_fields'] is not None:
kwargs['update_fields'] = set(kwargs['update_fields']) | {'assigned_control_team'}
if ctrl_member:
self.assigned_controller = ctrl_member
if 'update_fields' in kwargs and kwargs['update_fields'] is not None:
kwargs['update_fields'] = set(kwargs['update_fields']) | {'assigned_controller'}
except Exception:
import logging
logging.getLogger(__name__).exception(
f"Error auto-assigning control team for intervention {self.code}"
f"Error auto-assigning control team/controller for intervention {self.code}"
)
super().save(*args, **kwargs)

183
loko/interventions/services_control_assignment.py
View file

@ -12,95 +12,101 @@ def resolve_control_team_and_member(
thematic=None,
maintain_type=None,
locations=None,
target_team=None,
):
"""
Détermine l'équipe de contrôle (CompanyTeam) et le contrôleur (CompanyMember) :
1. Équipes candidates gérant l'opération 'controls' (OPERATION_CONTROLS).
2. Filtrage thématique si l'équipe restreint ses thématiques.
3. Filtrage par type de maintenance (corrective, preventive, ameliorative) si l'équipe restreint ses types.
4. Résolution spatiale si une géométrie est définie (cible et équipes) :
- team.geom.intersects(target_geom) ou team.geom.contains(point).
- Fallback sur toutes les équipes candidates si aucune géométrie n'est définie.
5. Calendrier d'assignation à la date scheduled_date :
1. Si target_team est fourni, restreint la recherche des membres à cette équipe.
Sinon :
- Équipes candidates gérant l'opération 'controls' (OPERATION_CONTROLS).
- Filtrage thématique si l'équipe restreint ses thématiques.
- Filtrage par type de maintenance (corrective, preventive, ameliorative) si l'équipe restreint ses types.
- Résolution spatiale si une géométrie est définie (cible et équipes) :
- team.geom.intersects(target_geom) ou team.geom.contains(point).
- Fallback sur toutes les équipes candidates si aucune géométrie n'est définie.
2. Calendrier d'assignation à la date scheduled_date :
- Évalue le nombre d'agents planifiés et non absents.
- Si aucun agent n'est disponible ou aucune date : retourne (None, None).
- Si aucun agent n'est disponible ou aucune date : retourne (target_team, None) ou (None, None).
- Sélectionne l'équipe ayant le plus d'agents disponibles.
- Sélectionne le premier membre disponible de cette équipe.
"""
from contracts.models import CompanyTeam, CompanyMember
from contracts.services_teams import get_teams_composition_for_period, get_member_absences_for_period
teams_qs = CompanyTeam.objects.all().select_related('company').prefetch_related('thematics', 'maintain_types')
candidate_teams = [t for t in teams_qs if t.handles_operation(CompanyTeam.OPERATION_CONTROLS)]
if not candidate_teams:
return None, None
# 1. Filtrage thématique si l'équipe restreint ses thématiques
if thematic:
candidate_teams = [
t for t in candidate_teams
if not t.thematics.all() or any(thm.id == thematic.id for thm in t.thematics.all())
]
if not candidate_teams:
return None, None
# 2. Filtrage par type de maintenance si l'équipe restreint ses types de maintenance
if maintain_type:
candidate_teams = [
t for t in candidate_teams
if t.handles_maintain_type(maintain_type)
]
if not candidate_teams:
return None, None
# 2. Résolution spatiale si géométrie définie
has_target_geom = bool(target_geom or point or locations)
teams_with_geom = [t for t in candidate_teams if t.geom is not None]
if has_target_geom and teams_with_geom:
spatial_matches = []
for team in teams_with_geom:
matched = False
if target_geom and team.geom:
try:
matched = team.geom.intersects(target_geom)
except Exception as e:
logger.debug(f"Error checking team {team.id} geom intersects: {e}")
if not matched and point and team.geom:
try:
matched = team.geom.contains(point)
except Exception as e:
logger.debug(f"Error checking team {team.id} geom contains point: {e}")
if not matched and locations and team.geom:
for loc in locations:
loc_geom = getattr(loc, 'geom', None)
if loc_geom:
try:
if team.geom.intersects(loc_geom):
matched = True
break
except Exception:
pass
if matched:
spatial_matches.append(team)
eligible_teams = spatial_matches
if target_team is not None:
eligible_teams = [target_team]
else:
# Fallback sans géométrie : toutes les équipes candidates
eligible_teams = candidate_teams
teams_qs = CompanyTeam.objects.all().select_related('company').prefetch_related('thematics', 'maintain_types')
candidate_teams = [t for t in teams_qs if t.handles_operation(CompanyTeam.OPERATION_CONTROLS)]
if not eligible_teams:
return None, None
if not candidate_teams:
return None, None
# 3. Calendrier d'assignation
# 1. Filtrage thématique si l'équipe restreint ses thématiques
if thematic:
candidate_teams = [
t for t in candidate_teams
if not t.thematics.all() or any(thm.id == thematic.id for thm in t.thematics.all())
]
if not candidate_teams:
return None, None
# 2. Filtrage par type de maintenance si l'équipe restreint ses types de maintenance
if maintain_type:
candidate_teams = [
t for t in candidate_teams
if t.handles_maintain_type(maintain_type)
]
if not candidate_teams:
return None, None
# 3. Résolution spatiale si géométrie définie
has_target_geom = bool(target_geom or point or locations)
teams_with_geom = [t for t in candidate_teams if t.geom is not None]
if has_target_geom and teams_with_geom:
spatial_matches = []
for team in teams_with_geom:
matched = False
if target_geom and team.geom:
try:
matched = team.geom.intersects(target_geom)
except Exception as e:
logger.debug(f"Error checking team {team.id} geom intersects: {e}")
if not matched and point and team.geom:
try:
matched = team.geom.contains(point)
except Exception as e:
logger.debug(f"Error checking team {team.id} geom contains point: {e}")
if not matched and locations and team.geom:
for loc in locations:
loc_geom = getattr(loc, 'geom', None)
if loc_geom:
try:
if team.geom.intersects(loc_geom):
matched = True
break
except Exception:
pass
if matched:
spatial_matches.append(team)
eligible_teams = spatial_matches
else:
# Fallback sans géométrie : toutes les équipes candidates
eligible_teams = candidate_teams
if not eligible_teams:
return None, None
# 4. Calendrier d'assignation
if not scheduled_date:
return None, None
return (target_team if target_team else (eligible_teams[0] if eligible_teams else None)), None
if hasattr(scheduled_date, 'date'):
target_d = scheduled_date.date()
@ -125,9 +131,9 @@ def resolve_control_team_and_member(
scored_teams.sort(key=lambda item: (-item[0], item[1].name))
best_len, best_team, best_members = scored_teams[0]
# Règle stricte : si aucun agent disponible ce jour-là, ne pas assigner
# Règle stricte pour le contrôleur : si aucun agent disponible ce jour-là, membre = None
if best_len == 0 or not best_members:
return None, None
return (target_team, None) if target_team else (None, None)
best_member_id = best_members[0]['id']
best_member = CompanyMember.objects.filter(id=best_member_id).first()
@ -135,7 +141,7 @@ def resolve_control_team_and_member(
return best_team, best_member
def resolve_control_team_and_member_for_intervention(intervention):
def resolve_control_team_and_member_for_intervention(intervention, target_date=None, target_team=None):
"""
Extrait la géométrie, la localisation et la date de l'intervention
puis appelle resolve_control_team_and_member.
@ -149,12 +155,20 @@ def resolve_control_team_and_member_for_intervention(intervention):
except Exception as e:
logger.debug(f"Error transforming intervention coordinates to 3812: {e}")
target_date = (
intervention.planned_begin_time
or intervention.begin_time
or intervention.creation_time
or timezone.now().date()
)
if target_date is None:
if intervention.status == 'finished':
target_date = (
getattr(intervention, 'end_time', None)
or getattr(intervention, 'planned_end_time', None)
or timezone.now().date()
)
else:
target_date = (
intervention.planned_begin_time
or intervention.begin_time
or intervention.creation_time
or timezone.now().date()
)
locs = list(intervention.locations.all()) if (intervention.pk and hasattr(intervention, 'locations')) else []
@ -165,4 +179,5 @@ def resolve_control_team_and_member_for_intervention(intervention):
thematic=intervention.thematic,
maintain_type=intervention.maintain_type,
locations=locs,
target_team=target_team,
)

87
loko/interventions/tests/test_control_assignment.py
View file

@ -104,8 +104,7 @@ class ControlAssignmentTests(TestCase):
self.contract_no_auto.thematics.add(self.thematic_structures)
def test_auto_assignment_structures_corrective(self):
"""Vérifie l'attribution automatique pour une intervention corrective Structures avec contrat activé."""
# Coordonnées au centre de la zone (Bruxelles ~ 4.35, 50.85 -> en Lambert 72 ~ 150000, 170000)
"""Vérifie l'attribution de l'équipe à la création et du contrôleur au statut terminé."""
p_poly = Polygon(((149000, 169000), (151000, 169000), (151000, 171000), (149000, 171000), (149000, 169000)))
geom = MultiPolygon([p_poly], srid=3812)
@ -117,6 +116,30 @@ class ControlAssignmentTests(TestCase):
planned_begin_time=timezone.now(),
)
# À la création (statut à traiter), seule l'équipe de contrôle est assignée
self.assertEqual(itv.assigned_control_team, self.team_ctrl)
self.assertIsNone(itv.assigned_controller)
# Lors du passage au statut 'terminé', le contrôleur disponible est automatiquement assigné
itv.status = 'finished'
itv.save()
itv.refresh_from_db()
self.assertEqual(itv.assigned_controller, self.member_ctrl)
def test_auto_assignment_when_created_directly_finished(self):
"""Vérifie l'attribution directe de l'équipe et du contrôleur si créée au statut terminé."""
p_poly = Polygon(((149000, 169000), (151000, 169000), (151000, 171000), (149000, 171000), (149000, 169000)))
geom = MultiPolygon([p_poly], srid=3812)
itv = Intervention.objects.create(
maintain_type='corrective',
thematic=self.thematic_structures,
contract=self.contract_auto,
geom=geom,
status='finished',
planned_begin_time=timezone.now(),
)
self.assertEqual(itv.assigned_control_team, self.team_ctrl)
self.assertEqual(itv.assigned_controller, self.member_ctrl)
@ -126,6 +149,7 @@ class ControlAssignmentTests(TestCase):
maintain_type='preventive',
thematic=self.thematic_structures,
contract=self.contract_auto,
status='finished',
planned_begin_time=timezone.now(),
)
self.assertIsNone(itv.assigned_control_team)
@ -138,6 +162,7 @@ class ControlAssignmentTests(TestCase):
maintain_type='preventive',
thematic=self.thematic_structures,
contract=self.contract_auto,
status='finished',
planned_begin_time=timezone.now(),
)
self.assertEqual(itv.assigned_control_team, self.team_ctrl)
@ -150,6 +175,7 @@ class ControlAssignmentTests(TestCase):
maintain_type='ameliorative',
thematic=self.thematic_structures,
contract=self.contract_auto,
status='finished',
planned_begin_time=timezone.now(),
)
self.assertEqual(itv.assigned_control_team, self.team_ctrl)
@ -161,6 +187,7 @@ class ControlAssignmentTests(TestCase):
maintain_type='corrective',
thematic=self.thematic_roads,
contract=self.contract_auto,
status='finished',
planned_begin_time=timezone.now(),
)
self.assertIsNone(itv.assigned_control_team)
@ -173,6 +200,7 @@ class ControlAssignmentTests(TestCase):
maintain_type='corrective',
thematic=self.thematic_roads,
contract=self.contract_auto,
status='finished',
planned_begin_time=timezone.now(),
)
self.assertEqual(itv.assigned_control_team, self.team_ctrl)
@ -184,6 +212,7 @@ class ControlAssignmentTests(TestCase):
maintain_type='corrective',
thematic=self.thematic_structures,
contract=self.contract_no_auto,
status='finished',
planned_begin_time=timezone.now(),
)
self.assertIsNone(itv.assigned_control_team)
@ -196,6 +225,7 @@ class ControlAssignmentTests(TestCase):
thematic=self.thematic_structures,
contract=self.contract_auto,
geom=None,
status='finished',
planned_begin_time=timezone.now(),
)
self.assertEqual(itv.assigned_control_team, self.team_ctrl)
@ -255,3 +285,56 @@ class ControlAssignmentTests(TestCase):
itv.status = 'in_progress'
itv.save(update_fields=['status'])
self.assertFalse(can_control_intervention(self.user_ctrl, itv))
def test_controller_not_assigned_during_in_progress_or_on_pause(self):
"""Le contrôleur ne doit pas être assigné aux statuts intermédiaires (to_be_processed, in_progress, on_pause)."""
itv = Intervention.objects.create(
maintain_type='corrective',
thematic=self.thematic_structures,
contract=self.contract_auto,
status='to_be_processed',
planned_begin_time=timezone.now(),
)
self.assertEqual(itv.assigned_control_team, self.team_ctrl)
self.assertIsNone(itv.assigned_controller)
# Passage à in_progress
itv.status = 'in_progress'
itv.save()
itv.refresh_from_db()
self.assertIsNone(itv.assigned_controller)
# Passage à on_pause
itv.status = 'on_pause'
itv.pause_reason = 'weather'
itv.save()
itv.refresh_from_db()
self.assertIsNone(itv.assigned_controller)
# Passage à finished -> contrôleur assigné
itv.status = 'finished'
itv.save()
itv.refresh_from_db()
self.assertEqual(itv.assigned_controller, self.member_ctrl)
def test_controller_manual_not_overwritten_on_finished(self):
"""Un contrôleur assigné manuellement ne doit pas être écrasé lors du passage à terminé."""
user_other = self.User.objects.create_user(username='other_ctrl', password='pwd')
other_ctrl = CompanyMember.objects.create(
name='Autre Contrôleur',
user=user_other
)
itv = Intervention.objects.create(
maintain_type='corrective',
thematic=self.thematic_structures,
contract=self.contract_auto,
status='in_progress',
assigned_control_team=self.team_ctrl,
assigned_controller=other_ctrl,
planned_begin_time=timezone.now(),
)
itv.status = 'finished'
itv.save()
itv.refresh_from_db()
self.assertEqual(itv.assigned_controller, other_ctrl)