refactor: remove automatic contract assignment logic from intervention detail view
This commit is contained in:
parent
1e7e8ae697
commit
4e35da62b8
6 changed files with 963 additions and 200 deletions
|
|
@ -123,10 +123,10 @@
|
|||
|
||||
<div class="mb-3">
|
||||
<label class="form-label fw-semibold">{% translate "Symptôme" %}</label>
|
||||
<select name="symptom_id" class="form-select">
|
||||
<select name="symptom_id" id="asset_intervention_symptom_select" class="form-select">
|
||||
<option value="">{% translate "-- Sélectionner un symptôme --" %}</option>
|
||||
{% for s in symptoms %}
|
||||
<option value="{{ s.id }}">{{ s.get_name }}</option>
|
||||
<option value="{{ s.id }}" data-name="{{ s.get_name }}" data-priority="{{ s.priority|default:'3' }}">{{ s.get_name }}</option>
|
||||
{% endfor %}
|
||||
</select>
|
||||
</div>
|
||||
|
|
@ -156,7 +156,7 @@
|
|||
|
||||
<div class="mb-3">
|
||||
<label class="form-label fw-semibold">{% translate "Priorité" %}</label>
|
||||
<select name="priority" class="form-select">
|
||||
<select name="priority" id="asset_intervention_priority_select" class="form-select">
|
||||
<option value="1">{% translate "Urgent" %}</option>
|
||||
<option value="2">{% translate "Élevé" %}</option>
|
||||
<option value="3" selected>{% translate "Normal" %}</option>
|
||||
|
|
@ -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();
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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'])
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue