Export des résultats en JSON

.gitignore

@@ -6,6 +6,7 @@ __pycache__
 *.sqlite3
 *.sqlite3-journal
 
+cache/
 data/
 
 migrations/versions/*

main.py

@@ -6,7 +6,7 @@ import os
 from dotenv import load_dotenv
 from sqlalchemy import create_engine
 
-from nupes.scripts import import_candidats_2024, import_geographie, import_resultats_web_2024
+from nupes.scripts import import_candidats_2024, import_geographie, import_resultats_web_2024, export_resultats_2024
 
 
 def parse_args():
@@ -16,7 +16,7 @@ def parse_args():
                                  "régionales", "départementales", "municipales"], default="européennes")
     parser.add_argument('--year', '-y', type=int, help="Année de l'élection", default=2024)
     parser.add_argument('action', help="Action à réaliser",
-                        choices=["import_candidats", "import_geographie", "import_resultats"])
+                        choices=["import_candidats", "import_geographie", "import_resultats", "export_resultats"])
     parser.add_argument('--debug', '-d', action='store_true', help="Mode debug")
     return parser.parse_args()
 
@@ -38,6 +38,9 @@ def main():
         case "import_resultats":
             print(f"Import des résultats pour les élections {args.type} {args.year}")
             import_resultats_web_2024.run(engine)
+        case "export_resultats":
+            print(f"Export des résultats pour les élections {args.type} {args.year}")
+            export_resultats_2024.run(engine)
         case _:
             print(f"Action {args.action} non reconnue")
 

nupes/cache.py

@@ -3,14 +3,14 @@ from pathlib import Path
 import requests
 
 
-DATA_DIR = Path(__file__).parent / 'data'
+CACHE_DIR = Path(__file__).parent / 'cache'
 
 
 def get_file(url, filename, force_etag: str = ""):
-    if not DATA_DIR.is_dir():
-        DATA_DIR.mkdir(parents=True)
+    if not CACHE_DIR.is_dir():
+        CACHE_DIR.mkdir(parents=True)
 
-    file_dir = DATA_DIR / filename
+    file_dir = CACHE_DIR / filename
     if not file_dir.is_dir():
         file_dir.mkdir(parents=True)
 

nupes/scripts/export_resultats_2024.py

@@ -0,0 +1,255 @@
+import json
+from pathlib import Path
+
+from sqlalchemy import Engine, select
+from sqlalchemy.orm import Session
+
+from nupes.models.geographie import Region, Departement, Commune
+from nupes.models.europeennes2024 import Bloc, Nuance, Liste, \
+    ResultatsFrance, ResultatsRegion, ResultatsDepartement, ResultatsCommune
+
+DATA_DIR = Path(__file__).parent.parent / 'data'
+
+
+def exporter_listes(engine: Engine) -> None:
+    with Session(engine) as session:
+        blocs = session.execute(select(Bloc)).scalars().all()
+        blocs_json = []
+
+        for bloc in blocs:
+            bloc_json = {'id': bloc.id, 'nom': bloc.nom, 'couleur': bloc.couleur}
+            blocs_json.append(bloc_json)
+
+        file = DATA_DIR / "resultats" / "europeennes2024" / "blocs.json"
+        if not file.parent.is_dir():
+            file.parent.mkdir(parents=True)
+
+        with file.open('w') as f:
+            json.dump(blocs_json, f)
+
+        nuances = session.execute(select(Nuance)).scalars().all()
+        nuances_json = []
+
+        for nuance in nuances:
+            nuance_json = {'code': nuance.code, 'nom': nuance.nom, 'couleur': nuance.couleur}
+            nuances_json.append(nuance_json)
+
+        file = DATA_DIR / "resultats" / "europeennes2024" / "nuances.json"
+        if not file.parent.is_dir():
+            file.parent.mkdir(parents=True)
+
+        with file.open('w') as f:
+            json.dump(nuances_json, f)
+
+        listes = session.execute(select(Liste)).scalars().all()
+        listes_json = []
+
+        for liste in listes:
+            candidats = [
+                {'ordre': candidat.ordre, 'nom': candidat.nom, 'prenom': candidat.prenom, 'sexe': candidat.sexe.value,
+                 'date_naissance': candidat.date_naissance.isoformat(), 'profession': candidat.profession,
+                 'code_personnalite': candidat.code_personnalite.value, 'sortant': candidat.sortant}
+                for candidat in liste.candidats
+            ]
+            liste_json = {'numero': liste.numero, 'nom': liste.nom, 'nuance': liste.nuance_id,
+                          'bloc': liste.bloc.nom, 'candidats': candidats}
+            listes_json.append(liste_json)
+
+        file = DATA_DIR / "resultats" / "europeennes2024" / "listes.json"
+        if not file.parent.is_dir():
+            file.parent.mkdir(parents=True)
+
+        with file.open('w') as f:
+            json.dump(listes_json, f)
+
+
+def exporter_resultats_france(engine: Engine) -> None:
+    with Session(engine) as session:
+        resultats_france = session.execute(select(ResultatsFrance)).scalar_one_or_none()
+        if not resultats_france:
+            resultats_france = ResultatsFrance()
+            session.add(resultats_france)
+
+        resultats_dict = {
+            "inscrits": resultats_france.inscrits,
+            "votants": resultats_france.votants,
+            "exprimes": resultats_france.exprimes,
+            "blancs": resultats_france.blancs,
+            "nuls": resultats_france.nuls,
+            'geometry': {},
+        }
+
+        resultats_listes = {}
+        resultats_dict['voix_listes'] = resultats_listes
+        for voix_liste in resultats_france.voix_listes:
+            resultats_listes[voix_liste.liste.numero] = voix_liste.voix
+
+        file = DATA_DIR / "resultats" / "europeennes2024" / "france.json"
+        if not file.parent.is_dir():
+            file.parent.mkdir(parents=True)
+
+        with file.open('w') as f:
+            json.dump(resultats_dict, f)
+
+        session.commit()
+
+
+def exporter_resultats_regions(engine: Engine) -> None:
+    with Session(engine) as session:
+        regions = session.execute(select(Region)).scalars().all()
+        regions_json = []
+
+        for region in regions:
+            region_json = {'code_insee': region.code_insee, 'nom': region.libelle,
+                           'departements': [dpt.code_insee for dpt in region.departements]}
+            regions_json.append(region_json)
+
+            resultats_region = session.execute(select(ResultatsRegion).filter_by(region_id=region.code_insee)) \
+                .scalar_one_or_none()
+            if not resultats_region:
+                resultats_france = session.execute(select(ResultatsFrance)).scalar_one()
+                resultats_region = ResultatsRegion(region_id=region.code_insee, resultats_france_id=resultats_france.id)
+                session.add(resultats_region)
+
+            resultats_dict = {
+                "inscrits": resultats_region.inscrits,
+                "votants": resultats_region.votants,
+                "exprimes": resultats_region.exprimes,
+                "blancs": resultats_region.blancs,
+                "nuls": resultats_region.nuls,
+                'region': region_json,
+                'geometry': region.geometry,
+            }
+
+            resultats_listes = {}
+            resultats_dict['voix_listes'] = resultats_listes
+            for voix_liste in resultats_region.voix_listes:
+                resultats_listes[voix_liste.liste.numero] = voix_liste.voix
+
+            file = DATA_DIR / "resultats" / "europeennes2024" / "regions" / f"{region.code_insee}.json"
+            if not file.parent.is_dir():
+                file.parent.mkdir(parents=True)
+
+            with file.open('w') as f:
+                json.dump(resultats_dict, f)
+
+        session.commit()
+
+    regions_file = DATA_DIR / "resultats" / "europeennes2024" / "regions" / "regions.json"
+    if not regions_file.parent.is_dir():
+        regions_file.parent.mkdir(parents=True)
+
+    with regions_file.open('w') as f:
+        json.dump(regions_json, f)
+
+
+def exporter_resultats_departements(engine: Engine) -> None:
+    with Session(engine) as session:
+        departements = session.execute(select(Departement)).scalars().all()
+        departements_json = []
+
+        for departement in departements:
+            departement_json = {'code_insee': departement.code_insee, 'nom': departement.libelle,
+                                'region': departement.region_code,
+                                'communes': [commune.code_insee for commune in departement.communes]}
+            departements_json.append(departement_json)
+
+            resultats_departement = session.execute(
+                select(ResultatsDepartement).filter_by(dpt_id=departement.code_insee)).scalar_one_or_none()
+            if not resultats_departement:
+                resultats_region = session.execute(select(ResultatsRegion)
+                                                   .filter_by(region_id=departement.region_code)).scalar_one()
+                resultats_departement = ResultatsDepartement(dpt_id=departement.code_insee,
+                                                             resultats_region_id=resultats_region.id)
+                session.add(resultats_departement)
+
+            resultats_dict = {
+                "inscrits": resultats_departement.inscrits,
+                "votants": resultats_departement.votants,
+                "exprimes": resultats_departement.exprimes,
+                "blancs": resultats_departement.blancs,
+                "nuls": resultats_departement.nuls,
+                'departement': departement_json,
+                'geometry': departement.geometry,
+            }
+
+            resultats_listes = {}
+            resultats_dict['voix_listes'] = resultats_listes
+            for voix_liste in resultats_departement.voix_listes:
+                resultats_listes[voix_liste.liste.numero] = voix_liste.voix
+
+            file = DATA_DIR / "resultats" / "europeennes2024" / "departements" / f"{departement.code_insee}.json"
+            if not file.parent.is_dir():
+                file.parent.mkdir(parents=True)
+
+            with file.open('w') as f:
+                json.dump(resultats_dict, f)
+
+        session.commit()
+
+    departements_file = DATA_DIR / "resultats" / "europeennes2024" / "departements" / "departements.json"
+    if not departements_file.parent.is_dir():
+        departements_file.parent.mkdir(parents=True)
+
+    with departements_file.open('w') as f:
+        json.dump(departements_json, f)
+
+
+def exporter_resultats_communes(engine: Engine) -> None:
+    with Session(engine) as session:
+        communes = session.execute(select(Commune)).scalars().all()
+        communes_json = []
+
+        for commune in communes:
+            commune_json = {'code_insee': commune.code_insee, 'nom': commune.libelle,
+                            'departement': commune.departement_code,
+                            'bureaux_vote': [bv.id for bv in commune.bureaux_vote]}
+            communes_json.append(commune_json)
+
+            resultats_commune = session.execute(
+                select(ResultatsCommune).filter_by(commune_id=commune.code_insee)).scalar_one_or_none()
+            if not resultats_commune:
+                resultats_departement = session.execute(select(ResultatsDepartement)
+                                                        .filter_by(dpt_id=commune.departement_code)).scalar_one()
+                resultats_commune = ResultatsCommune(commune_id=commune.code_insee,
+                                                     resultats_dpt_id=resultats_departement.id)
+                session.add(resultats_commune)
+
+            resultats_dict = {
+                "inscrits": resultats_commune.inscrits,
+                "votants": resultats_commune.votants,
+                "exprimes": resultats_commune.exprimes,
+                "blancs": resultats_commune.blancs,
+                "nuls": resultats_commune.nuls,
+                'commune': commune_json,
+                'geometry': commune.geometry,
+            }
+
+            resultats_listes = {}
+            resultats_dict['voix_listes'] = resultats_listes
+            for voix_liste in resultats_commune.voix_listes:
+                resultats_listes[voix_liste.liste.numero] = voix_liste.voix
+
+            file = DATA_DIR / "resultats" / "europeennes2024" / "communes" / f"{commune.code_insee}.json"
+            if not file.parent.is_dir():
+                file.parent.mkdir(parents=True)
+
+            with file.open('w') as f:
+                json.dump(resultats_dict, f)
+
+        session.commit()
+
+    communes_file = DATA_DIR / "resultats" / "europeennes2024" / "communes" / "communes.json"
+    if not communes_file.parent.is_dir():
+        communes_file.parent.mkdir(parents=True)
+
+    with communes_file.open('w') as f:
+        json.dump(communes_json, f)
+
+
+def run(engine: Engine) -> None:
+    exporter_listes(engine)
+    exporter_resultats_france(engine)
+    exporter_resultats_regions(engine)
+    exporter_resultats_departements(engine)
+    exporter_resultats_communes(engine)

nupes/scripts/import_candidats_2024.py

@@ -4,7 +4,7 @@ import datetime
 from sqlalchemy import Engine, select
 from sqlalchemy.orm import Session
 
-from nupes.data import get_file
+from nupes.cache import get_file
 from nupes.models.europeennes2024 import Bloc, Nuance, Liste, Candidat
 
 

nupes/scripts/import_geographie.py

@@ -5,7 +5,7 @@ import requests
 from sqlalchemy import Engine, select
 from sqlalchemy.orm import Session
 
-from nupes.data import get_file
+from nupes.cache import get_file
 from nupes.models.geographie import BureauVote, Circonscription, Commune, Departement, Region
 
 

nupes/scripts/import_resultats_web_2024.py

@@ -5,7 +5,7 @@ from sqlalchemy import Engine, select
 from sqlalchemy.orm import Session
 from tqdm import tqdm
 
-from nupes.data import get_file
+from nupes.cache import get_file
 from nupes.models.europeennes2024 import ResultatsFrance, ResultatsRegion, ResultatsDepartement, ResultatsCommune, \
     VoixListeFrance, VoixListeRegion, VoixListeDepartement, VoixListeCommune, Liste
 from nupes.models.geographie import Region, Departement, Commune
@@ -187,7 +187,9 @@ def importer_resultats_communes(engine: Engine) -> None:
                 # Pour ces collectivités d'outre-mer, qui ne sont pas dans des régions,
                 # on ne les regroupe pas dans des régions
                 reg_path = ""
-            if com_code in ["16355", "25282", "55050", "55189", "69152"]:
+            if com_code in ["08294", "16355", "18131", "25282", "25549", "35112", "49321", "55039", "55050", "55139",
+                            "55189", "55239", "55307", "64541", "69152", "75056", "85041", "85271", "86231", "95282",
+                            "98611", "98612", "98613"]:
                 # Aucun⋅e habitant⋅e donc pas de bureau de vote
                 # Ou alors la commune a récemment fusionné
                 continue
@@ -293,5 +295,3 @@ def run(engine: Engine) -> None:
     importer_resultats_regions(engine)
     importer_resultats_departements(engine)
     importer_resultats_communes(engine)
-
-    analyser_resultats("")