from django.shortcuts import render, get_object_or_404, redirect from django.http import JsonResponse, Http404, HttpResponseForbidden, HttpResponse 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_drop_off_stand_count' in request.POST and request.POST['verified_drop_off_stand_count'] != '': inspection.verified_drop_off_stand_count = int(request.POST['verified_drop_off_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: val = request.POST['verified_has_scooter_drop_zone'] if val == 'true': inspection.verified_has_scooter_drop_zone = True elif val == 'false': inspection.verified_has_scooter_drop_zone = False elif val == '': inspection.verified_has_scooter_drop_zone = None if 'verified_marking_quality' in request.POST and request.POST['verified_marking_quality'] != '': inspection.verified_marking_quality = request.POST['verified_marking_quality'] inspection.apply_physical_counts_to_asset = (request.POST.get('apply_physical_counts_to_asset') == 'true') 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]}) @login_required def get_asset_inspection_form_partial(request): """ API endpoint qui renvoie le snippet HTML du partial form d'inspection pour un asset donné. Paramètres GET : content_type_id & object_id (ou model_name). """ content_type_id = request.GET.get('content_type_id') object_id = request.GET.get('object_id') model_name = request.GET.get('model_name') asset = None if content_type_id and object_id: try: content_type = ContentType.objects.get(pk=content_type_id) asset = content_type.get_object_for_this_type(pk=object_id) model_name = content_type.model except Exception: pass from .registry import get_inspection_partial_template template_path = get_inspection_partial_template(asset or model_name) if not template_path: return HttpResponse('') context = {'asset': asset} return render(request, template_path, context)