|
| 1 | +from json import loads |
| 2 | +from pathlib import Path |
| 3 | + |
| 4 | +from django.core.management.base import BaseCommand |
| 5 | + |
| 6 | +from core.models import Collectivite, Commune, Departement, Region |
| 7 | + |
| 8 | +NUXT_DATA = ( |
| 9 | + Path(__file__).parent.parent.parent.parent.parent |
| 10 | + / "nuxt" |
| 11 | + / "server-middleware" |
| 12 | + / "Data" |
| 13 | +) |
| 14 | + |
| 15 | + |
| 16 | +class Command(BaseCommand): |
| 17 | + def handle(self, *args, **options) -> None: # noqa: ANN002, ANN003, ARG002 |
| 18 | + Commune.objects.all().delete() |
| 19 | + Collectivite.objects.all().delete() |
| 20 | + Departement.objects.all().delete() |
| 21 | + Region.objects.all().delete() |
| 22 | + self.stdout.write("All existing data deleted.") |
| 23 | + |
| 24 | + regions_json = loads((NUXT_DATA / "INSEE" / "regions.json").read_text()) |
| 25 | + regions = Region.objects.bulk_create( |
| 26 | + [ |
| 27 | + Region(code_insee=region["code"], nom=region["intituleSansArticle"]) |
| 28 | + for region in regions_json |
| 29 | + ] |
| 30 | + ) |
| 31 | + regions_by_code = {region.code_insee: region for region in regions} |
| 32 | + self.stdout.write(f"{len(regions)} regions loaded.") |
| 33 | + |
| 34 | + departements_json = loads( |
| 35 | + (NUXT_DATA / "INSEE" / "departements.json").read_text() |
| 36 | + ) |
| 37 | + departements = Departement.objects.bulk_create( |
| 38 | + [ |
| 39 | + Departement( |
| 40 | + code_insee=departement["code"], |
| 41 | + nom=departement["intituleSansArticle"], |
| 42 | + region=regions_by_code[departement["region"]["code"]], |
| 43 | + ) |
| 44 | + for departement in departements_json |
| 45 | + ] |
| 46 | + ) |
| 47 | + departements_by_code = { |
| 48 | + departement.code_insee: departement for departement in departements |
| 49 | + } |
| 50 | + self.stdout.write(f"{len(departements)} departements loaded.") |
| 51 | + |
| 52 | + groupements_json = loads( |
| 53 | + (NUXT_DATA / "referentiels" / "groupements_2024.json").read_text() |
| 54 | + ) |
| 55 | + groupements = Collectivite.objects.bulk_create( |
| 56 | + [ |
| 57 | + Collectivite( |
| 58 | + id=f"{groupement['code']}_{groupement['type']}", |
| 59 | + code_insee=groupement["code"], |
| 60 | + type=groupement["type"], |
| 61 | + nom=groupement["intitule"], |
| 62 | + departement=departements_by_code[groupement["departementCode"]], |
| 63 | + competence_plan=groupement["competencePLU"], |
| 64 | + competence_schema=groupement["competenceSCOT"], |
| 65 | + ) |
| 66 | + for groupement in groupements_json |
| 67 | + ] |
| 68 | + ) |
| 69 | + groupements_by_code = { |
| 70 | + groupement.code_insee: groupement for groupement in groupements |
| 71 | + } |
| 72 | + self.stdout.write(f"{len(groupements)} groupements loaded.") |
| 73 | + |
| 74 | + Collectivite.adhesions.through.objects.bulk_create( |
| 75 | + [ |
| 76 | + Collectivite.adhesions.through( |
| 77 | + from_collectivite=groupements_by_code[groupement["code"]], |
| 78 | + to_collectivite=groupements_by_code[membre_de["code"]], |
| 79 | + ) |
| 80 | + for groupement in groupements_json |
| 81 | + for membre_de in groupement["groupements"] |
| 82 | + ], |
| 83 | + ignore_conflicts=True, |
| 84 | + ) |
| 85 | + self.stdout.write("Groupement relationships loaded.") |
| 86 | + |
| 87 | + communes_json = loads( |
| 88 | + (NUXT_DATA / "referentiels" / "communes.json").read_text() |
| 89 | + ) |
| 90 | + communes_by_code = {} |
| 91 | + for commune in communes_json: |
| 92 | + if commune["type"] == "COM": |
| 93 | + commune_instance = Commune.objects.create( |
| 94 | + id=f"{commune['code']}_{commune['type']}", |
| 95 | + code_insee=commune["code"], |
| 96 | + type=commune["type"], |
| 97 | + nom=commune["intitule"], |
| 98 | + departement=departements_by_code[commune["departementCode"]], |
| 99 | + competence_plan=commune["competencePLU"], |
| 100 | + competence_schema=commune["competenceSCOT"], |
| 101 | + intercommunalite=groupements_by_code.get( |
| 102 | + commune["intercommunaliteCode"] |
| 103 | + ), |
| 104 | + ) |
| 105 | + communes_by_code[commune_instance.code_insee] = commune_instance |
| 106 | + self.stdout.write(f"{len(communes_by_code)} communes of type 'COM' loaded.") |
| 107 | + |
| 108 | + Collectivite.adhesions.through.objects.bulk_create( |
| 109 | + [ |
| 110 | + Collectivite.adhesions.through( |
| 111 | + from_collectivite=communes_by_code[commune["code"]], |
| 112 | + to_collectivite=groupements_by_code[membre_de["code"]], |
| 113 | + ) |
| 114 | + for commune in communes_json |
| 115 | + for membre_de in commune["groupements"] |
| 116 | + ] |
| 117 | + ) |
| 118 | + self.stdout.write("Commune relationships loaded.") |
| 119 | + |
| 120 | + for commune in communes_json: |
| 121 | + if commune["type"] != "COM": |
| 122 | + commune_parente = communes_by_code[commune["codeParent"]] |
| 123 | + Commune.objects.create( |
| 124 | + id=f"{commune['code']}_{commune['type']}", |
| 125 | + code_insee=commune["code"] |
| 126 | + if commune_parente.code_insee != commune["code"] |
| 127 | + else None, |
| 128 | + type=commune["type"], |
| 129 | + nom=commune["intitule"], |
| 130 | + nouvelle=commune_parente, |
| 131 | + departement=commune_parente.departement, |
| 132 | + competence_plan=commune["competencePLU"], |
| 133 | + competence_schema=commune["competenceSCOT"], |
| 134 | + ) |
| 135 | + self.stdout.write("Other types of communes loaded.") |
0 commit comments