Export données législatives 2024
This commit is contained in:
parent
7aa92d28d4
commit
df8c8db32d
GPG Key ID:
3A75C55819C8CF85
|
|
@ -5,7 +5,7 @@ from sqlalchemy import Engine, select
|
|||
from sqlalchemy.orm import Session
|
||||
from tqdm import tqdm
|
||||
|
||||
from nupes.models.geographie import Region, Departement, Commune, Circonscription, BureauVote
|
||||
from nupes.models.geographie import Circonscription, BureauVote
|
||||
from nupes.models.europeennes2024 import *
|
||||
|
||||
DATA_DIR = Path(__file__).parent.parent.parent / 'data'
|
||||
|
|
@ -76,17 +76,19 @@ def exporter_resultats_france(engine: Engine, verbose: bool = False) -> None:
|
|||
'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()],
|
||||
},
|
||||
"inscrits": resultats_france.inscrits,
|
||||
"votants": resultats_france.votants,
|
||||
"abstentions": resultats_france.abstentions,
|
||||
"exprimes": resultats_france.exprimes,
|
||||
"blancs": resultats_france.blancs,
|
||||
"nuls": resultats_france.nuls,
|
||||
"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,
|
||||
},
|
||||
'geometry': {},
|
||||
}
|
||||
|
||||
resultats_listes = {}
|
||||
resultats_dict['voix'] = 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
|
||||
|
||||
|
|
@ -124,18 +126,20 @@ def exporter_resultats_regions(engine: Engine, verbose: bool = False) -> None:
|
|||
session.add(resultats_region)
|
||||
|
||||
resultats_dict = {
|
||||
"inscrits": resultats_region.inscrits,
|
||||
"votants": resultats_region.votants,
|
||||
"abstentions": resultats_region.abstentions,
|
||||
"exprimes": resultats_region.exprimes,
|
||||
"blancs": resultats_region.blancs,
|
||||
"nuls": resultats_region.nuls,
|
||||
'region': 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,
|
||||
},
|
||||
'geometry': region.geometry,
|
||||
}
|
||||
|
||||
resultats_listes = {}
|
||||
resultats_dict['voix'] = 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
|
||||
|
||||
|
|
@ -180,18 +184,20 @@ def exporter_resultats_departements(engine: Engine, verbose: bool = False) -> No
|
|||
session.add(resultats_departement)
|
||||
|
||||
resultats_dict = {
|
||||
"inscrits": resultats_departement.inscrits,
|
||||
"votants": resultats_departement.votants,
|
||||
"abstentions": resultats_departement.abstentions,
|
||||
"exprimes": resultats_departement.exprimes,
|
||||
"blancs": resultats_departement.blancs,
|
||||
"nuls": resultats_departement.nuls,
|
||||
'departement': 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,
|
||||
},
|
||||
'geometry': departement.geometry,
|
||||
}
|
||||
|
||||
resultats_listes = {}
|
||||
resultats_dict['voix'] = 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
|
||||
|
||||
|
|
@ -236,18 +242,20 @@ def exporter_resultats_circonscriptions(engine: Engine, verbose: bool = False) -
|
|||
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,
|
||||
'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,
|
||||
},
|
||||
'geometry': circonscription.geometry,
|
||||
}
|
||||
|
||||
resultats_listes = {}
|
||||
resultats_dict['voix'] = 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
|
||||
|
||||
|
|
@ -291,18 +299,20 @@ def exporter_resultats_communes(engine: Engine, verbose: bool = False) -> None:
|
|||
session.add(resultats_commune)
|
||||
|
||||
resultats_dict = {
|
||||
"inscrits": resultats_commune.inscrits,
|
||||
"votants": resultats_commune.votants,
|
||||
"abstentions": resultats_commune.abstentions,
|
||||
"exprimes": resultats_commune.exprimes,
|
||||
"blancs": resultats_commune.blancs,
|
||||
"nuls": resultats_commune.nuls,
|
||||
'commune': 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,
|
||||
},
|
||||
'geometry': commune.geometry,
|
||||
}
|
||||
|
||||
resultats_listes = {}
|
||||
resultats_dict['voix'] = 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
|
||||
|
||||
|
|
@ -346,18 +356,20 @@ def exporter_resultats_bureaux_vote(engine: Engine, verbose: bool = False) -> No
|
|||
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,
|
||||
'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,
|
||||
},
|
||||
'geometry': bureau_vote.geometry,
|
||||
}
|
||||
|
||||
resultats_listes = {}
|
||||
resultats_dict['voix'] = 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
|
||||
|
||||
|
|
|
|||
|
|
@ -1 +1 @@
|
|||
from . import import_candidats, import_resultats
|
||||
from . import export_resultats, import_candidats, import_resultats
|
||||
|
|
|
|||
|
|
@ -0,0 +1,467 @@
|
|||
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 / '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_id}
|
||||
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,
|
||||
'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 = {
|
||||
'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()],
|
||||
},
|
||||
'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,
|
||||
},
|
||||
'geometry': {},
|
||||
}
|
||||
|
||||
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 = {'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()]}
|
||||
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 = {
|
||||
'region': 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,
|
||||
},
|
||||
'geometry': region.geometry,
|
||||
}
|
||||
|
||||
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 = {'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]}
|
||||
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 = {
|
||||
'departement': 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,
|
||||
},
|
||||
'geometry': departement.geometry,
|
||||
}
|
||||
|
||||
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 = {'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(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 = {
|
||||
'circonscription': 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,
|
||||
},
|
||||
'geometry': circonscription.geometry,
|
||||
}
|
||||
|
||||
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.id] = voix_candidat.voix_t1
|
||||
resultats_t2[voix_candidat.candidat.id] = 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 = {'code_insee': commune.code_insee, 'nom': commune.libelle,
|
||||
'departement': commune.departement_code,
|
||||
'bureaux_vote': [bv.id for bv in commune.bureaux_vote]}
|
||||
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 = {
|
||||
'commune': 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,
|
||||
},
|
||||
'geometry': commune.geometry,
|
||||
}
|
||||
|
||||
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 = {'id': bureau_vote.id,
|
||||
'libelle': bureau_vote.libelle,
|
||||
'commune': bureau_vote.commune_code,
|
||||
'circonscription': bureau_vote.circo_code}
|
||||
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 = {
|
||||
'bureau_vote': 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,
|
||||
},
|
||||
'geometry': bureau_vote.geometry,
|
||||
}
|
||||
|
||||
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.id] = voix_candidat.voix_t1
|
||||
resultats_t2[voix_candidat.candidat.id] = 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)
|
||||
|
|
@ -297,7 +297,8 @@ def importer_resultats_commune(engine: Engine, verbose: bool = False) -> None:
|
|||
.scalar_one_or_none()
|
||||
if not voix_nuance_commune:
|
||||
voix_nuance_commune = VoixCommuneLegislatives2022(resultats_commune_id=resultats_commune.id,
|
||||
nuance_id=nuance_code)
|
||||
nuance_id=nuance_code,
|
||||
voix_t1=0, voix_t2=0)
|
||||
session.add(voix_nuance_commune)
|
||||
|
||||
if tour == 1:
|
||||
|
|
@ -305,7 +306,7 @@ def importer_resultats_commune(engine: Engine, verbose: bool = False) -> None:
|
|||
elif tour == 2:
|
||||
voix_nuance_commune.voix_t2 += voix
|
||||
|
||||
session.commit()
|
||||
session.commit()
|
||||
|
||||
|
||||
def importer_resultats_circo(engine: Engine, verbose: bool = False) -> None:
|
||||
|
|
|
|||
Loading…
Reference in New Issue