from django.shortcuts import render, get_object_or_404, redirect from django.http import JsonResponse, Http404 from django.db.models import Func from django.db.models.functions import Lower from django.db.models import Q from django.db.models import CharField from django.contrib.gis.geos import Point, GEOSGeometry from django.contrib.gis.db.models.functions import Distance from django.views.decorators.http import require_GET, require_POST from django.utils.translation import gettext as _, get_language from django.urls import reverse import json import unicodedata from common.models import Thematic, UserThematics, Municipality from common.utils import check_thematic_access from assets.models import ArtworkMonument, ArtworkFountain, AssetCategory from assets.models.artworks import ( ArtworkStreet, MONUMENT_TYPE_CHOICES, FOUNTAIN_TYPE_CHOICES, ) from interventions.models import Intervention, InterventionAsset from interventions.views.drafts import InterventionDraftManager @check_thematic_access('artworks') def artworks_index(request): thematic = 'artworks' lang = get_language() selected_assets = [] selected_locations_ids = [] preventive_element_type = '' # Get filters from URL statuses = request.GET.getlist('status') selected_municipalities = request.GET.getlist('municipality') # Get preventive draft ID (if any) for selecting assets preventive_draft_id = request.GET.get('preventive_draft_id') if preventive_draft_id: drafts = InterventionDraftManager(request.session) draft_data = drafts.get(preventive_draft_id) form_data = draft_data.get('form_data', {}) preventive_element_type = form_data.get('type_of_selected_elements') or '' if preventive_element_type == 'assets': selected_assets = draft_data.get('assets', []) or [] else: selected_locations_ids = draft_data.get('locations', []) or [] thematic_choices_qs = ( UserThematics.objects .filter(user_config=request.user.config, can_view_assets=True) .select_related('thematic') .order_by(f'thematic__{"name_nl" if lang == "nl" else "name_fr"}') ) thematics = [tc.thematic for tc in thematic_choices_qs] selected_thematic = { "code": thematic, "name": next((t.name for t in thematics if t.code == thematic), None), "icon": next((t.icon for t in thematics if t.code == thematic), None), } municipality_order_field = 'name_nl' if lang == 'nl' else 'name_fr' municipalities = ( Municipality.objects .all() .order_by(municipality_order_field) ) municipality_choices = [(m.id, m.name) for m in municipalities] # Status choices for filters status_choices = [ ('active', _('Actif')), ('inactive', _('Inactif')), ('to_be_removed', _('À supprimer')), ] context = { 'thematics': thematics, 'municipality_choices': municipality_choices, 'selected_thematic': selected_thematic, 'selected_statuses': statuses, 'selected_municipalities': selected_municipalities, 'status_choices': status_choices, 'preventive_draft_id': preventive_draft_id, 'preventive_element_type': preventive_element_type, 'selected_locations_ids': selected_locations_ids, "selected_assets_json": json.dumps(selected_assets), 'locations_tab_title': _("Localisations"), 'locations_assets_title': _("Assets"), 'default_tab': 'assets', } return render(request, "assets/artworks_index.html", context) def ensure_list(value): if value is None: return [] if isinstance(value, (list, tuple)): return list(value) return [value] def _clean_int_list(values): cleaned = [] for val in values or []: try: cleaned.append(int(val)) except (TypeError, ValueError): continue return cleaned def _extract_lon_lat(obj, fallback_obj=None): lon = getattr(obj, 'lon', None) lat = getattr(obj, 'lat', None) geom = getattr(obj, 'geom', None) if (lon is None or lat is None) and geom is not None: geom_4326 = geom.transform(4326, clone=True) centroid = geom_4326 if geom_4326.geom_type == 'Point' else geom_4326.centroid lon = centroid.x lat = centroid.y if (lon is None or lat is None) and fallback_obj is not None: fallback_lon, fallback_lat = _extract_lon_lat(fallback_obj) if lon is None: lon = fallback_lon if lat is None: lat = fallback_lat return lon, lat def _serialize_geometry(obj): if getattr(obj, 'geojson', None): try: return json.loads(obj.geojson) except (TypeError, ValueError): pass geom = getattr(obj, 'geom', None) if geom: geom_4326 = geom.transform(4326, clone=True) return json.loads(geom_4326.geojson) return None def _street_to_feature(street): lon, lat = _extract_lon_lat(street) return { "type": "Feature", "id": street.id, "geometry": _serialize_geometry(street), "properties": { "code": street.code, "name_fr": street.name_fr, "name_nl": street.name_nl, "municipality_name": street.municipality.name if street.municipality else None, "road_code": street.road_code, "section_code": street.section_code, "street_level": street.street_level, "lon": lon, "lat": lat, "detail_url": street.get_absolute_url(), "element_type": "Voirie œuvres", "thematic": "artworks", "is_location": True, }, } def _artwork_asset_feature( obj, asset_kind, asset_kind_label, subtype_label, detail_url, fallback_category_id=None, fallback_category_label=None, ): from assets.views.core import get_element_type_label street = getattr(obj, 'street', None) lon, lat = _extract_lon_lat(obj, fallback_obj=street) geometry = _serialize_geometry(obj) if geometry is None and lon is not None and lat is not None: geometry = { "type": "Point", "coordinates": [lon, lat], } subtype_code = getattr(obj, 'artwork_type', None) if asset_kind == 'monument' else getattr(obj, 'fountain_type', None) category_obj = getattr(obj, 'category', None) category_id = category_obj.id if category_obj else fallback_category_id category_label = category_obj.get_name() if category_obj else (fallback_category_label or '') return { "type": "Feature", "id": obj.id, "geometry": geometry, "properties": { "code": obj.code, "name_fr": obj.name_fr, "name_nl": obj.name_nl, "status": obj.status, "asset_type": obj._meta.model_name, "asset_kind": asset_kind, "asset_kind_label": asset_kind_label, "asset_subtype": subtype_code, "asset_subtype_label": subtype_label or subtype_code, "category": category_label, "category_id": str(category_id) if category_id is not None else '', "lon": lon, "lat": lat, "detail_url": detail_url, "element_type": get_element_type_label(f"artwork{asset_kind}"), "thematic": "artworks", "is_location": False, }, } @check_thematic_access('artworks') def artworks_streets(request): """Placeholder view so that list URL exists (redirects to index).""" return redirect('assets:artworks_index') @check_thematic_access('artworks') def artworks_streets_detail(request, street_id): street = get_object_or_404( ArtworkStreet.objects.select_related('municipality'), pk=street_id ) assets = street.get_active_assets() or {} monuments = assets.get('monuments') or ArtworkMonument.objects.none() fountains = assets.get('fountains') or ArtworkFountain.objects.none() context = { 'street': street, 'monuments': monuments, 'fountains': fountains, 'monuments_count': monuments.count(), 'fountains_count': fountains.count(), } return render(request, "assets/artworks_street_detail.html", context) @check_thematic_access('artworks') def artworks_assets(request): """Placeholder view used as base URL for asset detail redirects.""" return redirect('assets:artworks_index') @check_thematic_access('artworks') def artworks_assets_detail_redirect(request, asset_model, asset_id): model_key = (asset_model or '').lower() model_map = { 'artworkmonument': (ArtworkMonument, 'artworks_monuments_detail', 'asset_id'), 'artworkfountain': (ArtworkFountain, 'artworks_fountains_detail', 'asset_id'), } config = model_map.get(model_key) if not config: raise Http404( _("Type d'asset inconnu: %(asset_model)s") % {"asset_model": asset_model} ) model, url_name, kwarg = config get_object_or_404(model, pk=asset_id) return redirect(f'assets:{url_name}', **{kwarg: asset_id}) @check_thematic_access('artworks') def artworks_streets_geojson(request): if request.method == "POST": try: data = json.loads(request.body) except json.JSONDecodeError: return JsonResponse({"error": "Invalid JSON"}, status=400) municipalities = ensure_list(data.get('municipality')) filter_geom_raw = data.get("filter_geom") filter_geom = None if filter_geom_raw: try: filter_geom = GEOSGeometry(json.dumps(filter_geom_raw['geometry']), srid=4326) filter_geom.transform(3812) except Exception as exc: return JsonResponse({"error": f"Invalid geometry: {exc}"}, status=400) else: municipalities = request.GET.getlist('municipality') filter_geom = None streets = ArtworkStreet.objects.select_related('municipality') municipality_ids = _clean_int_list(municipalities) if municipality_ids: streets = streets.filter(municipality_id__in=municipality_ids) if filter_geom: streets = streets.filter(geom__intersects=filter_geom) # Recherche plein texte if request.method == 'POST': from assets.views.core import apply_search streets = apply_search(streets, data) features = [_street_to_feature(street) for street in streets] geojson_data = { "type": "FeatureCollection", "features": features, } return JsonResponse(geojson_data, safe=False) @require_GET @check_thematic_access('artworks') def get_nearest_artworks_street(request): try: lon = float(request.GET.get('lon')) lat = float(request.GET.get('lat')) except (TypeError, ValueError): return JsonResponse({"error": _("Coordonnées invalides")}, status=400) user_point = Point(lon, lat, srid=4326) user_point.transform(3812) nearest = ( ArtworkStreet.objects .filter(geom__isnull=False) .annotate(distance=Distance('geom', user_point)) .filter(distance__lte=50) .order_by('distance') .first() ) if not nearest: return JsonResponse({"error": _("Aucune localisation trouvée")}, status=404) lon, lat = _extract_lon_lat(nearest) return JsonResponse({ "location_id": nearest.id, "location_label": f"{nearest.code} - {nearest.get_name()}", "lon": lon, "lat": lat, "detail_url": nearest.get_absolute_url(), }) @check_thematic_access('artworks') def artworks_assets_geojson(request): if request.method == "POST": try: data = json.loads(request.body) except json.JSONDecodeError: return JsonResponse({"error": "Invalid JSON"}, status=400) statuses = ensure_list(data.get('asset_status') or data.get('status')) locations_ids = _clean_int_list(ensure_list(data.get('locations_ids'))) categories = _clean_int_list(ensure_list(data.get('category'))) filter_geom_raw = data.get("filter_geom") filter_geom = None if filter_geom_raw: try: filter_geom = GEOSGeometry(json.dumps(filter_geom_raw['geometry']), srid=4326) filter_geom.transform(3812) except Exception as exc: return JsonResponse({"error": f"Invalid geometry: {exc}"}, status=400) else: statuses = request.GET.getlist('asset_status') or request.GET.getlist('status') locations_ids = _clean_int_list(request.GET.getlist('locations_ids')) categories = _clean_int_list(request.GET.getlist('category')) filter_geom = None features = [] artwork_categories = list(AssetCategory.objects.filter(thematic__code='artworks')) default_fountain_category = next( (c for c in artwork_categories if 'fontain' in strip_accents_lower(c.get_name())), None, ) default_monument_category = next( ( c for c in artwork_categories if any(keyword in strip_accents_lower(c.get_name()) for keyword in ['statue', 'monument', 'sculpt']) ), None, ) def build_queryset(model, default_category_id=None): qs = model.objects.all() if statuses: qs = qs.filter(status__in=statuses) if locations_ids: qs = qs.filter(street_id__in=locations_ids) if categories: category_filter = Q(category_id__in=categories) if default_category_id and default_category_id in categories: category_filter |= Q(category__isnull=True) qs = qs.filter(category_filter) if filter_geom: qs = qs.filter( Q(geom__intersects=filter_geom) | Q(geom__isnull=True, street__geom__intersects=filter_geom) ) if request.method == 'POST': from assets.views.core import apply_search qs = apply_search(qs, data) return qs monument_features = [ _artwork_asset_feature( obj, asset_kind='monument', asset_kind_label=_('Monument'), subtype_label=MONUMENT_TYPE_CHOICES.get(obj.artwork_type, obj.artwork_type), detail_url=reverse('assets:artworks_monuments_detail', kwargs={'asset_id': obj.id}), fallback_category_id=(default_monument_category.id if default_monument_category else None), fallback_category_label=(default_monument_category.get_name() if default_monument_category else _('Monument')), ) for obj in build_queryset( ArtworkMonument, default_category_id=(default_monument_category.id if default_monument_category else None), ) ] fountain_features = [ _artwork_asset_feature( obj, asset_kind='fountain', asset_kind_label=_('Fontaine'), subtype_label=FOUNTAIN_TYPE_CHOICES.get(obj.fountain_type, obj.fountain_type), detail_url=reverse('assets:artworks_fountains_detail', kwargs={'asset_id': obj.id}), fallback_category_id=(default_fountain_category.id if default_fountain_category else None), fallback_category_label=(default_fountain_category.get_name() if default_fountain_category else _('Fontaine')), ) for obj in build_queryset( ArtworkFountain, default_category_id=(default_fountain_category.id if default_fountain_category else None), ) ] features.extend(monument_features) features.extend(fountain_features) features.sort(key=lambda feature: feature["properties"].get("code") or "") geojson_data = { "type": "FeatureCollection", "features": features, } return JsonResponse(geojson_data, safe=False) def artworks_fountains(request): pass @check_thematic_access('artworks') def artworks_fountains_detail(request, asset_id): from django.contrib.contenttypes.models import ContentType from assets.views.core import get_asset_edit_context fountain = get_object_or_404(ArtworkFountain, id=asset_id) content_type = ContentType.objects.get_for_model(ArtworkFountain) interventions = Intervention.objects.filter( interventionasset__object_id=asset_id ).order_by('-creation_time') context = { 'asset': fountain, 'fountain': fountain, 'content_type': content_type, 'interventions': interventions, } context.update(get_asset_edit_context(request, fountain, content_type)) return render(request, "assets/detail_generic.html", context) def artworks_monuments(request): pass @check_thematic_access('artworks') def artworks_monuments_detail(request, asset_id): from django.contrib.contenttypes.models import ContentType from assets.views.core import get_asset_edit_context monument = get_object_or_404(ArtworkMonument, id=asset_id) content_type = ContentType.objects.get_for_model(ArtworkMonument) interventions = Intervention.objects.filter( interventionasset__object_id=asset_id ).order_by('-creation_time') context = { 'asset': monument, 'monument': monument, 'content_type': content_type, 'interventions': interventions, } context.update(get_asset_edit_context(request, monument, content_type)) return render(request, "assets/detail_generic.html", context) @check_thematic_access('artworks') def monuments_geojson(request): from django.contrib.gis.geos import GEOSGeometry if request.method == "POST": try: data = json.loads(request.body) except json.JSONDecodeError: return JsonResponse({"error": "Invalid JSON"}, status=400) statuses = data.get('asset_status', []) if isinstance(statuses, str): statuses = [statuses] # Récupération et parsing de la géométrie filter_geom_raw = data.get("filter_geom") if filter_geom_raw: try: filter_geom = GEOSGeometry(json.dumps(filter_geom_raw['geometry']), srid=4326) filter_geom.transform(3812) except Exception as e: return JsonResponse({"error": f"Invalid geometry: {str(e)}"}, status=400) else: filter_geom = None else: statuses = request.GET.getlist('asset_status') filter_geom = None monuments = ArtworkMonument.objects.all() # Filters if statuses: monuments = monuments.filter(status__in=statuses) if filter_geom: monuments = monuments.filter(geom__intersects=filter_geom) def obj_to_feature(obj): geometry = json.loads(obj.geojson) if obj.geojson else None return { "type": "Feature", "id": obj.id, "geometry": geometry, "properties": { "code": obj.code, "name_fr": obj.name_fr, "name_nl": obj.name_nl, "asset_type": obj.artwork_type, "status": obj.status, "validation_status": obj.validation_status, "lon": obj.lon, "lat": obj.lat }, } features = [] for a in monuments: features.append(obj_to_feature(a)) features = sorted(features, key=lambda f: f["properties"]["code"]) geojson_data = { "type": "FeatureCollection", "features": features, } return JsonResponse(geojson_data, safe=False) @check_thematic_access('artworks') def fountains_geojson(request): from django.contrib.gis.geos import GEOSGeometry if request.method == "POST": try: data = json.loads(request.body) except json.JSONDecodeError: return JsonResponse({"error": "Invalid JSON"}, status=400) statuses = data.get('status', []) if isinstance(statuses, str): statuses = [statuses] # Récupération et parsing de la géométrie filter_geom_raw = data.get("filter_geom") if filter_geom_raw: try: filter_geom = GEOSGeometry(json.dumps(filter_geom_raw['geometry']), srid=4326) filter_geom.transform(3812) except Exception as e: return JsonResponse({"error": f"Invalid geometry: {str(e)}"}, status=400) else: filter_geom = None else: statuses = request.GET.getlist('status') filter_geom = None fountains = ArtworkFountain.objects.all() # Filters if statuses: fountains = fountains.filter(status__in=statuses) if filter_geom: fountains = fountains.filter(geom__intersects=filter_geom) def obj_to_feature(obj): geometry = json.loads(obj.geojson) if obj.geojson else None return { "type": "Feature", "id": obj.id, "geometry": geometry, "properties": { "code": obj.code, "name_fr": obj.name_fr, "name_nl": obj.name_nl, "asset_type": obj.fountain_type, "status": obj.status, "validation_status": obj.validation_status, "lon": obj.lon, "lat": obj.lat }, } features = [] for a in fountains: features.append(obj_to_feature(a)) features = sorted(features, key=lambda f: f["properties"]["code"]) geojson_data = { "type": "FeatureCollection", "features": features, } return JsonResponse(geojson_data, safe=False) # Appelle la fonction PostgreSQL unaccent(...) class Unaccent(Func): function = 'public.unaccent_immutable' output_field = CharField() def strip_accents_lower(s: str) -> str: if not s: return '' s = ''.join(c for c in unicodedata.normalize('NFD', s) if unicodedata.category(c) != 'Mn') return s.lower() def _equipment_type_from_model(asset): class_name = asset.__class__.__name__ return class_name @require_GET def artworks_assets_autocomplete(request): raw_query_original = (request.GET.get('q') or '') raw_query = raw_query_original.strip() # Normalise la requête (équivalent de unaccent+lower côté Python) qn = strip_accents_lower(raw_query) has_search_term = len(qn) > 0 category_id = request.GET.get('asset_category') try: category_id = int(category_id) except (TypeError, ValueError): category_id = None def base_qs(model, extra_filter=None): ann = { 'code_u': Lower(Unaccent('code')), 'name_fr_u': Lower(Unaccent('name_fr')), 'name_nl_u': Lower(Unaccent('name_nl')), } q = model.objects.annotate(**ann) # Si un terme de recherche est fourni, filtrer dessus if has_search_term: q = q.filter( Q(code_u__contains=qn) | Q(name_fr_u__contains=qn) | Q(name_nl_u__contains=qn) ) # Filtrer uniquement les non supprimés q = q.exclude(status='removed') if extra_filter: q = q.filter(**extra_filter) if category_id is not None: q = q.filter( Q(category_id=category_id) | Q(category__isnull=True, model__category_id=category_id) ) return q # Récupère QS pour chaque type fountains_qs = base_qs(ArtworkFountain)[:100] monuments_qs = base_qs(ArtworkMonument)[:100] # Concatène en mémoire (on priorise ensuite) items = [] for a in list(fountains_qs) + list(monuments_qs): # priorisation locale (recalcule ici car on merge des modèles différents) code_u = getattr(a, 'code_u', None) name_fr_u = getattr(a, 'name_fr_u', None) name_nl_u = getattr(a, 'name_nl_u', None) prio = 7 # Priorisation uniquement si un terme de recherche est fourni if has_search_term: if code_u == qn: prio = 0 elif code_u and code_u.startswith(qn): prio = 1 elif (name_fr_u and name_fr_u.startswith(qn)) or (name_nl_u and name_nl_u.startswith(qn)): prio = 2 elif (code_u and qn in code_u) or (name_fr_u and qn in name_fr_u) or (name_nl_u and qn in name_nl_u): prio = 4 else: # Sans terme de recherche, on trie par code prio = 5 equipment_type = _equipment_type_from_model(a) parts = [a.code, a.get_name()] equipment_label = " - ".join(p for p in parts if p) code = getattr(a, 'code', '') or '' label = equipment_label lon = a.lon lat = a.lat items.append({ "prio": prio, "code": code or "", "id": getattr(a, 'id', None), "label": label, # attendus par ton autocomplete équipements "equipment_id": getattr(a, 'id', None), "equipment_label": equipment_label, "equipment_type": equipment_type, "lon": lon, "lat": lat, }) # Tri global par priorité puis code, coupe à 100 items.sort(key=lambda x: (x["prio"], x["code"])) data = items[:100] return JsonResponse(data, safe=False) @require_GET def get_nearest_artworks_asset(request): from django.contrib.contenttypes.models import ContentType try: lon = float(request.GET.get('lon')) lat = float(request.GET.get('lat')) except (TypeError, ValueError): return JsonResponse({"error": "Coordonnées invalides"}, status=400) user_point = Point(lon, lat, srid=4326) user_point.transform(3812) # Lambert 2008 → distances en mètres # plus proche des fontaines ≤ 50 m nearest_fountain = ( ArtworkFountain.objects .filter(geom__isnull=False) .annotate(distance=Distance('geom', user_point)) .filter(distance__lte=50) .order_by('distance') .first() ) # plus proche des monuments ≤ 50 m nearest_monument = ( ArtworkMonument.objects .filter(geom__isnull=False) .annotate(distance=Distance('geom', user_point)) .filter(distance__lte=50) .order_by('distance') .first() ) # garder uniquement ceux trouvés, puis prendre le plus proche candidates = [obj for obj in (nearest_fountain, nearest_monument) if obj is not None] if not candidates: return JsonResponse({"error": _("Aucun équipement trouvé")}, status=404) nearest = min(candidates, key=lambda o: o.distance) # Récupérer le ContentType de l'asset ct = ContentType.objects.get_for_model(nearest) return JsonResponse({ "equipment_id": nearest.id, "equipment_ct": ct.id, "equipment_label": f"{nearest.get_name()}", "lon": nearest.lon, "lat": nearest.lat, # "distance_m": round(nearest.distance.m, 2), "equipment_type": nearest.__class__.__name__, })