Import new communes

mesads/app/models.py

@@ -369,6 +369,7 @@ class ADS(SmartValidationMixin, CharFieldsStripperMixin, models.Model):
     class Meta:
         verbose_name = "ADS"
         verbose_name_plural = "ADS"
+
         constraints = [
             # To understand why we do not define violation_error_message here,
             # see the documentation of the method `unique_error_message`

mesads/app/test_reversion_diff.py

@@ -19,7 +19,7 @@ class TestReversionDiff(TestCase):
     def setUp(self):
         self.user = User.objects.create_user(email="test@test.com")
         self.commune = Commune.objects.create(
-            libelle="Melesse", insee="35173", departement="35"
+            type_commune="COM", libelle="Melesse", insee="35173", departement="35"
         )
         prefecture = Prefecture.objects.create(numero="35", libelle="Ille-et-Vilaine")
         administrator = ADSManagerAdministrator.objects.create(prefecture=prefecture)

mesads/app/views/test_ads.py

@@ -522,6 +522,7 @@ def test_get_incorrect_ads_manager(self):
         existing ADSManager, but not the one of the ADS, we want to make sure
         the user is redirected to the correct page."""
         commune = Commune.objects.create(
+            type_commune="COM",
             insee="xx",
             departement="xx",
             libelle="xx",

mesads/fradm/management/commands/load_communes.py → mesads/fradm/management/commands/fix_load_communes.py

@@ -1,3 +1,8 @@
+# When MesADS started, we loaded the commune published by INSEE from the file
+# https://www.insee.fr/fr/information/5057840. We didn't insert all the rows
+# that were not of type "COM".
+# This script loads all the other types.
+
 import argparse
 import csv
 import sys
@@ -9,8 +14,6 @@
 
 
 class Command(BaseCommand):
-    help = "Load communes from the CSV file published by INSEE (https://www.insee.fr/fr/information/5057840)"
-
     def add_arguments(self, parser):
         parser.add_argument("communes_file", type=argparse.FileType("r"))
 
@@ -23,20 +26,19 @@ def handle(self, *args, **options):
         print(self.style.SUCCESS(f"\nCreated: {created} new entries"))
 
     def insert_row(self, row):
-        if row["TYPECOM"] != "COM":
-            print(
-                self.style.SUCCESS(
-                    "\nSkip insert of row of type %s: %s" % (row["TYPECOM"], row["NCC"])
-                )
-            )
+        if row["TYPECOM"] == "COM":
             return 0
 
-        # The get_or_create below might fail in future updates if we attempt to
-        # reimport communes with different values for a row.
-        # If IntegrityError is raised, we need to find out why data has changed
-        # and change this loader accordingly.
+        parent = Commune.objects.get(insee=row["COMPARENT"])
+
+        if row["DEP"] != "" and row["DEP"] != parent.departement:
+            raise ValueError("Departement mismatch")
+
         commune, created = Commune.objects.get_or_create(
-            insee=row["COM"], departement=row["DEP"], libelle=row["LIBELLE"]
+            type_commune=row["TYPECOM"],
+            insee=row["COM"],
+            departement=parent.departement,
+            libelle=row["LIBELLE"],
         )
         sys.stdout.write(self.style.SUCCESS("."))
         sys.stdout.flush()

mesads/fradm/management/commands/update_communes.py

@@ -0,0 +1,222 @@
+import argparse
+import csv
+import os
+import re
+import sys
+import traceback
+import yaml
+
+from django.contrib.contenttypes.models import ContentType
+from django.core.management.base import BaseCommand, CommandError
+from django.db import connection, transaction
+
+from mesads.app.models import ADS, ADSManager, ADSManagerAdministrator
+from mesads.fradm.models import Commune
+from mesads.vehicules_relais.models import Vehicule
+
+
+class Command(BaseCommand):
+    help = "Mise à jour de la liste des communes dans la base de données."
+
+    def add_arguments(self, parser):
+        parser.add_argument("communes_file", type=argparse.FileType("r"))
+        parser.add_argument("datafix_file", type=argparse.FileType("r"))
+        parser.add_argument("--shell", action=argparse.BooleanOptionalAction)
+
+    def handle(self, *args, **options):
+        if not re.match(
+            r"v_commune_\d+.csv", os.path.basename(options["communes_file"].name)
+        ):
+            raise CommandError(
+                f'L\'INSEE appelle le fichier de communes "v_commune_<year>.csv". Le fichier fourni ({options["communes_file"].name}) est a priori incorrect.'
+            )
+        if not re.match(r".*.ya?ml$", os.path.basename(options["datafix_file"].name)):
+            raise CommandError(
+                f'Fichier datafix "{options["datafix_file"].name}" incorrect. Doit être un fichier YAML.'
+            )
+
+        datafix = yaml.safe_load(options["datafix_file"])
+
+        with transaction.atomic():
+            with connection.cursor() as cursor:
+                cursor.execute("SET CONSTRAINTS ALL DEFERRED")
+
+                try:
+                    self.update_communes(options["communes_file"], datafix)
+                except Exception as exc:
+                    if options["shell"]:
+                        traceback.print_exc()
+                        breakpoint()
+                    raise
+
+    def update_communes(self, new_communes_file, datafix):
+        existing_communes = []
+
+        # List all communes in existing_communes
+        for commune in Commune.all_objects.all():
+            existing_communes.append(commune)
+
+        # Iterate over the file of new communes
+        for row in csv.DictReader(new_communes_file):
+            # Try to find the new commune in the existing communes. If one
+            # exists, we remove it from the list of existing communes, and place
+            # it in the list of similar communes.
+            for commune in existing_communes:
+                if (
+                    commune.insee == row["COM"]
+                    and commune.type_commune == row["TYPECOM"]
+                    and commune.departement == row["DEP"]
+                    and commune.libelle == row["LIBELLE"]
+                ):
+                    existing_communes.remove(commune)
+                    break
+            # This case is for a commune that has not been found in the existing
+            # communes. We need to insert it.
+            else:
+                new_commune = Commune(
+                    insee=row["COM"],
+                    departement=row["DEP"],
+                    libelle=row["LIBELLE"],
+                    type_commune=row["TYPECOM"],
+                )
+                new_commune.save()
+
+                administrator = ADSManagerAdministrator.objects.get(
+                    # Don't ask me why, but some entries in the file have a DEP
+                    # column empty. In this case, the departement number is
+                    # retrieved from the first 2 chars of the INSEE code.
+                    prefecture__numero=row["DEP"]
+                    or row["COM"][:2]
+                )
+
+                ADSManager.objects.create(
+                    administrator=administrator,
+                    content_type=ContentType.objects.get_for_model(Commune),
+                    object_id=new_commune.id,
+                )
+
+        # existing_communes now contains all the communes that are in the database, but are not in the new file.
+        # For each of these communes, we check if there is an entry in the
+        # datafix file to move the resources linked to another commune.
+        #
+        # If there is no entry and the commune has linked resources, we raise an
+        # error. Otherwise, we can safely remove the commune.
+        #
+        # The related resources of a commune are:
+        # - ADS
+        # - ADSManagerDecree
+        # - ADSManagerRequest
+        # - Vehicule
+        ok = True
+        for commune in existing_communes:
+            # Commune de test
+            if commune.insee == "999":
+                continue
+
+            fix = datafix["communes"].get(int(commune.id))
+            if fix:
+                if fix["insee"] != commune.insee or fix["libelle"] != commune.libelle:
+                    raise ValueError(
+                        f'Problème dans le fichier datafix: insee or libelle différents de la commune dans la db. Fix INSEE: {fix["insee"]}, Fix libelle: {fix["libelle"]}, Commune INSEE: {commune.insee}, Commune libelle: {commune.libelle}'
+                    )
+                try:
+                    q = Commune.all_objects.filter(
+                        insee=fix["new_owner"]["insee"],
+                        libelle=fix["new_owner"]["libelle"],
+                    )
+                    if fix["new_owner"].get("type"):
+                        q = q.filter(type_commune=fix["new_owner"]["type"])
+                    new_commune = q.get()
+                except Commune.MultipleObjectsReturned as exc:
+                    raise ValueError(
+                        f'Dans le fichier datafix, trop de communes trouvées pour new_owner.insee={fix["new_owner"]["insee"]} new_owner.libelle={fix["new_owner"]["libelle"]}. Ajouter le type de la commune (par exemple type: COM) pour réstreindre la recherche.'
+                    ) from exc
+                except Commune.DoesNotExist as exc:
+                    raise ValueError(
+                        f'Dans le fichier datafix, new_owner.insee ({fix["new_owner"]["insee"]}) et new_owner.libelle ({fix["new_owner"]["libelle"]}) ne correspondent à aucune commune dans la base de données. Vérifiez que les données sont correctes.'
+                    ) from exc
+
+                self.move_related_resources(commune, new_commune)
+
+            ads_manager = ADSManager.objects.get(
+                content_type=ContentType.objects.get_for_model(Commune),
+                object_id=commune.id,
+            )
+
+            err = False
+
+            ads_manager_decrees_count = ads_manager.adsmanagerdecree_set.count()
+            if ads_manager_decrees_count != 0:
+                print(
+                    f"Impossible de supprimer la commune id={commune.id} ads_manager={ads_manager.id} insee={commune.insee} libelle={commune.libelle} — {ads_manager_decrees_count} décret(s) lié(s)"
+                )
+                err = True
+
+            ads_count = ads_manager.ads_set.count()
+            if ads_count != 0:
+                print(
+                    f"Impossible de supprimer la commune id={commune.id} ads_manager={ads_manager.id} insee={commune.insee} libelle={commune.libelle} — {ads_count} ADS liée(s)"
+                )
+                err = True
+
+            ads_manager_requests_count = ads_manager.adsmanagerrequest_set.count()
+            if ads_manager_requests_count != 0:
+                print(
+                    f"Impossible de supprimer la commune id={commune.id} ads_manager={ads_manager.id} insee={commune.insee} libelle={commune.libelle} — {ads_manager_requests_count} demande(s) liée(s) pour devenir gestionnaire liées"
+                )
+                err = True
+
+            vehicules_relais_count = Vehicule.objects.filter(
+                commune_localisation=commune
+            ).count()
+            if vehicules_relais_count != 0:
+                print(
+                    f"Impossible de supprimer la commune id={commune.id} ads_manager={ads_manager.id} insee={commune.insee} libelle={commune.libelle} — {vehicules_relais_count} véhicule(s) relais lié(s)"
+                )
+                err = True
+
+            if err:
+                ok = False
+                print("--")
+            else:
+                try:
+                    commune.delete()
+                except Exception as e:
+                    print(
+                        f"Impossible de supprimer la commune id={commune.id} ads_manager={ads_manager.id} insee={commune.insee} libelle={commune.libelle} — {e}"
+                    )
+                    print("--")
+                    ok = False
+
+        if not ok:
+            raise ValueError("Mettre à jour le fichier datafix pour pouvoir continuer")
+
+    def move_related_resources(self, old_commune, new_commune):
+        old_ads_manager = ADSManager.objects.get(
+            content_type=ContentType.objects.get_for_model(Commune),
+            object_id=old_commune.id,
+        )
+        new_ads_manager = ADSManager.objects.get(
+            content_type=ContentType.objects.get_for_model(Commune),
+            object_id=new_commune.id,
+        )
+
+        for decree in old_ads_manager.adsmanagerdecree_set.all():
+            decree.ads_manager = new_ads_manager
+            decree.save()
+
+        for ads in old_ads_manager.ads_set.all():
+            ads.ads_manager = new_ads_manager
+            ads.save()
+
+        for ads in ADS.objects.filter(epci_commune=old_commune):
+            ads.epci_commune = new_commune
+            ads.save()
+
+        for ads_manager_request in old_ads_manager.adsmanagerrequest_set.all():
+            ads_manager_request.ads_manager = new_ads_manager
+            ads_manager_request.save()
+
+        for vehicule in Vehicule.objects.filter(commune_localisation=old_commune):
+            vehicule.commune_localisation = new_commune
+            vehicule.save()

mesads/fradm/migrations/0003_alter_commune_unique_together_commune_type_commune.py

@@ -0,0 +1,32 @@
+# Generated by Django 5.0.6 on 2024-07-05 09:41
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("fradm", "0002_create_extension_unaccent"),
+    ]
+
+    operations = [
+        migrations.AlterUniqueTogether(
+            name="commune",
+            unique_together=set(),
+        ),
+        migrations.AddField(
+            model_name="commune",
+            name="type_commune",
+            field=models.CharField(
+                choices=[
+                    ("COM", "Commune"),
+                    ("COMA", "Commune associée"),
+                    ("COMD", "Commune déléguée"),
+                    ("ARM", "Arrondissement municipal"),
+                ],
+                default="COM",
+                max_length=16,
+                verbose_name="Type de commune",
+            ),
+            preserve_default=False,
+        ),
+    ]

mesads/fradm/migrations/0004_alter_commune_insee_alter_commune_unique_together.py

@@ -0,0 +1,21 @@
+# Generated by Django 5.0.6 on 2024-07-05 09:45
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("fradm", "0003_alter_commune_unique_together_commune_type_commune"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="commune",
+            name="insee",
+            field=models.CharField(max_length=16),
+        ),
+        migrations.AlterUniqueTogether(
+            name="commune",
+            unique_together={("type_commune", "insee")},
+        ),
+    ]

mesads/fradm/migrations/0005_alter_commune_unique_together_and_more.py

@@ -0,0 +1,25 @@
+# Generated by Django 5.0.6 on 2024-07-05 11:29
+
+import django.db.models.constraints
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("fradm", "0004_alter_commune_insee_alter_commune_unique_together"),
+    ]
+
+    operations = [
+        migrations.AlterUniqueTogether(
+            name="commune",
+            unique_together=set(),
+        ),
+        migrations.AddConstraint(
+            model_name="commune",
+            constraint=models.UniqueConstraint(
+                deferrable=django.db.models.constraints.Deferrable["IMMEDIATE"],
+                fields=("type_commune", "insee"),
+                name="fradm_commune_type_commune_insee",
+            ),
+        ),
+    ]