530 lines
22 KiB
Python
530 lines
22 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 .models import (
|
|
Inspection, ParkingSpotInspection, InspectionDocument,
|
|
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.
|
|
"""
|
|
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.
|
|
"""
|
|
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 filtres par défaut, recherche, thématiques et missions d'intervention.
|
|
"""
|
|
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', '')
|
|
mission_id = request.GET.get('mission_id', '')
|
|
checked_interv_id = request.GET.get('checked_interv_id', '')
|
|
search_q = request.GET.get('q', '').strip()
|
|
|
|
qs = Inspection.objects.all().select_related(
|
|
'asset_content_type', 'inspector', 'mission_intervention', 'checked_intervention'
|
|
).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 mission_id:
|
|
qs = qs.filter(mission_intervention_id=mission_id)
|
|
|
|
if checked_interv_id:
|
|
qs = qs.filter(checked_intervention_id=checked_interv_id)
|
|
|
|
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) |
|
|
Q(mission_intervention__title__icontains=search_q) |
|
|
Q(checked_intervention__title__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,
|
|
'selected_mission_id': mission_id,
|
|
'selected_checked_interv_id': checked_interv_id,
|
|
'search_q': search_q,
|
|
'thematics': thematics,
|
|
'can_inspect': can_inspect,
|
|
'result_choices': INSPECTION_RESULT_CHOICES,
|
|
'nature_choices': INSPECTION_NATURE_CHOICES,
|
|
}
|
|
return render(request, 'inspections/inspections_list.html', context)
|
|
|
|
|
|
def _get_asset_wgs84_lon_lat(obj):
|
|
if hasattr(obj, 'lon') and hasattr(obj, 'lat') and obj.lon is not None and obj.lat is not None:
|
|
try:
|
|
lon_f = float(obj.lon)
|
|
lat_f = float(obj.lat)
|
|
if -180 <= lon_f <= 180 and -90 <= lat_f <= 90 and not (lon_f == 0 and lat_f == 0):
|
|
return lon_f, lat_f
|
|
except (ValueError, TypeError):
|
|
pass
|
|
|
|
if hasattr(obj, 'geojson') and obj.geojson:
|
|
try:
|
|
import json
|
|
data = json.loads(obj.geojson) if isinstance(obj.geojson, str) else obj.geojson
|
|
geom_data = data.get('geometry', {}) if data.get('type') == 'Feature' else data
|
|
g_type = geom_data.get('type')
|
|
coords = geom_data.get('coordinates')
|
|
if g_type == 'Point' and coords:
|
|
return float(coords[0]), float(coords[1])
|
|
elif g_type == 'Polygon' and coords and len(coords) > 0:
|
|
ring = coords[0]
|
|
avg_lon = sum(pt[0] for pt in ring) / len(ring)
|
|
avg_lat = sum(pt[1] for pt in ring) / len(ring)
|
|
return avg_lon, avg_lat
|
|
elif g_type == 'MultiPolygon' and coords and len(coords) > 0:
|
|
all_pts = [pt for poly in coords for ring in poly for pt in ring]
|
|
if all_pts:
|
|
avg_lon = sum(pt[0] for pt in all_pts) / len(all_pts)
|
|
avg_lat = sum(pt[1] for pt in all_pts) / len(all_pts)
|
|
return avg_lon, avg_lat
|
|
except Exception:
|
|
pass
|
|
|
|
if hasattr(obj, 'geom') and obj.geom:
|
|
try:
|
|
g = obj.geom
|
|
srid = getattr(g, 'srid', None) or 3812
|
|
from django.contrib.gis.geos import GEOSGeometry
|
|
g_copy = GEOSGeometry(g.wkt, srid=srid)
|
|
g_4326 = g_copy.transform(4326, clone=True)
|
|
centroid = g_4326.centroid
|
|
return float(centroid.x), float(centroid.y)
|
|
except Exception:
|
|
pass
|
|
|
|
return None, None
|
|
|
|
|
|
@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
|
|
|
|
preselect_ct_id = request.GET.get('content_type_id')
|
|
preselect_obj_id = request.GET.get('object_id')
|
|
mission_id = request.GET.get('mission_id')
|
|
checked_interv_id = request.GET.get('checked_interv_id')
|
|
|
|
preselect_lon = None
|
|
preselect_lat = None
|
|
preselect_thematic_code = None
|
|
preselect_category_id = None
|
|
|
|
if preselect_ct_id and preselect_obj_id:
|
|
try:
|
|
ct = ContentType.objects.get(pk=preselect_ct_id)
|
|
obj = ct.get_object_for_this_type(pk=preselect_obj_id)
|
|
if obj:
|
|
from assets.permissions import _get_asset_thematic
|
|
th = _get_asset_thematic(obj)
|
|
if th:
|
|
preselect_thematic_code = th.code
|
|
if hasattr(obj, 'category') and obj.category:
|
|
preselect_category_id = obj.category.id
|
|
|
|
preselect_lon, preselect_lat = _get_asset_wgs84_lon_lat(obj)
|
|
except Exception:
|
|
pass
|
|
|
|
import json
|
|
|
|
context = {
|
|
'base_template': base_template,
|
|
'is_mobile': is_mobile,
|
|
'thematics': thematics,
|
|
'categories': categories,
|
|
'parking_spot_ct_id': parking_spot_ct_id,
|
|
'preselect_ct_id': preselect_ct_id or "null",
|
|
'preselect_obj_id': preselect_obj_id or "null",
|
|
'preselect_mission_id': mission_id or "null",
|
|
'preselect_checked_interv_id': checked_interv_id or "null",
|
|
'preselect_lon': json.dumps(preselect_lon) if preselect_lon is not None else "null",
|
|
'preselect_lat': json.dumps(preselect_lat) if preselect_lat is not None else "null",
|
|
'preselect_thematic_code': json.dumps(preselect_thematic_code) if preselect_thematic_code else "null",
|
|
'preselect_category_id': preselect_category_id or "null",
|
|
}
|
|
return render(request, 'inspections/quick_inspection_map.html', context)
|
|
|
|
|
|
@login_required
|
|
@require_POST
|
|
def create_inspection_api(request):
|
|
"""
|
|
API endpoint pour l'enregistrement d'une nouvelle inspection (avec prise en compte optionnelle de mission_intervention et checked_intervention).
|
|
"""
|
|
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'
|
|
|
|
mission_intervention_id = request.POST.get('mission_intervention_id') or None
|
|
checked_intervention_id = request.POST.get('checked_intervention_id') or None
|
|
|
|
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,
|
|
mission_intervention_id=mission_intervention_id,
|
|
checked_intervention_id=checked_intervention_id,
|
|
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 = Inspection.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,
|
|
mission_intervention_id=mission_intervention_id,
|
|
checked_intervention_id=checked_intervention_id,
|
|
verified_obsolescence=verified_obsolescence,
|
|
notes=notes,
|
|
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:
|
|
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
|
|
from django.contrib.gis.geos import MultiPolygon
|
|
interv_geom = getattr(target_obj, 'geom', None)
|
|
if interv_geom and interv_geom.geom_type == 'Polygon':
|
|
interv_geom = MultiPolygon(interv_geom)
|
|
|
|
intervention = Intervention.objects.create(
|
|
title=str(_("Intervention suite à inspection non conforme (%s)")) % str(target_obj),
|
|
thematic=thematic,
|
|
maintain_type='corrective',
|
|
priority='2' if result_status in ('damaged', 'missing') else '3',
|
|
init_description=notes or (str(_("Signalement automatique lors de l'inspection du %s.")) % timezone.now().strftime('%d/%m/%Y')),
|
|
status='in_preparation',
|
|
created_by=request.user,
|
|
geom=interv_geom,
|
|
)
|
|
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': str(_("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.
|
|
"""
|
|
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')
|
|
mission_id = data.get('mission_id')
|
|
|
|
if period == 'disabled':
|
|
return JsonResponse({'type': 'FeatureCollection', 'features': []})
|
|
|
|
qs = Inspection.objects.all().select_related('asset_content_type', 'inspector', 'mission_intervention').prefetch_related('documents')
|
|
|
|
if mission_id:
|
|
qs = qs.filter(mission_intervention_id=mission_id)
|
|
|
|
# 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))
|
|
|
|
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
|
|
|
|
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 = {c for c in categories.values_list('allowed_models', flat=True).distinct() if c}
|
|
except Thematic.DoesNotExist:
|
|
pass
|
|
|
|
if not ct_ids:
|
|
ct_ids = {c for c in qs.values_list('asset_content_type_id', flat=True).distinct() if c}
|
|
|
|
features = []
|
|
|
|
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)
|
|
|
|
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')
|
|
|
|
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]})
|