Export données par bureau de vote

This commit is contained in:
Emmy D'Anello 2024-06-15 20:28:23 +02:00
parent 032d589a34
commit ffc54be4fa
Signed by: ynerant
GPG Key ID: 3A75C55819C8CF85
1 changed files with 112 additions and 4 deletions

View File

@ -5,9 +5,10 @@ from sqlalchemy import Engine, select
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
from tqdm import tqdm 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, \ 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' 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) 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: def exporter_resultats_communes(engine: Engine, verbose: bool = False) -> None:
with Session(engine) as session: with Session(engine) as session:
communes = session.execute(select(Commune)).scalars().all() 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) 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: def run(engine: Engine, verbose: bool = False) -> None:
exporter_listes(engine, verbose) exporter_listes(engine, verbose)
exporter_resultats_france(engine, verbose) exporter_resultats_france(engine, verbose)
exporter_resultats_regions(engine, verbose) exporter_resultats_regions(engine, verbose)
exporter_resultats_departements(engine, verbose) exporter_resultats_departements(engine, verbose)
# FIXME Les communes prennent trop de temps exporter_resultats_circonscriptions(engine, verbose)
# exporter_resultats_communes(engine, verbose) exporter_resultats_communes(engine, verbose)
exporter_resultats_bureaux_vote(engine, verbose)