Multi epcis (#1010)

Makefile

@@ -78,12 +78,16 @@ createcachetable:
 
 .PHONY: createsuperuser
 createsuperuser:
-	$(DJANGO_ADMIN) createsuperuse
+	$(DJANGO_ADMIN) createsuperuser
 
 .PHONY: seed-database
 seed-database:
 	$(DJANGO_ADMIN) loaddata categories actions acteur_services acteur_types
 
+.PHONY: clear-cache
+clear-cache:
+	$(DJANGO_ADMIN) clear_cache --all
+
 # Dependencies management
 .PHONY: pip-update
 pip-update:

bin/post_deploy

@@ -3,3 +3,4 @@
 python manage.py createcachetable
 python manage.py migrate
 python manage.py clearsessions
+python manage.py clear_cache --all

dev-requirements.in

@@ -4,7 +4,6 @@ beautifulsoup4
 black
 django-browser-reload
 django-debug-toolbar
-django-extensions
 factory_boy
 honcho
 pre-commit

jinja2/qfdmo/iframe_configurator/base.html

@@ -22,6 +22,7 @@ <h2>Paramètres de l’intégration</h2>
             <hr>
             <div
                 class="qfdmo-max-w-screen-lg"
+                data-controller="scroll"
             >
                 <h2>Résultat</h2>
                 <h3>Code à implémenter</h3>

qfdmo/api.py

@@ -8,6 +8,7 @@
 from ninja import Field, FilterSchema, ModelSchema, Query, Router
 from ninja.pagination import paginate
 
+from qfdmo.geo_api import search_epci_code
 from qfdmo.models import (
     ActeurService,
     ActeurStatus,
@@ -170,3 +171,8 @@ def services(request):
 )
 def acteur(request, identifiant_unique: str):
     return get_object_or_404(DisplayedActeur, pk=id, statut=ActeurStatus.ACTIF)
+
+
+@router.get("/autocomplete/configurateur")
+def autocomplete_epcis(request, query: str):
+    return search_epci_code(query)

qfdmo/fields.py

@@ -1,7 +1,11 @@
+import logging
+
 from django import forms
 from django.template.loader import render_to_string
 from django.utils.safestring import mark_safe
 
+logger = logging.getLogger(__name__)
+
 
 class GroupeActionChoiceField(forms.ModelMultipleChoiceField):
     def label_from_instance(self, obj):
@@ -11,12 +15,3 @@ def label_from_instance(self, obj):
                 {"groupe_action": obj},
             )
         )
-
-
-class EPCIField(forms.ChoiceField):
-    def to_python(self, value):
-        # TODO : once multiple EPCI codes will be managed, this method will be useless
-        # and the frontend will be rewritten to support a more complex state with all
-        # values matching their labels.
-        value = super().to_python(value)
-        return value.split(" - ")[1]

qfdmo/forms.py

@@ -7,8 +7,8 @@
 from django.utils.safestring import mark_safe
 from dsfr.forms import DsfrBaseForm
 
-from qfdmo.fields import EPCIField, GroupeActionChoiceField
-from qfdmo.geo_api import all_epci_codes
+from qfdmo.fields import GroupeActionChoiceField
+from qfdmo.geo_api import epcis_from, formatted_epcis_as_list_of_tuple
 from qfdmo.models import DagRun, DagRunStatus, SousCategorieObjet
 from qfdmo.models.action import (
     Action,
@@ -301,7 +301,7 @@ def load_choices(
     )
 
     epci_codes = forms.MultipleChoiceField(
-        choices=all_epci_codes,
+        choices=[(code, code) for code in cast(List[str], epcis_from(["code"]))],
         widget=forms.MultipleHiddenInput(),
         required=False,
     )
@@ -367,27 +367,29 @@ class ConfiguratorForm(DsfrBaseForm):
         "faire une carte que sur les points de collecte ou de réparation, il vous "
         "suffit de décocher toutes les autres actions possibles.",
     )
