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)