From 4e35da62b858203a8f6e89bd1f0b1747fdf3b2ea Mon Sep 17 00:00:00 2001 From: kdeterme Date: Mon, 17 Aug 2026 09:57:09 +0200 Subject: [PATCH] refactor: remove automatic contract assignment logic from intervention detail view --- .../components/asset_interventions.html | 36 +- loko/assets/templatetags/asset_tags.py | 41 +- loko/assets/tests.py | 342 ++++++++++++++ loko/assets/views/asset_editing.py | 144 +++++- loko/interventions/views/detail.py | 184 +------- loko/interventions/views/helpers.py | 416 +++++++++++++++++- 6 files changed, 963 insertions(+), 200 deletions(-) diff --git a/loko/assets/templates/assets/components/asset_interventions.html b/loko/assets/templates/assets/components/asset_interventions.html index 6a1f703..2d2ff16 100644 --- a/loko/assets/templates/assets/components/asset_interventions.html +++ b/loko/assets/templates/assets/components/asset_interventions.html @@ -123,10 +123,10 @@
- {% for s in symptoms %} - + {% endfor %}
@@ -156,7 +156,7 @@
- @@ -185,6 +185,36 @@ (function() { const form = document.getElementById('createInterventionForAssetForm'); if (!form) return; + + const symptomSelect = document.getElementById('asset_intervention_symptom_select'); + const titleInput = form.querySelector('input[name="title"]'); + const prioritySelect = document.getElementById('asset_intervention_priority_select'); + let lastAutoTitle = ''; + + if (symptomSelect && titleInput) { + symptomSelect.addEventListener('change', function() { + const selectedOpt = symptomSelect.options[symptomSelect.selectedIndex]; + if (selectedOpt && selectedOpt.value) { + const symName = selectedOpt.getAttribute('data-name') || selectedOpt.text.trim(); + const symPriority = selectedOpt.getAttribute('data-priority'); + + if (!titleInput.value || titleInput.value === lastAutoTitle) { + titleInput.value = symName; + lastAutoTitle = symName; + } + + if (symPriority && prioritySelect) { + prioritySelect.value = symPriority; + } + } + }); + + titleInput.addEventListener('input', function() { + if (titleInput.value !== lastAutoTitle) { + lastAutoTitle = ''; + } + }); + } form.addEventListener('submit', function(e) { e.preventDefault(); diff --git a/loko/assets/templatetags/asset_tags.py b/loko/assets/templatetags/asset_tags.py index 953fbec..b1b065f 100644 --- a/loko/assets/templatetags/asset_tags.py +++ b/loko/assets/templatetags/asset_tags.py @@ -495,9 +495,11 @@ def asset_interventions(context, asset): 'interventions': [], 'can_add_intervention': False, 'symptoms': [], + 'symptoms_json': '[]', } from django.contrib.contenttypes.models import ContentType + from django.utils.translation import get_language from interventions.models import Intervention, InterventionAsset, Symptom, STATUS_COLORS, STATUS_COLORS_DARK_TEXT from interventions.permissions import can_add_intervention @@ -513,22 +515,50 @@ def asset_interventions(context, asset): 'symptom', 'contract', 'assigned_provider', 'assigned_team', 'created_by' ).order_by('-creation_time')) - thematic = None - if hasattr(asset, 'category') and asset.category and asset.category.thematic: - thematic = asset.category.thematic + from assets.permissions import _get_asset_thematic + thematic = _get_asset_thematic(asset) request = context.get('request') user = request.user if request else None can_add = False symptoms = [] + symptoms_json = '[]' if user and user.is_authenticated: can_add = can_add_intervention(user, thematic) if can_add: if thematic: - symptoms = list(Symptom.objects.filter(thematic=thematic, is_active=True).order_by('order_number', 'name_fr')) + symptoms_qs = Symptom.objects.filter(thematic=thematic, is_active=True).prefetch_related('asset_categories') else: - symptoms = list(Symptom.objects.filter(is_active=True).order_by('name_fr')) + symptoms_qs = Symptom.objects.filter(is_active=True).prefetch_related('asset_categories') + + lang = (get_language() or "fr").lower() + symptoms_list = list(symptoms_qs) + + def get_display_name(s): + if lang.startswith("nl") and s.name_nl: + return s.name_nl + return s.name_fr + + def sort_key(s): + has_order = 0 if s.order_number is not None else 1 + order_num = s.order_number if s.order_number is not None else 999999 + name = (get_display_name(s) or "").lower() + return (has_order, order_num, name, s.pk) + + symptoms_list.sort(key=sort_key) + symptoms = symptoms_list + + symptoms_data = [ + { + 'id': s.id, + 'name': get_display_name(s), + 'priority': s.priority or '3', + 'origin_type': s.origin_type or '', + } + for s in symptoms + ] + symptoms_json = json.dumps(symptoms_data) return { 'asset': asset, @@ -536,6 +566,7 @@ def asset_interventions(context, asset): 'interventions': interventions, 'can_add_intervention': can_add, 'symptoms': symptoms, + 'symptoms_json': symptoms_json, 'status_colors': STATUS_COLORS, 'status_colors_dark_text': STATUS_COLORS_DARK_TEXT, } diff --git a/loko/assets/tests.py b/loko/assets/tests.py index ab7d2a3..8095d6b 100644 --- a/loko/assets/tests.py +++ b/loko/assets/tests.py @@ -2760,3 +2760,345 @@ class ParentChildAssetLocationAndGeoSyncTest(TestCase): lantern.refresh_from_db() self.assertEqual(lantern.geom, new_point) + + +class CreateInterventionForAssetWorkflowTest(TestCase): + def setUp(self): + from django.contrib.auth.models import User, Permission + from django.contrib.contenttypes.models import ContentType + from django.contrib.gis.geos import Point + from common.models import Thematic, UserThematics, UserConfig + from contracts.models import Contract, Company + from interventions.models import ( + Intervention, InterventionAsset, InterventionLocation, + InterventionTimeLine, InterventionPlanificationTimeLine, InterventionNote, + Symptom, DirectContractRouting + ) + from assets.models import ( + AssetCategory, + PublicLightingStreet, PublicLightingPole, PublicLightingPoleModel, + TrafficLightIntersection, TrafficLightContract, TrafficLightPole, TrafficLightPoleModel, + Structure, StructureContract, StructureLocation + ) + + self.thematic_pl, _ = Thematic.objects.get_or_create(code="publiclighting", defaults={'name_fr': "Éclairage public"}) + self.thematic_tl, _ = Thematic.objects.get_or_create(code="trafficlights", defaults={'name_fr': "Signalisation lumineuse"}) + self.thematic_struct, _ = Thematic.objects.get_or_create(code="structures", defaults={'name_fr': "Ouvrages d'art"}) + + self.user = User.objects.create_user(username="asset_agent", password="secret_password", email="agent@example.com") + self.user.is_staff = True + self.user.save() + # Donner la permission add_intervention + perm = Permission.objects.get(codename="add_intervention") + self.user.user_permissions.add(perm) + + # Configurer UserConfig et thématiques + self.user_config = UserConfig.objects.create( + user=self.user, + is_intern=True, + limit_interventions_to_contracts=False + ) + UserThematics.objects.create(user_config=self.user_config, thematic=self.thematic_pl, can_edit_interventions=True) + UserThematics.objects.create(user_config=self.user_config, thematic=self.thematic_tl, can_edit_interventions=True) + UserThematics.objects.create(user_config=self.user_config, thematic=self.thematic_struct, can_edit_interventions=True) + + import datetime + # Prestataires et contrats + self.company_a = Company.objects.create(name="Company Alpha") + self.company_b = Company.objects.create(name="Company Beta") + self.contract_a = Contract.objects.create( + company=self.company_a, + contract_number="CTR-PL-001", + start_date=datetime.date(2025, 1, 1), + end_date=datetime.date(2030, 1, 1), + description="Contrat PL Alpha" + ) + self.contract_tl = Contract.objects.create( + company=self.company_b, + contract_number="CTR-TL-001", + start_date=datetime.date(2025, 1, 1), + end_date=datetime.date(2030, 1, 1), + description="Contrat TL Curatif" + ) + + # Assets et localisations + self.pt_pl = Point(648819, 670788, srid=3812) + self.pl_category = AssetCategory.objects.create(name_fr="Support PL", code="CAT-PL-1", thematic=self.thematic_pl) + self.pl_street = PublicLightingStreet.objects.create(code="ST-PL-1", name_fr="Rue Royale", geom=MultiPolygon(self.pt_pl.buffer(5))) + self.pl_model = PublicLightingPoleModel.objects.create(code="PLM-1", name_fr="Candélabre 8m", category=self.pl_category) + self.pl_pole = PublicLightingPole.objects.create( + code="POLE-PL-001", + street=self.pl_street, + model=self.pl_model, + geom=self.pt_pl + ) + + self.client.login(username="asset_agent", password="secret_password") + + def test_create_intervention_full_workflow_from_asset(self): + from interventions.models import Symptom, Intervention, InterventionTimeLine, InterventionPlanificationTimeLine, InterventionNote + from django.urls import reverse + + symptom = Symptom.objects.create( + thematic=self.thematic_pl, + name_fr="Candélabre accidenté", + priority="1", + origin_type="damage", + contract=self.contract_a, + is_active=True + ) + + url = reverse('assets:create_intervention_for_asset', kwargs={ + 'asset_model': 'publiclightingpole', + 'asset_id': self.pl_pole.pk + }) + + response = self.client.post(url, { + 'title': '', # vide, doit être rempli automatiquement par le symptôme + 'symptom_id': symptom.id, + 'maintain_type': 'corrective', + 'init_description': 'Candélabre heurté par un camion hier soir.', + }) + + self.assertEqual(response.status_code, 200) + data = response.json() + self.assertTrue(data.get('success')) + intervention_id = data.get('intervention_id') + self.assertIsNotNone(intervention_id) + + intervention = Intervention.objects.get(pk=intervention_id) + + # 1. Titre déduit du symptôme + self.assertEqual(intervention.title, "Candélabre accidenté") + + # 2. Prestataire et contrat résolus depuis le symptôme + self.assertEqual(intervention.contract, self.contract_a) + self.assertEqual(intervention.assigned_provider, self.company_a) + + # 3. Géométrie et localisation créées depuis l'asset + self.assertIsNotNone(intervention.geom) + self.assertEqual(intervention.geom.srid, 3812) + self.assertEqual(intervention.geom.geom_type, 'MultiPolygon') + self.assertIsNotNone(intervention.lon) + self.assertIsNotNone(intervention.lat) + self.assertEqual(intervention.location_code, "ST-PL-1") + self.assertEqual(intervention.address, "Rue Royale") + + # 4. Type, catégorie, priorité et type d'origine remplis automatiquement + self.assertEqual(intervention.type, 'fixing') + self.assertEqual(intervention.origin_type, 'damage') + self.assertEqual(intervention.priority, '1') + self.assertEqual(intervention.asset_category, self.pl_category) + + # 5. Date de début prévue (+2h) & timeline de planification + self.assertIsNotNone(intervention.expected_begin_time) + plan_event = InterventionPlanificationTimeLine.objects.filter( + intervention=intervention, + event_type='expected_begin' + ).first() + self.assertIsNotNone(plan_event) + self.assertEqual(plan_event.event_user, self.user) + + # 6. Historique de création (Timeline) & Origine + timeline_event = InterventionTimeLine.objects.filter( + intervention=intervention, + event_type='creation' + ).first() + self.assertIsNotNone(timeline_event) + self.assertEqual(timeline_event.event_user, self.user) + self.assertEqual(timeline_event.to_status, 'in_preparation') + + # 7. Note d'origine créée + origin_note = InterventionNote.objects.filter( + intervention=intervention, + note_type=InterventionNote.NoteType.ORIGIN + ).first() + self.assertIsNotNone(origin_note) + self.assertEqual(origin_note.content, 'Candélabre heurté par un camion hier soir.') + self.assertEqual(origin_note.note_author, self.user) + + def test_create_intervention_trafficlight_contract_resolution(self): + from assets.models.trafficlights import TrafficLightIntersection, TrafficLightPole, TrafficLightPoleModel, TrafficLightContract + from interventions.models import Symptom, Intervention + from django.urls import reverse + from django.contrib.gis.geos import Point + + intersection = TrafficLightIntersection.objects.create( + code="TL-INT-100", + name_fr="Carrefour Arts-Loi", + geom=Point(649000, 671000, srid=3812) + ) + tl_pole_model = TrafficLightPoleModel.objects.create(code="TLPM-100", name_fr="Modèle Poteau TL") + tl_pole = TrafficLightPole.objects.create( + code="TL-POL-100", + intersection=intersection, + model=tl_pole_model, + geom=Point(649000, 671000, srid=3812) + ) + + tl_contract_rel = TrafficLightContract.objects.create( + intersection=intersection, + contract=self.contract_tl, + maintenance_type='corrective', + status='active' + ) + + symptom_tl = Symptom.objects.create( + thematic=self.thematic_tl, + name_fr="Feu éteint", + maintenance_type='corrective', + is_active=True + ) + + url = reverse('assets:create_intervention_for_asset', kwargs={ + 'asset_model': 'trafficlightpole', + 'asset_id': tl_pole.pk + }) + + response = self.client.post(url, { + 'title': 'Panne de signalisation', + 'symptom_id': symptom_tl.id, + 'maintain_type': 'corrective', + }) + + self.assertEqual(response.status_code, 200) + data = response.json() + self.assertTrue(data.get('success')) + + intervention = Intervention.objects.get(pk=data['intervention_id']) + self.assertEqual(intervention.contract, self.contract_tl) + self.assertEqual(intervention.assigned_provider, self.company_b) + self.assertEqual(intervention.thematic, self.thematic_tl) + + def test_create_intervention_structure_direct_contract_routing(self): + from assets.models.structures import Structure, StructureLocation + from contracts.models import Contract + from interventions.models import Symptom, Intervention, DirectContractRouting + from django.urls import reverse + from django.contrib.gis.geos import Point, MultiPolygon + import datetime + + struct_loc = StructureLocation.objects.create( + code="OA-001", + name_fr="Pont de Buda", + geom=MultiPolygon(Point(650000, 672000, srid=3812).buffer(10)) + ) + structure = Structure.objects.create( + code="STR-OA-001", + name_fr="Pont de Buda", + main_location=struct_loc + ) + + contract_struct = Contract.objects.create( + company=self.company_a, + contract_number="CTR-STRUCT-001", + start_date=datetime.date(2025, 1, 1), + end_date=datetime.date(2030, 1, 1), + description="Contrat Ouvrages d'Art" + ) + + symptom_struct = Symptom.objects.create( + thematic=self.thematic_struct, + name_fr="Fissure tablier", + is_active=True + ) + + DirectContractRouting.objects.create( + thematic='structures', + asset_code="STR-OA-001", + symptom=symptom_struct, + contract=contract_struct + ) + + url = reverse('assets:create_intervention_for_asset', kwargs={ + 'asset_model': 'structure', + 'asset_id': structure.pk + }) + + response = self.client.post(url, { + 'title': "Inspection fissure", + 'symptom_id': symptom_struct.id, + 'maintain_type': 'corrective', + }) + + self.assertEqual(response.status_code, 200) + data = response.json() + self.assertTrue(data.get('success')) + + intervention = Intervention.objects.get(pk=data['intervention_id']) + self.assertEqual(intervention.contract, contract_struct) + self.assertEqual(intervention.assigned_provider, self.company_a) + self.assertEqual(intervention.thematic, self.thematic_struct) + + def test_create_intervention_user_default_contract_fallback(self): + from interventions.models import Symptom, Intervention + from django.urls import reverse + + self.user_config.limit_interventions_to_contracts = True + self.user_config.default_contract = self.contract_a + self.user_config.save() + + symptom = Symptom.objects.create( + thematic=self.thematic_pl, + name_fr="Ampoule grillée", + is_active=True + ) + + url = reverse('assets:create_intervention_for_asset', kwargs={ + 'asset_model': 'publiclightingpole', + 'asset_id': self.pl_pole.pk + }) + + response = self.client.post(url, { + 'title': "Ampoule HS", + 'symptom_id': symptom.id, + 'maintain_type': 'corrective', + }) + + self.assertEqual(response.status_code, 200) + data = response.json() + self.assertTrue(data.get('success')) + + intervention = Intervention.objects.get(pk=data['intervention_id']) + self.assertEqual(intervention.contract, self.contract_a) + self.assertEqual(intervention.assigned_provider, self.company_a) + + def test_create_intervention_external_user_forces_to_be_approved(self): + from interventions.models import Symptom, Intervention + from django.urls import reverse + + # Utilisateur externe sans accès au contrat du symptôme + self.user_config.is_intern = False + self.user_config.limit_interventions_to_contracts = True + self.user_config.save() + + symptom = Symptom.objects.create( + thematic=self.thematic_pl, + name_fr="Candélabre endommagé", + contract=self.contract_a, + is_active=True + ) + + url = reverse('assets:create_intervention_for_asset', kwargs={ + 'asset_model': 'publiclightingpole', + 'asset_id': self.pl_pole.pk + }) + + response = self.client.post(url, { + 'title': "Candélabre endommagé", + 'symptom_id': symptom.id, + 'maintain_type': 'corrective', + }) + + self.assertEqual(response.status_code, 200) + data = response.json() + self.assertTrue(data.get('success')) + + intervention = Intervention.objects.get(pk=data['intervention_id']) + self.assertEqual(intervention.status, 'to_be_approved') + # Vérifier que le premier événement de timeline reflète to_be_approved + first_event = intervention.events.first() + self.assertEqual(first_event.to_status, 'to_be_approved') + self.assertEqual(first_event.event_user, self.user) + + diff --git a/loko/assets/views/asset_editing.py b/loko/assets/views/asset_editing.py index adafff4..9833f89 100644 --- a/loko/assets/views/asset_editing.py +++ b/loko/assets/views/asset_editing.py @@ -2388,36 +2388,94 @@ def create_intervention_for_asset(request, asset_model, asset_id): asset, model_class, content_type = _get_asset_by_ct(asset_model, asset_id) - thematic = None - if hasattr(asset, 'category') and asset.category and asset.category.thematic: - thematic = asset.category.thematic + from assets.permissions import _get_asset_thematic + thematic = _get_asset_thematic(asset) from interventions.permissions import can_add_intervention if not can_add_intervention(request.user, thematic): return JsonResponse({'success': False, 'error': _("Permission refusée pour créer une intervention.")}, status=403) - title = request.POST.get('title', '').strip() - if not title: - return JsonResponse({'success': False, 'error': _("Le titre est obligatoire.")}, status=400) + from interventions.models import ( + Intervention, InterventionAsset, Symptom, InterventionLocation, + InterventionTimeLine, InterventionPlanificationTimeLine, InterventionNote, + STATUS_ORDERS + ) + from interventions.views.helpers import ( + resolve_intervention_contract_and_provider, + resolve_asset_spatial_data, + resolve_asset_category, + auto_check_if_eligible, + ) + from datetime import timedelta + from django.utils.timezone import now symptom_id = request.POST.get('symptom_id') - maintain_type = request.POST.get('maintain_type', 'corrective') - priority = request.POST.get('priority', '3') - init_description = request.POST.get('init_description', '').strip() - - from interventions.models import Intervention, InterventionAsset, Symptom, InterventionLocation - symptom = None if symptom_id: symptom = Symptom.objects.filter(pk=symptom_id, is_active=True).first() - geom = getattr(asset, 'geom', None) - if geom is not None: - from django.contrib.gis.geos import Polygon, MultiPolygon - if isinstance(geom, Polygon): - geom = MultiPolygon(geom) - elif not isinstance(geom, MultiPolygon): - geom = None + title = request.POST.get('title', '').strip() + if not title and symptom: + title = symptom.get_name() + if not title: + return JsonResponse({'success': False, 'error': _("Le titre est obligatoire.")}, status=400) + + maintain_type = request.POST.get('maintain_type', 'corrective') + priority = request.POST.get('priority') + intervention_type = request.POST.get('type') + origin_type = request.POST.get('origin_type') + init_description = request.POST.get('init_description', '').strip() + + # Déduction automatique de la priorité + if not priority or priority == '3': + if symptom and symptom.priority: + priority = symptom.priority + else: + priority = priority or '3' + + # Déduction automatique du type d'intervention + if not intervention_type: + if maintain_type == 'corrective': + intervention_type = 'fixing' + elif maintain_type == 'ameliorative': + intervention_type = 'repair' + elif maintain_type == 'preventive': + intervention_type = 'inspection' + + # Déduction automatique de l'origin_type + if not origin_type: + if symptom and symptom.origin_type: + origin_type = symptom.origin_type + elif maintain_type == 'ameliorative': + origin_type = 'demand' + else: + origin_type = 'damage' + + # Déduction automatique de la catégorie d'asset + asset_category = resolve_asset_category(asset, symptom) + + # Déduction des données spatiales et localisation + spatial_data = resolve_asset_spatial_data(asset=asset) + geom = spatial_data['geom'] + lon = spatial_data['lon'] + lat = spatial_data['lat'] + address = spatial_data['address'] + location_code = spatial_data['location_code'] + loc_obj = spatial_data['location_obj'] + + # Résolution contrat et prestataire + contract, assigned_provider, force_to_be_approved = resolve_intervention_contract_and_provider( + thematic=thematic, + symptom=symptom, + asset=asset, + location=loc_obj, + user=request.user + ) + + status = 'to_be_approved' if force_to_be_approved else 'in_preparation' + status_order = STATUS_ORDERS.get(status, 1) + + expected_begin_time = now() + timedelta(hours=2) try: with transaction.atomic(): @@ -2426,20 +2484,28 @@ def create_intervention_for_asset(request, asset_model, asset_id): thematic=thematic, symptom=symptom, maintain_type=maintain_type, + type=intervention_type, + origin_type=origin_type, priority=priority, init_description=init_description, - status='in_preparation', + status=status, + status_order=status_order, created_by=request.user, + asset_category=asset_category, + contract=contract, + assigned_provider=assigned_provider, geom=geom, + lon=lon, + lat=lat, + address=address, + location_code=location_code, + expected_begin_time=expected_begin_time, ) InterventionAsset.objects.create( intervention=intervention, content_type=content_type, object_id=asset.pk ) - loc_obj = ( - asset.get_location() if hasattr(asset, 'get_location') else None - ) or getattr(asset, 'street', None) or getattr(asset, 'location', None) or getattr(asset, 'intersection', None) if loc_obj: loc_ct = ContentType.objects.get_for_model(loc_obj) InterventionLocation.objects.get_or_create( @@ -2448,6 +2514,38 @@ def create_intervention_for_asset(request, asset_model, asset_id): object_id=loc_obj.pk ) + # Auto-vérification si l'utilisateur a le droit can_check_interventions + auto_check_if_eligible(request.user, intervention) + + # Enregistrement dans l'historique (Timeline) & Section Origine + InterventionTimeLine.objects.create( + intervention=intervention, + event_user=request.user, + event_time=now(), + event_type='creation', + from_status=None, + to_status=intervention.status, + ) + + # Date de début prévue (Timeline planification) + InterventionPlanificationTimeLine.objects.create( + intervention=intervention, + event_user=request.user, + event_time=expected_begin_time, + event_type='expected_begin', + event_description=_("Début prévu initial."), + ) + + # Note d'origine si description initiale fournie + if init_description: + InterventionNote.objects.create( + intervention=intervention, + note_author=request.user, + note_time=now(), + content=init_description, + note_type=InterventionNote.NoteType.ORIGIN, + ) + redirect_url = reverse('interventions:intervention_detail', args=[intervention.id]) return JsonResponse({ 'success': True, diff --git a/loko/interventions/views/detail.py b/loko/interventions/views/detail.py index a7d3a71..1dda605 100644 --- a/loko/interventions/views/detail.py +++ b/loko/interventions/views/detail.py @@ -1638,106 +1638,12 @@ def interventions_add(request, thematic_code): # mémoriser le 1er objet pour updater location_code if first_loc_obj is None: Model = ct.model_class() - first_loc_obj = Model.objects.filter(pk=obj_id).only('code').first() + first_loc_obj = Model.objects.filter(pk=obj_id).first() if first_loc_obj and hasattr(first_loc_obj, 'code'): intervention.location_code = first_loc_obj.code intervention.save(update_fields=['location_code']) - # -------- Recherche contrat -------- - # Si thématique 'structures', recherche d'une règle de routage spécifique dans contract_routing.json - if thematic.code == 'structures': - structure_location_ct = ContentType.objects.get_for_model(StructureLocation) - structure_loc = InterventionLocation.objects.filter( - intervention=intervention, - content_type=structure_location_ct - ).first() - if structure_loc: - structure_location = StructureLocation.objects.filter(id=structure_loc.object_id).only('code').first() - structure = structure_location.structure if structure_location else None - if structure: - from interventions.models import DirectContractRouting - # 1. Tenter de trouver une règle avec le symptôme spécifique - routing = DirectContractRouting.objects.filter( - thematic='structures', - asset_code=structure.code, - symptom=intervention.symptom - ).select_related('contract').first() - - # 2. Si non trouvé, tenter de trouver la règle par défaut pour cet asset - if not routing: - routing = DirectContractRouting.objects.filter( - thematic='structures', - asset_code=structure.code, - symptom__isnull=True - ).select_related('contract').first() - - if routing and routing.contract: - contract = routing.contract - intervention.contract = contract - intervention.assigned_provider = contract.company - intervention.save(update_fields=['contract', 'assigned_provider']) - - # Si pas de contrat défini et que le symptôme a un maintenance_type, - # chercher le contrat actif lié à ce type de maintenance. - if ( - not intervention.contract - and intervention.symptom - and intervention.symptom.maintenance_type - ): - contract = None - if thematic.code == 'trafficlights': - # Récupérer l'intersection depuis les localisations - tl_intersection_ct = ContentType.objects.get_for_model(TrafficLightIntersection) - intersection_location = InterventionLocation.objects.filter( - intervention=intervention, - content_type=tl_intersection_ct - ).first() - - if intersection_location: - # Chercher le TrafficLightContract correspondant au maintenance_type du symptôme - contract = TrafficLightContract.objects.filter( - intersection_id=intersection_location.object_id, - maintenance_type=intervention.symptom.maintenance_type, - status='active' - ).select_related('contract__company').first() - - elif thematic.code == 'structures': - structure_location_ct = ContentType.objects.get_for_model(StructureLocation) - structure_loc = InterventionLocation.objects.filter( - intervention=intervention, - content_type=structure_location_ct - ).first() - structure_location = StructureLocation.objects.filter(id=structure_loc.object_id).first() if structure_loc else None - structure = structure_location.structure if structure_location else None - if structure: - contract = StructureContract.objects.filter( - structure_id=structure.id, - maintenance_type=intervention.symptom.maintenance_type, - status='active' - ).select_related('contract__company').first() - - elif thematic.code == 'clean': - location_ct = ContentType.objects.get_for_model(CleanLocation) - clean_loc = InterventionLocation.objects.filter( - intervention=intervention, - content_type=location_ct - ).first() - clean_location = CleanLocation.objects.filter(id=clean_loc.object_id).first() if clean_loc else None - if clean_location: - contract = CleanLocationContract.objects.filter( - location_id=clean_location.id, - maintenance_type=intervention.symptom.maintenance_type, - status='active' - ).select_related('contract__company').first() - - if contract: - intervention.contract = contract.contract - if not intervention.assigned_provider: - intervention.assigned_provider = contract.contract.company - intervention.save(update_fields=['contract', 'assigned_provider']) - - # -------- Assets (optionnel) -------- # 1) JSON 'assets' = [{ct,obj}] (recommandé) asset_pairs = _parse_pairs_json(request.POST.get('assets')) @@ -1754,6 +1660,7 @@ def interventions_add(request, thematic_code): ct = ContentType.objects.get(model=equip_type.lower()) asset_pairs = [(ct.id, int(equip_id))] + first_asset_obj = None for ct_id, obj_id in asset_pairs: ct = ContentType.objects.get_for_id(ct_id) InterventionAsset.objects.get_or_create( @@ -1761,6 +1668,9 @@ def interventions_add(request, thematic_code): content_type=ct, object_id=obj_id, ) + if first_asset_obj is None: + Model = ct.model_class() + first_asset_obj = Model.objects.filter(pk=obj_id).first() if inspection_config: templates = list(inspection_config.operation_templates.all()) @@ -1772,78 +1682,26 @@ def interventions_add(request, thematic_code): resolved_assets = _resolve_assets_from_pairs(asset_pairs) _create_equipment_operations_for_assets(intervention, resolved_assets, ignore_filters=True) - # -------- Recherche contrat ITS (après sauvegarde des assets) -------- - if ( - thematic.code == 'its' - and not intervention.contract - and intervention.symptom - and intervention.symptom.maintenance_type - ): - its_equipment_ct = ContentType.objects.get_for_model(ITSGeoAsset) - its_equipment = InterventionAsset.objects.filter( - intervention=intervention, - content_type=its_equipment_ct - ).first() - - if its_equipment: - its_contract = ITSGeoAssetContract.objects.filter( - asset_id=its_equipment.object_id, - maintenance_type=intervention.symptom.maintenance_type, - status='active' - ).select_related('contract__company').first() - - if its_contract: - intervention.contract = its_contract.contract - if not intervention.assigned_provider: - intervention.assigned_provider = its_contract.contract.company - intervention.save(update_fields=['contract', 'assigned_provider']) - - # Fallback contrat par défaut pour les utilisateurs à visibilité limitée aux contrats - if not intervention.contract and user_config and user_config.limit_interventions_to_contracts: - default_contract = user_config.get_default_contract() - if default_contract: - intervention.contract = default_contract - if not intervention.assigned_provider: - intervention.assigned_provider = default_contract.company + # -------- Résolution contrat et prestataire -------- + if not intervention.contract: + resolved_contract, resolved_provider, force_to_be_approved = resolve_intervention_contract_and_provider( + thematic=thematic, + symptom=intervention.symptom, + asset=first_asset_obj, + location=first_loc_obj, + user=request.user + ) + if resolved_contract: + intervention.contract = resolved_contract + if resolved_provider and not intervention.assigned_provider: + intervention.assigned_provider = resolved_provider + if resolved_contract or resolved_provider: intervention.save(update_fields=['contract', 'assigned_provider']) - # -------- Forcer le statut 'à approuver' pour les utilisateurs externes -------- - # Applicable si limit_interventions_to_contracts est activé et que l'utilisateur - # n'a pas accès au contrat résolu (ou à celui du symptôme). - if user_config and not user_config.is_intern and user_config.limit_interventions_to_contracts: - _force_approved = False - _symptom = intervention.symptom - - if _symptom: - if _symptom.contract_id: - # Cas 1 : le symptôme est explicitement lié à un contrat - # → l'utilisateur doit avoir accès à ce contrat pour éviter to_be_approved - _has_access = UserContractAccess.objects.filter( - user_config=user_config, - contract_id=_symptom.contract_id, - ).exists() - if not _has_access: - _force_approved = True - - elif _symptom.maintenance_type: - # Cas 2 : pas de contrat sur le symptôme mais maintenance_type défini - # → vérifier si le contrat résolu (depuis la localisation/asset) est accessible - if intervention.contract_id: - _has_access = UserContractAccess.objects.filter( - user_config=user_config, - contract_id=intervention.contract_id, - ).exists() - if not _has_access: - _force_approved = True - else: - # Aucun contrat résolu : la localisation/asset n'est pas liée - # à un contrat de l'utilisateur → approbation requise - _force_approved = True - - if _force_approved and intervention.status != 'to_be_approved': + if force_to_be_approved and intervention.status != 'to_be_approved': intervention.status = 'to_be_approved' + intervention.status_order = STATUS_ORDERS.get('to_be_approved', 2) intervention.save(update_fields=['status', 'status_order']) - # Mettre à jour l'entrée de timeline créée lors de la création event.to_status = 'to_be_approved' event.save(update_fields=['to_status']) diff --git a/loko/interventions/views/helpers.py b/loko/interventions/views/helpers.py index 7eac521..755e835 100644 --- a/loko/interventions/views/helpers.py +++ b/loko/interventions/views/helpers.py @@ -60,7 +60,8 @@ from interventions.models import (InterventionSubscription, Intervention, Interv InterventionOccupation, Operation, OperationTemplate, OperationTemplateItem, OperationStatus, OperationStatusSet, MeasurementDefinition, GuidedOperationTemplate, GuidedOperationData, GuidedOperationMedia, - InterventionPostPeriodQuantity, InterventionEmailLog, InterventionAssignment) + InterventionPostPeriodQuantity, InterventionEmailLog, InterventionAssignment, + DirectContractRouting) from interventions.models import STATUS_CHOICES, STATUS_ORDERS, MAINTAIN_CHOICES, TYPE_CHOICES, ORIGIN_TYPE_CHOICES, PAUSE_REASON_CHOICES from interventions.views.forms import (InterventionForm, InterventionNoteForm, InterventionDocumentForm, InterventionOccupationForm, PreventiveInterventionForm, AssetsLinkForm, @@ -95,6 +96,7 @@ from assets.models import ( AbstractAsset, AbstractAssetModel, ITSGeoAsset, ITSGeoAssetContract, + ITSLocation, ITSLocationContract, CleanLocation, CleanLocationContract, ) from assets.permissions import get_allowed_actions_for_asset @@ -211,6 +213,10 @@ __all__ = [ 'get_intervention_attachments', 'inspection_configuration_geojson', 'is_smtp_available', + 'resolve_asset_category', + 'resolve_asset_spatial_data', + 'resolve_coords_from_hierarchy', + 'resolve_intervention_contract_and_provider', 'resolve_selected_assets', 'to_multipolygon', 'update_operation_comment' @@ -3083,16 +3089,414 @@ def to_multipolygon(geom, srid=None): gtype = geom.geom_type # 'Polygon', 'MultiPolygon', 'GeometryCollection', etc. if gtype == 'Polygon': - return MultiPolygon(geom) # ou MultiPolygon([geom]) + return MultiPolygon(geom) if gtype == 'MultiPolygon': - return geom # ne pas ré-envelopper + return geom if gtype == 'GeometryCollection': - polys = [g for g in geom if g.geom_type == 'Polygon'] - return MultiPolygon(polys) if polys else None - # Optionnel: gérer LineString/Point, sinon considérer comme erreur + polys = [g for g in geom if g.geom_type in ['Polygon', 'MultiPolygon']] + if polys: + all_polys = [] + for p in polys: + if p.geom_type == 'MultiPolygon': + all_polys.extend(list(p)) + else: + all_polys.append(p) + return MultiPolygon(all_polys) + return None + if gtype in ['Point', 'MultiPoint', 'LineString', 'MultiLineString']: + buffered = geom.buffer(1) + if buffered.geom_type == 'Polygon': + return MultiPolygon(buffered) + elif buffered.geom_type == 'MultiPolygon': + return buffered raise ValueError(f"Type géométrique non supporté: {gtype}") +def resolve_coords_from_hierarchy(loc, max_depth=10): + """ + Remonte la hiérarchie parent_location pour trouver lon/lat/geom. + Retourne (lon, lat, geom) - les valeurs peuvent être None. + """ + current = loc + depth = 0 + lon, lat, geom = None, None, None + + while current is not None and depth < max_depth: + current_lon = getattr(current, "lon", None) or getattr(current, "longitude", None) + current_lat = getattr(current, "lat", None) or getattr(current, "latitude", None) + current_geom = getattr(current, "geom", None) + + if lon is None and current_lon is not None: + lon = current_lon + if lat is None and current_lat is not None: + lat = current_lat + if geom is None and current_geom is not None: + geom = current_geom + + if lon is not None and lat is not None and geom is not None: + break + + parent = getattr(current, "parent_location", None) + if parent is None: + struct = getattr(current, "structure", None) + parent = getattr(struct, "main_location", None) if struct else None + + current = parent + depth += 1 + + return lon, lat, geom + + +def resolve_intervention_contract_and_provider(thematic, symptom, asset=None, location=None, user=None): + """ + Détermine le contrat et le prestataire associés pour une intervention, ainsi que le flag force_to_be_approved + si l'utilisateur externe n'a pas accès au contrat résolu. + + Retourne un tuple (contract, assigned_provider, force_to_be_approved) + """ + contract = None + assigned_provider = None + force_to_be_approved = False + + thematic_code = thematic.code if hasattr(thematic, 'code') else (str(thematic) if thematic else None) + + # 1. Contrat / prestataire direct du symptôme + if symptom: + if symptom.contract: + contract = symptom.contract + assigned_provider = symptom.provider or (contract.company if contract else None) + elif symptom.provider: + assigned_provider = symptom.provider + + # 2. Routage direct (thématique 'structures') + if not contract and thematic_code == 'structures': + structure = None + if asset: + if isinstance(asset, Structure): + structure = asset + elif hasattr(asset, 'structure') and asset.structure: + structure = asset.structure + if not structure and location: + if isinstance(location, Structure): + structure = location + elif hasattr(location, 'structure') and location.structure: + structure = location.structure + + if structure and getattr(structure, 'code', None): + routing = DirectContractRouting.objects.filter( + thematic='structures', + asset_code=structure.code, + symptom=symptom + ).select_related('contract__company').first() + + if not routing: + routing = DirectContractRouting.objects.filter( + thematic='structures', + asset_code=structure.code, + symptom__isnull=True + ).select_related('contract__company').first() + + if routing and routing.contract: + contract = routing.contract + if not assigned_provider: + assigned_provider = contract.company + + # 3. Contrat actif lié au maintenance_type du symptôme + if not contract and symptom and symptom.maintenance_type: + mt = symptom.maintenance_type + if thematic_code == 'trafficlights': + intersection_id = None + if location: + if isinstance(location, TrafficLightIntersection): + intersection_id = location.id + elif hasattr(location, 'intersection_id') and location.intersection_id: + intersection_id = location.intersection_id + if not intersection_id and asset: + if isinstance(asset, TrafficLightIntersection): + intersection_id = asset.id + elif hasattr(asset, 'intersection_id') and asset.intersection_id: + intersection_id = asset.intersection_id + + if intersection_id: + tl_contract = TrafficLightContract.objects.filter( + intersection_id=intersection_id, + maintenance_type=mt, + status='active' + ).select_related('contract__company').first() + if tl_contract and tl_contract.contract: + contract = tl_contract.contract + if not assigned_provider: + assigned_provider = contract.company + + elif thematic_code == 'structures': + structure_id = None + if location: + if isinstance(location, Structure): + structure_id = location.id + elif hasattr(location, 'structure_id') and location.structure_id: + structure_id = location.structure_id + if not structure_id and asset: + if isinstance(asset, Structure): + structure_id = asset.id + elif hasattr(asset, 'structure_id') and asset.structure_id: + structure_id = asset.structure_id + + if structure_id: + struct_contract = StructureContract.objects.filter( + structure_id=structure_id, + maintenance_type=mt, + status='active' + ).select_related('contract__company').first() + if struct_contract and struct_contract.contract: + contract = struct_contract.contract + if not assigned_provider: + assigned_provider = contract.company + + elif thematic_code == 'clean': + clean_loc_id = None + if location: + if isinstance(location, CleanLocation): + clean_loc_id = location.id + elif hasattr(location, 'location_id') and location.location_id: + clean_loc_id = location.location_id + if not clean_loc_id and asset: + if isinstance(asset, CleanLocation): + clean_loc_id = asset.id + elif hasattr(asset, 'location_id') and asset.location_id: + clean_loc_id = asset.location_id + + if clean_loc_id: + clean_contract = CleanLocationContract.objects.filter( + location_id=clean_loc_id, + maintenance_type=mt, + status='active' + ).select_related('contract__company').first() + if clean_contract and clean_contract.contract: + contract = clean_contract.contract + if not assigned_provider: + assigned_provider = contract.company + + elif thematic_code == 'its': + its_contract = None + if asset and isinstance(asset, ITSGeoAsset): + its_contract = ITSGeoAssetContract.objects.filter( + asset_id=asset.id, + maintenance_type=mt, + status='active' + ).select_related('contract__company').first() + elif location and isinstance(location, ITSLocation): + its_contract = ITSLocationContract.objects.filter( + location_id=location.id, + maintenance_type=mt, + status='active' + ).select_related('contract__company').first() + + if its_contract and its_contract.contract: + contract = its_contract.contract + if not assigned_provider: + assigned_provider = contract.company + + # 4. Fallback contrat par défaut de l'utilisateur (limit_interventions_to_contracts) + user_config = None + if user: + try: + user_config = getattr(user, 'config', None) or UserConfig.objects.filter(user=user).first() + except Exception: + user_config = None + + if not contract and user_config and user_config.limit_interventions_to_contracts: + default_contract = user_config.get_default_contract() + if default_contract: + contract = default_contract + if not assigned_provider: + assigned_provider = default_contract.company + + # 5. Vérification d'accès contrat pour utilisateur externe (limit_interventions_to_contracts) + if user_config and not user_config.is_intern and user_config.limit_interventions_to_contracts: + if symptom: + if symptom.contract_id: + has_access = UserContractAccess.objects.filter( + user_config=user_config, + contract_id=symptom.contract_id, + ).exists() + if not has_access: + force_to_be_approved = True + elif symptom.maintenance_type: + if contract: + has_access = UserContractAccess.objects.filter( + user_config=user_config, + contract_id=contract.id, + ).exists() + if not has_access: + force_to_be_approved = True + else: + force_to_be_approved = True + + return contract, assigned_provider, force_to_be_approved + + +def resolve_asset_spatial_data(asset=None, location=None): + """ + Extrait et calcule la géométrie (MultiPolygon SRID 3812), les coordonnées (lon, lat WGS84 4326), + l'adresse, le location_code et l'objet de localisation parent. + + Retourne un dictionnaire: + { + 'geom': MultiPolygon | None (SRID 3812), + 'lon': float | None (WGS84 4326), + 'lat': float | None (WGS84 4326), + 'address': str | None, + 'location_code': str | None, + 'location_obj': Model | None, + } + """ + geom = None + lon = None + lat = None + address = None + location_code = None + loc_obj = location + + # 1. Déterminer l'objet localisation parent si non fourni + if not loc_obj and asset: + loc_obj = ( + asset.get_location() if hasattr(asset, 'get_location') else None + ) or getattr(asset, 'street', None) or getattr(asset, 'location', None) or getattr(asset, 'intersection', None) or getattr(asset, 'structure', None) or getattr(asset, 'section', None) + + # 2. Extraction adresse et location_code + if asset: + address = getattr(asset, 'address', None) + if not address and hasattr(asset, 'get_address'): + try: + address = asset.get_address() + except Exception: + address = None + location_code = getattr(asset, 'code', None) + + if not address and loc_obj: + address = getattr(loc_obj, 'address', None) + if not address and hasattr(loc_obj, 'get_name'): + try: + address = loc_obj.get_name() + except Exception: + address = None + + if loc_obj and hasattr(loc_obj, 'code') and loc_obj.code: + location_code = loc_obj.code + + # 3. Extraction de la géométrie source + raw_geom = getattr(asset, 'geom', None) if asset else None + if raw_geom is None and loc_obj: + raw_geom = getattr(loc_obj, 'geom', None) + if raw_geom is None: + _, _, raw_geom = resolve_coords_from_hierarchy(loc_obj) + + # Coordonnées directes (lat/lon) + if asset: + lon = getattr(asset, 'lon', None) or getattr(asset, 'longitude', None) + lat = getattr(asset, 'lat', None) or getattr(asset, 'latitude', None) + if lon is None and lat is None and hasattr(asset, 'get_coordinates'): + try: + coords = asset.get_coordinates() + if coords and len(coords) >= 2: + lat, lon = coords[0], coords[1] + except Exception: + pass + + if (lon is None or lat is None) and loc_obj: + h_lon, h_lat, _ = resolve_coords_from_hierarchy(loc_obj) + if lon is None: + lon = h_lon + if lat is None: + lat = h_lat + + # 4. Traitement de la géométrie vers MultiPolygon SRID 3812 + if raw_geom is not None and not raw_geom.empty: + try: + geom_3812 = raw_geom.clone() + if getattr(geom_3812, 'srid', None) and geom_3812.srid != 3812: + geom_3812.transform(3812) + elif not getattr(geom_3812, 'srid', None): + geom_3812.srid = 3812 + + gtype = geom_3812.geom_type + if gtype in ['Point', 'MultiPoint', 'LineString', 'MultiLineString']: + buffered = geom_3812.buffer(1) + geom = to_multipolygon(buffered, srid=3812) + elif gtype == 'Polygon': + geom = MultiPolygon(geom_3812) + geom.srid = 3812 + elif gtype == 'MultiPolygon': + geom = geom_3812 + elif gtype == 'GeometryCollection': + polys = [g for g in geom_3812 if g.geom_type in ['Polygon', 'MultiPolygon']] + if polys: + all_polys = [] + for p in polys: + if p.geom_type == 'MultiPolygon': + all_polys.extend(list(p)) + else: + all_polys.append(p) + geom = MultiPolygon(all_polys) + geom.srid = 3812 + + # Extraire lon/lat si non encore définis + if lon is None or lat is None: + geom_4326 = raw_geom.clone() + if getattr(geom_4326, 'srid', None) != 4326: + if not getattr(geom_4326, 'srid', None): + geom_4326.srid = 3812 + geom_4326.transform(4326) + centroid = geom_4326.centroid + if lon is None: + lon = centroid.x + if lat is None: + lat = centroid.y + except Exception as e: + logger.warning(f"Impossible de convertir la géométrie : {e}") + + # Si pas de géométrie valide mais lon/lat sont présents : créer un buffer de 1m + if geom is None and lon is not None and lat is not None: + try: + pt = Point(float(lon), float(lat), srid=4326) + pt.transform(3812) + geom = MultiPolygon(pt.buffer(1)) + geom.srid = 3812 + except Exception as e: + logger.warning(f"Impossible de générer le buffer géométrique depuis lon/lat : {e}") + + return { + 'geom': geom, + 'lon': float(lon) if lon is not None else None, + 'lat': float(lat) if lat is not None else None, + 'address': address, + 'location_code': location_code, + 'location_obj': loc_obj, + } + + +def resolve_asset_category(asset, symptom=None): + """ + Résout la catégorie d'asset appropriée pour une intervention liée à un asset. + """ + category = getattr(asset, 'category', None) + if not category: + # Vérifier si l'asset a un modèle avec catégorie + for model_attr in ('pole_model', 'lantern_model', 'controller_model', 'cabinet_model', 'detector_model', 'radar_model', 'model'): + model_obj = getattr(asset, model_attr, None) + if model_obj and getattr(model_obj, 'category', None): + category = model_obj.category + break + + # Si toujours pas de catégorie et qu'un symptôme est fourni avec une seule catégorie + if not category and symptom: + symptom_cats = list(symptom.asset_categories.all()) + if len(symptom_cats) == 1: + category = symptom_cats[0] + + return category + +