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, data_round: int = 0, verbose: bool = False) -> None: tours = [(1, "https://www.data.gouv.fr/fr/datasets/r/6813fb28-7ec0-42ff-a528-2bc3d82d7dcd"), (2, "https://www.data.gouv.fr/fr/datasets/r/ca974f04-cfd9-4da8-8554-4a868a09c6c2")] 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 for bv in session.execute(select(BureauVote)).scalars().all()} for tour, file_url in tours: if data_round > 0 and tour != data_round: # On saute ce tour continue 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 = row['Code commune'] 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 = bv_ids[(com_code, bv_code)] bv_id = bv.id circo_id = bv.circo_code 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['Exprimés']) 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['Exprimés']) 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, data_round: int = 0, verbose: bool = False) -> None: tours = [(1, "https://www.data.gouv.fr/fr/datasets/r/bd32fcd3-53df-47ac-bf1d-8d8003fe23a1"), (2, "https://www.data.gouv.fr/fr/datasets/r/5a8088fd-8168-402a-9f40-c48daab88cd1")] with Session(engine) as session: for resultats_commune in session.execute(select(ResultatsCommuneLegislatives2024)).scalars().all(): if data_round == 0 or data_round == 1: 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 if data_round == 0 or data_round == 2: 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: if data_round == 0 or data_round == 1: voix_nuance.voix_t1 = 0 if data_round == 0 or data_round == 2: voix_nuance.voix_t2 = 0 session.commit() com_codes = {commune.code_insee for commune in session.execute(select(Commune)).scalars().all()} for tour, file_url in tours: if data_round > 0 and tour != data_round: # On saute ce tour continue 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 = row['Code commune'] if com_code == "98601": # 3 royaumes à Wallis-et-Futuna, mais un seul résultat de commune com_code = "98611" if com_code not in com_codes: print(f"Commune {com_code} non trouvée") continue 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, data_round: int = 0, verbose: bool = False) -> None: tours = [(1, "https://www.data.gouv.fr/fr/datasets/r/5163f2e3-1362-4c35-89a0-1934bb74f2d9"), (2, "https://www.data.gouv.fr/fr/datasets/r/41ed46cd-77c2-4ecc-b8eb-374aa953ca39")] for tour, file_url in tours: if data_round > 0 and tour != data_round: # On saute ce tour continue 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, data_round: int = 0, verbose: bool = False) -> None: tours = [(1, "https://www.data.gouv.fr/fr/datasets/r/78c708c5-5bc5-438d-8379-f432beae3f2b"), (2, "https://www.data.gouv.fr/fr/datasets/r/8d4a6927-c96f-4cf5-b757-ea745eca26bd")] for tour, file_url in tours: if data_round > 0 and tour != data_round: # On saute ce tour continue 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, data_round: int = 0, verbose: bool = False) -> None: tours = [(1, "https://www.data.gouv.fr/fr/datasets/r/f69ffab7-fe37-494e-ad6d-a7cfc80ddc1f"), (2, "https://www.data.gouv.fr/fr/datasets/r/ac4e272a-3ce2-4f20-941e-fb0ded444a5d")] for tour, file_url in tours: if data_round > 0 and tour != data_round: # On saute ce tour continue 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, data_round: int = 0, verbose: bool = False) -> None: # Erreur avec les données France Entière du 1er tour, qui sera j'espère corrigée un jour tours = [# (1, "https://www.data.gouv.fr/fr/datasets/r/386fd5ac-e7f1-4e0f-8929-12d2c5391081"), (2, "https://www.data.gouv.fr/fr/datasets/r/f64d2781-4ffa-414b-8f6d-e9d849f2f5e7")] for tour, file_url in tours: if data_round > 0 and tour != data_round: # On saute ce tour continue 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['Exprimés']) 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['Exprimés']) 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, data_round: int = 1, verbose: bool = False) -> None: importer_resultats_france(engine, data_round, verbose) importer_resultats_region(engine, data_round, verbose) importer_resultats_departement(engine, data_round, verbose) importer_resultats_circo(engine, data_round, verbose) importer_resultats_commune(engine, data_round, verbose) importer_resultats_bv(engine, data_round, verbose)