1453 lines
65 KiB
Python
1453 lines
65 KiB
Python
from django.shortcuts import render, get_object_or_404, redirect
|
|
from django.http import JsonResponse, Http404, HttpResponseForbidden, HttpResponse, HttpResponseRedirect
|
|
from django.contrib.auth.decorators import login_required
|
|
from django.views.decorators.http import require_POST, require_GET
|
|
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 _, get_language
|
|
import json
|
|
|
|
from common.models import Thematic
|
|
from common.utils import get_short_name
|
|
from assets.models import AssetCategory
|
|
from .models import (
|
|
Inspection, ParkingSpotInspection, GiepInspection, SignPanelInspection, 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
|
|
|
|
|
|
def _filter_inspections_queryset(request):
|
|
"""
|
|
Construit et filtre le QuerySet des inspections selon les paramètres de requête GET.
|
|
"""
|
|
from .permissions import filter_viewable_inspections_for_user
|
|
qs = filter_viewable_inspections_for_user(request.user)
|
|
qs = qs.select_related(
|
|
'asset_content_type', 'inspector', 'mission_intervention', 'checked_intervention',
|
|
'parkingspotinspection', 'giepinspection', 'signpanelinspection'
|
|
).prefetch_related('documents')
|
|
|
|
thematics = request.GET.getlist('thematic')
|
|
if thematics:
|
|
thematic_objs = Thematic.objects.filter(code__in=thematics)
|
|
categories_for_thematics = AssetCategory.objects.filter(thematic__in=thematic_objs)
|
|
cts = set(ContentType.objects.filter(asset_categories__in=categories_for_thematics).distinct())
|
|
if 'water' in thematics:
|
|
from assets.models.nature import NatureRWIZ, NatureRWIAsset
|
|
cts.add(ContentType.objects.get_for_model(NatureRWIZ))
|
|
cts.add(ContentType.objects.get_for_model(NatureRWIAsset))
|
|
if 'sign' in thematics:
|
|
from assets.models import SignPanel, SignPole
|
|
cts.add(ContentType.objects.get_for_model(SignPanel))
|
|
cts.add(ContentType.objects.get_for_model(SignPole))
|
|
if 'parking' in thematics:
|
|
from assets.models import ParkingSpot, ParkingLocation
|
|
cts.add(ContentType.objects.get_for_model(ParkingSpot))
|
|
cts.add(ContentType.objects.get_for_model(ParkingLocation))
|
|
qs = qs.filter(asset_content_type__in=cts)
|
|
|
|
categories = request.GET.getlist('category')
|
|
if categories:
|
|
child_ids = list(AssetCategory.objects.filter(parent_id__in=categories).values_list('id', flat=True))
|
|
expanded_cat_ids = list(set(categories) | set(str(c) for c in child_ids))
|
|
category_cts = set(ContentType.objects.filter(asset_categories__id__in=expanded_cat_ids).distinct())
|
|
|
|
cat_asset_q = Q(asset_content_type__in=category_cts)
|
|
for ct in category_cts:
|
|
ModelClass = ct.model_class()
|
|
if ModelClass and hasattr(ModelClass, 'category'):
|
|
match_ids = list(ModelClass.objects.filter(category_id__in=expanded_cat_ids).values_list('pk', flat=True)[:1000])
|
|
if match_ids:
|
|
cat_asset_q |= Q(asset_content_type=ct, asset_object_id__in=match_ids)
|
|
qs = qs.filter(cat_asset_q)
|
|
|
|
search_query = (request.GET.get('q') or '').strip()
|
|
search_mode = (request.GET.get('search_mode') or 'and').lower()
|
|
if search_mode not in ('and', 'or'):
|
|
search_mode = 'and'
|
|
|
|
if search_query:
|
|
words = search_query.split()
|
|
|
|
def _word_q(word):
|
|
return (
|
|
Q(notes__icontains=word) |
|
|
Q(inspector__first_name__icontains=word) |
|
|
Q(inspector__last_name__icontains=word) |
|
|
Q(inspector__username__icontains=word) |
|
|
Q(mission_intervention__title__icontains=word) |
|
|
Q(checked_intervention__title__icontains=word)
|
|
)
|
|
|
|
asset_q_conditions = Q()
|
|
for ct in ContentType.objects.filter(app_label__in=['assets', 'interventions']):
|
|
ModelClass = ct.model_class()
|
|
if not ModelClass:
|
|
continue
|
|
|
|
model_fields = [f.name for f in ModelClass._meta.get_fields()]
|
|
asset_search = Q()
|
|
for word in words:
|
|
word_asset_search = Q()
|
|
if 'code' in model_fields:
|
|
word_asset_search |= Q(code__icontains=word)
|
|
if 'name' in model_fields:
|
|
word_asset_search |= Q(name__icontains=word)
|
|
if 'name_fr' in model_fields:
|
|
word_asset_search |= Q(name_fr__icontains=word)
|
|
if 'address' in model_fields:
|
|
word_asset_search |= Q(address__icontains=word)
|
|
if word_asset_search:
|
|
asset_search |= word_asset_search
|
|
|
|
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)
|
|
|
|
combined = Q()
|
|
if search_mode == 'and':
|
|
for w in words:
|
|
combined &= _word_q(w)
|
|
else:
|
|
for w in words:
|
|
combined |= _word_q(w)
|
|
|
|
if asset_q_conditions:
|
|
combined |= asset_q_conditions
|
|
|
|
qs = qs.filter(combined)
|
|
|
|
statuses = request.GET.getlist('status') or request.GET.getlist('result_status')
|
|
if statuses:
|
|
qs = qs.filter(result_status__in=statuses)
|
|
|
|
natures = request.GET.getlist('nature')
|
|
if natures:
|
|
qs = qs.filter(nature__in=natures)
|
|
|
|
show_mine_only = request.GET.get('mine') == '1' or request.GET.get('tab') == 'my'
|
|
if show_mine_only:
|
|
qs = qs.filter(inspector=request.user)
|
|
|
|
date_from = request.GET.get('date_from', '').strip()
|
|
date_to = request.GET.get('date_to', '').strip()
|
|
if date_from:
|
|
qs = qs.filter(inspection_date__date__gte=date_from)
|
|
if date_to:
|
|
qs = qs.filter(inspection_date__date__lte=date_to)
|
|
|
|
mission_id = request.GET.get('mission_id')
|
|
if mission_id:
|
|
qs = qs.filter(mission_intervention_id=mission_id)
|
|
|
|
checked_interv_id = request.GET.get('checked_interv_id')
|
|
if checked_interv_id:
|
|
qs = qs.filter(checked_intervention_id=checked_interv_id)
|
|
|
|
return qs.order_by('-inspection_date')
|
|
|
|
|
|
@login_required
|
|
def inspections_list(request):
|
|
"""
|
|
Vue de liste des inspections :
|
|
- Desktop : vue partagée Tableau interactif (Bootstrap Table) + Carte (MapLibre GL) avec filtres, recherche et export.
|
|
- Mobile : vue en cartes (inspections_list_mobile.html).
|
|
"""
|
|
is_mobile = request.path.startswith('/mobile/')
|
|
|
|
thematics = get_user_viewable_thematics(request.user) or Thematic.objects.none()
|
|
insp_thematics = get_user_inspectable_thematics(request.user)
|
|
can_inspect = bool(insp_thematics and insp_thematics.exists())
|
|
|
|
if is_mobile:
|
|
default_tab = 'my' if can_inspect else 'all'
|
|
filter_type = request.GET.get('tab', default_tab)
|
|
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()
|
|
|
|
from .permissions import filter_viewable_inspections_for_user
|
|
qs = filter_viewable_inspections_for_user(request.user)
|
|
qs = qs.select_related(
|
|
'asset_content_type', 'inspector', 'mission_intervention', 'checked_intervention',
|
|
'parkingspotinspection', 'giepinspection', 'signpanelinspection'
|
|
).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 = set(ContentType.objects.filter(asset_categories__in=categories).distinct())
|
|
if thematic_code == 'water':
|
|
from assets.models.nature import NatureRWIZ, NatureRWIAsset
|
|
cts.add(ContentType.objects.get_for_model(NatureRWIZ))
|
|
cts.add(ContentType.objects.get_for_model(NatureRWIAsset))
|
|
if thematic_code == 'sign':
|
|
from assets.models import SignPanel, SignPole
|
|
cts.add(ContentType.objects.get_for_model(SignPanel))
|
|
cts.add(ContentType.objects.get_for_model(SignPole))
|
|
if thematic_code == 'parking':
|
|
from assets.models import ParkingSpot, ParkingLocation
|
|
cts.add(ContentType.objects.get_for_model(ParkingSpot))
|
|
cts.add(ContentType.objects.get_for_model(ParkingLocation))
|
|
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)
|
|
|
|
context = {
|
|
'base_template': "mobile/mobile_base.html",
|
|
'is_mobile': True,
|
|
'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,
|
|
'INSPECTION_RESULT_CHOICES_CONFIG': Inspection.get_result_choices_config(),
|
|
}
|
|
return render(request, 'inspections/inspections_list_mobile.html', context)
|
|
|
|
# Desktop view
|
|
lang = get_language()
|
|
order_field = f'name_{lang}' if lang == 'nl' else 'name_fr'
|
|
|
|
user_config = getattr(request.user, 'config', None)
|
|
saved_filters = (user_config.inspection_list_filters or {}) if user_config else {}
|
|
saved_sort = (user_config.inspection_list_sort or {}) if user_config else {}
|
|
|
|
# Redirect on first visit with saved preferences
|
|
if not request.GET and saved_filters:
|
|
query_params = request.GET.copy()
|
|
for key, value in saved_filters.items():
|
|
if isinstance(value, list):
|
|
query_params.setlist(key, value)
|
|
elif value is not None:
|
|
query_params[key] = value
|
|
return HttpResponseRedirect(f"{request.path}?{query_params.urlencode()}")
|
|
|
|
categories = AssetCategory.objects.filter(thematic__in=thematics).order_by(order_field)
|
|
|
|
selected_thematics = request.GET.getlist('thematic')
|
|
selected_categories = request.GET.getlist('category')
|
|
selected_statuses = request.GET.getlist('status') or request.GET.getlist('result_status')
|
|
selected_natures = request.GET.getlist('nature')
|
|
search_query = request.GET.get('q', '')
|
|
show_mine_only = request.GET.get('mine') == '1' or request.GET.get('tab') == 'my'
|
|
date_from = request.GET.get('date_from', '')
|
|
date_to = request.GET.get('date_to', '')
|
|
|
|
category_choices = sorted(
|
|
[(c.id, c.get_name(), c.thematic.code if c.thematic else '', c.parent_id) for c in categories],
|
|
key=lambda t: (t[3] if t[3] is not None else t[0], 1 if t[3] is not None else 0, t[1])
|
|
)
|
|
|
|
result_choices_config = Inspection.get_result_choices_config()
|
|
_insp_status_color_map = {
|
|
'compliant': 'success',
|
|
'non_compliant': 'warning',
|
|
'damaged': 'danger',
|
|
'missing': 'dark',
|
|
'cancelled': 'secondary',
|
|
}
|
|
result_status_choices = [
|
|
(cfg['code'], cfg['label'], _insp_status_color_map.get(cfg['code'], 'secondary'), cfg.get('icon', 'bi-question-circle'))
|
|
for cfg in result_choices_config
|
|
]
|
|
|
|
context = {
|
|
'thematic_choices': [(t.code, t.name, t.icon or 'bi-geo-alt') for t in thematics],
|
|
'selected_thematics': selected_thematics,
|
|
'category_choices': category_choices,
|
|
'selected_categories': selected_categories,
|
|
'result_choices': result_choices_config,
|
|
'result_status_choices': result_status_choices,
|
|
'selected_statuses': selected_statuses,
|
|
'selected_result_statuses': selected_statuses,
|
|
'nature_choices': INSPECTION_NATURE_CHOICES,
|
|
'selected_natures': selected_natures,
|
|
'search_query': search_query,
|
|
'show_mine_only': show_mine_only,
|
|
'date_from': date_from,
|
|
'date_to': date_to,
|
|
'can_inspect': can_inspect,
|
|
'saved_sort': json.dumps(saved_sort, ensure_ascii=False),
|
|
'INSPECTION_RESULT_CHOICES_CONFIG': result_choices_config,
|
|
}
|
|
return render(request, 'inspections/inspections_index.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, SignPanel, SignPole
|
|
parking_spot_ct_id = ContentType.objects.get_for_model(ParkingSpot).id
|
|
sign_panel_ct_id = ContentType.objects.get_for_model(SignPanel).id
|
|
sign_pole_ct_id = ContentType.objects.get_for_model(SignPole).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,
|
|
'sign_panel_ct_id': sign_panel_ct_id,
|
|
'sign_pole_ct_id': sign_pole_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')
|
|
is_giep = (content_type.model == 'naturerwiz')
|
|
is_signpanel = (content_type.model == 'signpanel')
|
|
|
|
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()
|
|
elif is_giep:
|
|
import json
|
|
date_derniere_pluie = request.POST.get('date_derniere_pluie', 'null')
|
|
orage_val = request.POST.get('orage_moins_une_heure', 'null')
|
|
orage_moins_une_heure = True if orage_val == 'true' else (False if orage_val == 'false' else None)
|
|
|
|
giep_data_raw = request.POST.get('giep_data', '{}')
|
|
try:
|
|
parsed_giep_data = json.loads(giep_data_raw)
|
|
except Exception:
|
|
parsed_giep_data = {}
|
|
|
|
inspection = GiepInspection(
|
|
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,
|
|
date_derniere_pluie=date_derniere_pluie,
|
|
orage_moins_une_heure=orage_moins_une_heure,
|
|
data=parsed_giep_data,
|
|
)
|
|
inspection.save()
|
|
elif is_signpanel:
|
|
from assets.models import SignPanelType
|
|
try:
|
|
from sign.ai.catalog import SIGN_CATALOG
|
|
except ImportError:
|
|
SIGN_CATALOG = {}
|
|
|
|
verified_type_id = request.POST.get('verified_signpanel_type_id') or None
|
|
ai_code = request.POST.get('ai_detected_code')
|
|
verified_type = None
|
|
|
|
if verified_type_id:
|
|
if str(verified_type_id).isdigit():
|
|
verified_type = SignPanelType.objects.filter(pk=int(verified_type_id)).first()
|
|
elif str(verified_type_id).startswith('code:'):
|
|
code = str(verified_type_id).split(':', 1)[1].strip()
|
|
verified_type = SignPanelType.objects.filter(code__iexact=code).first()
|
|
else:
|
|
verified_type = SignPanelType.objects.filter(code__iexact=str(verified_type_id).strip()).first()
|
|
|
|
if not verified_type and ai_code:
|
|
verified_type = SignPanelType.objects.filter(code__iexact=ai_code.strip()).first()
|
|
|
|
if not verified_type and (verified_type_id or ai_code):
|
|
target_code = str(verified_type_id).replace('code:', '').strip() if verified_type_id and not str(verified_type_id).isdigit() else (ai_code or '').strip()
|
|
if target_code:
|
|
cat_info = SIGN_CATALOG.get(target_code.upper(), {})
|
|
verified_type, _created = SignPanelType.objects.get_or_create(
|
|
code=target_code.upper(),
|
|
defaults={
|
|
'name_fr': cat_info.get('name_fr', f'Panneau {target_code.upper()}'),
|
|
'name_nl': cat_info.get('name_nl', f'Verkeersbord {target_code.upper()}'),
|
|
}
|
|
)
|
|
|
|
verified_text = request.POST.get('verified_signpanel_text')
|
|
verified_cleanliness = request.POST.get('verified_cleanliness', 'clean')
|
|
ai_conf_raw = request.POST.get('ai_confidence')
|
|
ai_conf = float(ai_conf_raw) if ai_conf_raw else None
|
|
|
|
inspection = SignPanelInspection(
|
|
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,
|
|
verified_signpanel_type=verified_type,
|
|
verified_signpanel_text=verified_text,
|
|
verified_cleanliness=verified_cleanliness,
|
|
ai_detected_code=ai_code,
|
|
ai_confidence=ai_conf,
|
|
apply_sign_updates_to_asset=apply_changes,
|
|
)
|
|
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')
|
|
field_keys = request.POST.getlist('field_keys')
|
|
if uploaded_files:
|
|
for idx, f in enumerate(uploaded_files):
|
|
f_key = field_keys[idx] if idx < len(field_keys) else ''
|
|
InspectionDocument.objects.create(
|
|
inspection=inspection,
|
|
file=f,
|
|
field_key=f_key
|
|
)
|
|
|
|
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
|
|
def get_inspection_detail_api(request, inspection_id):
|
|
"""
|
|
API endpoint qui renvoie les détails d'une inspection sous forme JSON pour pré-remplir un formulaire d'édition.
|
|
"""
|
|
inspection_obj = get_object_or_404(Inspection, pk=inspection_id)
|
|
leaf = inspection_obj.detailed_inspection
|
|
|
|
from .permissions import can_view_inspection
|
|
if not can_view_inspection(request.user, leaf):
|
|
return JsonResponse({
|
|
'success': False,
|
|
'error': str(_("Permission refusée. Vous n'avez pas accès à cette inspection."))
|
|
}, status=403)
|
|
|
|
docs_data = []
|
|
for doc in leaf.documents.all():
|
|
docs_data.append({
|
|
'id': doc.id,
|
|
'filename': doc.filename,
|
|
'url': doc.file_url,
|
|
'is_image': doc.is_image,
|
|
})
|
|
|
|
from django.template.loader import render_to_string
|
|
from .registry import get_inspection_detail_partial_template
|
|
from assets.permissions import _get_asset_thematic
|
|
|
|
asset = leaf.asset
|
|
th = _get_asset_thematic(asset) if asset else None
|
|
thematic_name = th.name if th else ''
|
|
thematic_icon = th.icon if th and th.icon else 'bi-geo-alt'
|
|
|
|
detail_template = get_inspection_detail_partial_template(leaf)
|
|
rendered_detail_html = ''
|
|
if detail_template:
|
|
try:
|
|
rendered_detail_html = render_to_string(detail_template, {'insp': leaf}, request=request)
|
|
except Exception:
|
|
rendered_detail_html = ''
|
|
|
|
data = {
|
|
'success': True,
|
|
'id': leaf.id,
|
|
'content_type_id': leaf.asset_content_type_id,
|
|
'object_id': leaf.asset_object_id,
|
|
'target_display': leaf.target_display,
|
|
'asset_url': getattr(asset, 'get_absolute_url', lambda: '')() if asset and hasattr(asset, 'get_absolute_url') else '',
|
|
'mobile_asset_url': leaf.mobile_asset_url,
|
|
'thematic_name': thematic_name,
|
|
'thematic_icon': thematic_icon,
|
|
'asset_type_name': leaf.asset_content_type.name if leaf.asset_content_type else '',
|
|
'result_status': leaf.result_status,
|
|
'result_status_display': str(leaf.get_result_status_display()),
|
|
'status_badge_class': leaf.status_badge_class,
|
|
'status_icon': leaf.status_icon,
|
|
'nature': leaf.nature,
|
|
'nature_display': str(leaf.get_nature_display()),
|
|
'inspector_name': get_short_name(leaf.inspector) if leaf.inspector else '-',
|
|
'inspector_role': leaf.inspector_role,
|
|
'inspector_role_display': str(leaf.get_inspector_role_display()) if leaf.inspector_role else '',
|
|
'notes': leaf.notes or '',
|
|
'verified_obsolescence': leaf.verified_obsolescence or '',
|
|
'details_summary': leaf.details_summary,
|
|
'details_html': rendered_detail_html,
|
|
'apply_changes_to_asset': leaf.apply_changes_to_asset,
|
|
'is_cancelled': leaf.is_cancelled,
|
|
'inspection_date': leaf.inspection_date.strftime('%d/%m/%Y %H:%M') if leaf.inspection_date else '',
|
|
'created_at': leaf.created_at.strftime('%d/%m/%Y %H:%M') if leaf.created_at else '',
|
|
'updated_at': leaf.updated_at.strftime('%d/%m/%Y %H:%M') if leaf.updated_at else '',
|
|
'can_edit': leaf.can_edit(request.user),
|
|
'documents': docs_data,
|
|
}
|
|
|
|
if isinstance(leaf, ParkingSpotInspection):
|
|
data['is_parking_spot'] = True
|
|
data['verified_standard_stand_count'] = leaf.verified_standard_stand_count
|
|
data['verified_cargo_stand_count'] = leaf.verified_cargo_stand_count
|
|
data['verified_drop_off_stand_count'] = leaf.verified_drop_off_stand_count
|
|
data['verified_stand_type'] = leaf.verified_stand_type or ''
|
|
data['verified_has_scooter_drop_zone'] = leaf.verified_has_scooter_drop_zone
|
|
data['verified_marking_quality'] = leaf.verified_marking_quality or ''
|
|
data['apply_physical_counts_to_asset'] = leaf.apply_physical_counts_to_asset
|
|
|
|
elif isinstance(leaf, GiepInspection):
|
|
data['is_giep'] = True
|
|
data['date_derniere_pluie'] = leaf.date_derniere_pluie or 'null'
|
|
data['orage_moins_une_heure'] = leaf.orage_moins_une_heure
|
|
data['giep_data'] = leaf.data or {}
|
|
|
|
elif isinstance(leaf, SignPanelInspection):
|
|
data['is_signpanel'] = True
|
|
data['verified_signpanel_type_id'] = leaf.verified_signpanel_type_id
|
|
data['verified_signpanel_type_code'] = leaf.verified_signpanel_type.code if leaf.verified_signpanel_type else ''
|
|
data['verified_signpanel_type_name'] = leaf.verified_signpanel_type.get_name() if leaf.verified_signpanel_type else ''
|
|
data['verified_signpanel_text'] = leaf.verified_signpanel_text or ''
|
|
data['verified_cleanliness'] = leaf.verified_cleanliness or 'clean'
|
|
data['ai_detected_code'] = leaf.ai_detected_code or ''
|
|
data['ai_confidence'] = leaf.ai_confidence
|
|
data['apply_sign_updates_to_asset'] = leaf.apply_sign_updates_to_asset
|
|
|
|
return JsonResponse(data)
|
|
|
|
|
|
@login_required
|
|
@require_POST
|
|
def update_inspection_api(request, inspection_id):
|
|
"""
|
|
API endpoint pour modifier une inspection existante. Seuls l'auteur et les administrateurs ont le droit de modifier.
|
|
La date de modification (updated_at) est sauvegardée automatiquement lors du save().
|
|
"""
|
|
inspection_obj = get_object_or_404(Inspection, pk=inspection_id)
|
|
inspection = inspection_obj.detailed_inspection
|
|
|
|
if not inspection.can_edit(request.user):
|
|
return JsonResponse({
|
|
'success': False,
|
|
'error': str(_("Permission refusée. Seuls l'auteur de l'inspection et les administrateurs peuvent la modifier."))
|
|
}, status=403)
|
|
|
|
try:
|
|
if 'result_status' in request.POST:
|
|
inspection.result_status = request.POST['result_status']
|
|
if 'nature' in request.POST:
|
|
inspection.nature = request.POST['nature']
|
|
if 'inspector_role' in request.POST:
|
|
inspection.inspector_role = request.POST['inspector_role']
|
|
if 'notes' in request.POST:
|
|
inspection.notes = request.POST['notes'].strip()
|
|
if 'verified_obsolescence' in request.POST:
|
|
inspection.verified_obsolescence = request.POST['verified_obsolescence'].strip() or None
|
|
inspection.apply_changes_to_asset = (request.POST.get('apply_changes_to_asset') == 'true')
|
|
|
|
if isinstance(inspection, ParkingSpotInspection):
|
|
if 'verified_standard_stand_count' in request.POST:
|
|
val = request.POST['verified_standard_stand_count']
|
|
inspection.verified_standard_stand_count = int(val) if val != '' else None
|
|
if 'verified_cargo_stand_count' in request.POST:
|
|
val = request.POST['verified_cargo_stand_count']
|
|
inspection.verified_cargo_stand_count = int(val) if val != '' else None
|
|
if 'verified_drop_off_stand_count' in request.POST:
|
|
val = request.POST['verified_drop_off_stand_count']
|
|
inspection.verified_drop_off_stand_count = int(val) if val != '' else None
|
|
if 'verified_stand_type' in request.POST:
|
|
val = request.POST['verified_stand_type']
|
|
inspection.verified_stand_type = val if val != '' else None
|
|
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:
|
|
val = request.POST['verified_marking_quality']
|
|
inspection.verified_marking_quality = val if val != '' else None
|
|
if 'apply_physical_counts_to_asset' in request.POST:
|
|
inspection.apply_physical_counts_to_asset = (request.POST['apply_physical_counts_to_asset'] == 'true')
|
|
elif isinstance(inspection, GiepInspection):
|
|
import json
|
|
if 'date_derniere_pluie' in request.POST:
|
|
inspection.date_derniere_pluie = request.POST['date_derniere_pluie']
|
|
if 'orage_moins_une_heure' in request.POST:
|
|
val = request.POST['orage_moins_une_heure']
|
|
inspection.orage_moins_une_heure = True if val == 'true' else (False if val == 'false' else None)
|
|
if 'giep_data' in request.POST:
|
|
try:
|
|
inspection.data = json.loads(request.POST['giep_data'])
|
|
except Exception:
|
|
pass
|
|
elif isinstance(inspection, SignPanelInspection):
|
|
from assets.models import SignPanelType
|
|
if 'verified_signpanel_type_id' in request.POST:
|
|
v_type_id = request.POST.get('verified_signpanel_type_id') or None
|
|
inspection.verified_signpanel_type = SignPanelType.objects.filter(pk=v_type_id).first() if v_type_id else None
|
|
if 'verified_signpanel_text' in request.POST:
|
|
inspection.verified_signpanel_text = request.POST.get('verified_signpanel_text')
|
|
if 'verified_cleanliness' in request.POST:
|
|
inspection.verified_cleanliness = request.POST.get('verified_cleanliness')
|
|
if 'apply_sign_updates_to_asset' in request.POST:
|
|
inspection.apply_sign_updates_to_asset = (request.POST.get('apply_sign_updates_to_asset') == 'true')
|
|
|
|
# Traitement des nouvelles pièces jointes / photos
|
|
uploaded_files = request.FILES.getlist('photos') + request.FILES.getlist('documents') + request.FILES.getlist('files')
|
|
field_keys = request.POST.getlist('field_keys')
|
|
if uploaded_files:
|
|
for idx, f in enumerate(uploaded_files):
|
|
f_key = field_keys[idx] if idx < len(field_keys) else ''
|
|
InspectionDocument.objects.create(
|
|
inspection=inspection,
|
|
file=f,
|
|
field_key=f_key
|
|
)
|
|
|
|
# Sauvegarde (met automatiquement à jour updated_at)
|
|
inspection.save()
|
|
|
|
return JsonResponse({
|
|
'success': True,
|
|
'inspection_id': inspection.id,
|
|
'updated_at': inspection.updated_at.strftime('%d/%m/%Y %H:%M'),
|
|
'message': str(_("Inspection mise à jour avec succès."))
|
|
})
|
|
except Exception as e:
|
|
return JsonResponse({'success': False, 'error': str(e)}, status=500)
|
|
|
|
|
|
@login_required
|
|
@require_POST
|
|
def delete_inspection_document_api(request, document_id):
|
|
"""
|
|
API endpoint pour supprimer une photo / document attaché à une inspection.
|
|
Seuls l'auteur et les administrateurs ont la permission.
|
|
"""
|
|
doc = get_object_or_404(InspectionDocument, pk=document_id)
|
|
inspection = doc.inspection.detailed_inspection
|
|
|
|
if not inspection.can_edit(request.user):
|
|
return JsonResponse({
|
|
'success': False,
|
|
'error': str(_("Permission refusée. Seuls l'auteur de l'inspection et les administrateurs peuvent la modifier."))
|
|
}, status=403)
|
|
|
|
doc.delete()
|
|
return JsonResponse({'success': True, 'document_id': document_id})
|
|
|
|
|
|
@login_required
|
|
@require_POST
|
|
def toggle_cancel_inspection_api(request, inspection_id):
|
|
"""
|
|
API endpoint pour annuler ou rétablir une inspection.
|
|
Seuls l'auteur et les administrateurs ont le droit d'annuler/rétablir une inspection.
|
|
"""
|
|
inspection_obj = get_object_or_404(Inspection, pk=inspection_id)
|
|
inspection = inspection_obj.detailed_inspection
|
|
|
|
if not inspection.can_edit(request.user):
|
|
return JsonResponse({
|
|
'success': False,
|
|
'error': str(_("Permission refusée. Seuls l'auteur de l'inspection et les administrateurs peuvent la modifier."))
|
|
}, status=403)
|
|
|
|
action = request.POST.get('action') # 'cancel', 'restore', ou toggle automatique si non spécifié
|
|
if action == 'cancel' or (action is None and not inspection.is_cancelled):
|
|
inspection.result_status = 'cancelled'
|
|
elif action == 'restore' or (action is None and inspection.is_cancelled):
|
|
target_status = request.POST.get('result_status', 'compliant')
|
|
inspection.result_status = target_status if target_status != 'cancelled' else 'compliant'
|
|
|
|
inspection.save()
|
|
|
|
return JsonResponse({
|
|
'success': True,
|
|
'inspection_id': inspection.id,
|
|
'is_cancelled': inspection.is_cancelled,
|
|
'result_status': inspection.result_status,
|
|
'result_status_display': str(inspection.get_result_status_display()),
|
|
'status_badge_class': inspection.status_badge_class,
|
|
'status_icon': inspection.status_icon,
|
|
'status_btn_class': inspection.status_btn_class,
|
|
'message': str(_("L'inspection a été annulée avec succès.")) if inspection.is_cancelled else str(_("L'inspection a été rétablie avec succès."))
|
|
})
|
|
|
|
|
|
@login_required
|
|
def inspections_geojson(request):
|
|
"""
|
|
Retourne la collection GeoJSON des inspections existantes filtrées.
|
|
Supporte :
|
|
- GET (utilisé par la vue desktop tableau/carte)
|
|
- POST avec JSON body (utilisé par quick_inspection_map et rétro-compatibilité)
|
|
"""
|
|
import json
|
|
from datetime import timedelta
|
|
from django.contrib.gis.db.models.functions import Transform, Centroid, AsGeoJSON
|
|
from django.contrib.gis.geos import GEOSGeometry
|
|
from assets.permissions import _get_asset_thematic
|
|
|
|
# Si c'est une requête POST avec payload JSON (format quick_inspection_map)
|
|
if request.method == 'POST' and (request.content_type == 'application/json' or request.body):
|
|
try:
|
|
data = json.loads(request.body)
|
|
except Exception:
|
|
data = {}
|
|
|
|
if data.get('period') or data.get('filter_geom') or 'thematic' in 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': []})
|
|
|
|
from .permissions import filter_viewable_inspections_for_user, get_contract_filtered_asset_qs
|
|
qs = filter_viewable_inspections_for_user(request.user)
|
|
qs = qs.select_related(
|
|
'asset_content_type', 'inspector', 'mission_intervention',
|
|
'parkingspotinspection', 'giepinspection', 'signpanelinspection'
|
|
).prefetch_related('documents')
|
|
|
|
if mission_id:
|
|
qs = qs.filter(mission_intervention_id=mission_id)
|
|
|
|
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}
|
|
if thematic_code == 'water':
|
|
from assets.models.nature import NatureRWIZ, NatureRWIAsset
|
|
ct_ids.add(ContentType.objects.get_for_model(NatureRWIZ).id)
|
|
ct_ids.add(ContentType.objects.get_for_model(NatureRWIAsset).id)
|
|
elif thematic_code == 'sign':
|
|
from assets.models import SignPanel, SignPole
|
|
ct_ids.add(ContentType.objects.get_for_model(SignPanel).id)
|
|
ct_ids.add(ContentType.objects.get_for_model(SignPole).id)
|
|
elif thematic_code == 'parking':
|
|
from assets.models import ParkingSpot, ParkingLocation
|
|
ct_ids.add(ContentType.objects.get_for_model(ParkingSpot))
|
|
ct_ids.add(ContentType.objects.get_for_model(ParkingLocation))
|
|
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 = []
|
|
user_config = getattr(request.user, 'config', None)
|
|
is_admin_user = (
|
|
request.user.is_superuser or
|
|
request.user.is_staff or
|
|
(user_config and (user_config.has_role('admin') or user_config.has_role('top_manager')))
|
|
)
|
|
is_external = (user_config is None) or (not user_config.is_intern)
|
|
must_limit = is_external and (not user_config or user_config.limit_assets_to_contracts)
|
|
accessible_contract_ids = []
|
|
if must_limit and user_config:
|
|
from common.models import UserContractAccess
|
|
accessible_contract_ids = list(
|
|
UserContractAccess.objects.filter(
|
|
user_config=user_config,
|
|
can_view_assets=True
|
|
).values_list('contract_id', flat=True)
|
|
)
|
|
|
|
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 must_limit:
|
|
contract_filtered_qs = get_contract_filtered_asset_qs(ModelClass, accessible_contract_ids)
|
|
if contract_filtered_qs is not None:
|
|
asset_qs = asset_qs.filter(pk__in=contract_filtered_qs.values('pk'))
|
|
|
|
if bbox_geom:
|
|
bbox_geom_3812 = bbox_geom.transform(3812, clone=True)
|
|
asset_qs = asset_qs.filter(geom__intersects=bbox_geom_3812)
|
|
|
|
asset_qs = asset_qs.annotate(
|
|
wgs84_json=AsGeoJSON(Transform(Centroid('geom'), 4326))
|
|
)
|
|
|
|
asset_objs = list(asset_qs)
|
|
asset_geom_map = {obj.id: obj.wgs84_json for obj in asset_objs}
|
|
asset_obj_map = {obj.id: obj for obj in asset_objs}
|
|
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
|
|
|
|
active_insps = [i for i in insp_list if i.result_status != 'cancelled']
|
|
top_insp = active_insps[0] if active_insps else insp_list[0]
|
|
target_obj = asset_obj_map.get(asset_id)
|
|
|
|
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
|
|
})
|
|
|
|
can_edit = is_admin_user or (insp.inspector_id == request.user.id)
|
|
|
|
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,
|
|
'status_icon': insp.status_icon,
|
|
'is_cancelled': insp.is_cancelled,
|
|
'nature_display': insp.get_nature_display(),
|
|
'inspector_name': get_short_name(insp.inspector) 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,
|
|
'can_edit': can_edit,
|
|
'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': top_insp.target_display,
|
|
'latest_result_status': top_insp.result_status,
|
|
'latest_result_status_display': top_insp.get_result_status_display(),
|
|
'latest_status_badge_class': top_insp.status_badge_class,
|
|
'is_cancelled': top_insp.is_cancelled,
|
|
'inspections_count': len(inspections_data),
|
|
'inspections': inspections_data,
|
|
}
|
|
}
|
|
features.append(feature)
|
|
except Exception:
|
|
continue
|
|
|
|
return JsonResponse({'type': 'FeatureCollection', 'features': features[:1000]})
|
|
|
|
# GET / standard query request (desktop index / table / map)
|
|
qs = _filter_inspections_queryset(request)[:5000]
|
|
inspections_list = list(qs)
|
|
|
|
user_config = getattr(request.user, 'config', None)
|
|
is_admin_user = (
|
|
request.user.is_superuser or
|
|
request.user.is_staff or
|
|
(user_config and (user_config.has_role('admin') or user_config.has_role('top_manager')))
|
|
)
|
|
thematics_by_code = {t.code: t for t in Thematic.objects.all()}
|
|
|
|
ct_to_ids = {}
|
|
for insp in inspections_list:
|
|
ct_to_ids.setdefault(insp.asset_content_type_id, set()).add(insp.asset_object_id)
|
|
|
|
assets_by_ct_and_id = {}
|
|
for ct_id, obj_ids in ct_to_ids.items():
|
|
try:
|
|
ct = ContentType.objects.get_for_id(ct_id)
|
|
ModelClass = ct.model_class()
|
|
if ModelClass:
|
|
asset_qs = ModelClass.objects.filter(id__in=obj_ids)
|
|
if hasattr(ModelClass, 'geom'):
|
|
asset_qs = asset_qs.annotate(
|
|
wgs84_geojson=AsGeoJSON(Transform(Centroid('geom'), 4326))
|
|
)
|
|
if hasattr(ModelClass, 'category'):
|
|
asset_qs = asset_qs.select_related('category', 'category__thematic')
|
|
for asset_obj in asset_qs:
|
|
assets_by_ct_and_id[(ct_id, asset_obj.id)] = asset_obj
|
|
except Exception:
|
|
pass
|
|
|
|
features = []
|
|
for insp in inspections_list:
|
|
asset = assets_by_ct_and_id.get((insp.asset_content_type_id, insp.asset_object_id))
|
|
|
|
geometry = None
|
|
if asset:
|
|
if hasattr(asset, 'wgs84_geojson') and asset.wgs84_geojson:
|
|
try:
|
|
geometry = json.loads(asset.wgs84_geojson)
|
|
except Exception:
|
|
geometry = None
|
|
if not geometry:
|
|
lon, lat = _get_asset_wgs84_lon_lat(asset)
|
|
if lon is not None and lat is not None:
|
|
geometry = {"type": "Point", "coordinates": [lon, lat]}
|
|
|
|
th = _get_asset_thematic(asset, thematics_by_code=thematics_by_code) if asset else None
|
|
thematic_name = th.name if th else ''
|
|
thematic_code = th.code if th else ''
|
|
thematic_icon = th.icon if th and th.icon else 'bi-geo-alt'
|
|
|
|
cat_name = ''
|
|
if asset and hasattr(asset, 'category') and asset.category:
|
|
cat_name = asset.category.get_name()
|
|
elif asset and hasattr(asset, 'category_name'):
|
|
cat_name = asset.category_name
|
|
|
|
target_display = str(asset) if asset else f"Cible #{insp.asset_object_id}"
|
|
asset_url = getattr(asset, 'get_absolute_url', lambda: '')() if asset and hasattr(asset, 'get_absolute_url') else ''
|
|
|
|
can_edit = is_admin_user or (insp.inspector_id == request.user.id)
|
|
|
|
features.append({
|
|
"type": "Feature",
|
|
"id": insp.id,
|
|
"geometry": geometry,
|
|
"properties": {
|
|
"id": insp.id,
|
|
"code": f"INSP#{insp.id:05d}",
|
|
"target_display": target_display,
|
|
"asset_id": insp.asset_object_id,
|
|
"content_type_id": insp.asset_content_type_id,
|
|
"asset_model": insp.asset_content_type.model if insp.asset_content_type else '',
|
|
"asset_model_name": insp.asset_content_type.name if insp.asset_content_type else '',
|
|
"asset_url": asset_url,
|
|
"thematic": thematic_name,
|
|
"thematic_code": thematic_code,
|
|
"thematic_icon": thematic_icon,
|
|
"category_name": cat_name,
|
|
"result_status": insp.result_status,
|
|
"result_status_label": str(insp.get_result_status_display()),
|
|
"result_status_badge": insp.status_badge_class,
|
|
"result_status_color": insp.status_color_hex,
|
|
"result_status_icon": insp.status_icon,
|
|
"nature": insp.nature,
|
|
"nature_label": str(insp.get_nature_display()),
|
|
"inspector": get_short_name(insp.inspector) if insp.inspector else '-',
|
|
"inspector_username": insp.inspector.username if insp.inspector else '',
|
|
"inspection_date": insp.inspection_date.isoformat() if insp.inspection_date else '',
|
|
"inspection_date_formatted": insp.inspection_date.strftime('%d/%m/%Y %H:%M') if insp.inspection_date else '',
|
|
"notes": insp.notes or '',
|
|
"details_summary": insp.details_summary if insp.details_summary and insp.details_summary != '-' else '',
|
|
"verified_obsolescence": insp.verified_obsolescence or '',
|
|
"is_cancelled": insp.is_cancelled,
|
|
"documents_count": len(insp.documents.all()),
|
|
"can_edit": can_edit,
|
|
}
|
|
})
|
|
|
|
return JsonResponse({
|
|
"type": "FeatureCollection",
|
|
"features": features
|
|
}, safe=False)
|
|
|
|
|
|
@require_GET
|
|
@login_required
|
|
def inspections_export_excel(request):
|
|
"""
|
|
Exporte les inspections filtrées au format Excel (.xlsx).
|
|
"""
|
|
import io
|
|
from openpyxl import Workbook
|
|
from openpyxl.styles import Font, Alignment, PatternFill, Border, Side
|
|
from openpyxl.utils import get_column_letter
|
|
from assets.permissions import _get_asset_thematic
|
|
|
|
qs = _filter_inspections_queryset(request)[:5000]
|
|
inspections_list = list(qs)
|
|
thematics_by_code = {t.code: t for t in Thematic.objects.all()}
|
|
|
|
ct_to_ids = {}
|
|
for insp in inspections_list:
|
|
ct_to_ids.setdefault(insp.asset_content_type_id, set()).add(insp.asset_object_id)
|
|
|
|
assets_by_ct_and_id = {}
|
|
for ct_id, obj_ids in ct_to_ids.items():
|
|
try:
|
|
ct = ContentType.objects.get_for_id(ct_id)
|
|
ModelClass = ct.model_class()
|
|
if ModelClass:
|
|
asset_qs = ModelClass.objects.filter(id__in=obj_ids)
|
|
if hasattr(ModelClass, 'category'):
|
|
asset_qs = asset_qs.select_related('category', 'category__thematic')
|
|
for asset_obj in asset_qs:
|
|
assets_by_ct_and_id[(ct_id, asset_obj.id)] = asset_obj
|
|
except Exception:
|
|
pass
|
|
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
ws.title = str(_("Inspections"))
|
|
|
|
headers = [
|
|
str(_("ID")),
|
|
str(_("Date")),
|
|
str(_("Thématique")),
|
|
str(_("Type d'asset")),
|
|
str(_("Cible")),
|
|
str(_("Résultat du contrôle")),
|
|
str(_("Nature")),
|
|
str(_("Inspecteur")),
|
|
str(_("Obsolescence")),
|
|
str(_("Constats spécifiques")),
|
|
str(_("Remarques")),
|
|
str(_("Nb photos / docs")),
|
|
str(_("Annulée")),
|
|
]
|
|
|
|
header_font = Font(name="Arial", size=11, bold=True, color="FFFFFF")
|
|
header_fill = PatternFill(start_color="1E3A8A", end_color="1E3A8A", fill_type="solid")
|
|
header_alignment = Alignment(horizontal="center", vertical="center", wrap_text=True)
|
|
|
|
thin_border = Border(
|
|
left=Side(style='thin', color='D3D3D3'),
|
|
right=Side(style='thin', color='D3D3D3'),
|
|
top=Side(style='thin', color='D3D3D3'),
|
|
bottom=Side(style='thin', color='D3D3D3')
|
|
)
|
|
|
|
ws.append(headers)
|
|
for col_num in range(1, len(headers) + 1):
|
|
cell = ws.cell(row=1, column=col_num)
|
|
cell.font = header_font
|
|
cell.fill = header_fill
|
|
cell.alignment = header_alignment
|
|
cell.border = thin_border
|
|
ws.row_dimensions[1].height = 28
|
|
|
|
data_font = Font(name="Arial", size=10)
|
|
data_alignment = Alignment(vertical="center")
|
|
|
|
for row_num, insp in enumerate(inspections_list, start=2):
|
|
asset = assets_by_ct_and_id.get((insp.asset_content_type_id, insp.asset_object_id))
|
|
th = _get_asset_thematic(asset) if asset else None
|
|
target_display = str(asset) if asset else f"Cible #{insp.asset_object_id}"
|
|
|
|
row_data = [
|
|
f"INSP#{insp.id:05d}",
|
|
insp.inspection_date.strftime('%d/%m/%Y %H:%M') if insp.inspection_date else '',
|
|
th.name if th else '',
|
|
insp.asset_content_type.name if insp.asset_content_type else '',
|
|
target_display,
|
|
str(insp.get_result_status_display()),
|
|
str(insp.get_nature_display()),
|
|
get_short_name(insp.inspector) if insp.inspector else '-',
|
|
insp.verified_obsolescence or '',
|
|
insp.details_summary if insp.details_summary and insp.details_summary != '-' else '',
|
|
insp.notes or '',
|
|
insp.documents.count(),
|
|
str(_("Oui")) if insp.is_cancelled else str(_("Non")),
|
|
]
|
|
|
|
ws.append(row_data)
|
|
for col_num in range(1, len(headers) + 1):
|
|
cell = ws.cell(row=row_num, column=col_num)
|
|
cell.font = data_font
|
|
cell.alignment = data_alignment
|
|
cell.border = thin_border
|
|
ws.row_dimensions[row_num].height = 20
|
|
|
|
for col in ws.columns:
|
|
max_len = 0
|
|
col_letter = get_column_letter(col[0].column)
|
|
for cell in col:
|
|
val_str = str(cell.value or '')
|
|
if '\n' in val_str:
|
|
val_str = max(val_str.split('\n'), key=len)
|
|
max_len = max(max_len, len(val_str))
|
|
ws.column_dimensions[col_letter].width = max(max_len + 3, 12)
|
|
|
|
output = io.BytesIO()
|
|
wb.save(output)
|
|
output.seek(0)
|
|
|
|
filename = f"inspections_export_{timezone.now().strftime('%Y%m%d_%H%M%S')}.xlsx"
|
|
response = HttpResponse(
|
|
output.getvalue(),
|
|
content_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'
|
|
)
|
|
response['Content-Disposition'] = f'attachment; filename="{filename}"'
|
|
return response
|
|
|
|
|
|
@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}
|
|
if model_name == 'signpanel':
|
|
from assets.models import SignPanelType
|
|
context['panel_types'] = SignPanelType.objects.all().order_by('code')
|
|
|
|
return render(request, template_path, context)
|
|
|