From 9543df0d8da36994120d739052156899adc6a775 Mon Sep 17 00:00:00 2001 From: Emmy D'Anello Date: Tue, 2 Jul 2024 14:43:48 +0200 Subject: [PATCH] =?UTF-8?q?Importation=20des=20r=C3=A9sultats=20des=20l?= =?UTF-8?q?=C3=A9gislatives=202024?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../legislatives2022/import_resultats.py | 3 +- nupes/scripts/legislatives2024/__init__.py | 2 +- .../legislatives2024/import_candidats.py | 4 +- .../legislatives2024/import_resultats.py | 604 ++++++++++++++++++ 4 files changed, 608 insertions(+), 5 deletions(-) create mode 100644 nupes/scripts/legislatives2024/import_resultats.py diff --git a/nupes/scripts/legislatives2022/import_resultats.py b/nupes/scripts/legislatives2022/import_resultats.py index a76c7ae..2554752 100644 --- a/nupes/scripts/legislatives2022/import_resultats.py +++ b/nupes/scripts/legislatives2022/import_resultats.py @@ -20,8 +20,7 @@ def importer_resultats_bv(engine: Engine, verbose: bool = False) -> None: for bv in session.execute(select(BureauVote)).scalars().all()} for tour, file_url in tours: - file = get_file("https://www.data.gouv.fr/fr/datasets/r/a1f73b85-8194-44f4-a2b7-c343edb47d32", - "resultats-legislatives-2022-t1-par-bureau-de-vote.csv") + file = get_file(file_url, f"resultats-legislatives-2022-t{tour}-par-bureau-de-vote.csv") with file.open('r', encoding="ISO-8859-1") as f: reader = csv.reader(f, delimiter=';') diff --git a/nupes/scripts/legislatives2024/__init__.py b/nupes/scripts/legislatives2024/__init__.py index 2f08ab4..2704fde 100644 --- a/nupes/scripts/legislatives2024/__init__.py +++ b/nupes/scripts/legislatives2024/__init__.py @@ -1 +1 @@ -from . import export_resultats, import_candidats +from . import export_resultats, import_candidats, import_resultats diff --git a/nupes/scripts/legislatives2024/import_candidats.py b/nupes/scripts/legislatives2024/import_candidats.py index 8f4906b..9b6e729 100644 --- a/nupes/scripts/legislatives2024/import_candidats.py +++ b/nupes/scripts/legislatives2024/import_candidats.py @@ -71,12 +71,12 @@ def creer_nuances(engine: Engine, verbose: bool = False) -> None: def importer_candidats(engine: Engine, verbose: bool = False) -> None: - DATASET_URL = "https://www.data.gouv.fr/fr/datasets/r/e02a62a1-9f4b-49d6-bb32-05dc4da8c581" + DATASET_URL = "https://www.data.gouv.fr/fr/datasets/r/cd5b39e6-5feb-4864-b274-3431ed81442c" file = get_file(DATASET_URL, "candidats_legislatives_2024.csv") with file.open('r') as f: with Session(engine) as session: - reader = csv.DictReader(f) + reader = csv.DictReader(f, delimiter=";") for line in tqdm(reader, desc="Candidat⋅es", disable=not verbose): line: dict[str, str] numero_panneau = int(line.get("Numéro de panneau")) diff --git a/nupes/scripts/legislatives2024/import_resultats.py b/nupes/scripts/legislatives2024/import_resultats.py new file mode 100644 index 0000000..9e0fb3a --- /dev/null +++ b/nupes/scripts/legislatives2024/import_resultats.py @@ -0,0 +1,604 @@ +import csv + +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 +from nupes.models.legislatives2024 import * + + +def importer_resultats_bv(engine: Engine, verbose: bool = False) -> None: + tours = [(1, "https://www.data.gouv.fr/fr/datasets/r/6813fb28-7ec0-42ff-a528-2bc3d82d7dcd")] + + with Session(engine) as session: + candidats = {(candidat.circonscription_id, candidat.numero): candidat + for candidat in session.execute(select(CandidatLegislatives2024)).scalars().all()} + bv_ids = {(bv.commune_code, bv.code_bureau): bv.id + for bv in session.execute(select(BureauVote)).scalars().all()} + + for tour, file_url in tours: + file = get_file(file_url, f"resultats-legislatives-2024-t{tour}-par-bureau-de-vote.csv") + + with file.open('r') as f: + reader = csv.DictReader(f, delimiter=';') + with Session(engine) as session: + for row in tqdm(reader, desc=f"Bureau de vote tour {tour}", disable=not verbose): + dpt_code = row['Code département'] + + match dpt_code: + case "ZX": + dpt_code = "977" + + if dpt_code == "987" or dpt_code == "988": + # Les communes de Polynésie française ne sont pas importées + continue + + com_code = str(dpt_code + row['Code commune'][(1 if len(dpt_code) == 3 else 0):]).zfill(5) + + if com_code == "98601": + # 3 royaumes à Wallis-et-Futuna, mais un seul résultat de commune + com_code = "98611" + + if com_code.startswith("987"): # Les communes de Polynésie française ne sont pas importées + continue + + bv_code = row['Code BV'] + + if (com_code, bv_code) not in bv_ids: + print(f"Bureau de vote {com_code}_{bv_code} non trouvé") + continue + bv_id = bv_ids[(com_code, bv_code)] + + circo_code = row['Code circonscription législative'][-2:] + circo_id = f"{dpt_code.zfill(2)}-{circo_code.zfill(2)}" + + resultats_bv = session.execute(select(ResultatsBureauVoteLegislatives2024) + .filter_by(bv_id=bv_id)).scalar_one_or_none() + if not resultats_bv: + resultats_commune = session.execute(select(ResultatsCommuneLegislatives2024) + .filter_by(commune_id=com_code)).scalar_one_or_none() + if not resultats_commune: + resultats_dpt = session.execute(select(ResultatsDepartementLegislatives2024) + .filter_by(dpt_id=dpt_code)) \ + .scalar_one_or_none() + if not resultats_dpt: + dpt = session.execute(select(Departement).filter_by(code_insee=dpt_code)) \ + .scalar_one() + resultats_reg = session.execute( + select(ResultatsRegionLegislatives2024).filter_by(region_id=dpt.region_code)) \ + .scalar_one_or_none() + if not resultats_reg: + resultats_france = session.execute(select(ResultatsFranceLegislatives2024)) \ + .scalar_one_or_none() + if not resultats_france: + session.add(ResultatsFranceLegislatives2024()) + resultats_france = session.execute(select(ResultatsFranceLegislatives2024)) \ + .scalar_one() + + resultats_reg = ResultatsRegionLegislatives2024( + region_id=str(dpt.region_code), resultats_france_id=resultats_france.id) + session.add(resultats_reg) + resultats_reg = session.execute(select(ResultatsRegionLegislatives2024) + .filter_by(region_id=dpt.region_code)).scalar_one() + + resultats_dpt = ResultatsDepartementLegislatives2024( + dpt_id=dpt_code, resultats_region_id=resultats_reg.id) + session.add(resultats_dpt) + resultats_dpt = session.execute(select(ResultatsDepartementLegislatives2024) + .filter_by(dpt_id=dpt_code)).scalar_one() + + resultats_commune = ResultatsCommuneLegislatives2024(commune_id=com_code, + resultats_dpt_id=resultats_dpt.id) + session.add(resultats_commune) + resultats_commune = session.execute(select(ResultatsCommuneLegislatives2024) + .filter_by(commune_id=com_code)).scalar_one() + + resultats_circo = session.execute(select(ResultatsCirconscriptionLegislatives2024) + .filter_by(circo_id=circo_id)).scalar_one_or_none() + + resultats_bv = ResultatsBureauVoteLegislatives2024(bv_id=bv_id, + resultats_commune_id=resultats_commune.id, + resultats_circo_id=resultats_circo.id) + session.add(resultats_bv) + resultats_bv = session.execute(select(ResultatsBureauVoteLegislatives2024) + .filter_by(bv_id=bv_id)).scalar_one() + + if tour == 1: + resultats_bv.inscrits_t1 = int(row['Inscrits']) + resultats_bv.votants_t1 = int(row['Votants']) + resultats_bv.abstentions_t1 = int(row['Abstentions']) + resultats_bv.exprimes_t1 = int(row['Exprimes']) + resultats_bv.blancs_t1 = int(row['Blancs']) + resultats_bv.nuls_t1 = int(row['Nuls']) + elif tour == 2: + resultats_bv.inscrits_t2 = int(row['Inscrits']) + resultats_bv.votants_t2 = int(row['Votants']) + resultats_bv.abstentions_t2 = int(row['Abstentions']) + resultats_bv.exprimes_t2 = int(row['Exprimes']) + resultats_bv.blancs_t2 = int(row['Blancs']) + resultats_bv.nuls_t2 = int(row['Nuls']) + + for i in range(1, 23): + if f'Nuance candidat {i}' not in row: + # On a épuisé toutes les nuances + break + + numero = row[f'Numéro de panneau {i}'] + if not numero: + continue + + voix = int(row[f'Voix {i}']) + + candidat = candidats[(circo_id, int(numero))] + + voix_candidat_bv = session.execute(select(VoixBureauVoteLegislatives2024) + .filter_by(resultats_bureau_vote_id=resultats_bv.id, + candidat_id=candidat.id)) \ + .scalar_one_or_none() + if not voix_candidat_bv: + voix_candidat_bv = VoixBureauVoteLegislatives2024( + resultats_bureau_vote_id=resultats_bv.id, candidat_id=candidat.id) + session.add(voix_candidat_bv) + + if tour == 1: + voix_candidat_bv.voix_t1 = voix + elif tour == 2: + voix_candidat_bv.voix_t2 = voix + + session.commit() + + +def importer_resultats_commune(engine: Engine, verbose: bool = False) -> None: + tours = [(1, "https://www.data.gouv.fr/fr/datasets/r/bd32fcd3-53df-47ac-bf1d-8d8003fe23a1")] + + with Session(engine) as session: + for resultats_commune in session.execute(select(ResultatsCommuneLegislatives2024)).scalars().all(): + resultats_commune.inscrits_t1 = 0 + resultats_commune.votants_t1 = 0 + resultats_commune.abstentions_t1 = 0 + resultats_commune.exprimes_t1 = 0 + resultats_commune.blancs_t1 = 0 + resultats_commune.nuls_t1 = 0 + resultats_commune.inscrits_t2 = 0 + resultats_commune.votants_t2 = 0 + resultats_commune.abstentions_t2 = 0 + resultats_commune.exprimes_t2 = 0 + resultats_commune.blancs_t2 = 0 + resultats_commune.nuls_t2 = 0 + + for voix_nuance in resultats_commune.voix: + voix_nuance.voix_t1 = 0 + voix_nuance.voix_t2 = 0 + + session.commit() + + for tour, file_url in tours: + file = get_file(file_url, + f"resultats-legislatives-2024-t{tour}-par-commune.csv") + + with file.open('r') as f: + reader = csv.DictReader(f, delimiter=';') + with Session(engine) as session: + for row in tqdm(reader, desc=f"Commune tour {tour}", disable=not verbose): + dpt_code = row['Code département'] + + match dpt_code: + case "ZX": + dpt_code = "977" + + if dpt_code == "987" or dpt_code == "988": + # Les communes de Polynésie française ne sont pas importées + continue + + com_code = str(dpt_code + row['Code commune'][(1 if len(dpt_code) == 3 else 0):]).zfill(5) + + if com_code == "98601": + # 3 royaumes à Wallis-et-Futuna, mais un seul résultat de commune + com_code = "98611" + + resultats_commune = session.execute(select(ResultatsCommuneLegislatives2024) + .filter_by(commune_id=com_code)).scalar_one_or_none() + if not resultats_commune: + resultats_dpt = session.execute(select(ResultatsDepartementLegislatives2024) + .filter_by(dpt_id=dpt_code)).scalar_one_or_none() + if not resultats_dpt: + dpt = session.execute(select(Departement) + .filter_by(code_insee=f"{dpt_code.zfill(2)}")).scalar_one() + resultats_reg = session.execute(select(ResultatsRegionLegislatives2024) + .filter_by(region_id=dpt.region_code)).scalar_one_or_none() + if not resultats_reg: + resultats_france = session.execute(select(ResultatsFranceLegislatives2024)) \ + .scalar_one_or_none() + if not resultats_france: + session.add(ResultatsFranceLegislatives2024()) + resultats_france = session.execute(select(ResultatsFranceLegislatives2024)) \ + .scalar_one() + + resultats_reg = ResultatsRegionLegislatives2024(region_id=str(dpt.region_code), + resultats_france_id=resultats_france.id) + session.add(resultats_reg) + resultats_reg = session.execute(select(ResultatsRegionLegislatives2024) + .filter_by(region_id=dpt.region_code)).scalar_one() + + resultats_dpt = ResultatsDepartementLegislatives2024(dpt_id=dpt_code, + resultats_region_id=resultats_reg.id) + session.add(resultats_dpt) + resultats_dpt = session.execute(select(ResultatsDepartementLegislatives2024) + .filter_by(dpt_id=dpt_code)).scalar_one() + + resultats_commune = ResultatsCommuneLegislatives2024(commune_id=com_code, + resultats_dpt_id=resultats_dpt.id) + session.add(resultats_commune) + resultats_commune = session.execute(select(ResultatsCommuneLegislatives2024) + .filter_by(commune_id=com_code)).scalar_one() + + if tour == 1: + resultats_commune.inscrits_t1 += int(row['Inscrits']) + resultats_commune.votants_t1 += int(row['Votants']) + resultats_commune.abstentions_t1 += int(row['Abstentions']) + resultats_commune.exprimes_t1 += int(row['Exprimés']) + resultats_commune.blancs_t1 += int(row['Blancs']) + resultats_commune.nuls_t1 += int(row['Nuls']) + elif tour == 2: + resultats_commune.inscrits_t2 += int(row['Inscrits']) + resultats_commune.votants_t2 += int(row['Votants']) + resultats_commune.abstentions_t2 += int(row['Abstentions']) + resultats_commune.exprimes_t2 += int(row['Exprimés']) + resultats_commune.blancs_t2 += int(row['Blancs']) + resultats_commune.nuls_t2 += int(row['Nuls']) + + for i in range(1, 23): + if f'Nuance candidat {i}' not in row: + # On a épuisé toutes les nuances + break + + nuance_code = row[f'Nuance candidat {i}'] + if not nuance_code: + continue + + voix = int(row[f'Voix {i}']) + + voix_nuance_commune = session.execute(select(VoixCommuneLegislatives2024) + .filter_by(resultats_commune_id=resultats_commune.id, + nuance_id=nuance_code)) \ + .scalar_one_or_none() + if not voix_nuance_commune: + voix_nuance_commune = VoixCommuneLegislatives2024(resultats_commune_id=resultats_commune.id, + nuance_id=nuance_code, + voix_t1=0, voix_t2=0) + session.add(voix_nuance_commune) + + if tour == 1: + voix_nuance_commune.voix_t1 += voix + elif tour == 2: + voix_nuance_commune.voix_t2 += voix + + session.commit() + + +def importer_resultats_circo(engine: Engine, verbose: bool = False) -> None: + tours = [(1, "https://www.data.gouv.fr/fr/datasets/r/5163f2e3-1362-4c35-89a0-1934bb74f2d9")] + + for tour, file_url in tours: + file = get_file(file_url, f"resultats-legislatives-2024-t{tour}-par-circonscription.csv") + + with file.open('r') as f: + reader = csv.DictReader(f, delimiter=';') + with Session(engine) as session: + for row in tqdm(reader, desc=f"Circonscription tour {tour}", disable=not verbose): + dpt_code = row['Code département'] + circo_code = row['Code circonscription législative'][-2:] + + match dpt_code: + case "ZX": + dpt_code = "977" + + circo_id = f"{dpt_code.zfill(2)}-{circo_code.zfill(2)}" + + resultats_circo = session.execute(select(ResultatsCirconscriptionLegislatives2024) + .filter_by(circo_id=circo_id)).scalar_one_or_none() + if not resultats_circo: + resultats_dpt = session.execute(select(ResultatsDepartementLegislatives2024) + .filter_by(dpt_id=dpt_code)).scalar_one_or_none() + if not resultats_dpt: + dpt = session.execute(select(Departement).filter_by(code_insee=f"{dpt_code.zfill(2)}")) \ + .scalar_one() + resultats_reg = session.execute(select(ResultatsRegionLegislatives2024) + .filter_by(region_id=dpt.region_code)) \ + .scalar_one_or_none() + if not resultats_reg: + resultats_france = session.execute(select(ResultatsFranceLegislatives2024)) \ + .scalar_one_or_none() + if not resultats_france: + session.add(ResultatsFranceLegislatives2024()) + resultats_france = session.execute(select(ResultatsFranceLegislatives2024)).scalar_one() + + resultats_reg = ResultatsRegionLegislatives2024(region_id=str(dpt.region_code), + resultats_france_id=resultats_france.id) + session.add(resultats_reg) + resultats_reg = session.execute(select(ResultatsRegionLegislatives2024) + .filter_by(region_id=dpt.region_code)).scalar_one() + + resultats_dpt = ResultatsDepartementLegislatives2024(dpt_id=dpt_code, + resultats_region_id=resultats_reg.id) + session.add(resultats_dpt) + resultats_dpt = session.execute(select(ResultatsDepartementLegislatives2024) + .filter_by(dpt_id=dpt_code)).scalar_one() + + resultats_circo = ResultatsCirconscriptionLegislatives2024( + circo_id=circo_id, resultats_departement_id=resultats_dpt.id) + session.add(resultats_circo) + resultats_circo = session.execute(select(ResultatsCirconscriptionLegislatives2024) + .filter_by(circo_id=circo_id)).scalar_one() + + if tour == 1: + resultats_circo.inscrits_t1 = int(row['Inscrits']) + resultats_circo.votants_t1 = int(row['Votants']) + resultats_circo.abstentions_t1 = int(row['Abstentions']) + resultats_circo.exprimes_t1 = int(row['Exprimés']) + resultats_circo.blancs_t1 = int(row['Blancs']) + resultats_circo.nuls_t1 = int(row['Nuls']) + elif tour == 2: + resultats_circo.inscrits_t2 = int(row['Inscrits']) + resultats_circo.votants_t2 = int(row['Votants']) + resultats_circo.abstentions_t2 = int(row['Abstentions']) + resultats_circo.exprimes_t2 = int(row['Exprimés']) + resultats_circo.blancs_t2 = int(row['Blancs']) + resultats_circo.nuls_t2 = int(row['Nuls']) + + for i in range(1, 23): + if f'Nuance candidat {i}' not in row: + # On a épuisé toutes les nuances + break + + numero = row[f'Numéro de panneau {i}'] + if not numero: + continue + + voix = int(row[f'Voix {i}']) + + candidat = session.execute(select(CandidatLegislatives2024) + .filter_by(circonscription_id=circo_id, numero=numero)) \ + .scalar_one_or_none() + + voix_candidat_circo = session.execute(select(VoixCirconscriptionLegislatives2024) + .filter_by(resultats_circonscription_id=resultats_circo.id, + candidat_id=candidat.id)) \ + .scalar_one_or_none() + if not voix_candidat_circo: + voix_candidat_circo = VoixCirconscriptionLegislatives2024( + resultats_circonscription_id=resultats_circo.id, candidat_id=candidat.id) + session.add(voix_candidat_circo) + + if tour == 1: + voix_candidat_circo.voix_t1 = voix + elif tour == 2: + voix_candidat_circo.voix_t2 = voix + + session.commit() + + +def importer_resultats_departement(engine: Engine, verbose: bool = False) -> None: + tours = [(1, "https://www.data.gouv.fr/fr/datasets/r/78c708c5-5bc5-438d-8379-f432beae3f2b")] + + for tour, file_url in tours: + file = get_file(file_url, f"resultats-legislatives-2024-t{tour}-par-departement.csv") + + with file.open('r') as f: + reader = csv.DictReader(f, delimiter=';') + with Session(engine) as session: + for row in tqdm(reader, desc=f"Département tour {tour}", disable=not verbose): + dpt_code = row['Code département'] + + match dpt_code: + case "ZX": + dpt_code = "977" + + resultats_dpt = session.execute(select(ResultatsDepartementLegislatives2024) + .filter_by(dpt_id=dpt_code)).scalar_one_or_none() + if not resultats_dpt: + dpt = session.execute(select(Departement) + .filter_by(code_insee=f"{dpt_code.zfill(2)}")) \ + .scalar_one() + resultats_reg = session.execute(select(ResultatsRegionLegislatives2024) + .filter_by(region_id=dpt.region_code)) \ + .scalar_one_or_none() + if not resultats_reg: + resultats_france = session.execute(select(ResultatsFranceLegislatives2024)) \ + .scalar_one_or_none() + if not resultats_france: + session.add(ResultatsFranceLegislatives2024()) + resultats_france = session.execute(select(ResultatsFranceLegislatives2024)).scalar_one() + + resultats_reg = ResultatsRegionLegislatives2024(region_id=str(dpt.region_code), + resultats_france_id=resultats_france.id) + session.add(resultats_reg) + resultats_reg = session.execute(select(ResultatsRegionLegislatives2024) + .filter_by(region_id=dpt.region_code)).scalar_one() + + resultats_dpt = ResultatsDepartementLegislatives2024(dpt_id=dpt_code, + resultats_region_id=resultats_reg.id) + session.add(resultats_dpt) + resultats_dpt = session.execute(select(ResultatsDepartementLegislatives2024) + .filter_by(dpt_id=dpt_code)).scalar_one() + + if tour == 1: + resultats_dpt.inscrits_t1 = int(row['Inscrits']) + resultats_dpt.votants_t1 = int(row['Votants']) + resultats_dpt.abstentions_t1 = int(row['Abstentions']) + resultats_dpt.exprimes_t1 = int(row['Exprimés']) + resultats_dpt.blancs_t1 = int(row['Blancs']) + resultats_dpt.nuls_t1 = int(row['Nuls']) + elif tour == 2: + resultats_dpt.inscrits_t2 = int(row['Inscrits']) + resultats_dpt.votants_t2 = int(row['Votants']) + resultats_dpt.abstentions_t2 = int(row['Abstentions']) + resultats_dpt.exprimes_t2 = int(row['Exprimés']) + resultats_dpt.blancs_t2 = int(row['Blancs']) + resultats_dpt.nuls_t2 = int(row['Nuls']) + + for i in range(1, 23): + if f'Nuance candidat {i}' not in row: + # On a épuisé toutes les nuances + break + + nuance_code = row[f'Nuance candidat {i}'] + if not nuance_code: + continue + + voix = int(row[f'Voix {i}']) + + voix_nuance_dpt = session.execute(select(VoixDepartementLegislatives2024) + .filter_by(resultats_departement_id=resultats_dpt.id, + nuance_id=nuance_code)) \ + .scalar_one_or_none() + if not voix_nuance_dpt: + voix_nuance_dpt = VoixDepartementLegislatives2024(resultats_departement_id=resultats_dpt.id, + nuance_id=nuance_code) + session.add(voix_nuance_dpt) + + if tour == 1: + voix_nuance_dpt.voix_t1 = voix + elif tour == 2: + voix_nuance_dpt.voix_t2 = voix + + session.commit() + + +def importer_resultats_region(engine: Engine, verbose: bool = False) -> None: + tours = [(1, "https://www.data.gouv.fr/fr/datasets/r/f69ffab7-fe37-494e-ad6d-a7cfc80ddc1f")] + + for tour, file_url in tours: + file = get_file(file_url, f"resultats-legislatives-2024-t{tour}-par-region.csv") + + with file.open('r') as f: + reader = csv.DictReader(f, delimiter=';') + + with Session(engine) as session: + for row in tqdm(reader, desc=f"Région tour {tour}", disable=not verbose): + reg_code = row['Code région'] + + resultats_reg = session.execute(select(ResultatsRegionLegislatives2024) + .filter_by(region_id=reg_code)).scalar_one_or_none() + if not resultats_reg: + resultats_france = session.execute(select(ResultatsFranceLegislatives2024)).scalar_one_or_none() + if not resultats_france: + session.add(ResultatsFranceLegislatives2024()) + resultats_france = session.execute(select(ResultatsFranceLegislatives2024)).scalar_one() + + resultats_reg = ResultatsRegionLegislatives2024(region_id=reg_code, + resultats_france_id=resultats_france.id) + session.add(resultats_reg) + session.commit() + resultats_reg = session.execute(select(ResultatsRegionLegislatives2024) + .filter_by(region_id=reg_code)).scalar_one() + + if tour == 1: + resultats_reg.inscrits_t1 = int(row['Inscrits']) + resultats_reg.votants_t1 = int(row['Votants']) + resultats_reg.abstentions_t1 = int(row['Abstentions']) + resultats_reg.exprimes_t1 = int(row['Exprimés']) + resultats_reg.blancs_t1 = int(row['Blancs']) + resultats_reg.nuls_t1 = int(row['Nuls']) + elif tour == 2: + resultats_reg.inscrits_t2 = int(row['Inscrits']) + resultats_reg.votants_t2 = int(row['Votants']) + resultats_reg.abstentions_t2 = int(row['Abstentions']) + resultats_reg.exprimes_t2 = int(row['Exprimés']) + resultats_reg.blancs_t2 = int(row['Blancs']) + resultats_reg.nuls_t2 = int(row['Nuls']) + + for i in range(1, 23): + if f'Nuance candidat {i}' not in row: + # On a épuisé toutes les nuances + break + + nuance_code = row[f'Nuance candidat {i}'] + if not nuance_code: + continue + + voix = int(row[f'Voix {i}']) + + voix_nuance_reg = session.execute(select(VoixRegionLegislatives2024) + .filter_by(resultats_region_id=resultats_reg.id, + nuance_id=nuance_code)) \ + .scalar_one_or_none() + if not voix_nuance_reg: + voix_nuance_reg = VoixRegionLegislatives2024(resultats_region_id=resultats_reg.id, + nuance_id=nuance_code) + session.add(voix_nuance_reg) + + if tour == 1: + voix_nuance_reg.voix_t1 = voix + elif tour == 2: + voix_nuance_reg.voix_t2 = voix + + session.commit() + + +def importer_resultats_france(engine: Engine, verbose: bool = False) -> None: + tours = [(1, "https://www.data.gouv.fr/fr/datasets/r/386fd5ac-e7f1-4e0f-8929-12d2c5391081")] + for tour, file_url in tours: + file = get_file(file_url, f"resultats-legislatives-2024-t{tour}-france-entiere.csv") + + with file.open('r') as f: + reader = csv.DictReader(f, delimiter=';') + + with Session(engine) as session: + for row in tqdm(reader, desc=f"France Entière tour {tour}", disable=not verbose): + resultats_france = session.execute(select(ResultatsFranceLegislatives2024)).scalar_one_or_none() + + if not resultats_france: + session.add(ResultatsFranceLegislatives2024()) + session.commit() + resultats_france = session.execute(select(ResultatsFranceLegislatives2024)).scalar_one() + + if tour == 1: + resultats_france.inscrits_t1 = int(row['Inscrits']) + resultats_france.votants_t1 = int(row['Votants']) + resultats_france.abstentions_t1 = int(row['Abstentions']) + resultats_france.exprimes_t1 = int(row['Votants']) + resultats_france.blancs_t1 = int(row['Blancs']) + resultats_france.nuls_t1 = int(row['Nuls']) + elif tour == 2: + resultats_france.inscrits_t2 = int(row['Inscrits']) + resultats_france.votants_t2 = int(row['Votants']) + resultats_france.abstentions_t2 = int(row['Abstentions']) + resultats_france.exprimes_t2 = int(row['Votants']) + resultats_france.blancs_t2 = int(row['Blancs']) + resultats_france.nuls_t2 = int(row['Nuls']) + + for i in range(1, 23): + if f'Nuance candidat {i}' not in row: + # On a épuisé toutes les nuances + break + + nuance_code = row[f'Nuance candidat {i}'] + voix = int(row[f'Voix {i}']) + + voix_nuance_france = session.execute(select(VoixFranceLegislatives2024) + .filter_by(resultats_france_id=resultats_france.id, + nuance_id=nuance_code)) \ + .scalar_one_or_none() + if not voix_nuance_france: + voix_nuance_france = VoixFranceLegislatives2024(resultats_france_id=resultats_france.id, + nuance_id=nuance_code) + session.add(voix_nuance_france) + + if tour == 1: + voix_nuance_france.voix_t1 = voix + elif tour == 2: + voix_nuance_france.voix_t2 = voix + + session.commit() + + +def run(engine: Engine, verbose: bool = False) -> None: + importer_resultats_france(engine, verbose) + importer_resultats_region(engine, verbose) + importer_resultats_departement(engine, verbose) + importer_resultats_circo(engine, verbose) + importer_resultats_commune(engine, verbose) + importer_resultats_bv(engine, verbose)