import json from pathlib import Path from sqlalchemy import Engine, select from sqlalchemy.orm import Session from tqdm import tqdm from nupes.models.geographie import Circonscription, BureauVote from nupes.models.europeennes2024 import * DATA_DIR = Path(__file__).parent.parent.parent.parent / 'data' def exporter_listes(engine: Engine, verbose: bool = False) -> None: with Session(engine) as session: blocs = session.execute(select(BlocEuropeennes2024)).scalars().all() blocs_json = [] for bloc in blocs: bloc_json = { 'id': bloc.id, 'nom': bloc.nom, 'couleur': bloc.couleur, } blocs_json.append(bloc_json) file = DATA_DIR / "resultats" / "europeennes" / "2024" / "blocs.json" if not file.parent.is_dir(): file.parent.mkdir(parents=True) with file.open('w') as f: json.dump(blocs_json, f) nuances = session.execute(select(NuanceEuropeennes2024)).scalars().all() nuances_json = [] for nuance in nuances: nuance_json = { 'code': nuance.code, 'nom': nuance.nom, 'couleur': nuance.couleur, } nuances_json.append(nuance_json) file = DATA_DIR / "resultats" / "europeennes" / "2024" / "nuances.json" if not file.parent.is_dir(): file.parent.mkdir(parents=True) with file.open('w') as f: json.dump(nuances_json, f) listes = session.execute(select(ListeEuropeennes2024)).scalars().all() listes_json = [] for liste in listes: candidats = [ { 'ordre': candidat.ordre, 'nom': candidat.nom, 'prenom': candidat.prenom, 'sexe': candidat.sexe.value, 'date_naissance': candidat.date_naissance.isoformat(), 'profession': candidat.profession, 'code_personnalite': candidat.code_personnalite.value, 'sortant': candidat.sortant } for candidat in liste.candidats ] liste_json = { 'numero': liste.numero, 'nom': liste.nom, 'nuance': liste.nuance_id, 'bloc': liste.bloc.nom, 'candidats': candidats, } listes_json.append(liste_json) file = DATA_DIR / "resultats" / "europeennes" / "2024" / "listes.json" if not file.parent.is_dir(): file.parent.mkdir(parents=True) with file.open('w') as f: json.dump(listes_json, f) def exporter_resultats_france(engine: Engine, verbose: bool = False) -> None: with Session(engine) as session: resultats_france = session.execute(select(ResultatsFranceEuropeennes2024)).scalar_one_or_none() if not resultats_france: resultats_france = ResultatsFranceEuropeennes2024() session.add(resultats_france) resultats_dict = { 'zone': { 'type': "france", 'regions': [reg.code_insee for reg in session.execute(select(Region)).scalars().all()], 'departements': [dpt.code_insee for dpt in session.execute(select(Departement)).scalars().all()], 'circonscriptions': [circo.id for circo in session.execute(select(Circonscription)).scalars().all()], 'geometry': {}, }, 'tour1': { 'inscrits': resultats_france.inscrits, 'votants': resultats_france.votants, 'abstentions': resultats_france.abstentions, 'exprimes': resultats_france.exprimes, 'blancs': resultats_france.blancs, 'nuls': resultats_france.nuls, }, } resultats_listes = {} resultats_dict['tour1']['voix'] = resultats_listes for voix_liste in resultats_france.voix_listes: resultats_listes[voix_liste.liste.numero] = voix_liste.voix file = DATA_DIR / "resultats" / "europeennes" / "2024" / "france.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() 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 verbose else regions for region in regions_iterator: region_json = { 'type': "region", 'code_insee': region.code_insee, 'nom': region.libelle, 'departements': [dpt.code_insee for dpt in region.departements], 'circonscriptions': [ circo.id for circo in session.execute( select(Circonscription).join(Departement).filter_by(region_code=region.code_insee)) .scalars().all() ], 'geometry': region.geometry, } regions_json.append(region_json) resultats_region = session.execute(select(ResultatsRegionEuropeennes2024) .filter_by(region_id=region.code_insee)).scalar_one_or_none() if not resultats_region: resultats_france = session.execute(select(ResultatsFranceEuropeennes2024)).scalar_one() resultats_region = ResultatsRegionEuropeennes2024(region_id=region.code_insee, resultats_france_id=resultats_france.id) session.add(resultats_region) resultats_dict = { 'zone': region_json, 'tour1': { 'inscrits': resultats_region.inscrits, 'votants': resultats_region.votants, 'abstentions': resultats_region.abstentions, 'exprimes': resultats_region.exprimes, 'blancs': resultats_region.blancs, 'nuls': resultats_region.nuls, }, } resultats_listes = {} resultats_dict['tour1']['voix'] = resultats_listes for voix_liste in resultats_region.voix_listes: resultats_listes[voix_liste.liste.numero] = voix_liste.voix file = DATA_DIR / "resultats" / "europeennes" / "2024" / "region" / f"{region.code_insee}.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() regions_file = DATA_DIR / "resultats" / "europeennes" / "2024" / "region" / "regions.json" if not regions_file.parent.is_dir(): regions_file.parent.mkdir(parents=True) with regions_file.open('w') as f: json.dump(regions_json, f) 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 verbose else departements for departement in iterator: departement_json = { 'type': "departement", 'code_insee': departement.code_insee, 'nom': departement.libelle, 'region': departement.region_code, 'circonscriptions': [circo.id for circo in departement.circonscriptions], 'communes': [commune.code_insee for commune in departement.communes], 'geometry': departement.geometry, } departements_json.append(departement_json) resultats_departement = session.execute( select(ResultatsDepartementEuropeennes2024) .filter_by(dpt_id=departement.code_insee)).scalar_one_or_none() if not resultats_departement: resultats_region = session.execute(select(ResultatsRegionEuropeennes2024) .filter_by(region_id=departement.region_code)).scalar_one() resultats_departement = ResultatsDepartementEuropeennes2024(dpt_id=departement.code_insee, resultats_region_id=resultats_region.id) session.add(resultats_departement) resultats_dict = { 'zone': departement_json, 'tour1': { 'inscrits': resultats_departement.inscrits, 'votants': resultats_departement.votants, 'abstentions': resultats_departement.abstentions, 'exprimes': resultats_departement.exprimes, 'blancs': resultats_departement.blancs, 'nuls': resultats_departement.nuls, }, } resultats_listes = {} resultats_dict['tour1']['voix'] = resultats_listes for voix_liste in resultats_departement.voix_listes: resultats_listes[voix_liste.liste.numero] = voix_liste.voix file = DATA_DIR / "resultats" / "europeennes" / "2024" / "departement" / f"{departement.code_insee}.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() departements_file = DATA_DIR / "resultats" / "europeennes" / "2024" / "departement" / "departements.json" if not departements_file.parent.is_dir(): departements_file.parent.mkdir(parents=True) with departements_file.open('w') as f: 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 = { 'type': "circonscription", 'id': circonscription.id, 'departement': circonscription.departement_code, 'numero': circonscription.numero, 'bureaux_vote': [bv.id for bv in circonscription.bureaux_vote], 'geometry': circonscription.geometry, } circonscriptions_json.append(circonscription_json) resultats_circonscription = session.execute( select(ResultatsCirconscriptionEuropeennes2024) .filter_by(circo_id=circonscription.id)).scalar_one_or_none() if not resultats_circonscription: resultats_departement = session.execute(select(ResultatsDepartementEuropeennes2024) .filter_by(dpt_id=circonscription.departement_code)) \ .scalar_one() resultats_circonscription = ResultatsCirconscriptionEuropeennes2024( circo_id=circonscription.id, resultats_departement_id=resultats_departement.id) session.add(resultats_circonscription) resultats_dict = { 'zone': circonscription_json, 'tour1': { 'inscrits': resultats_circonscription.inscrits, 'votants': resultats_circonscription.votants, 'abstentions': resultats_circonscription.abstentions, 'exprimes': resultats_circonscription.exprimes, 'blancs': resultats_circonscription.blancs, 'nuls': resultats_circonscription.nuls, }, } resultats_listes = {} resultats_dict['tour1']['voix'] = resultats_listes for voix_liste in resultats_circonscription.voix_listes: resultats_listes[voix_liste.liste.numero] = voix_liste.voix file = DATA_DIR / "resultats" / "europeennes" / "2024" / "circonscription" / 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" / "europeennes" / "2024" / "circonscription" / "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() communes_json = [] iterator = tqdm(communes, desc="Communes") if verbose else communes for commune in iterator: commune_json = { 'type': "commune", 'code_insee': commune.code_insee, 'nom': commune.libelle, 'departement': commune.departement_code, 'bureaux_vote': [bv.id for bv in commune.bureaux_vote], 'geometry': commune.geometry, } communes_json.append(commune_json) resultats_commune = session.execute( select(ResultatsCommuneEuropeennes2024).filter_by(commune_id=commune.code_insee)).scalar_one_or_none() if not resultats_commune: resultats_departement = session.execute(select(ResultatsDepartementEuropeennes2024) .filter_by(dpt_id=commune.departement_code)).scalar_one() resultats_commune = ResultatsCommuneEuropeennes2024(commune_id=commune.code_insee, resultats_dpt_id=resultats_departement.id) session.add(resultats_commune) resultats_dict = { 'zone': commune_json, 'tour1': { 'inscrits': resultats_commune.inscrits, 'votants': resultats_commune.votants, 'abstentions': resultats_commune.abstentions, 'exprimes': resultats_commune.exprimes, 'blancs': resultats_commune.blancs, 'nuls': resultats_commune.nuls, }, } resultats_listes = {} resultats_dict['tour1']['voix'] = resultats_listes for voix_liste in resultats_commune.voix_listes: resultats_listes[voix_liste.liste.numero] = voix_liste.voix file = DATA_DIR / "resultats" / "europeennes" / "2024" / "commune" / f"{commune.code_insee}.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() communes_file = DATA_DIR / "resultats" / "europeennes" / "2024" / "commune" / "communes.json" if not communes_file.parent.is_dir(): communes_file.parent.mkdir(parents=True) with communes_file.open('w') as f: 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 = { 'type': "bureau_vote", 'id': bureau_vote.id, 'libelle': bureau_vote.libelle, 'commune': bureau_vote.commune_code, 'circonscription': bureau_vote.circo_code, 'geometry': bureau_vote.geometry, } bureaux_vote_json.append(bureau_vote_json) resultats_bureau_vote = session.execute( select(ResultatsBureauVoteEuropeennes2024).filter_by(bv_id=bureau_vote.id)).scalar_one_or_none() if not resultats_bureau_vote: resultats_commune = session.execute(select(ResultatsCommuneEuropeennes2024) .filter_by(commune_id=bureau_vote.commune_code)).scalar_one() resultats_bureau_vote = ResultatsBureauVoteEuropeennes2024(bv_id=bureau_vote.id, resultats_commune_id=resultats_commune.id) session.add(resultats_bureau_vote) resultats_dict = { 'zone': bureau_vote_json, 'tour1': { '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, }, } resultats_listes = {} resultats_dict['tour1']['voix'] = resultats_listes for voix_liste in resultats_bureau_vote.voix_listes: resultats_listes[voix_liste.liste.numero] = voix_liste.voix file = DATA_DIR / "resultats" / "europeennes" / "2024" / "bureau_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" / "europeennes" / "2024" / "bureau_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) exporter_resultats_circonscriptions(engine, verbose) exporter_resultats_communes(engine, verbose) exporter_resultats_bureaux_vote(engine, verbose)