ynerant/nupes-elections
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Déplacement des scripts

Browse Source
This commit is contained in:
Emmy D'Anello
2024-06-18 14:37:06 +02:00
parent ac0b5e5262
commit 66b8fb8103
6 changed files with 15 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
nupes/scripts/europeennes2024/__init__.py Normal file
View File

@ -0,0 +1 @@
from . import export_resultats, import_candidats, import_resultats, import_resultats_web

388
nupes/scripts/europeennes2024/export_resultats.py Normal file
View File

@ -0,0 +1,388 @@
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 Region, Departement, Commune, Circonscription, BureauVote
from nupes.models.europeennes2024 import *
DATA_DIR = Path(__file__).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 = {
'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()],
},
"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
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 = {'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(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 = {
"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,
'geometry': region.geometry,
}
resultats_listes = {}
resultats_dict['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 = {'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(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 = {
"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,
'geometry': departement.geometry,
}
resultats_listes = {}
resultats_dict['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 = {'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(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 = {
"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'] = 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 = {'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(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 = {
"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,
'geometry': commune.geometry,
}
resultats_listes = {}
resultats_dict['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 = {'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(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 = {
"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'] = 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)

177
nupes/scripts/europeennes2024/import_candidats.py Normal file
View File

@ -0,0 +1,177 @@
import csv
import datetime
from sqlalchemy import Engine, select
from sqlalchemy.orm import Session
from nupes.cache import get_file
from nupes.models.europeennes2024 import BlocEuropeennes2024, NuanceEuropeennes2024, ListeEuropeennes2024, \
CandidatEuropeennes2024
def creer_blocs(engine: Engine, verbose: bool = False) -> None:
blocs = [
{"id": 1, "nom": "Gauche", "couleur": "#BB1840"},
{"id": 2, "nom": "Droite libérale", "couleur": "#FFEB00"},
{"id": 3, "nom": "Droite conservatrice", "couleur": "#0066CC"},
{"id": 4, "nom": "Extrême droite", "couleur": "#404040"},
{"id": 5, "nom": "Autres", "couleur": "#DCBFA3"}
]
with Session(engine) as session:
for bloc_dict in blocs:
if bloc := session.execute(select(BlocEuropeennes2024).filter_by(id=bloc_dict["id"])).scalar_one_or_none():
bloc.nom = bloc_dict["nom"]
bloc.couleur = bloc_dict["couleur"]
else:
session.add(BlocEuropeennes2024(**bloc_dict))
session.commit()
def creer_nuances(engine: Engine, verbose: bool = False) -> None:
nuances = [
{"code": "LEXG", "nom": "Liste d'extrême gauche", "couleur": "#BB0000"},
{"code": "LCOM", "nom": "Liste du Parti communiste français", "couleur": "#DD0000"},
{"code": "LFI", "nom": "Liste de La France insoumise", "couleur": "#CC2443"},
{"code": "LUG", "nom": "Liste d'union de la gauche", "couleur": "#FF8080"},
{"code": "LVEC", "nom": "Liste Les Écologistes", "couleur": "#00C000"},
{"code": "LDVG", "nom": "Liste divers gauche", "couleur": "#FFC0C0"},
{"code": "LECO", "nom": "Liste écologiste", "couleur": "#77FF77"},
{"code": "LDIV", "nom": "Liste Divers", "couleur": "#DCDCDC"},
{"code": "LENS", "nom": "Liste Ensemble", "couleur": "#FFEB00"},
{"code": "LLR", "nom": "Liste des Républicains", "couleur": "#0066CC"},
{"code": "LDVD", "nom": "Liste divers droite", "couleur": "#26C4EC"},
{"code": "LRN", "nom": "Liste du Rassemblement national", "couleur": "#0D378A"},
{"code": "LREC", "nom": "Liste Reconquête !", "couleur": "#404040"},
{"code": "LEXD", "nom": "Liste d'extrême droite", "couleur": "#404040"},
]
with Session(engine) as session:
for nuance_dict in nuances:
if nuance := session.execute(select(NuanceEuropeennes2024)
.filter_by(code=nuance_dict["code"])).scalar_one_or_none():
nuance.nom = nuance_dict["nom"]
nuance.couleur = nuance_dict["couleur"]
else:
session.add(NuanceEuropeennes2024(**nuance_dict))
session.commit()
def importer_listes(engine: Engine, verbose: bool = False) -> None:
listes = [
{"numero": 1, "nom": "Pour une humanité souveraine", "nuance_id": "LDIV", "bloc_id": 5},
{"numero": 2, "nom": "Pour une démocratie réelle : Décidons nous-mêmes !",
"nuance_id": "LDIV", "bloc_id": 5},
{"numero": 3, "nom": "La France fière, menée par Marion Maréchal et soutenue par Éric Zemmour",
"nuance_id": "LREC", "bloc_id": 4},
{"numero": 4, "nom": "La France insoumise - Union populaire", "nuance_id": "LFI", "bloc_id": 1},
{"numero": 5, "nom": "La France revient ! Avec Jordan Bardella et Marine Le Pen",
"nuance_id": "LRN", "bloc_id": 4},
{"numero": 6, "nom": "Europe Écologie", "nuance_id": "LVEC", "bloc_id": 1},
{"numero": 7, "nom": "Free Palestine", "nuance_id": "LDIV", "bloc_id": 1},
{"numero": 8, "nom": "Parti animaliste - Les animaux comptent, votre voix aussi",
"nuance_id": "LDIV", "bloc_id": 5},
{"numero": 9, "nom": "Parti révolutionnaire Communistes", "nuance_id": "LEXG", "bloc_id": 1},
{"numero": 10, "nom": "Parti pirate", "nuance_id": "LDIV", "bloc_id": 1},
{"numero": 11, "nom": "Besoin d'Europe", "nuance_id": "LENS", "bloc_id": 2},
{"numero": 12, "nom": "PACE - Parti des citoyens européens, pour l'armée européenne, "
"pour l'Europe sociale, pour la planète !", "nuance_id": "LDIV", "bloc_id": 1},
{"numero": 13, "nom": "Équinoxe : écologie pratique et renouveau démocratique",
"nuance_id": "LECO", "bloc_id": 5},
{"numero": 14, "nom": "Écologie positive et territoires", "nuance_id": "LECO", "bloc_id": 5},
{"numero": 15, "nom": "Liste Asselineau-Frexit, pour le pouvoir d'achat et pour la paix",
"nuance_id": "LDIV", "bloc_id": 4},
{"numero": 16, "nom": "Paix et Décroissance", "nuance_id": "LEXG", "bloc_id": 1},
{"numero": 17, "nom": "Pour une autre Europe", "nuance_id": "LDIV", "bloc_id": 5},
{"numero": 18, "nom": "La droite pour faire entendre la voix de la France en Europe",
"nuance_id": "LLR", "bloc_id": 3},
{"numero": 19, "nom": "Lutte ouvrière le camp des travailleurs", "nuance_id": "LEXG", "bloc_id": 1},
{"numero": 20, "nom": "Changer l'Europe", "nuance_id": "LDVG", "bloc_id": 1},
{"numero": 21, "nom": "Nous le peuple", "nuance_id": "LDIV", "bloc_id": 5},
{"numero": 22, "nom": "Pour un monde sans frontières ni patrons, urgence révolution !",
"nuance_id": "LEXG", "bloc_id": 1},
{"numero": 23, "nom": "\"Pour le pain, la paix, la liberté !\" présentée par le Parti des travailleurs",
"nuance_id": "LEXG", "bloc_id": 1},
{"numero": 24, "nom": "L'Europe ça suffit !", "nuance_id": "LEXD", "bloc_id": 4},
{"numero": 25, "nom": "Non ! Prenons-nous en mains", "nuance_id": "LDIV", "bloc_id": 4},
{"numero": 26, "nom": "Forteresse Europe — Liste d'unité nationaliste", "nuance_id": "LEXD", "bloc_id": 4},
{"numero": 27, "nom": "Réveiller l'Europe", "nuance_id": "LUG", "bloc_id": 1},
{"numero": 28, "nom": "Non à l'UE et à l'OTAN, communistes pour la paix et le progrès social",
"nuance_id": "LEXG", "bloc_id": 1},
{"numero": 29, "nom": "Alliance rurale", "nuance_id": "LDVD", "bloc_id": 3},
{"numero": 30, "nom": "France libre", "nuance_id": "LDIV", "bloc_id": 4},
{"numero": 31, "nom": "Europe Territoires Écologie", "nuance_id": "LDVG", "bloc_id": 1},
{"numero": 32, "nom": "La ruche citoyenne", "nuance_id": "LDIV", "bloc_id": 5},
{"numero": 33, "nom": "Gauche unie pour le monde du travail soutenue par Fabien Roussel",
"nuance_id": "LCOM", "bloc_id": 1},
{"numero": 34, "nom": "Défendre les enfants", "nuance_id": "LDIV", "bloc_id": 5},
{"numero": 35, "nom": "Écologie au centre", "nuance_id": "LECO", "bloc_id": 2},
{"numero": 36, "nom": "Démocratie représentative", "nuance_id": "LDIV", "bloc_id": 1},
{"numero": 37, "nom": "Espéranto langue commune", "nuance_id": "LDIV", "bloc_id": 5},
{"numero": 38, "nom": "Liberté démocratique française", "nuance_id": "LDIV", "bloc_id": 4},
]
with Session(engine) as session:
for liste_dict in listes:
if liste := session.execute(select(ListeEuropeennes2024)
.filter_by(numero=liste_dict["numero"])).scalar_one_or_none():
liste.nom = liste_dict["nom"]
liste.nuance_id = liste_dict["nuance_id"]
liste.bloc_id = liste_dict["bloc_id"]
else:
liste = ListeEuropeennes2024(numero=liste_dict['numero'], nom=liste_dict['nom'],
nom_majuscules=liste_dict['nom'].upper(),
nuance_id=liste_dict['nuance_id'], bloc_id=liste_dict['bloc_id'])
session.add(liste)
session.commit()
def importer_candidats(engine: Engine, verbose: bool = False) -> None:
DATASET_URL = "https://www.data.gouv.fr/fr/datasets/r/483cd4bd-1b0e-4b52-a923-aadadf9c8f62"
file = get_file(DATASET_URL, "candidats_2024.csv")
with file.open('r') as f:
with Session(engine) as session:
reader = csv.DictReader(f, delimiter=';')
for line in reader:
line: dict[str, str]
numero_panneau = line.get("Numéro de panneau")
liste = session.execute(select(ListeEuropeennes2024).filter_by(numero=numero_panneau)).scalar_one()
liste.nom_majuscules = line['Libellé de la liste']
if candidat := session.execute(select(CandidatEuropeennes2024)
.filter_by(liste_id=liste.id, ordre=line['Ordre'])) \
.scalar_one_or_none():
candidat.liste_id = liste.id
candidat.nom = line['Nom sur le bulletin de vote']
candidat.prenom = line['Prénom sur le bulletin de vote']
candidat.sexe = CandidatEuropeennes2024.Genre(line['Sexe'])
candidat.date_naissance = datetime.datetime.strptime(line['Date de naissance'], "%d/%m/%Y").date()
candidat.profession = line['Profession']
candidat.code_personnalite = CandidatEuropeennes2024.Personnalite(line['Code personnalité'])
candidat.sortant = line['Sortant'] == "OUI"
else:
candidat = CandidatEuropeennes2024(
liste_id=liste.id,
ordre=int(line['Ordre']),
nom=line['Nom sur le bulletin de vote'],
prenom=line['Prénom sur le bulletin de vote'],
sexe=CandidatEuropeennes2024.Genre(line['Sexe']).name,
date_naissance=datetime.datetime.strptime(line['Date de naissance'], "%d/%m/%Y").date(),
profession=line['Profession'],
code_personnalite=CandidatEuropeennes2024.Personnalite(line['Code personnalité']).name,
sortant=line['Sortant'] == "OUI",
)
session.add(candidat)
session.commit()
def run(engine: Engine, verbose: bool = False) -> None:
creer_blocs(engine, verbose)
creer_nuances(engine, verbose)
importer_listes(engine, verbose)
importer_candidats(engine, verbose)

418
nupes/scripts/europeennes2024/import_resultats.py Normal file
View File

@ -0,0 +1,418 @@
from csv import DictReader
from sqlalchemy import Engine, select
from sqlalchemy.orm import Session
from tqdm import tqdm
from nupes.cache import get_file
from nupes.models import BureauVote, Departement
from nupes.models.europeennes2024 import *
def importer_resultats_bv(engine: Engine, verbose: bool = False) -> None:
file = get_file("https://www.data.gouv.fr/fr/datasets/r/cc1883d9-1265-4365-b754-fb6aef22d82e",
"resultats-europeennes-par-bureau-de-vote.csv")
with file.open('r') as f:
reader = DictReader(f, delimiter=';')
with Session(engine) as session:
for row in tqdm(reader, desc="Bureau de vote", disable=not verbose):
com_code = str(row['Code commune']).zfill(5)
bv_code = row['Code BV']
if com_code == "60694":
com_code = "60054" # Les Hauts-Talicans, données pas à jour
elif com_code in ["85165", "85212"]:
# Communes nouvelles ayant défusionné en 2024
# TODO Gérer
continue
elif com_code == "98601":
# 3 royaumes à Wallis-et-Futuna, mais un seul résultat de commune
com_code = "98611"
if bv := session.execute(select(BureauVote).filter_by(commune_code=com_code, code_bureau=bv_code)) \
.scalar_one_or_none():
bv_id = bv.id
else:
print(f"Bureau de vote {com_code}_{bv_code} non trouvé")
continue
resultats_bv = session.execute(select(ResultatsBureauVoteEuropeennes2024)
.filter_by(bv_id=bv_id)).scalar_one_or_none()
if not resultats_bv:
resultats_commune = session.execute(select(ResultatsCommuneEuropeennes2024)
.filter_by(commune_id=com_code)).scalar_one_or_none()
if not resultats_commune:
resultats_dpt = session.execute(select(ResultatsDepartementEuropeennes2024)
.filter_by(dpt_id=row['Code département'])) \
.scalar_one_or_none()
if not resultats_dpt:
dpt = session.execute(select(Departement)
.filter_by(code_insee=f"{row['Code département'].zfill(2)}")) \
.scalar_one()
resultats_reg = session.execute(select(ResultatsRegionEuropeennes2024)
.filter_by(region_id=dpt.region_code)).scalar_one_or_none()
if not resultats_reg:
resultats_france = session.execute(select(ResultatsFranceEuropeennes2024)) \
.scalar_one_or_none()
if not resultats_france:
session.add(ResultatsFranceEuropeennes2024())
resultats_france = session.execute(select(ResultatsFranceEuropeennes2024)) \
.scalar_one()
resultats_reg = ResultatsRegionEuropeennes2024(region_id=str(dpt.region_code),
resultats_france_id=resultats_france.id)
session.add(resultats_reg)
resultats_reg = session.execute(select(ResultatsRegionEuropeennes2024)
.filter_by(region_id=dpt.region_code)).scalar_one()
resultats_dpt = ResultatsDepartementEuropeennes2024(dpt_id=row['Code département'],
resultats_region_id=resultats_reg.id)
session.add(resultats_dpt)
resultats_dpt = session.execute(select(ResultatsDepartementEuropeennes2024)
.filter_by(dpt_id=row['Code département'])).scalar_one()
resultats_commune = ResultatsCommuneEuropeennes2024(commune_id=com_code,
resultats_dpt_id=resultats_dpt.id)
session.add(resultats_commune)
resultats_commune = session.execute(select(ResultatsCommuneEuropeennes2024)
.filter_by(commune_id=com_code)).scalar_one()
resultats_bv = ResultatsBureauVoteEuropeennes2024(bv_id=bv_id,
resultats_commune_id=resultats_commune.id)
session.add(resultats_bv)
resultats_bv = session.execute(select(ResultatsBureauVoteEuropeennes2024)
.filter_by(bv_id=bv_id)).scalar_one()
resultats_bv.inscrits = int(row['Inscrits'])
resultats_bv.votants = int(row['Votants'])
resultats_bv.abstentions = int(row['Abstentions'])
resultats_bv.exprimes = int(row['Exprimés'])
resultats_bv.blancs = int(row['Blancs'])
resultats_bv.nuls = int(row['Nuls'])
for liste in session.execute(select(ListeEuropeennes2024)).scalars().all():
voix_liste_bv = session.execute(select(VoixListeBureauVoteEuropeennes2024)
.filter_by(resultats_bureau_vote_id=resultats_bv.id,
liste_id=liste.id)) \
.scalar_one_or_none()
if not voix_liste_bv:
voix_liste_bv = VoixListeBureauVoteEuropeennes2024(resultats_bureau_vote_id=resultats_bv.id,
liste_id=liste.id)
session.add(voix_liste_bv)
voix_liste_bv.voix = int(row[f"Voix {liste.id}"])
session.commit()
def importer_resultats_commune(engine: Engine, verbose: bool = False) -> None:
file = get_file("https://www.data.gouv.fr/fr/datasets/r/6a782ef9-8ad6-4e66-832d-338b1041a42d",
"resultats-europeennes-par-commune.csv")
with file.open('r') as f:
reader = DictReader(f, delimiter=';')
with Session(engine) as session:
for row in tqdm(reader, desc="Commune", disable=not verbose):
com_code = str(row['Code commune']).zfill(5)
dpt_code = row['Code département']
if dpt_code == "ZX":
dpt_code = "977"
if com_code == "60694":
com_code = "60054" # Les Hauts-Talicans, données pas à jour
elif com_code in ["85165", "85212"]:
# Communes nouvelles ayant défusionné en 2024
# TODO Gérer
continue
elif com_code == "98601":
# 3 royaumes à Wallis-et-Futuna, mais un seul résultat de commune
com_code = "98611"
resultats_commune = session.execute(select(ResultatsCommuneEuropeennes2024)
.filter_by(commune_id=com_code)).scalar_one_or_none()
if not resultats_commune:
resultats_dpt = session.execute(select(ResultatsDepartementEuropeennes2024)
.filter_by(dpt_id=dpt_code)).scalar_one_or_none()
if not resultats_dpt:
dpt = session.execute(select(Departement)
.filter_by(code_insee=f"{dpt_code.zfill(2)}")).scalar_one()
resultats_reg = session.execute(select(ResultatsRegionEuropeennes2024)
.filter_by(region_id=dpt.region_code)).scalar_one_or_none()
if not resultats_reg:
resultats_france = session.execute(select(ResultatsFranceEuropeennes2024)) \
.scalar_one_or_none()
if not resultats_france:
session.add(ResultatsFranceEuropeennes2024())
resultats_france = session.execute(select(ResultatsFranceEuropeennes2024)).scalar_one()
resultats_reg = ResultatsRegionEuropeennes2024(region_id=str(dpt.region_code),
resultats_france_id=resultats_france.id)
session.add(resultats_reg)
resultats_reg = session.execute(select(ResultatsRegionEuropeennes2024)
.filter_by(region_id=dpt.region_code)).scalar_one()
resultats_dpt = ResultatsDepartementEuropeennes2024(dpt_id=dpt_code,
resultats_region_id=resultats_reg.id)
session.add(resultats_dpt)
resultats_dpt = session.execute(select(ResultatsDepartementEuropeennes2024)
.filter_by(dpt_id=dpt_code)).scalar_one()
resultats_commune = ResultatsCommuneEuropeennes2024(commune_id=com_code,
resultats_dpt_id=resultats_dpt.id)
session.add(resultats_commune)
resultats_commune = session.execute(select(ResultatsCommuneEuropeennes2024)
.filter_by(commune_id=com_code)).scalar_one()
resultats_commune.inscrits = int(row['Inscrits'])
resultats_commune.votants = int(row['Votants'])
resultats_commune.abstentions = int(row['Abstentions'])
resultats_commune.exprimes = int(row['Exprimés'])
resultats_commune.blancs = int(row['Blancs'])
resultats_commune.nuls = int(row['Nuls'])
for liste in session.execute(select(ListeEuropeennes2024)).scalars().all():
voix_liste_com = session.execute(select(VoixListeCommuneEuropeennes2024)
.filter_by(resultats_commune_id=resultats_commune.id,
liste_id=liste.id)) \
.scalar_one_or_none()
if not voix_liste_com:
voix_liste_com = VoixListeCommuneEuropeennes2024(resultats_commune_id=resultats_commune.id,
liste_id=liste.id)
session.add(voix_liste_com)
voix_liste_com.voix = int(row[f"Voix {liste.id}"])
session.commit()
def importer_resultats_circo(engine: Engine, verbose: bool = False) -> None:
file = get_file("https://www.data.gouv.fr/fr/datasets/r/ee37cbef-3d2a-4efe-a395-530b85a63028",
"resultats-europeennes-par-circonscription.csv")
with file.open('r') as f:
reader = DictReader(f, delimiter=';')
with Session(engine) as session:
for row in tqdm(reader, desc="Circonscription", disable=not verbose):
circo_code = row['Code circonscription législative']
dpt_code, circo_code = circo_code[:-2], circo_code[-2:]
if dpt_code == "ZX":
dpt_code = "977"
circo_id = f"{dpt_code.zfill(2)}-{circo_code.zfill(2)}"
resultats_circo = session.execute(select(ResultatsCirconscriptionEuropeennes2024)
.filter_by(circo_id=circo_id)).scalar_one_or_none()
if not resultats_circo:
resultats_dpt = session.execute(select(ResultatsDepartementEuropeennes2024)
.filter_by(dpt_id=dpt_code)).scalar_one_or_none()
if not resultats_dpt:
dpt = session.execute(select(Departement).filter_by(code_insee=f"{dpt_code.zfill(2)}")) \
.scalar_one()
resultats_reg = session.execute(select(ResultatsRegionEuropeennes2024)
.filter_by(region_id=dpt.region_code)) \
.scalar_one_or_none()
if not resultats_reg:
resultats_france = session.execute(select(ResultatsFranceEuropeennes2024)) \
.scalar_one_or_none()
if not resultats_france:
session.add(ResultatsFranceEuropeennes2024())
resultats_france = session.execute(select(ResultatsFranceEuropeennes2024)).scalar_one()
resultats_reg = ResultatsRegionEuropeennes2024(region_id=str(dpt.region_code),
resultats_france_id=resultats_france.id)
session.add(resultats_reg)
resultats_reg = session.execute(select(ResultatsRegionEuropeennes2024)
.filter_by(region_id=dpt.region_code)).scalar_one()
resultats_dpt = ResultatsDepartementEuropeennes2024(dpt_id=dpt_code,
resultats_region_id=resultats_reg.id)
session.add(resultats_dpt)
resultats_dpt = session.execute(select(ResultatsDepartementEuropeennes2024)
.filter_by(dpt_id=dpt_code)).scalar_one()
resultats_circo = ResultatsCirconscriptionEuropeennes2024(circo_id=circo_id,
resultats_departement_id=resultats_dpt.id)
session.add(resultats_circo)
resultats_circo = session.execute(select(ResultatsCirconscriptionEuropeennes2024)
.filter_by(circo_id=circo_id)).scalar_one()
resultats_circo.inscrits = int(row['Inscrits'])
resultats_circo.votants = int(row['Votants'])
resultats_circo.abstentions = int(row['Abstentions'])
resultats_circo.exprimes = int(row['Exprimés'])
resultats_circo.blancs = int(row['Blancs'])
resultats_circo.nuls = int(row['Nuls'])
for liste in session.execute(select(ListeEuropeennes2024)).scalars().all():
voix_liste_circo = session.execute(select(VoixListeCirconscriptionEuropeennes2024)
.filter_by(resultats_circonscription_id=resultats_circo.id,
liste_id=liste.id)).scalar_one_or_none()
if not voix_liste_circo:
voix_liste_circo = VoixListeCirconscriptionEuropeennes2024(
resultats_circonscription_id=resultats_circo.id, liste_id=liste.id)
session.add(voix_liste_circo)
voix_liste_circo.voix = int(row[f"Voix {liste.id}"])
session.commit()
def importer_resultats_departement(engine: Engine, verbose: bool = False) -> None:
file = get_file("https://www.data.gouv.fr/fr/datasets/r/b77cc4da-644f-4323-b6f7-ae6fe9b33f86",
"resultats-europeennes-par-departement.csv")
with file.open('r') as f:
reader = DictReader(f, delimiter=';')
with Session(engine) as session:
for row in tqdm(reader, desc="Département", disable=not verbose):
dpt_code = row['Code département']
if dpt_code == "ZX":
dpt_code = "977"
resultats_dpt = session.execute(select(ResultatsDepartementEuropeennes2024)
.filter_by(dpt_id=dpt_code)).scalar_one_or_none()
if not resultats_dpt:
dpt = session.execute(select(Departement)
.filter_by(code_insee=f"{dpt_code.zfill(2)}")) \
.scalar_one()
resultats_reg = session.execute(select(ResultatsRegionEuropeennes2024)
.filter_by(region_id=dpt.region_code)) \
.scalar_one_or_none()
if not resultats_reg:
resultats_france = session.execute(select(ResultatsFranceEuropeennes2024)).scalar_one_or_none()
if not resultats_france:
session.add(ResultatsFranceEuropeennes2024())
resultats_france = session.execute(select(ResultatsFranceEuropeennes2024)).scalar_one()
resultats_reg = ResultatsRegionEuropeennes2024(region_id=str(dpt.region_code),
resultats_france_id=resultats_france.id)
session.add(resultats_reg)
resultats_reg = session.execute(select(ResultatsRegionEuropeennes2024)
.filter_by(region_id=dpt.region_code)).scalar_one()
resultats_dpt = ResultatsDepartementEuropeennes2024(dpt_id=dpt_code,
resultats_region_id=resultats_reg.id)
session.add(resultats_dpt)
resultats_dpt = session.execute(select(ResultatsDepartementEuropeennes2024)
.filter_by(dpt_id=dpt_code)).scalar_one()
resultats_dpt.inscrits = int(row['Inscrits'])
resultats_dpt.votants = int(row['Votants'])
resultats_dpt.abstentions = int(row['Abstentions'])
resultats_dpt.exprimes = int(row['Exprimés'])
resultats_dpt.blancs = int(row['Blancs'])
resultats_dpt.nuls = int(row['Nuls'])
for liste in session.execute(select(ListeEuropeennes2024)).scalars().all():
voix_liste_dpt = session.execute(select(VoixListeDepartementEuropeennes2024)
.filter_by(resultats_departement_id=resultats_dpt.id,
liste_id=liste.id)) \
.scalar_one_or_none()
if not voix_liste_dpt:
voix_liste_dpt = VoixListeDepartementEuropeennes2024(resultats_departement_id=resultats_dpt.id,
liste_id=liste.id)
session.add(voix_liste_dpt)
voix_liste_dpt.voix = int(row[f"Voix {liste.id}"])
session.commit()
def importer_resultats_region(engine: Engine, verbose: bool = False) -> None:
file = get_file("https://www.data.gouv.fr/fr/datasets/r/7c3a854b-7344-4c68-b1f9-9d651b4ca823",
"resultats-europeennes-par-region.csv")
with file.open('r') as f:
reader = DictReader(f, delimiter=';')
with Session(engine) as session:
for row in tqdm(reader, desc="Région", disable=not verbose):
reg_code = row['Code région']
resultats_reg = session.execute(select(ResultatsRegionEuropeennes2024).filter_by(region_id=reg_code)) \
.scalar_one_or_none()
if not resultats_reg:
resultats_france = session.execute(select(ResultatsFranceEuropeennes2024)).scalar_one_or_none()
if not resultats_france:
session.add(ResultatsFranceEuropeennes2024())
resultats_france = session.execute(select(ResultatsFranceEuropeennes2024)).scalar_one()
resultats_reg = ResultatsRegionEuropeennes2024(region_id=reg_code,
resultats_france_id=resultats_france.id)
session.add(resultats_reg)
session.commit()
resultats_reg = session.execute(select(ResultatsRegionEuropeennes2024)
.filter_by(region_id=reg_code)).scalar_one()
resultats_reg.inscrits = int(row['Inscrits'])
resultats_reg.votants = int(row['Votants'])
resultats_reg.abstentions = int(row['Abstentions'])
resultats_reg.exprimes = int(row['Exprimés'])
resultats_reg.blancs = int(row['Blancs'])
resultats_reg.nuls = int(row['Nuls'])
for liste in session.execute(select(ListeEuropeennes2024)).scalars().all():
voix_liste_reg = session.execute(select(VoixListeRegionEuropeennes2024)
.filter_by(resultats_region_id=resultats_reg.id,
liste_id=liste.id)) \
.scalar_one_or_none()
if not voix_liste_reg:
voix_liste_reg = VoixListeRegionEuropeennes2024(resultats_region_id=resultats_reg.id,
liste_id=liste.id)
session.add(voix_liste_reg)
voix_liste_reg.voix = int(row[f"Voix {liste.id}"])
session.commit()
def calculer_resultats_france(engine: Engine, verbose: bool = False) -> None:
with Session(engine) as session:
resultats_france = session.execute(select(ResultatsFranceEuropeennes2024)).scalar_one()
resultats_france.inscrits = 0
resultats_france.votants = 0
resultats_france.abstentions = 0
resultats_france.exprimes = 0
resultats_france.blancs = 0
resultats_france.nuls = 0
for voix_liste_france in session.execute(
select(VoixListeFranceEuropeennes2024)
.filter_by(resultats_france_id=resultats_france.id)).scalars().all():
voix_liste_france.voix = 0 # Réinitialisation des voix
for resultats_dpt in session.execute(select(ResultatsDepartementEuropeennes2024)).scalars().all():
resultats_france.inscrits += resultats_dpt.inscrits
resultats_france.votants += resultats_dpt.votants
resultats_france.abstentions += resultats_dpt.abstentions
resultats_france.exprimes += resultats_dpt.exprimes
resultats_france.blancs += resultats_dpt.blancs
resultats_france.nuls += resultats_dpt.nuls
for voix_liste_dpt in session.execute(select(VoixListeDepartementEuropeennes2024)
.filter_by(resultats_departement_id=resultats_dpt.id)) \
.scalars().all():
if voix_liste_france := session.execute(
select(VoixListeFranceEuropeennes2024)
.filter_by(resultats_france_id=resultats_france.id,
liste_id=voix_liste_dpt.liste_id)).scalar_one_or_none():
voix_liste_france.voix += voix_liste_dpt.voix
else:
session.add(VoixListeFranceEuropeennes2024(resultats_france_id=resultats_france.id,
liste_id=voix_liste_dpt.liste_id,
voix=voix_liste_dpt.voix))
session.commit()
def run(engine: Engine, verbose: bool = False) -> None:
importer_resultats_region(engine, verbose)
importer_resultats_departement(engine, verbose)
calculer_resultats_france(engine, verbose)
importer_resultats_circo(engine, verbose)
importer_resultats_commune(engine, verbose)
importer_resultats_bv(engine, verbose)

318
nupes/scripts/europeennes2024/import_resultats_web.py Normal file
View File

@ -0,0 +1,318 @@
import re
from bs4 import BeautifulSoup
from sqlalchemy import Engine, select
from sqlalchemy.orm import Session
from tqdm import tqdm
from nupes.cache import get_file
from nupes.models.europeennes2024 import ResultatsFranceEuropeennes2024, ResultatsRegionEuropeennes2024, ResultatsDepartementEuropeennes2024, ResultatsCommuneEuropeennes2024, \
VoixListeFranceEuropeennes2024, VoixListeRegionEuropeennes2024, VoixListeDepartementEuropeennes2024, VoixListeCommuneEuropeennes2024, ListeEuropeennes2024
from nupes.models.geographie import Region, Departement, Commune
BASE_URL = "https://www.resultats-elections.interieur.gouv.fr/europeennes2024/ensemble_geographique"
def importer_resultats_france(engine: Engine, verbose: bool = False) -> None:
file = get_file(f"{BASE_URL}/index.html", "resultats2024/resultats.html")
with file.open() as f:
resultats = analyser_resultats(f)
if not resultats:
return
with Session(engine) as session:
if resultats_france := session.execute(select(ResultatsFranceEuropeennes2024)).scalar_one_or_none():
resultats_france.inscrits = resultats["inscrits"]
resultats_france.abstentions = resultats["abstentions"]
resultats_france.votants = resultats["votants"]
resultats_france.blancs = resultats["blancs"]
resultats_france.nuls = resultats["nuls"]
resultats_france.exprimes = resultats["exprimes"]
else:
resultats_france = ResultatsFranceEuropeennes2024(
inscrits=resultats["inscrits"],
abstentions=resultats["abstentions"],
votants=resultats["votants"],
blancs=resultats["blancs"],
nuls=resultats["nuls"],
exprimes=resultats["exprimes"],
)
session.add(resultats_france)
for voix_liste in resultats["resultats_par_liste"]:
if voix_liste_france := session.execute(
select(VoixListeFranceEuropeennes2024).join(ListeEuropeennes2024).filter_by(nom_majuscules=voix_liste["nom"].upper())
).scalar_one_or_none():
voix_liste_france.voix = voix_liste["voix"]
else:
liste = session.execute(select(ListeEuropeennes2024).filter_by(nom_majuscules=voix_liste["nom"].upper())).scalar_one()
voix_liste_france = VoixListeFranceEuropeennes2024(liste_id=liste.id, resultats_france_id=resultats_france.id,
voix=voix_liste["voix"])
session.add(voix_liste_france)
session.commit()
def importer_resultats_regions(engine: Engine, verbose: bool = False) -> None:
with Session(engine) as session:
regions = session.execute(select(Region)).scalars().all()
regions_iterator = tqdm(regions, desc="Régions") if verbose else regions
for region in regions_iterator:
reg_code = region.code_insee
if reg_code == "984" or reg_code == "989":
# TAAF + Île de Clipperton, pas d'élection
continue
elif reg_code == "977" or reg_code == "978":
# Saint-Barthélemy et Saint-Martin, pas mis dans une région mais dans un unique département
continue
file = get_file(f"{BASE_URL}/{reg_code}/index.html",
f"resultats2024/region_{reg_code}/resultats.html")
with file.open() as f:
resultats = analyser_resultats(f)
if not resultats:
continue
if resultats_region := session.execute(select(ResultatsRegionEuropeennes2024).filter_by(region_id=reg_code)) \
.scalar_one_or_none():
resultats_region.inscrits = resultats["inscrits"]
resultats_region.abstentions = resultats["abstentions"]
resultats_region.votants = resultats["votants"]
resultats_region.blancs = resultats["blancs"]
resultats_region.nuls = resultats["nuls"]
resultats_region.exprimes = resultats["exprimes"]
else:
resultats_france = session.execute(select(ResultatsFranceEuropeennes2024)).scalar_one()
resultats_region = ResultatsRegionEuropeennes2024(
region_id=reg_code,
resultats_france_id=resultats_france.id,
inscrits=resultats["inscrits"],
abstentions=resultats["abstentions"],
votants=resultats["votants"],
blancs=resultats["blancs"],
nuls=resultats["nuls"],
exprimes=resultats["exprimes"],
)
session.add(resultats_region)
for voix_liste in resultats["resultats_par_liste"]:
if voix_liste_region := session.execute(
select(VoixListeRegionEuropeennes2024).join(ListeEuropeennes2024).filter_by(nom_majuscules=voix_liste["nom"].upper())
).scalar_one_or_none():
voix_liste_region.voix = voix_liste["voix"]
else:
liste = session.execute(select(ListeEuropeennes2024).filter_by(nom_majuscules=voix_liste["nom"].upper())) \
.scalar_one()
voix_liste_region = VoixListeRegionEuropeennes2024(liste_id=liste.id, resultats_region_id=resultats_region.id,
voix=voix_liste["voix"])
session.add(voix_liste_region)
session.commit()
def importer_resultats_departements(engine: Engine, verbose: bool = False) -> None:
with Session(engine) as session:
departements = session.execute(select(Departement)).scalars().all()
iterator = tqdm(departements, desc="Départements") if verbose else departements
for dpt in iterator:
reg_code = dpt.region_code
reg_path = f"{reg_code}/"
dpt_code = dpt.code_insee
if dpt_code == "984" or dpt_code == "989":
# TAAF + Île de Clipperton, pas d'élection
continue
elif dpt_code == "977" or dpt_code == "978":
# Pour une raison obscure, Saint-Barthélemy et Saint-Martin sont dans le département ZX
reg_path = ""
dpt_code = "ZX"
if dpt_code in ["975", "977", "978", "986", "987", "988"]:
# Pour ces collectivités d'outre-mer, qui ne sont pas dans des régions,
# on ne les regroupe pas dans des régions
reg_path = ""
file = get_file(f"{BASE_URL}/{reg_path}{dpt_code}/index.html",
f"resultats2024/region_{reg_code}/dpt_{dpt_code}/resultats.html")
with file.open() as f:
resultats = analyser_resultats(f)
if not resultats:
continue
if resultats_departement := session.execute(
select(ResultatsDepartementEuropeennes2024).filter_by(dpt_id=dpt.code_insee)).scalar_one_or_none():
resultats_departement.inscrits = resultats["inscrits"]
resultats_departement.abstentions = resultats["abstentions"]
resultats_departement.votants = resultats["votants"]
resultats_departement.blancs = resultats["blancs"]
resultats_departement.nuls = resultats["nuls"]
resultats_departement.exprimes = resultats["exprimes"]
else:
resultats_region = session.execute(select(ResultatsRegionEuropeennes2024).filter_by(region_id=dpt.region_code)) \
.scalar_one_or_none()
resultats_departement = ResultatsDepartementEuropeennes2024(
dpt_id=dpt.code_insee,
resultats_region_id=resultats_region.id if resultats_region else None,
inscrits=resultats["inscrits"],
abstentions=resultats["abstentions"],
votants=resultats["votants"],
blancs=resultats["blancs"],
nuls=resultats["nuls"],
exprimes=resultats["exprimes"],
)
session.add(resultats_departement)
for voix_liste in resultats["resultats_par_liste"]:
if voix_liste_departement := session.execute(
select(VoixListeDepartementEuropeennes2024).join(ListeEuropeennes2024).filter_by(nom_majuscules=voix_liste["nom"].upper())
).scalar_one_or_none():
voix_liste_departement.voix = voix_liste["voix"]
else:
liste = session.execute(select(ListeEuropeennes2024).filter_by(nom_majuscules=voix_liste["nom"].upper())) \
.scalar_one()
voix_liste_departement = VoixListeDepartementEuropeennes2024(liste_id=liste.id,
resultats_departement_id=resultats_departement.id,
voix=voix_liste["voix"])
session.add(voix_liste_departement)
session.commit()
def importer_resultats_communes(engine: Engine, verbose: bool = False) -> None:
with Session(engine) as session:
communes = session.execute(select(Commune)).scalars().all()
iterator = tqdm(communes, desc="Communes") if verbose else communes
for commune in iterator:
reg_code = commune.departement.region_code
reg_path = f"{reg_code}/"
dpt_code = commune.departement_code
com_code = commune.code_insee
if dpt_code == "984" or dpt_code == "989":
# TAAF + Île de Clipperton, pas d'élection
continue
elif dpt_code == "977" or dpt_code == "978":
# Pour une raison obscure, Saint-Barthélemy et Saint-Martin sont dans le département ZX
reg_path = ""
dpt_code = "ZX"
com_code = f"ZX{com_code[2:]}"
if dpt_code in ["975", "977", "978", "986", "987", "988"]:
# Pour ces collectivités d'outre-mer, qui ne sont pas dans des régions,
# on ne les regroupe pas dans des régions
reg_path = ""
if com_code in ["08294", "16355", "18131", "25282", "25549", "35112", "49321", "55039", "55050", "55139",
"55189", "55239", "55307", "64541", "69152", "75056", "85041", "85271", "86231", "95282",
"98611", "98612", "98613"]:
# Aucun⋅e habitant⋅e donc pas de bureau de vote
# Ou alors la commune a récemment fusionné
continue
file = get_file(f"{BASE_URL}/{reg_path}{dpt_code}/{com_code}/index.html",
f"resultats2024/region_{reg_code}/dpt_{dpt_code}"
f"/commune_{com_code}/resultats.html")
if not file:
continue
with file.open() as f:
resultats = analyser_resultats(f)
if not resultats:
continue
if resultats_commune := session.execute(
select(ResultatsCommuneEuropeennes2024).filter_by(commune_id=com_code)).scalar_one_or_none():
resultats_commune.inscrits = resultats["inscrits"]
resultats_commune.abstentions = resultats["abstentions"]
resultats_commune.votants = resultats["votants"]
resultats_commune.blancs = resultats["blancs"]
resultats_commune.nuls = resultats["nuls"]
resultats_commune.exprimes = resultats["exprimes"]
else:
resultats_departement = session.execute(select(ResultatsDepartementEuropeennes2024).filter_by(dpt_id=dpt_code)) \
.scalar_one_or_none()
resultats_commune = ResultatsCommuneEuropeennes2024(
commune_id=com_code,
resultats_dpt_id=resultats_departement.id if resultats_departement else None,
inscrits=resultats["inscrits"],
abstentions=resultats["abstentions"],
votants=resultats["votants"],
blancs=resultats["blancs"],
nuls=resultats["nuls"],
exprimes=resultats["exprimes"],
)
session.add(resultats_commune)
for voix_liste in resultats["resultats_par_liste"]:
if voix_liste_commune := session.execute(
select(VoixListeCommuneEuropeennes2024).join(ListeEuropeennes2024).filter_by(nom_majuscules=voix_liste["nom"].upper())
).scalar_one_or_none():
voix_liste_commune.voix = voix_liste["voix"]
else:
liste = session.execute(select(ListeEuropeennes2024).filter_by(nom_majuscules=voix_liste["nom"].upper())) \
.scalar_one()
voix_liste_commune = VoixListeCommuneEuropeennes2024(liste_id=liste.id, resultats_commune_id=resultats_commune.id,
voix=voix_liste["voix"])
session.add(voix_liste_commune)
session.commit()
def analyser_resultats(file) -> dict:
parsed_data = {}
soup = BeautifulSoup(file, 'html.parser')
tables = soup.find_all('table')
for table in tables:
if table.find('th', text=re.compile("Voix")):
resultats_par_liste = []
parsed_data['resultats_par_liste'] = resultats_par_liste
tbody = table.tbody
for line in tbody.find_all('tr'):
cells = line.find_all('td')
if len(cells) == 0:
continue
liste = {}
resultats_par_liste.append(liste)
liste['nom'] = cells[0].text.replace("\n", "").replace("\u202f", " ").strip()
liste['voix'] = int(cells[2].string.replace("\n", "").replace(" ", "").replace("\u202f", "").strip())
elif table.find('td', text=re.compile("Inscrits")):
tbody = table.tbody
for line in tbody.find_all('tr'):
cells = line.find_all('td')
if len(cells) == 0:
continue
nom = cells[0].string.replace("\n", "").replace(" ", "").replace("\u202f", "").strip()
number = int(cells[1].string.replace("\n", "").replace(" ", "").replace("\u202f", "").strip())
match nom:
case "Inscrits":
parsed_data["inscrits"] = number
case "Abstentions":
parsed_data["abstentions"] = number
case "Votants":
parsed_data["votants"] = number
case "Blancs":
parsed_data["blancs"] = number
case "Nuls":
parsed_data["nuls"] = number
case "Exprimés":
parsed_data["exprimes"] = number
case _:
print(f"Unknown cell: {nom}")
return parsed_data
def run(engine: Engine, verbose: bool = False) -> None:
importer_resultats_france(engine, verbose)
importer_resultats_regions(engine, verbose)
importer_resultats_departements(engine, verbose)
# FIXME Les communes prennent trop de temps
# importer_resultats_communes(engine, verbose)