-    epci_codes = EPCIField(
+    epci_codes = forms.MultipleChoiceField(
         label=mark_safe(
             """
         <hr/>
         <h3>Paramètres de la carte</h3>
-        1. Choisir l’EPCI affiché par défaut sur la carte"""
+        1. Choisir les EPCI affichés par défaut sur la carte"""
         ),
         help_text="Commencez à taper un nom d’EPCI et sélectionnez un EPCI parmi "
         "les propositions de la liste.",
-        choices=all_epci_codes,
-        initial="",
+        # TODO: voir comment évaluer cela "lazily"
+        # L'utilisation de lazy(all_epci_codes(...)) génère une erreur côté Django DSFR
+        choices=formatted_epcis_as_list_of_tuple(),
         widget=GenericAutoCompleteInput(
-            additionnal_info=mark_safe(
-                render_to_string(
-                    "forms/widgets/epci_codes_additionnal_info.html",
-                )
-            ),
-            attrs={
-                "class": "fr-input",
-                "wrapper_classes": "qfdmo-max-w-[576]",
-                "autocomplete": "off",
+            attrs={"data-autocomplete-target": "hiddenInput", "class": "qfdmo-hidden"},
+            extra_attrs={
+                "selected_label": "Vos EPCI sélectionnés",
+                "empty_label": "Il n’y a pas d’EPCI sélectionné pour le moment",
+                "endpoint": "/api/qfdmo/autocomplete/configurateur?query=",
+                "additionnal_info": mark_safe(
+                    render_to_string(
+                        "forms/widgets/epci_codes_additionnal_info.html",
+                    )
+                ),
             },
         ),
     )

qfdmo/geo_api.py

@@ -1,41 +1,72 @@
-from typing import List, Tuple, cast
+import itertools
+import logging
+from typing import List, Tuple, Union, cast
 
 import requests
 from django.contrib.gis.geos import GEOSGeometry
 from django.core.cache import caches
+from rapidfuzz import fuzz, process
 
 db_cache = caches["database"]
 
 
-def fetch_epci_codes() -> List[Tuple[str, str]]:
-    """Retrieves EPCI codes from geo.api"""
-    response = requests.get("https://geo.api.gouv.fr/epcis/?fields=code,nom")
-    codes = [
-        (item["code"], f"{item['nom']} - {item['code']}") for item in response.json()
-    ]
-    return codes
-
-
-def all_epci_codes():
-    return cast(
-        List[Tuple[str, str]],
-        db_cache.get_or_set(
-            "all_epci_codes", fetch_epci_codes, timeout=3600 * 24 * 365
-        ),
+logger = logging.getLogger(__name__)
+
+BASE_URL = "https://geo.api.gouv.fr"
+
+
+def formatted_epcis_list() -> List[str]:
+    formatted = [f"{nom} - {code}" for nom, code in epcis_from(["nom", "code"])]
+    return formatted
+
+
+def formatted_epcis_as_list_of_tuple() -> List[Tuple[str, str]]:
+    return [(item, item) for item in formatted_epcis_list()]
+
+
+def search_epci_code(query) -> List[str]:
+    results = process.extract(
+        query.lower(),
+        formatted_epcis_list(),
+        scorer=fuzz.WRatio,
+        limit=5,
     )
+    return [match for match, score, index in results]
 
 
-def retrieve_epci_geojson(epci):
-    all_epcis_codes = db_cache.get_or_set(
-        "all_epci_codes", fetch_epci_codes, timeout=3600 * 24 * 365
+def fetch_epci_codes() -> List[str]:
+    """Retrieves EPCI codes from geo.api"""
+    response = requests.get(f"{BASE_URL}/epcis/?fields=code,nom")
+    return response.json()
+
+
+def epcis_from(fields: List[str] = []) -> Union[List, List[tuple]]:
+    """Retrieves a list of fields from the geo.api
+    At the moment only nom and code fields are supported.
+    The fields are returned in the same order they are passed in the parameter
+    """
+    raw_codes = cast(
+        List,
+        db_cache.get_or_set("all_epci_codes", fetch_epci_codes, timeout=3600 * 24 * 30),
     )
+    epcis = [[code.get(field) for field in fields] for code in raw_codes]
 
-    if epci not in [k for k, v in cast(List[Tuple[str, str]], all_epcis_codes)]:
+    # This can be handy if we just need a list of codes and not
+    # a list of list of codes.
+    if len(fields) == 1:
+        return list(itertools.chain.from_iterable(epcis))
+
+    return epcis
+
+
+def retrieve_epci_geojson(epci):
+    all_codes = epcis_from(["code"])
+    if epci not in all_codes:
         raise ValueError(f"The provided EPCI code does not seem to exist | {epci}")
 
     def fetch_epci_bounding_box():
         response = requests.get(
-            f"https://geo.api.gouv.fr/epcis/{epci}?nom=Nan&fields=code,nom,contour"
+            f"{BASE_URL}/epcis/{epci}?nom=Nan&fields=code,nom,contour"
         )
         contour = response.json()["contour"]
         return contour

qfdmo/leaflet.py

@@ -1,7 +1,10 @@
 import json
+import logging
 from json.decoder import JSONDecodeError
 from typing import List, TypedDict
 
+logger = logging.getLogger(__name__)
+
 
 class PointDict(TypedDict):
     lat: float
@@ -25,17 +28,17 @@ def center_from_leaflet_bbox(custom_bbox_as_string: str) -> List[float]:
 
 
 def sanitize_leaflet_bbox(custom_bbox_as_string: str) -> List[float] | None:
-    custom_bbox: LeafletBbox = json.loads(custom_bbox_as_string)
-
     try:
+        custom_bbox: LeafletBbox = json.loads(custom_bbox_as_string)
         # Handle center
         return [
             custom_bbox["southWest"]["lng"],
             custom_bbox["southWest"]["lat"],
             custom_bbox["northEast"]["lng"],
             custom_bbox["northEast"]["lat"],
         ]
-    except KeyError:
+    except (KeyError, JSONDecodeError):
+        logger.error("An error occured while sanitizing a leaflet bbox. It was ignored")
         # TODO : gérer l'erreur
         return []
 

qfdmo/models/acteur.py

@@ -13,7 +13,7 @@
 from django.contrib.gis.geos.geometry import GEOSGeometry
 from django.core.cache import cache
 from django.core.files.images import get_image_dimensions
-from django.db.models import Exists, Min, OuterRef
+from django.db.models import Exists, Min, OuterRef, Q
 from django.db.models.functions import Now
 from django.forms import ValidationError, model_to_dict
 from django.http import HttpRequest
@@ -209,8 +209,15 @@ def in_geojson(self, geojson):
             # TODO : test
             return self.physical()
 
-        geometry = GEOSGeometry(geojson)
-        return self.physical().filter(location__within=geometry).order_by("?")
+        if type(geojson) is not list:
+            geojson = [geojson]
+
+        geometries = [GEOSGeometry(geojson) for geojson in geojson]
+        query = Q()
+        for geometry in geometries:
+            query |= Q(location__within=geometry)
+
+        return self.physical().filter(query).order_by("?")
 
     def in_bbox(self, bbox):
         if not bbox:

qfdmo/views/adresses.py

@@ -272,9 +272,9 @@ def _bbox_and_acteurs_from_location_or_epci(self, acteurs):
             geojson_list = [retrieve_epci_geojson(code) for code in epci_codes]
             bbox = bbox_from_list_of_geojson(geojson_list, buffer=0)
             if geojson_list:
-                # TODO: handle case with multiples EPCI codes passed in URL
-                geojson = json.dumps(geojson_list[0])
-                acteurs = acteurs.in_geojson(geojson)
+                acteurs = acteurs.in_geojson(
+                    [json.dumps(geojson) for geojson in geojson_list]
+                )
             return compile_leaflet_bbox(bbox), acteurs
 
         return custom_bbox, acteurs.none()