nupes-elections/nupes/scripts/legislatives2022/export_resultats.py

495 lines
22 KiB
Python

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 BureauVote
from nupes.models.legislatives2022 import *
DATA_DIR = Path(__file__).parent.parent.parent.parent / 'data'
def exporter_nuances(engine: Engine, verbose: bool = False) -> None:
with Session(engine) as session:
blocs = session.execute(select(BlocLegislatives2022)).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" / "legislatives" / "2022" / "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(NuanceLegislatives2022)).scalars().all()
nuances_json = []
for nuance in nuances:
nuance_json = {
'code': nuance.code,
'nom': nuance.nom,
'couleur': nuance.couleur,
"bloc": nuance.bloc.nom,
}
nuances_json.append(nuance_json)
file = DATA_DIR / "resultats" / "legislatives" / "2022" / "nuances.json"
if not file.parent.is_dir():
file.parent.mkdir(parents=True)
with file.open('w') as f:
json.dump(nuances_json, f)
def exporter_candidats(engine: Engine, verbose: bool = False) -> None:
with Session(engine) as session:
circonscriptions = session.execute(select(Circonscription)).scalars().all()
for circonscription in tqdm(circonscriptions, desc="Candidat⋅es", disable=not verbose):
candidats = session.execute(select(CandidatLegislatives2022)
.filter_by(circonscription_id=circonscription.id)).scalars().all()
candidats_json = []
for candidat in candidats:
candidat_json = {
'numero': candidat.numero,
'nuance': candidat.nuance_id,
'bloc': candidat.nuance.bloc.nom,
'nom': candidat.nom,
'prenom': candidat.prenom,
'nom_suppleance': candidat.nom_suppleance,
'prenom_suppleance': candidat.prenom_suppleance,
}
candidats_json.append(candidat_json)
file = DATA_DIR / "resultats" / "legislatives" / "2022" / "candidats" / f"{circonscription.id}.json"
if not file.parent.is_dir():
file.parent.mkdir(parents=True)
with file.open('w') as f:
json.dump(candidats_json, f)
def exporter_resultats_france(engine: Engine, verbose: bool = False) -> None:
with Session(engine) as session:
resultats_france = session.execute(select(ResultatsFranceLegislatives2022)).scalar_one_or_none()
if not resultats_france:
resultats_france = ResultatsFranceLegislatives2022()
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_t1,
'votants': resultats_france.votants_t1,
'abstentions': resultats_france.abstentions_t1,
'exprimes': resultats_france.exprimes_t1,
'blancs': resultats_france.blancs_t1,
'nuls': resultats_france.nuls_t1,
},
'tour2': {
'inscrits': resultats_france.inscrits_t2,
'votants': resultats_france.votants_t2,
'abstentions': resultats_france.abstentions_t2,
'exprimes': resultats_france.exprimes_t2,
'blancs': resultats_france.blancs_t2,
'nuls': resultats_france.nuls_t2,
},
}
resultats_t1, resultats_t2 = {}, {}
resultats_dict['tour1']['voix'] = resultats_t1
resultats_dict['tour2']['voix'] = resultats_t2
for voix_nuance in resultats_france.voix:
resultats_t1[voix_nuance.nuance_id] = voix_nuance.voix_t1
resultats_t2[voix_nuance.nuance_id] = voix_nuance.voix_t2
file = DATA_DIR / "resultats" / "legislatives" / "2022" / "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(ResultatsRegionLegislatives2022)
.filter_by(region_id=region.code_insee)).scalar_one_or_none()
if not resultats_region:
resultats_france = session.execute(select(ResultatsFranceLegislatives2022)).scalar_one()
resultats_region = ResultatsRegionLegislatives2022(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_t1,
'votants': resultats_region.votants_t1,
'abstentions': resultats_region.abstentions_t1,
'exprimes': resultats_region.exprimes_t1,
'blancs': resultats_region.blancs_t1,
'nuls': resultats_region.nuls_t1,
},
'tour2': {
'inscrits': resultats_region.inscrits_t2,
'votants': resultats_region.votants_t2,
'abstentions': resultats_region.abstentions_t2,
'exprimes': resultats_region.exprimes_t2,
'blancs': resultats_region.blancs_t2,
'nuls': resultats_region.nuls_t2,
},
}
resultats_t1, resultats_t2 = {}, {}
resultats_dict['tour1']['voix'] = resultats_t1
resultats_dict['tour2']['voix'] = resultats_t2
for voix_nuance in resultats_region.voix:
resultats_t1[voix_nuance.nuance_id] = voix_nuance.voix_t1
resultats_t2[voix_nuance.nuance_id] = voix_nuance.voix_t2
file = DATA_DIR / "resultats" / "legislatives" / "2022" / "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" / "legislatives" / "2022" / "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(ResultatsDepartementLegislatives2022)
.filter_by(dpt_id=departement.code_insee)).scalar_one_or_none()
if not resultats_departement:
resultats_region = session.execute(select(ResultatsRegionLegislatives2022)
.filter_by(region_id=departement.region_code)).scalar_one()
resultats_departement = ResultatsDepartementLegislatives2022(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_t1,
'votants': resultats_departement.votants_t1,
'abstentions': resultats_departement.abstentions_t1,
'exprimes': resultats_departement.exprimes_t1,
'blancs': resultats_departement.blancs_t1,
'nuls': resultats_departement.nuls_t1,
},
'tour2': {
'inscrits': resultats_departement.inscrits_t2,
'votants': resultats_departement.votants_t2,
'abstentions': resultats_departement.abstentions_t2,
'exprimes': resultats_departement.exprimes_t2,
'blancs': resultats_departement.blancs_t2,
'nuls': resultats_departement.nuls_t2,
},
}
resultats_t1, resultats_t2 = {}, {}
resultats_dict['tour1']['voix'] = resultats_t1
resultats_dict['tour2']['voix'] = resultats_t2
for voix_nuance in resultats_departement.voix:
resultats_t1[voix_nuance.nuance_id] = voix_nuance.voix_t1
resultats_t2[voix_nuance.nuance_id] = voix_nuance.voix_t2
file = DATA_DIR / "resultats" / "legislatives" / "2022" / "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" / "legislatives" / "2022" / "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(ResultatsCirconscriptionLegislatives2022)
.filter_by(circo_id=circonscription.id)).scalar_one_or_none()
if not resultats_circonscription:
resultats_departement = session.execute(select(ResultatsDepartementLegislatives2022)
.filter_by(dpt_id=circonscription.departement_code)) \
.scalar_one()
resultats_circonscription = ResultatsCirconscriptionLegislatives2022(
circo_id=circonscription.id, resultats_departement_id=resultats_departement.id)
session.add(resultats_circonscription)
resultats_dict = {
'zone': circonscription_json,
'tour1': {
'inscrits': resultats_circonscription.inscrits_t1,
'votants': resultats_circonscription.votants_t1,
'abstentions': resultats_circonscription.abstentions_t1,
'exprimes': resultats_circonscription.exprimes_t1,
'blancs': resultats_circonscription.blancs_t1,
'nuls': resultats_circonscription.nuls_t1,
},
'tour2': {
'inscrits': resultats_circonscription.inscrits_t2,
'votants': resultats_circonscription.votants_t2,
'abstentions': resultats_circonscription.abstentions_t2,
'exprimes': resultats_circonscription.exprimes_t2,
'blancs': resultats_circonscription.blancs_t2,
'nuls': resultats_circonscription.nuls_t2,
},
}
resultats_t1, resultats_t2 = {}, {}
resultats_dict['tour1']['voix'] = resultats_t1
resultats_dict['tour2']['voix'] = resultats_t2
for voix_candidat in resultats_circonscription.voix:
resultats_t1[voix_candidat.candidat.numero] = voix_candidat.voix_t1
resultats_t2[voix_candidat.candidat.numero] = voix_candidat.voix_t2
file = DATA_DIR / "resultats" / "legislatives" / "2022" / "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" / "legislatives" / "2022"
/ "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(ResultatsCommuneLegislatives2022).filter_by(commune_id=commune.code_insee)).scalar_one_or_none()
if not resultats_commune:
resultats_departement = session.execute(select(ResultatsDepartementLegislatives2022)
.filter_by(dpt_id=commune.departement_code)).scalar_one()
resultats_commune = ResultatsCommuneLegislatives2022(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_t1,
'votants': resultats_commune.votants_t1,
'abstentions': resultats_commune.abstentions_t1,
'exprimes': resultats_commune.exprimes_t1,
'blancs': resultats_commune.blancs_t1,
'nuls': resultats_commune.nuls_t1,
},
'tour2': {
'inscrits': resultats_commune.inscrits_t2,
'votants': resultats_commune.votants_t2,
'abstentions': resultats_commune.abstentions_t2,
'exprimes': resultats_commune.exprimes_t2,
'blancs': resultats_commune.blancs_t2,
'nuls': resultats_commune.nuls_t2,
},
}
resultats_t1, resultats_t2 = {}, {}
resultats_dict['tour1']['voix'] = resultats_t1
resultats_dict['tour2']['voix'] = resultats_t2
for voix_nuance in resultats_commune.voix:
resultats_t1[voix_nuance.nuance_id] = voix_nuance.voix_t1
resultats_t2[voix_nuance.nuance_id] = voix_nuance.voix_t2
file = DATA_DIR / "resultats" / "legislatives" / "2022" / "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" / "legislatives" / "2022" / "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(ResultatsBureauVoteLegislatives2022).filter_by(bv_id=bureau_vote.id)).scalar_one_or_none()
if not resultats_bureau_vote:
resultats_commune = session.execute(select(ResultatsCommuneLegislatives2022)
.filter_by(commune_id=bureau_vote.commune_code)).scalar_one()
resultats_bureau_vote = ResultatsBureauVoteLegislatives2022(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_t1,
'votants': resultats_bureau_vote.votants_t1,
'abstentions': resultats_bureau_vote.abstentions_t1,
'exprimes': resultats_bureau_vote.exprimes_t1,
'blancs': resultats_bureau_vote.blancs_t1,
'nuls': resultats_bureau_vote.nuls_t1,
},
'tour2': {
'inscrits': resultats_bureau_vote.inscrits_t2,
'votants': resultats_bureau_vote.votants_t2,
'abstentions': resultats_bureau_vote.abstentions_t2,
'exprimes': resultats_bureau_vote.exprimes_t2,
'blancs': resultats_bureau_vote.blancs_t2,
'nuls': resultats_bureau_vote.nuls_t2,
},
}
resultats_t1, resultats_t2 = {}, {}
resultats_dict['tour1']['voix'] = resultats_t1
resultats_dict['tour2']['voix'] = resultats_t2
for voix_candidat in resultats_bureau_vote.voix:
resultats_t1[voix_candidat.candidat.numero] = voix_candidat.voix_t1
resultats_t2[voix_candidat.candidat.numero] = voix_candidat.voix_t2
file = DATA_DIR / "resultats" / "legislatives" / "2022" / "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" / "legislatives" / "2022" / "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_nuances(engine, verbose)
exporter_candidats(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)