diff --git a/nupes/scripts/export_resultats_2024.py b/nupes/scripts/export_resultats_2024.py index 88d2440..40a316c 100644 --- a/nupes/scripts/export_resultats_2024.py +++ b/nupes/scripts/export_resultats_2024.py @@ -5,9 +5,10 @@ from sqlalchemy import Engine, select from sqlalchemy.orm import Session from tqdm import tqdm -from nupes.models.geographie import Region, Departement, Commune +from nupes.models.geographie import Region, Departement, Commune, Circonscription, BureauVote from nupes.models.europeennes2024 import Bloc, Nuance, Liste, \ - ResultatsFrance, ResultatsRegion, ResultatsDepartement, ResultatsCommune + ResultatsFrance, ResultatsRegion, ResultatsDepartement, ResultatsCommune, \ + ResultatsCirconscription, ResultatsBureauVote DATA_DIR = Path(__file__).parent.parent.parent / 'data' @@ -201,6 +202,60 @@ def exporter_resultats_departements(engine: Engine, verbose: bool = False) -> No json.dump(departements_json, f) +def exporter_resultats_circonscriptions(engine: Engine, verbose: bool = False) -> None: + with Session(engine) as session: + circonscriptions = session.execute(select(Circonscription)).scalars().all() + circonscriptions_json = [] + + iterator = tqdm(circonscriptions, desc="Circonscriptions") if verbose else circonscriptions + for circonscription in iterator: + circonscription_json = {'id': circonscription.id, 'departement': circonscription.departement_code, + 'numero': circonscription.numero, + 'bureaux_vote': [bv.id for bv in circonscription.bureaux_vote]} + circonscriptions_json.append(circonscription_json) + + resultats_circonscription = session.execute( + select(ResultatsCirconscription).filter_by(circo_id=circonscription.id)).scalar_one_or_none() + if not resultats_circonscription: + resultats_departement = session.execute(select(ResultatsDepartement) + .filter_by(dpt_id=circonscription.departement_code)).scalar_one() + resultats_circonscription = ResultatsCirconscription(circo_id=circonscription.id, + resultats_departement_id=resultats_departement.id) + session.add(resultats_circonscription) + + resultats_dict = { + "inscrits": resultats_circonscription.inscrits, + "votants": resultats_circonscription.votants, + "abstentions": resultats_circonscription.abstentions, + "exprimes": resultats_circonscription.exprimes, + "blancs": resultats_circonscription.blancs, + "nuls": resultats_circonscription.nuls, + 'circonscription': circonscription_json, + 'geometry': circonscription.geometry, + } + + resultats_listes = {} + resultats_dict['voix_listes'] = resultats_listes + for voix_liste in resultats_circonscription.voix_listes: + resultats_listes[voix_liste.liste.numero] = voix_liste.voix + + file = DATA_DIR / "resultats" / "europeennes2024" / "circonscriptions" / f"{circonscription.id}.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() + + circonscriptions_file = DATA_DIR / "resultats" / "europeennes2024" / "circonscriptions" / "circonscriptions.json" + if not circonscriptions_file.parent.is_dir(): + circonscriptions_file.parent.mkdir(parents=True) + + with circonscriptions_file.open('w') as f: + json.dump(circonscriptions_json, f) + + def exporter_resultats_communes(engine: Engine, verbose: bool = False) -> None: with Session(engine) as session: communes = session.execute(select(Commune)).scalars().all() @@ -255,10 +310,63 @@ def exporter_resultats_communes(engine: Engine, verbose: bool = False) -> None: json.dump(communes_json, f) +def exporter_resultats_bureaux_vote(engine: Engine, verbose: bool = False) -> None: + with Session(engine) as session: + bureaux_vote = session.execute(select(BureauVote)).scalars().all() + bureaux_vote_json = [] + + iterator = tqdm(bureaux_vote, desc="Bureaux de vote") if verbose else bureaux_vote + for bureau_vote in iterator: + bureau_vote_json = {'id': bureau_vote.id, 'commune': bureau_vote.commune_code} + bureaux_vote_json.append(bureau_vote_json) + + resultats_bureau_vote = session.execute( + select(ResultatsBureauVote).filter_by(bv_id=bureau_vote.id)).scalar_one_or_none() + if not resultats_bureau_vote: + resultats_commune = session.execute(select(ResultatsCommune) + .filter_by(commune_id=bureau_vote.commune_code)).scalar_one() + resultats_bureau_vote = ResultatsBureauVote(bv_id=bureau_vote.id, + resultats_commune_id=resultats_commune.id) + session.add(resultats_bureau_vote) + + resultats_dict = { + "inscrits": resultats_bureau_vote.inscrits, + "votants": resultats_bureau_vote.votants, + "abstentions": resultats_bureau_vote.abstentions, + "exprimes": resultats_bureau_vote.exprimes, + "blancs": resultats_bureau_vote.blancs, + "nuls": resultats_bureau_vote.nuls, + 'bureau_vote': bureau_vote_json, + 'geometry': bureau_vote.geometry, + } + + resultats_listes = {} + resultats_dict['voix_listes'] = resultats_listes + for voix_liste in resultats_bureau_vote.voix_listes: + resultats_listes[voix_liste.liste.numero] = voix_liste.voix + + file = DATA_DIR / "resultats" / "europeennes2024" / "bureaux_vote" / f"{bureau_vote.id}.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() + + bureaux_vote_file = DATA_DIR / "resultats" / "europeennes2024" / "bureaux_vote" / "bureaux_vote.json" + if not bureaux_vote_file.parent.is_dir(): + bureaux_vote_file.parent.mkdir(parents=True) + + with bureaux_vote_file.open('w') as f: + json.dump(bureaux_vote_json, f) + + def run(engine: Engine, verbose: bool = False) -> None: exporter_listes(engine, verbose) exporter_resultats_france(engine, verbose) exporter_resultats_regions(engine, verbose) exporter_resultats_departements(engine, verbose) - # FIXME Les communes prennent trop de temps - # exporter_resultats_communes(engine, verbose) + exporter_resultats_circonscriptions(engine, verbose) + exporter_resultats_communes(engine, verbose) + exporter_resultats_bureaux_vote(engine, verbose)