On n'importe pas tous les bureaux de vote

main.py

@@ -6,7 +6,8 @@ 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, export_resultats_2024
+from nupes.scripts import import_candidats_2024, import_geographie, import_resultats_2024, \
+    import_resultats_web_2024, export_resultats_2024
 
 
 def parse_args():
@@ -16,8 +17,11 @@ 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", "export_resultats"])
+                        choices=["import_candidats", "import_geographie", "import_resultats",
-    parser.add_argument('--debug', '-d', action='store_true', help="Mode debug")
+                                 "import_resultats_web", "export_resultats"])
+    parser.add_argument('--debug', '-d', action='store_true',
+                        help="Mode debug (affiche toutes les requêtes SQL)")
+    parser.add_argument('--verbose', '-v', action='store_true', help="Mode verbeux")
     return parser.parse_args()
 
 
@@ -26,26 +30,31 @@ def main():
 
     args = parse_args()
     debug = args.debug
+    verbose = args.verbose
 
     engine = create_engine(os.getenv("SQLALCHEMY_URL"), echo=args.debug)
 
     match args.action:
         case "import_geographie":
-            if debug:
+            if verbose:
                 print(f"Import des données géographiques pour les élections {args.type} {args.year}")
-            import_geographie.run(engine, debug)
+            import_geographie.run(engine, verbose)
         case "import_candidats":
-            if debug:
+            if verbose:
                 print(f"Import des candidats pour les élections {args.type} {args.year}")
-            import_candidats_2024.run(engine, debug)
+            import_candidats_2024.run(engine, verbose)
         case "import_resultats":
-            if debug:
+            if verbose:
                 print(f"Import des résultats pour les élections {args.type} {args.year}")
-            import_resultats_web_2024.run(engine, debug)
+            import_resultats_2024.run(engine, verbose)
+        case "import_resultats_web":
+            if verbose:
+                print(f"Import des résultats depuis le web pour les élections {args.type} {args.year}")
+            import_resultats_web_2024.run(engine, verbose)
         case "export_resultats":
-            if debug:
+            if verbose:
                 print(f"Export des résultats pour les élections {args.type} {args.year}")
-            export_resultats_2024.run(engine, debug)
+            export_resultats_2024.run(engine, verbose)
         case _:
             print(f"Action {args.action} non reconnue")
 

nupes/scripts/export_resultats_2024.py

@@ -12,7 +12,7 @@ from nupes.models.europeennes2024 import Bloc, Nuance, Liste, \
 DATA_DIR = Path(__file__).parent.parent.parent / 'data'
 
 
-def exporter_listes(engine: Engine, debug: bool = False) -> None:
+def exporter_listes(engine: Engine, verbose: bool = False) -> None:
     with Session(engine) as session:
         blocs = session.execute(select(Bloc)).scalars().all()
         blocs_json = []
@@ -64,7 +64,7 @@ def exporter_listes(engine: Engine, debug: bool = False) -> None:
             json.dump(listes_json, f)
 
 
-def exporter_resultats_france(engine: Engine, debug: bool = False) -> None:
+def exporter_resultats_france(engine: Engine, verbose: bool = False) -> None:
     with Session(engine) as session:
         resultats_france = session.execute(select(ResultatsFrance)).scalar_one_or_none()
         if not resultats_france:
@@ -96,12 +96,12 @@ def exporter_resultats_france(engine: Engine, debug: bool = False) -> None:
         session.commit()
 
 
-def exporter_resultats_regions(engine: Engine, debug: bool = False) -> None:
+def exporter_resultats_regions(engine: Engine, verbose: bool = False) -> None:
     with Session(engine) as session:
         regions = session.execute(select(Region)).scalars().all()
         regions_json = []
 
-        regions_iterator = tqdm(regions, desc="Régions") if debug else regions
+        regions_iterator = tqdm(regions, desc="Régions") if verbose else regions
         for region in regions_iterator:
             region_json = {'code_insee': region.code_insee, 'nom': region.libelle,
                            'departements': [dpt.code_insee for dpt in region.departements]}
@@ -147,12 +147,12 @@ def exporter_resultats_regions(engine: Engine, debug: bool = False) -> None:
         json.dump(regions_json, f)
 
 
-def exporter_resultats_departements(engine: Engine, debug: bool = False) -> None:
+def exporter_resultats_departements(engine: Engine, verbose: bool = False) -> None:
     with Session(engine) as session:
         departements = session.execute(select(Departement)).scalars().all()
         departements_json = []
 
