# locations_registry.py from django.contrib.contenttypes.models import ContentType from django.core.exceptions import FieldDoesNotExist from django.db.models import Q, Case, When, IntegerField, Value, CharField from django.db.models.functions import Lower, Coalesce, Concat from django.utils.translation import get_language, gettext as _ from django.views.decorators.http import require_GET from django.http import JsonResponse from common.utils import strip_accents_lower, Unaccent from typing import Callable, Optional from assets.models import StructureLocation from assets.models import TrafficLightIntersection from assets.models import PublicLightingStreet from assets.models import NatureLocation, NatureRWIZ from assets.models import ControlCenterLocation from assets.models import SignStreet from assets.models import RoadStreet from assets.models import CleanLocation LOCATION_REGISTRY = { "structures_location": {"thematic": "structures", "model": StructureLocation, "label": _("Ouvrages"), "parent": {"attr": "structure"},}, "trafficlights_intersection": {"thematic": "trafficlights", "model": TrafficLightIntersection,"label": _("Feux")}, "publiclighting_street":{"thematic":"publiclighting", "model": PublicLightingStreet, "label": _("Éclairage")}, "green_section": {"thematic": "nature", "model": NatureLocation, "label": _("Nature")}, "naturerwiz": {"thematic": "water", "model": NatureRWIZ, "label": _("GIEP")}, "controlcenters_location": {"thematic": "controlcenters", "model": ControlCenterLocation, "label": _("Centres de contrôle"), "parent": {"attr": "control_center"},}, "sign_street":{"thematic":"sign", "model": SignStreet, "label": _("Signalisation")}, "roads_street":{"thematic":"roads", "model": RoadStreet, "label": _("Voirie")}, "clean_location":{"thematic":"clean", "model": CleanLocation, "label": _("Propreté")}, } ASSETS_LOCATION_REGISTRY = ['location', 'intersection', 'street'] def registry_items_for_thematics(thematics: list[str] | None): items = LOCATION_REGISTRY.items() if thematics: items = [(k,v) for (k,v) in items if v["thematic"] in thematics] return items def content_type_for_model(Model): return ContentType.objects.get_for_model(Model) SEARCH_LIMIT_PER_TYPE = 25 SEARCH_LIMIT_TOTAL = 50 def base_search_qs(Model, qn: str, is_nl: bool): qs = ( Model.objects .annotate( code_u=Lower(Unaccent('code')), name_fr_u=Lower(Unaccent('name_fr')), name_nl_u=Lower(Unaccent('name_nl')), ) .filter( Q(code_u__contains=qn) | Q(name_fr_u__contains=qn) | Q(name_nl_u__contains=qn) ) ) prio = Case( # code exact / startswith d'abord When(code_u=qn, then=0), When(code_u__startswith=qn, then=1), # puis startswith/contains dans la langue active avant l'autre *(( When(name_nl_u__startswith=qn, then=2), When(name_fr_u__startswith=qn, then=3), When(code_u__contains=qn, then=4), When(name_nl_u__contains=qn, then=5), When(name_fr_u__contains=qn, then=6), ) if is_nl else ( When(name_fr_u__startswith=qn, then=2), When(name_nl_u__startswith=qn, then=3), When(code_u__contains=qn, then=4), When(name_fr_u__contains=qn, then=5), When(name_nl_u__contains=qn, then=6), )), default=7, output_field=IntegerField(), ) return qs.annotate(_prio=prio) def search_locations( raw_query: str, thematics: list[str] | None = None, label_builder: Optional[Callable[[str | None, str | None, dict], str]] = None, prioritize: str | None = None, ): """ Retourne une liste d'items prêts pour l'UI: { "type": "...", "type_label": "...", "ct": , "obj": , "name": , "code": , "label": " ( - )", "lon": ..., "lat": ... } """ qn = strip_accents_lower((raw_query or "").strip()) if len(qn) < 2: return [] lang = (get_language() or "fr").split("-")[0] is_nl = (lang == "nl") # Label par défaut: "name_lang (code)" — surchargable via label_builder def _default_label_builder(code: str | None, name_lang: str | None, ctx: dict) -> str: name_lang = (name_lang or "").strip() code = (code or "").strip() return f"{name_lang} ({code})" if name_lang and code else (name_lang or code) build_label = label_builder or _default_label_builder items = registry_items_for_thematics(thematics) if not items: return [] ancestor_coords_cache: dict[tuple[str, int], tuple[float | None, float | None]] = {} def _coords_from_ancestors(Model, location_id: int | None) -> tuple[float | None, float | None]: """Retourne (lon, lat) d'un parent (récursif) en cachant les lectures DB.""" if not location_id: return (None, None) cache_key = (Model._meta.label_lower, location_id) if cache_key in ancestor_coords_cache: return ancestor_coords_cache[cache_key] data = ( Model.objects .filter(pk=location_id) .values("lon", "lat", "parent_location_id") .first() ) if not data: ancestor_coords_cache[cache_key] = (None, None) return (None, None) lon, lat = data["lon"], data["lat"] parent_id = data.get("parent_location_id") if (lon is None or lat is None) and parent_id: parent_lon, parent_lat = _coords_from_ancestors(Model, parent_id) if lon is None: lon = parent_lon if lat is None: lat = parent_lat ancestor_coords_cache[cache_key] = (lon, lat) return lon, lat out = [] for type_key, meta in items: Model = meta["model"] ct = content_type_for_model(Model) # --- champs dynamiques à récupérer --- fields = ["id", "code", "name_fr", "name_nl", "lon", "lat"] parent_attr = (meta.get("parent") or {}).get("attr") if parent_attr: fields += [ f"{parent_attr}_id", f"{parent_attr}__name_fr", f"{parent_attr}__name_nl", f"{parent_attr}__code", ] has_parent_location = False try: Model._meta.get_field("parent_location") has_parent_location = True fields.append("parent_location_id") except FieldDoesNotExist: has_parent_location = False rows = ( base_search_qs(Model, qn, is_nl) .order_by("_prio", "code") .values(*fields)[:SEARCH_LIMIT_PER_TYPE] ) for r in rows: name_lang = r["name_nl"] if is_nl else r["name_fr"] if not name_lang: name_lang = r["name_fr"] or r["name_nl"] or "" # --- parent de référence (optionnel) --- parent_ref_id = None parent_ref_label = None if parent_attr: pid = r.get(f"{parent_attr}_id") if pid: # nom du parent dans la langue active (+ fallback) pname = r.get(f"{parent_attr}__name_nl") if is_nl else r.get(f"{parent_attr}__name_fr") if not pname: pname = r.get(f"{parent_attr}__name_fr") or r.get(f"{parent_attr}__name_nl") or "" pcode = r.get(f"{parent_attr}__code") parent_ref_id = pid parent_ref_label = f"{pname} ({pcode})" if pcode else pname lon = r["lon"] lat = r["lat"] if (lon is None or lat is None) and has_parent_location: ancestor_lon, ancestor_lat = _coords_from_ancestors(Model, r.get("parent_location_id")) if lon is None: lon = ancestor_lon if lat is None: lat = ancestor_lat ctx = { "type": type_key, "type_label": meta["label"], "ct": ct.id, "obj": r["id"], "lon": lon, "lat": lat, } out.append({ "type": type_key, "type_label": meta["label"], "ct": ct.id, "obj": r["id"], "name": name_lang, "code": r["code"], "label": build_label(r["code"], name_lang, ctx), "lon": lon, "lat": lat, "thematic": meta.get("thematic"), "parent_ref_id": parent_ref_id, "parent_ref_label": parent_ref_label, }) # Tri global léger par label (les per-type sont déjà ordonnés par priorité) prioritize_type = None if prioritize: p_lower = prioritize.lower().strip() if p_lower in LOCATION_REGISTRY: prioritize_type = p_lower elif "rwi" in p_lower or "rwiz" in p_lower or "giep" in p_lower: prioritize_type = "naturerwiz" else: for key, meta in LOCATION_REGISTRY.items(): if p_lower in key.lower() or p_lower in meta["model"].__name__.lower(): prioritize_type = key break if prioritize_type: out.sort(key=lambda r: (0 if r["type"] == prioritize_type else 1, r["label"] or "")) else: out.sort(key=lambda r: (r["label"] or "")) return out[:SEARCH_LIMIT_TOTAL] def _parse_thematics_param(request): thematics = request.GET.getlist("thematic") parsed_thematics = [] for t in thematics: for item in t.split(","): item_stripped = item.strip() if item_stripped: parsed_thematics.append(item_stripped) return parsed_thematics or None @require_GET def locations_autocomplete(request): q = (request.GET.get("q") or "").strip() thematics = _parse_thematics_param(request) prioritize = request.GET.get("prioritize") def lbl(code, name, ctx): base = f"{name} ({code})" if name and code else (name or code or "") # return f"{base} — {ctx['type_label']}" if base else ctx["type_label"] return base results = search_locations(q, thematics=thematics, label_builder=lbl, prioritize=prioritize) payload = [{ "label": d["label"], # ex. "Schuman (INT-001) — Feux" "value": {"ct": d["ct"], "obj": d["obj"]}, # jQuery UI attend {label, value} "type": d["type"], "type_label": d["type_label"], "lon": d["lon"], "lat": d["lat"], "name": d.get("name"), "code": d.get("code"), # (facultatif) utile pour purger côté UI quand les thématiques changent : "thematic": d.get("thematic"), # ↓ parent de référence envoyé au front "parent_ref_id": d.get("parent_ref_id"), "parent_ref_label": d.get("parent_ref_label"), } for d in results] return JsonResponse(payload, safe=False)