[Backend] chore: import_aveugles (#2364)

* chore: import_aveugles

* Fix case with `[]`

* chore: telephone serializer empty edge case fix

* chore: settrace removed

* Change PATCH

- `def validate` was putting all accessibility to None even those not provided
in payload. Dedicated management in case of partial.
- in PATCH through API we should be able to change a value but not set it
to None

---------

Co-authored-by: Marie-Laure Vernay <mlvernay@gmail.com>

Author: Destinyyyy

erp/imports/serializers.py

@@ -202,6 +202,9 @@ def validate_code_postal(self, obj):
 
     def validate_telephone(self, obj):
         # weird invisible char
+        if not obj:
+            return obj
+
         obj = obj.replace(" ", "")
         obj = obj.replace(" ", "")
         return obj
@@ -235,81 +238,85 @@ def validate(self, obj):
         if "accessibilite" not in obj:
             raise serializers.ValidationError("Veuillez fournir les données d'accessibilité.")
 
-        if self.instance:
-            # if we are updating an ERP, only accessibility, asp_id and import_email are editable
-            self.instance.import_email = obj.get("import_email")
-            self.instance.asp_id = obj.get("asp_id")
-            accessibilite = Accessibilite(**obj["accessibilite"])
-            accessibilite.full_clean()
-
-            sources_data_list = []
-            sources_data = obj.get("sources") or []
-            for source_data in sources_data:
-                external_source = ExternalSource(**source_data)
-                external_source.full_clean(exclude=("erp",))
-                sources_data_list.append(model_to_dict(external_source))
-
-            return (
-                model_to_dict(self.instance)
-                | {"accessibilite": model_to_dict(accessibilite)}
-                | {"sources": sources_data_list}
-            )
-
-        if not obj.get("voie") and not obj.get("lieu_dit"):
-            raise serializers.ValidationError("Veuillez entrer une voie OU un lieu-dit")
-
-        for i in range(3):
-            try:
-                address = get_address_query_to_geocode(obj)
-                locdata = geocoder.geocode(address, postcode=obj["code_postal"])
-                if not locdata:
-                    raise RuntimeError
-                self._geom = locdata["geom"]
-                obj["voie"] = locdata["voie"]
-                obj["lieu_dit"] = locdata["lieu_dit"]
-                obj["code_postal"] = locdata["code_postal"]
-                obj["commune"] = locdata["commune"]
-                obj["code_insee"] = locdata["code_insee"]
-                obj["geoloc_provider"] = locdata["provider"]
-                obj["numero"] = locdata["numero"]
-                obj["ban_id"] = locdata.get("ban_id")
-                obj.pop("latitude", None)
-                obj.pop("longitude", None)
-                break
-            except (RuntimeError, KeyError):
-                if i < 2:
-                    continue
-
-                if obj.get("latitude") is not None and obj.get("longitude") is not None:
-                    self._geom = Point((obj["longitude"], obj["latitude"]), srid=4326)
-                    obj.pop("latitude")
-                    obj.pop("longitude")
+        if not self.partial:
+            if self.instance:
+                # if we are updating an ERP, only accessibility, asp_id and import_email are editable
+                self.instance.import_email = obj.get("import_email")
+                self.instance.asp_id = obj.get("asp_id")
+                accessibilite = Accessibilite(**obj["accessibilite"])
+                accessibilite.full_clean()
+
+                sources_data_list = []
+                sources_data = obj.get("sources") or []
+                for source_data in sources_data:
+                    external_source = ExternalSource(**source_data)
+                    external_source.full_clean(exclude=("erp",))
+                    sources_data_list.append(model_to_dict(external_source))
+
+                return (
+                    model_to_dict(self.instance)
+                    | {"accessibilite": model_to_dict(accessibilite)}
+                    | {"sources": sources_data_list}
+                )
+
+            if not obj.get("voie") and not obj.get("lieu_dit"):
+                raise serializers.ValidationError("Veuillez entrer une voie OU un lieu-dit")
+
+            for i in range(3):
+                try:
+                    address = get_address_query_to_geocode(obj)
+                    locdata = geocoder.geocode(address, postcode=obj["code_postal"])
+                    if not locdata:
+                        raise RuntimeError
+                    self._geom = locdata["geom"]
+                    obj["voie"] = locdata["voie"]
+                    obj["lieu_dit"] = locdata["lieu_dit"]
+                    obj["code_postal"] = locdata["code_postal"]
+                    obj["commune"] = locdata["commune"]
+                    obj["code_insee"] = locdata["code_insee"]
+                    obj["geoloc_provider"] = locdata["provider"]
+                    obj["numero"] = locdata["numero"]
+                    obj["ban_id"] = locdata.get("ban_id")
+                    obj.pop("latitude", None)
+                    obj.pop("longitude", None)
                     break
+                except (RuntimeError, KeyError):
+                    if i < 2:
+                        continue
+
+                    if obj.get("latitude") is not None and obj.get("longitude") is not None:
+                        self._geom = Point((obj["longitude"], obj["latitude"]), srid=4326)
+                        obj.pop("latitude")
+                        obj.pop("longitude")
+                        break
+
+                    raise serializers.ValidationError(f"Adresse non localisable: {address}")
 
-                raise serializers.ValidationError(f"Adresse non localisable: {address}")
+            obj["commune_ext"] = Commune.objects.filter(
+                nom__iexact=obj["commune"], code_postaux__contains=[obj["code_postal"]]
+            ).first()
 
-        obj["commune_ext"] = Commune.objects.filter(
-            nom__iexact=obj["commune"], code_postaux__contains=[obj["code_postal"]]
-        ).first()
+            self._ensure_no_duplicate(obj)
 
-        self._ensure_no_duplicate(obj)
+            erp_data = obj.copy()
+            erp_data.pop("accessibilite")
+            erp_data.pop("sources")
 
-        erp_data = obj.copy()
-        erp_data.pop("accessibilite")
-        erp_data.pop("sources")
+            Erp(**erp_data).full_clean(exclude=("source_id", "asp_id", "user", "metadata", "search_vector"))
 
-        Erp(**erp_data).full_clean(exclude=("source_id", "asp_id", "user", "metadata", "search_vector"))
-        Accessibilite(**obj["accessibilite"]).full_clean()
+        if "accessibilite" in obj:
+            Accessibilite(**obj["accessibilite"]).full_clean()
 
         sources_data = obj.get("sources") or []
         for source_data in sources_data:
             ExternalSource(**source_data).full_clean(exclude=("erp",))
 
         return obj
 
-    def update(self, instance, validated_data, partial=True):
-        # If enrich_only, it won't update any access info already there
+    def update(self, instance, validated_data):
         enrich_only = self.context.get("enrich_only") or False
+        raw_data = self.initial_data
+
         # if we are updating an ERP, only accessibility, asp_id and import_email are editable
         if validated_data.get("import_email"):
             instance.import_email = validated_data["import_email"]
@@ -320,15 +327,20 @@ def update(self, instance, validated_data, partial=True):
             instance.save(update_fields=["asp_id"])
 
         accessibilite = instance.accessibilite
-        for attr in ("id", "erp"):
-            validated_data["accessibilite"].pop(attr, False)
+        acc_data = validated_data.get("accessibilite", {})
+        raw_acc_data = raw_data.get("accessibilite", {})
+
+        for attr, new_value in acc_data.items():
+            if attr in ("id", "erp"):
+                continue
 
-        for attr in validated_data["accessibilite"]:
             if enrich_only and getattr(accessibilite, attr, None) is not None:
                 continue
 
-            new_value = validated_data["accessibilite"][attr]
-            if new_value not in (None, [], ()):
+            if self.partial and attr in raw_acc_data and raw_acc_data[attr] in (None, "", [], ()):
+                continue
+
+            if not enrich_only or new_value not in (None, [], ()):
                 setattr(accessibilite, attr, new_value)
                 self._handle_children_reinit(accessibilite, attr)
 

erp/management/commands/import_aveugles.py

@@ -0,0 +1,81 @@
+import csv
+
+import reversion
+from django.core.management.base import BaseCommand
+from django.db import IntegrityError
+from rest_framework.exceptions import ValidationError
+
+from erp.imports.serializers import ErpImportSerializer
+from erp.models import Erp
+
+
+class Command(BaseCommand):
+    help = "Import cinema aveugles accessibility"
+
+    def handle(self, *args, **kwargs):
+        with open("cinemas_audiodescription.csv", "r") as file:
+            reader = csv.DictReader(file, delimiter=";")
+
+            for row in reader:
+                if not row["erp_id"]:
+                    continue
+
+                accessibilite = dict()
+                accessibilite["erp_id"] = row["erp_id"]
+
+                if (
+                    row["accueil_audiodescription"]
+                    == "avec équipement permanent, casques et boîtiers disponibles à l’accueil"
+                ):
+                    accessibilite["accueil_audiodescription"] = ["avec_équipement_permanent"]
+
+                if (
+                    row["accueil_audiodescription"]
+                    == "avec équipement permanent nécessitant le téléchargement d'une application sur smartphone"
+                ):
+                    accessibilite["accueil_audiodescription"] = ["avec_app"]
+
+                if row["accueil_audiodescription"] == "avec équipement occasionnel selon la programmation":
+                    accessibilite["accueil_audiodescription"] = ["avec_équipement_occasionnel"]
+
+                accessibilite["accueil_personnels"] = row["accueil_personnels"]
+                accessibilite["entree_balise_sonore"] = True if row["entree_balise_sonore"] == "True" else False
+                accessibilite["accueil_audiodescription_presence"] = (
+                    True if row["accueil_audiodescription_presence"] == "True" else False
+                )
+                accessibilite["commentaire"] = row["commentaire"]
+
+                if not accessibilite["accueil_audiodescription_presence"]:
+                    accessibilite["accueil_audiodescription"] = []
+
+                erp = Erp.objects.get(pk=row["erp_id"])
+
+                self._update_erp(
+                    data={"accessibilite": accessibilite, "activite": erp.activite.nom, **erp.__dict__}, erp=erp
+                )
+
+    def _update_erp(self, data, erp=None):
+        serializer = ErpImportSerializer(instance=erp, data=data)
+        try:
+            serializer.is_valid(raise_exception=True)
+        except Exception as e:
+            if (
+                isinstance(e, ValidationError)
+                and "non_field_errors" in e.get_codes()
+                and "duplicate" in e.get_codes()["non_field_errors"]
+            ):
+                existing_erp_pk = int(e.detail["non_field_errors"][1])
+                existing_erp = Erp.objects.get(pk=existing_erp_pk)
+
+                return self._update_erp(data=data, erp=existing_erp)
+            return
+        try:
+            with reversion.create_revision():
+                erp = serializer.save()
+                reversion.set_comment("Created via cinema blindness accessibility import")
+        except reversion.errors.RevertError:
+            erp = serializer.save()
+        except IntegrityError as integrity_error:
+            raise integrity_error
+
+        print(f"ERP {erp.id} updated available at: {erp.get_absolute_uri()}")

tests/api/tests.py

@@ -546,7 +546,7 @@ def test_post_patch(self, api_client, activite):
         response = api_client.patch(reverse("erp-detail", kwargs={"slug": erp.slug}), data=payload, format="json")
         assert response.status_code == 200, response.json()
         erp.accessibilite.refresh_from_db()
-        assert erp.accessibilite.transport_station_presence is False, "Should kept access info"
+        assert erp.accessibilite.transport_station_presence is False, "Should have kept access info"
         assert erp.accessibilite.commentaire == "Updated comment"
 
         payload = {"accessibilite": {"transport_station_presence": None}}