From 00ba2ad527233872fe0771f28e1520309b8b9e64 Mon Sep 17 00:00:00 2001 From: Emmy D'Anello Date: Thu, 13 Jun 2024 12:19:02 +0200 Subject: [PATCH] =?UTF-8?q?Import=20des=20r=C3=A9ultats=20temporaires=20de?= =?UTF-8?q?s=20=C3=A9lections=20europ=C3=A9ennes?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main.py | 10 ++- nupes/scripts/import_resultats_2024.py | 119 +++++++++++++++++++++++++ 2 files changed, 127 insertions(+), 2 deletions(-) create mode 100644 nupes/scripts/import_resultats_2024.py diff --git a/main.py b/main.py index 524b803..39ff4e3 100755 --- a/main.py +++ b/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,7 +17,8 @@ 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", + "import_resultats_web", "export_resultats"]) parser.add_argument('--debug', '-d', action='store_true', help="Mode debug") return parser.parse_args() @@ -41,6 +43,10 @@ def main(): case "import_resultats": if debug: print(f"Import des résultats pour les élections {args.type} {args.year}") + import_resultats_2024.run(engine, debug) + case "import_resultats_web": + if debug: + print(f"Import des résultats depuis le web pour les élections {args.type} {args.year}") import_resultats_web_2024.run(engine, debug) case "export_resultats": if debug: diff --git a/nupes/scripts/import_resultats_2024.py b/nupes/scripts/import_resultats_2024.py new file mode 100644 index 0000000..79e49f8 --- /dev/null +++ b/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, debug: 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 debug): + 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, debug: bool = False) -> None: + pass + + +def calculer_resultats_circo(engine: Engine, debug: bool = False) -> None: + pass + + +def calculer_resultats_departement(engine: Engine, debug: bool = False) -> None: + pass + + +def calculer_resultats_region(engine: Engine, debug: bool = False) -> None: + pass + + +def calculer_resultats_national(engine: Engine, debug: bool = False) -> None: + pass + + +def run(engine: Engine, debug: bool = False) -> None: + importer_resultats_bv(engine, debug) + calculer_resultats_commune(engine, debug) + # calculer_resultats_circo(engine, debug) + calculer_resultats_departement(engine, debug) + calculer_resultats_region(engine, debug) + calculer_resultats_national(engine, debug)