424 lines
18 KiB
Python
424 lines
18 KiB
Python
from django.shortcuts import render, get_object_or_404, redirect
|
|
from django.http import JsonResponse, Http404, HttpResponseForbidden
|
|
from django.contrib.auth.decorators import login_required
|
|
from django.views.decorators.http import require_POST
|
|
from django.db.models import Q
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.utils import timezone
|
|
from django.utils.translation import gettext as _
|
|
|
|
from common.models import Thematic
|
|
from assets.models import AssetCategory
|
|
from assets.models.inspection import (
|
|
AssetInspection, ParkingSpotInspection,
|
|
INSPECTION_RESULT_CHOICES, INSPECTION_NATURE_CHOICES, INSPECTOR_ROLE_CHOICES
|
|
)
|
|
|
|
|
|
def get_user_inspectable_thematics(user):
|
|
"""
|
|
Retourne les thématiques pour lesquelles l'utilisateur détient la permission can_inspect.
|
|
Si l'utilisateur n'a pas de UserConfig, retourne None.
|
|
"""
|
|
if hasattr(user, 'config') and user.config:
|
|
res = user.config.get_inspectable_thematics()
|
|
return res.order_by('name_fr') if res is not None else None
|
|
return None
|
|
|
|
|
|
def get_user_viewable_thematics(user):
|
|
"""
|
|
Retourne les thématiques pour lesquelles l'utilisateur détient la permission de visualisation (can_view_assets).
|
|
Si l'utilisateur n'a pas de UserConfig, retourne None.
|
|
"""
|
|
if hasattr(user, 'config') and user.config:
|
|
res = user.config.get_viewable_assets_thematics()
|
|
return res.order_by('name_fr') if res is not None else None
|
|
return None
|
|
|
|
|
|
@login_required
|
|
def inspections_list(request):
|
|
"""
|
|
Vue de liste et de gestion des inspections avec filtre par défaut 'Mes inspections'.
|
|
"""
|
|
filter_type = request.GET.get('tab', 'my') # 'my' ou 'all'
|
|
thematic_code = request.GET.get('thematic', '')
|
|
status_filter = request.GET.get('status', '')
|
|
nature_filter = request.GET.get('nature', '')
|
|
search_q = request.GET.get('q', '').strip()
|
|
|
|
qs = AssetInspection.objects.all().select_related('asset_content_type', 'inspector').prefetch_related('documents').order_by('-inspection_date')
|
|
|
|
if filter_type == 'my':
|
|
qs = qs.filter(inspector=request.user)
|
|
|
|
if thematic_code:
|
|
try:
|
|
thematic = Thematic.objects.get(code=thematic_code)
|
|
categories = AssetCategory.objects.filter(thematic=thematic)
|
|
cts = ContentType.objects.filter(assetcategory__in=categories).distinct()
|
|
qs = qs.filter(asset_content_type__in=cts)
|
|
except Thematic.DoesNotExist:
|
|
pass
|
|
|
|
if status_filter:
|
|
qs = qs.filter(result_status=status_filter)
|
|
|
|
if nature_filter:
|
|
qs = qs.filter(nature=nature_filter)
|
|
|
|
if search_q:
|
|
search_filter = (
|
|
Q(notes__icontains=search_q) |
|
|
Q(inspector__first_name__icontains=search_q) |
|
|
Q(inspector__last_name__icontains=search_q) |
|
|
Q(inspector__username__icontains=search_q)
|
|
)
|
|
|
|
asset_q_conditions = Q()
|
|
for ct in ContentType.objects.filter(app_label='assets'):
|
|
ModelClass = ct.model_class()
|
|
if not ModelClass:
|
|
continue
|
|
|
|
model_fields = [f.name for f in ModelClass._meta.get_fields()]
|
|
asset_search = Q()
|
|
if 'code' in model_fields:
|
|
asset_search |= Q(code__icontains=search_q)
|
|
if 'name' in model_fields:
|
|
asset_search |= Q(name__icontains=search_q)
|
|
if 'name_fr' in model_fields:
|
|
asset_search |= Q(name_fr__icontains=search_q)
|
|
|
|
if asset_search:
|
|
matching_ids = list(ModelClass.objects.filter(asset_search).values_list('pk', flat=True)[:500])
|
|
if matching_ids:
|
|
asset_q_conditions |= Q(asset_content_type=ct, asset_object_id__in=matching_ids)
|
|
|
|
if asset_q_conditions:
|
|
search_filter |= asset_q_conditions
|
|
|
|
qs = qs.filter(search_filter)
|
|
|
|
thematics = get_user_viewable_thematics(request.user)
|
|
insp_thematics = get_user_inspectable_thematics(request.user)
|
|
can_inspect = bool(insp_thematics and insp_thematics.exists())
|
|
|
|
is_mobile = request.path.startswith('/mobile/')
|
|
base_template = "mobile/mobile_base.html" if is_mobile else "base.html"
|
|
|
|
context = {
|
|
'base_template': base_template,
|
|
'is_mobile': is_mobile,
|
|
'inspections': qs[:100],
|
|
'total_count': qs.count(),
|
|
'filter_type': filter_type,
|
|
'selected_thematic': thematic_code,
|
|
'selected_status': status_filter,
|
|
'selected_nature': nature_filter,
|
|
'search_q': search_q,
|
|
'thematics': thematics,
|
|
'can_inspect': can_inspect,
|
|
'result_choices': INSPECTION_RESULT_CHOICES,
|
|
'nature_choices': INSPECTION_NATURE_CHOICES,
|
|
}
|
|
return render(request, 'assets/inspections_list.html', context)
|
|
|
|
|
|
@login_required
|
|
def quick_inspection_map(request):
|
|
"""
|
|
Vue cartographique d'inspection rapide optimisée mobile & GPS.
|
|
"""
|
|
thematics = get_user_inspectable_thematics(request.user)
|
|
if not thematics or not thematics.exists():
|
|
return HttpResponseForbidden(_("Vous ne disposez d'aucune thématique autorisée pour réaliser des inspections."))
|
|
|
|
categories = AssetCategory.objects.filter(thematic__in=thematics).select_related('thematic').order_by('name_fr')
|
|
|
|
is_mobile = request.path.startswith('/mobile/')
|
|
base_template = "mobile/mobile_base.html" if is_mobile else "base.html"
|
|
|
|
from assets.models import ParkingSpot
|
|
parking_spot_ct_id = ContentType.objects.get_for_model(ParkingSpot).id
|
|
|
|
context = {
|
|
'base_template': base_template,
|
|
'is_mobile': is_mobile,
|
|
'thematics': thematics,
|
|
'categories': categories,
|
|
'parking_spot_ct_id': parking_spot_ct_id,
|
|
}
|
|
return render(request, 'assets/quick_inspection_map.html', context)
|
|
|
|
|
|
@login_required
|
|
@require_POST
|
|
def create_inspection_api(request):
|
|
"""
|
|
API endpoint pour l'enregistrement d'une nouvelle inspection.
|
|
"""
|
|
content_type_id = request.POST.get('content_type_id')
|
|
object_id = request.POST.get('object_id')
|
|
result_status = request.POST.get('result_status', 'compliant')
|
|
nature = request.POST.get('nature', 'visual')
|
|
inspector_role = request.POST.get('inspector_role', 'observer')
|
|
notes = request.POST.get('notes', '').strip()
|
|
verified_obsolescence = request.POST.get('verified_obsolescence', '').strip() or None
|
|
apply_changes = request.POST.get('apply_changes_to_asset', 'true') == 'true'
|
|
|
|
if not content_type_id or not object_id:
|
|
return JsonResponse({'success': False, 'error': _("Cible d'inspection non spécifiée.")}, status=400)
|
|
|
|
try:
|
|
content_type = ContentType.objects.get(pk=content_type_id)
|
|
target_obj = content_type.get_object_for_this_type(pk=object_id)
|
|
except Exception as e:
|
|
return JsonResponse({'success': False, 'error': _("Cible introuvable : %s") % str(e)}, status=404)
|
|
|
|
is_parking_spot = (content_type.model == 'parkingspot')
|
|
|
|
try:
|
|
if is_parking_spot:
|
|
inspection = ParkingSpotInspection(
|
|
asset_content_type=content_type,
|
|
asset_object_id=target_obj.pk,
|
|
inspector=request.user,
|
|
inspection_date=timezone.now(),
|
|
result_status=result_status,
|
|
nature=nature,
|
|
inspector_role=inspector_role,
|
|
verified_obsolescence=verified_obsolescence,
|
|
notes=notes,
|
|
apply_changes_to_asset=apply_changes,
|
|
)
|
|
|
|
if 'verified_standard_stand_count' in request.POST and request.POST['verified_standard_stand_count'] != '':
|
|
inspection.verified_standard_stand_count = int(request.POST['verified_standard_stand_count'])
|
|
if 'verified_cargo_stand_count' in request.POST and request.POST['verified_cargo_stand_count'] != '':
|
|
inspection.verified_cargo_stand_count = int(request.POST['verified_cargo_stand_count'])
|
|
if 'verified_stand_type' in request.POST and request.POST['verified_stand_type'] != '':
|
|
inspection.verified_stand_type = request.POST['verified_stand_type']
|
|
if 'verified_has_scooter_drop_zone' in request.POST:
|
|
inspection.verified_has_scooter_drop_zone = (request.POST['verified_has_scooter_drop_zone'] == 'true')
|
|
if 'verified_marking_quality' in request.POST and request.POST['verified_marking_quality'] != '':
|
|
inspection.verified_marking_quality = request.POST['verified_marking_quality']
|
|
|
|
inspection.save()
|
|
else:
|
|
inspection = AssetInspection.objects.create(
|
|
asset_content_type=content_type,
|
|
asset_object_id=target_obj.pk,
|
|
inspector=request.user,
|
|
inspection_date=timezone.now(),
|
|
result_status=result_status,
|
|
nature=nature,
|
|
inspector_role=inspector_role,
|
|
verified_obsolescence=verified_obsolescence,
|
|
apply_changes_to_asset=apply_changes,
|
|
)
|
|
|
|
# Enregistrement des photos et documents joints
|
|
uploaded_files = request.FILES.getlist('photos') + request.FILES.getlist('documents') + request.FILES.getlist('files')
|
|
if uploaded_files:
|
|
from assets.models.inspection import InspectionDocument
|
|
for f in uploaded_files:
|
|
InspectionDocument.objects.create(
|
|
inspection=inspection,
|
|
file=f
|
|
)
|
|
|
|
created_intervention_code = None
|
|
if request.POST.get('create_intervention') == 'true' and result_status in ('non_compliant', 'damaged', 'missing'):
|
|
from assets.permissions import _get_asset_thematic
|
|
thematic = _get_asset_thematic(target_obj)
|
|
if not thematic:
|
|
if is_parking_spot:
|
|
thematic = Thematic.objects.filter(code='parking').first()
|
|
else:
|
|
thematic = Thematic.objects.first()
|
|
|
|
from interventions.models import Intervention, InterventionAsset
|
|
intervention = Intervention.objects.create(
|
|
title=_("Intervention suite à inspection non conforme (%s)") % target_obj,
|
|
thematic=thematic,
|
|
maintain_type='corrective',
|
|
priority='2' if result_status in ('damaged', 'missing') else '3',
|
|
init_description=notes or _("Signalement automatique lors de l'inspection du %s.") % timezone.now().strftime('%d/%m/%Y'),
|
|
status='in_preparation',
|
|
created_by=request.user,
|
|
geom=getattr(target_obj, 'geom', None),
|
|
)
|
|
InterventionAsset.objects.create(
|
|
intervention=intervention,
|
|
content_type=content_type,
|
|
object_id=target_obj.pk
|
|
)
|
|
created_intervention_code = intervention.code
|
|
|
|
return JsonResponse({
|
|
'success': True,
|
|
'inspection_id': inspection.id,
|
|
'created_intervention_code': created_intervention_code,
|
|
'message': _("Inspection enregistrée avec succès.")
|
|
})
|
|
except Exception as e:
|
|
return JsonResponse({'success': False, 'error': str(e)}, status=500)
|
|
|
|
|
|
@login_required
|
|
@require_POST
|
|
def inspections_geojson(request):
|
|
"""
|
|
Retourne la collection GeoJSON des inspections existantes filtrées par thématique, catégorie, période et emprise cartographique.
|
|
Reprojection (ST_Transform), calcul de centroïde (ST_Centroid) et conversion GeoJSON (ST_AsGeoJSON) effectuées nativement par PostGIS.
|
|
"""
|
|
import json
|
|
from datetime import timedelta
|
|
from django.contrib.gis.db.models.functions import Transform, Centroid, AsGeoJSON
|
|
from django.contrib.gis.geos import GEOSGeometry
|
|
|
|
try:
|
|
data = json.loads(request.body)
|
|
except Exception:
|
|
data = {}
|
|
|
|
thematic_code = data.get('thematic', '')
|
|
category_id = data.get('category')
|
|
if isinstance(category_id, list) and category_id:
|
|
category_id = category_id[0]
|
|
period = data.get('period', '3m')
|
|
|
|
if period == 'disabled':
|
|
return JsonResponse({'type': 'FeatureCollection', 'features': []})
|
|
|
|
qs = AssetInspection.objects.all().select_related('asset_content_type', 'inspector').prefetch_related('documents')
|
|
|
|
# Filtre par période temporelle
|
|
now = timezone.now()
|
|
if period == '1m':
|
|
qs = qs.filter(inspection_date__gte=now - timedelta(days=30))
|
|
elif period == '3m':
|
|
qs = qs.filter(inspection_date__gte=now - timedelta(days=90))
|
|
elif period == '6m':
|
|
qs = qs.filter(inspection_date__gte=now - timedelta(days=180))
|
|
elif period == '1y':
|
|
qs = qs.filter(inspection_date__gte=now - timedelta(days=365))
|
|
|
|
# Récupération de la géométrie d'emprise BBOX 4326
|
|
bbox_geom = None
|
|
filter_geom_data = data.get('filter_geom')
|
|
if filter_geom_data:
|
|
try:
|
|
geom_json = filter_geom_data.get('geometry', filter_geom_data)
|
|
bbox_geom = GEOSGeometry(json.dumps(geom_json), srid=4326)
|
|
except Exception:
|
|
pass
|
|
|
|
# Identification des ContentTypes cibles
|
|
ct_ids = set()
|
|
if thematic_code:
|
|
try:
|
|
thematic = Thematic.objects.get(code=thematic_code)
|
|
categories = AssetCategory.objects.filter(thematic=thematic)
|
|
if category_id:
|
|
categories = categories.filter(id=category_id)
|
|
ct_ids = set(categories.values_list('allowed_models', flat=True).distinct())
|
|
except Thematic.DoesNotExist:
|
|
pass
|
|
|
|
if not ct_ids:
|
|
ct_ids = set(qs.values_list('asset_content_type_id', flat=True).distinct())
|
|
|
|
features = []
|
|
|
|
# Pour chaque ContentType d'asset, requêter PostGIS pour ST_Transform, ST_Centroid & ST_AsGeoJSON
|
|
for ct_id in ct_ids:
|
|
if not ct_id:
|
|
continue
|
|
try:
|
|
ct = ContentType.objects.get(id=ct_id)
|
|
ModelClass = ct.model_class()
|
|
if not ModelClass or not hasattr(ModelClass, 'geom'):
|
|
continue
|
|
|
|
asset_qs = ModelClass.objects.filter(geom__isnull=False)
|
|
if bbox_geom:
|
|
asset_qs = asset_qs.filter(geom__intersects=bbox_geom)
|
|
|
|
# Traitement natif par PostGIS: ST_AsGeoJSON(ST_Transform(ST_Centroid(geom), 4326))
|
|
asset_qs = asset_qs.annotate(
|
|
wgs84_json=AsGeoJSON(Transform(Centroid('geom'), 4326))
|
|
)
|
|
|
|
asset_geom_map = dict(asset_qs.values_list('id', 'wgs84_json'))
|
|
if not asset_geom_map:
|
|
continue
|
|
|
|
ct_inspections = qs.filter(
|
|
asset_content_type=ct,
|
|
asset_object_id__in=asset_geom_map.keys()
|
|
).order_by('-inspection_date')
|
|
|
|
# Regroupement des inspections par asset (clé object_id)
|
|
asset_inspections_dict = {}
|
|
for insp in ct_inspections:
|
|
asset_inspections_dict.setdefault(insp.asset_object_id, []).append(insp)
|
|
|
|
for asset_id, insp_list in asset_inspections_dict.items():
|
|
geojson_str = asset_geom_map.get(asset_id)
|
|
if not geojson_str or not insp_list:
|
|
continue
|
|
|
|
latest_insp = insp_list[0]
|
|
target_obj = latest_insp.asset
|
|
|
|
inspections_data = []
|
|
for insp in insp_list:
|
|
docs_data = []
|
|
for doc in insp.documents.all():
|
|
docs_data.append({
|
|
'id': doc.id,
|
|
'url': doc.file_url if doc.file else '',
|
|
'filename': doc.filename,
|
|
'is_image': doc.is_image
|
|
})
|
|
|
|
inspections_data.append({
|
|
'id': insp.id,
|
|
'date': insp.inspection_date.strftime('%d/%m/%Y %H:%M') if insp.inspection_date else '',
|
|
'result_status': insp.result_status,
|
|
'result_status_display': insp.get_result_status_display(),
|
|
'status_badge_class': insp.status_badge_class,
|
|
'nature_display': insp.get_nature_display(),
|
|
'inspector_name': insp.inspector.get_full_name() if insp.inspector else '-',
|
|
'inspector_role_display': insp.get_inspector_role_display() if insp.inspector_role else '',
|
|
'details_summary': insp.details_summary,
|
|
'notes': insp.notes or '',
|
|
'target_display': insp.target_display,
|
|
'mobile_asset_url': insp.mobile_asset_url,
|
|
'documents': docs_data,
|
|
})
|
|
|
|
feature = {
|
|
'type': 'Feature',
|
|
'geometry': json.loads(geojson_str),
|
|
'properties': {
|
|
'content_type_id': ct.id,
|
|
'object_id': asset_id,
|
|
'model_name': ct.model,
|
|
'spot_type': getattr(target_obj, 'spot_type', None) if target_obj else None,
|
|
'target_display': latest_insp.target_display,
|
|
'latest_result_status': latest_insp.result_status,
|
|
'latest_result_status_display': latest_insp.get_result_status_display(),
|
|
'latest_status_badge_class': latest_insp.status_badge_class,
|
|
'inspections_count': len(inspections_data),
|
|
'inspections': inspections_data,
|
|
}
|
|
}
|
|
features.append(feature)
|
|
except Exception:
|
|
continue
|
|
|
|
return JsonResponse({'type': 'FeatureCollection', 'features': features[:1000]})
|