-        iterator = tqdm(departements, desc="Départements") if debug else departements
+        iterator = tqdm(departements, desc="Départements") if verbose else departements
         for departement in iterator:
             departement_json = {'code_insee': departement.code_insee, 'nom': departement.libelle,
                                 'region': departement.region_code,
@@ -201,12 +201,12 @@ def exporter_resultats_departements(engine: Engine, debug: bool = False) -> None
         json.dump(departements_json, f)
 
 
-def exporter_resultats_communes(engine: Engine, debug: bool = False) -> None:
+def exporter_resultats_communes(engine: Engine, verbose: bool = False) -> None:
     with Session(engine) as session:
         communes = session.execute(select(Commune)).scalars().all()
         communes_json = []
 
-        iterator = tqdm(communes, desc="Communes") if debug else communes
+        iterator = tqdm(communes, desc="Communes") if verbose else communes
         for commune in iterator:
             commune_json = {'code_insee': commune.code_insee, 'nom': commune.libelle,
                             'departement': commune.departement_code,
@@ -255,10 +255,10 @@ def exporter_resultats_communes(engine: Engine, debug: bool = False) -> None:
         json.dump(communes_json, f)
 
 
-def run(engine: Engine, debug: bool = False) -> None:
+def run(engine: Engine, verbose: bool = False) -> None:
-    exporter_listes(engine, debug)
+    exporter_listes(engine, verbose)
-    exporter_resultats_france(engine, debug)
+    exporter_resultats_france(engine, verbose)
-    exporter_resultats_regions(engine, debug)
+    exporter_resultats_regions(engine, verbose)
-    exporter_resultats_departements(engine, debug)
+    exporter_resultats_departements(engine, verbose)
     # FIXME Les communes prennent trop de temps
-    # exporter_resultats_communes(engine, debug)
+    # exporter_resultats_communes(engine, verbose)

nupes/scripts/import_candidats_2024.py

@@ -8,7 +8,7 @@ from nupes.cache import get_file
 from nupes.models.europeennes2024 import Bloc, Nuance, Liste, Candidat
 
 
-def creer_blocs(engine: Engine, debug: bool = False) -> None:
+def creer_blocs(engine: Engine, verbose: bool = False) -> None:
     blocs = [
         {"id": 1, "nom": "Gauche", "couleur": "#BB1840"},
         {"id": 2, "nom": "Droite libérale", "couleur": "#FFEB00"},
@@ -28,7 +28,7 @@ def creer_blocs(engine: Engine, debug: bool = False) -> None:
         session.commit()
 
 
-def creer_nuances(engine: Engine, debug: bool = False) -> None:
+def creer_nuances(engine: Engine, verbose: bool = False) -> None:
     nuances = [
         {"code": "LEXG", "nom": "Liste d'extrême gauche", "couleur": "#BB0000"},
         {"code": "LCOM", "nom": "Liste du Parti communiste français", "couleur": "#DD0000"},
@@ -57,7 +57,7 @@ def creer_nuances(engine: Engine, debug: bool = False) -> None:
         session.commit()
 
 
-def importer_listes(engine: Engine, debug: bool = False) -> None:
+def importer_listes(engine: Engine, verbose: bool = False) -> None:
     listes = [
         {"numero": 1, "nom": "Pour une humanité souveraine", "nuance_id": "LDIV", "bloc_id": 5},
         {"numero": 2, "nom": "Pour une démocratie réelle : Décidons nous-mêmes !",
@@ -126,7 +126,7 @@ def importer_listes(engine: Engine, debug: bool = False) -> None:
         session.commit()
 
 
-def importer_candidats(engine: Engine, debug: bool = False) -> None:
+def importer_candidats(engine: Engine, verbose: bool = False) -> None:
     DATASET_URL = "https://www.data.gouv.fr/fr/datasets/r/483cd4bd-1b0e-4b52-a923-aadadf9c8f62"
     file = get_file(DATASET_URL, "candidats_2024.csv")
 
@@ -166,8 +166,8 @@ def importer_candidats(engine: Engine, debug: bool = False) -> None:
             session.commit()
 
 
-def run(engine: Engine, debug: bool = False) -> None:
+def run(engine: Engine, verbose: bool = False) -> None:
-    creer_blocs(engine, debug)
+    creer_blocs(engine, verbose)
-    creer_nuances(engine, debug)
+    creer_nuances(engine, verbose)
-    importer_listes(engine, debug)
+    importer_listes(engine, verbose)
-    importer_candidats(engine, debug)
+    importer_candidats(engine, verbose)

nupes/scripts/import_geographie.py

@@ -9,7 +9,7 @@ from nupes.cache import get_file
 from nupes.models.geographie import BureauVote, Circonscription, Commune, Departement, Region
 
 
-def importer_regions(engine: Engine, debug: bool = False) -> None:
+def importer_regions(engine: Engine, verbose: bool = False) -> None:
     etag = requests.get(
         "https://public.opendatasoft.com/api/explore/v2.1/catalog/datasets"
         "/georef-france-region?select=data_processed").json()['data_processed']
@@ -36,7 +36,7 @@ def importer_regions(engine: Engine, debug: bool = False) -> None:
         session.commit()
 
 
-def importer_departements(engine: Engine, debug: bool = False) -> None:
+def importer_departements(engine: Engine, verbose: bool = False) -> None:
     etag = requests.get(
         "https://public.opendatasoft.com/api/explore/v2.1/catalog/datasets"
         "/georef-france-departement?select=data_processed").json()['data_processed']
@@ -66,7 +66,7 @@ def importer_departements(engine: Engine, debug: bool = False) -> None:
         session.commit()
 
 
-def importer_communes(engine: Engine, debug: bool = False) -> None:
+def importer_communes(engine: Engine, verbose: bool = False) -> None:
     etag = requests.get(
         "https://public.opendatasoft.com/api/explore/v2.1/catalog/datasets"
         "/georef-france-commune-arrondissement-municipal?select=data_processed").json()['data_processed']
@@ -95,7 +95,7 @@ def importer_communes(engine: Engine, debug: bool = False) -> None:
         session.commit()
 
 
-def importer_bureaux_vote(engine: Engine, debug: bool = False) -> None:
+def importer_bureaux_vote(engine: Engine, verbose: bool = False) -> None:
     etag = requests.get(
         "https://public.opendatasoft.com/api/explore/v2.1/catalog/datasets"
         "/elections-france-bureau-vote-2022?select=data_processed").json()['data_processed']
@@ -128,6 +128,10 @@ def importer_bureaux_vote(engine: Engine, debug: bool = False) -> None:
                     session.add(Commune(code_insee=code_commune, libelle=bv_dict['com_name'], departement_code=dpt_code,
                                         geometry={}))
 
+            if not session.execute(select(Commune).filter_by(code_insee=code_commune)).scalar_one_or_none():
+                print("Commune non trouvée avec le code", code_commune, "et le nom", bv_dict['com_name'])
+                continue
+
             if not session.execute(select(Circonscription).filter_by(id=code_circo)).scalar_one_or_none():
                 session.add(Circonscription(id=code_circo, departement_code=dpt_code, numero=numero_circo))
 
@@ -146,7 +150,7 @@ def importer_bureaux_vote(engine: Engine, debug: bool = False) -> None:
         session.commit()
 
 
-def importer_contours_bureaux_vote(engine: Engine, debug: bool = False) -> None:
+def importer_contours_bureaux_vote(engine: Engine, verbose: bool = False) -> None:
     file = get_file("https://www.data.gouv.fr/fr/datasets/r/f98165a7-7c37-4705-a181-bcfc943edc73",
                     "contours-bureaux-vote.geojson")
 
@@ -177,9 +181,9 @@ def importer_contours_bureaux_vote(engine: Engine, debug: bool = False) -> None:
         session.commit()
 
 
-def run(engine: Engine, debug: bool = False) -> None:
+def run(engine: Engine, verbose: bool = False) -> None:
-    importer_regions(engine, debug)
+    importer_regions(engine, verbose)
-    importer_departements(engine, debug)
+    importer_departements(engine, verbose)
-    importer_communes(engine, debug)
+    importer_communes(engine, verbose)
-    importer_bureaux_vote(engine, debug)
+    importer_bureaux_vote(engine, verbose)
-    importer_contours_bureaux_vote(engine, debug)
+    importer_contours_bureaux_vote(engine, verbose)

nupes/scripts/import_resultats_2024.py

@@ -0,0 +1,119 @@
+from csv import DictReader
+
+from sqlalchemy import Engine, select
+from sqlalchemy.orm import Session
+from tqdm import tqdm
+
+from nupes.cache import get_file
+from nupes.models import BureauVote, Departement
+from nupes.models.europeennes2024 import ResultatsBureauVote, Liste, VoixListeBureauVote, ResultatsCommune, \
+    ResultatsDepartement, ResultatsRegion, ResultatsFrance
+
+
+def importer_resultats_bv(engine: Engine, verbose: bool = False) -> None:
+    file = get_file("https://www.data.gouv.fr/fr/datasets/r/937bb638-a487-40cd-9a0b-610d539a4207",
+                    "resultats-temporaires-par-bureau-de-vote.csv")
+
+    with file.open('r') as f:
+        next(f)  # On saute la première ligne
+        reader = DictReader(f)
+        with Session(engine) as session:
+            for row in tqdm(reader, desc="Bureau de vote", disable=not verbose):
+                com_code = str(row['Code commune']).zfill(5)
+                bv_code = row['Code BV']
+                bv_id = f"{com_code}_{bv_code}"
+
+                if not session.execute(select(BureauVote).filter_by(id=bv_id)).scalar_one_or_none():
+                    print(f"Bureau de vote {bv_id} non trouvé")
+                    continue
+
+                resultats_bv = session.execute(select(ResultatsBureauVote).filter_by(bv_id=bv_id)).scalar_one_or_none()
+                if not resultats_bv:
+                    resultats_commune = session.execute(select(ResultatsCommune).filter_by(commune_id=com_code)) \
+                        .scalar_one_or_none()
+                    if not resultats_commune:
+                        resultats_dpt = session.execute(select(ResultatsDepartement)
+                                                        .filter_by(dpt_id=row['Code département'])) \
+                            .scalar_one_or_none()
+                        if not resultats_dpt:
+                            dpt = session.execute(select(Departement)
+                                                  .filter_by(code_insee=f"{row['Code département'].zfill(2)}")) \
+                                .scalar_one()
+                            resultats_reg = session.execute(select(ResultatsRegion)
+                                                            .filter_by(region_id=dpt.region_code)) \
+                                .scalar_one_or_none()
+                            if not resultats_reg:
+                                resultats_france = session.execute(select(ResultatsFrance)).scalar_one_or_none()
+                                if not resultats_france:
+                                    session.add(ResultatsFrance())
+                                    resultats_france = session.execute(select(ResultatsFrance)).scalar_one()
+
+                                resultats_reg = ResultatsRegion(region_id=str(dpt.region_code),
+                                                                resultats_france_id=resultats_france.id)
+                                session.add(resultats_reg)
+                                resultats_reg = session.execute(select(ResultatsRegion)
+                                                                .filter_by(region_id=dpt.region_code)).scalar_one()
+
+                            resultats_dpt = ResultatsDepartement(dpt_id=row['Code département'],
+                                                                 resultats_region_id=resultats_reg.id)
+                            session.add(resultats_dpt)
+                            resultats_dpt = session.execute(select(ResultatsDepartement)
+                                                            .filter_by(dpt_id=row['Code département'])).scalar_one()
+
+                        resultats_commune = ResultatsCommune(commune_id=com_code, resultats_dpt_id=resultats_dpt.id)
+                        session.add(resultats_commune)
+                        resultats_commune = session.execute(select(ResultatsCommune).filter_by(commune_id=com_code)) \
+                            .scalar_one()
+
+                    resultats_bv = ResultatsBureauVote(bv_id=bv_id, resultats_commune_id=resultats_commune.id)
+                    session.add(resultats_bv)
+                    resultats_bv = session.execute(select(ResultatsBureauVote).filter_by(bv_id=bv_id)).scalar_one()
+
+                resultats_bv.inscrits = int(row['Inscrits'])
+                resultats_bv.votants = int(row['Votants'])
+                resultats_bv.abstentions = int(row['Abstentions'])
+                resultats_bv.exprimes = int(row['Exprimés'])
+                resultats_bv.blancs = int(row['Blancs'])
+                resultats_bv.nuls = int(row['Nuls'])
+
+                for liste in session.execute(select(Liste)).scalars().all():
+                    voix_liste_bv = session.execute(select(VoixListeBureauVote)
+                                                    .filter_by(resultats_bureau_vote_id=resultats_bv.id,
+                                                               liste_id=liste.id)) \
+                        .scalar_one_or_none()
+                    if not voix_liste_bv:
+                        voix_liste_bv = VoixListeBureauVote(resultats_bureau_vote_id=resultats_bv.id, liste_id=liste.id)
+                        session.add(voix_liste_bv)
+
+                    voix_liste_bv.voix = int(row[f"Voix {liste.id}"])
+
+            session.commit()
+
+
+def calculer_resultats_commune(engine: Engine, verbose: bool = False) -> None:
+    pass
+
+
+def calculer_resultats_circo(engine: Engine, verbose: bool = False) -> None:
+    pass
+
+
+def calculer_resultats_departement(engine: Engine, verbose: bool = False) -> None:
+    pass
+
+
+def calculer_resultats_region(engine: Engine, verbose: bool = False) -> None:
+    pass
+
+
+def calculer_resultats_national(engine: Engine, verbose: bool = False) -> None:
+    pass
+
+
+def run(engine: Engine, verbose: bool = False) -> None:
+    importer_resultats_bv(engine, verbose)
+    calculer_resultats_commune(engine, verbose)
+    # calculer_resultats_circo(engine, verbose)
+    calculer_resultats_departement(engine, verbose)
+    calculer_resultats_region(engine, verbose)
+    calculer_resultats_national(engine, verbose)

nupes/scripts/import_resultats_web_2024.py

@@ -13,7 +13,7 @@ from nupes.models.geographie import Region, Departement, Commune
 BASE_URL = "https://www.resultats-elections.interieur.gouv.fr/europeennes2024/ensemble_geographique"
 
 
-def importer_resultats_france(engine: Engine, debug: bool = False) -> None:
+def importer_resultats_france(engine: Engine, verbose: bool = False) -> None:
     file = get_file(f"{BASE_URL}/index.html", "resultats2024/resultats.html")
 
     with file.open() as f:
@@ -54,10 +54,10 @@ def importer_resultats_france(engine: Engine, debug: bool = False) -> None:
 
         session.commit()
 
-def importer_resultats_regions(engine: Engine, debug: bool = False) -> None:
+def importer_resultats_regions(engine: Engine, verbose: bool = False) -> None:
     with Session(engine) as session:
         regions = session.execute(select(Region)).scalars().all()
-        regions_iterator = tqdm(regions, desc="Régions") if debug else regions
+        regions_iterator = tqdm(regions, desc="Régions") if verbose else regions
         for region in regions_iterator:
             reg_code = region.code_insee
             if reg_code == "984" or reg_code == "989":
@@ -113,10 +113,10 @@ def importer_resultats_regions(engine: Engine, debug: bool = False) -> None:
         session.commit()
 
 
-def importer_resultats_departements(engine: Engine, debug: bool = False) -> None:
+def importer_resultats_departements(engine: Engine, verbose: bool = False) -> None:
     with Session(engine) as session:
         departements = session.execute(select(Departement)).scalars().all()
-        iterator = tqdm(departements, desc="Départements") if debug else departements
+        iterator = tqdm(departements, desc="Départements") if verbose else departements
         for dpt in iterator:
             reg_code = dpt.region_code
             reg_path = f"{reg_code}/"
@@ -182,10 +182,10 @@ def importer_resultats_departements(engine: Engine, debug: bool = False) -> None
         session.commit()
 
 
-def importer_resultats_communes(engine: Engine, debug: bool = False) -> None:
+def importer_resultats_communes(engine: Engine, verbose: bool = False) -> None:
     with Session(engine) as session:
         communes = session.execute(select(Commune)).scalars().all()
-        iterator = tqdm(communes, desc="Communes") if debug else communes
+        iterator = tqdm(communes, desc="Communes") if verbose else communes
         for commune in iterator:
             reg_code = commune.departement.region_code
             reg_path = f"{reg_code}/"
@@ -310,9 +310,9 @@ def analyser_resultats(file) -> dict:
     return parsed_data
 
 
-def run(engine: Engine, debug: bool = False) -> None:
+def run(engine: Engine, verbose: bool = False) -> None:
-    importer_resultats_france(engine, debug)
+    importer_resultats_france(engine, verbose)
-    importer_resultats_regions(engine, debug)
+    importer_resultats_regions(engine, verbose)
-    importer_resultats_departements(engine, debug)
+    importer_resultats_departements(engine, verbose)
     # FIXME Les communes prennent trop de temps
-    # importer_resultats_communes(engine, debug)
+    # importer_resultats_communes(engine, verbose)