2024-06-13 10:19:02 +00:00
|
|
|
from csv import DictReader
|
|
|
|
|
|
|
|
from sqlalchemy import Engine, select
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from tqdm import tqdm
|
|
|
|
|
|
|
|
from nupes.cache import get_file
|
2024-06-18 21:29:44 +00:00
|
|
|
from nupes.models import BureauVote
|
2024-06-18 12:30:14 +00:00
|
|
|
from nupes.models.europeennes2024 import *
|
2024-06-13 10:19:02 +00:00
|
|
|
|
|
|
|
|
2024-06-13 10:21:11 +00:00
|
|
|
def importer_resultats_bv(engine: Engine, verbose: bool = False) -> None:
|
2024-06-14 21:58:36 +00:00
|
|
|
file = get_file("https://www.data.gouv.fr/fr/datasets/r/cc1883d9-1265-4365-b754-fb6aef22d82e",
|
2024-06-20 12:45:53 +00:00
|
|
|
"resultats-europeennes-2024-par-bureau-de-vote.csv")
|
2024-06-13 10:19:02 +00:00
|
|
|
|
|
|
|
with file.open('r') as f:
|
2024-06-14 21:58:36 +00:00
|
|
|
reader = DictReader(f, delimiter=';')
|
2024-06-13 10:19:02 +00:00
|
|
|
with Session(engine) as session:
|
2024-06-13 10:21:11 +00:00
|
|
|
for row in tqdm(reader, desc="Bureau de vote", disable=not verbose):
|
2024-06-13 10:19:02 +00:00
|
|
|
com_code = str(row['Code commune']).zfill(5)
|
|
|
|
bv_code = row['Code BV']
|
|
|
|
|
2024-06-20 12:45:53 +00:00
|
|
|
if com_code.startswith("987") or com_code.startswith("988"):
|
|
|
|
# Les communes de Polynésie française ne sont pas importées
|
2024-06-18 21:29:44 +00:00
|
|
|
continue
|
|
|
|
|
2024-06-16 06:44:32 +00:00
|
|
|
if com_code == "60694":
|
|
|
|
com_code = "60054" # Les Hauts-Talicans, données pas à jour
|
2024-06-16 07:30:47 +00:00
|
|
|
elif com_code in ["85165", "85212"]:
|
|
|
|
# Communes nouvelles ayant défusionné en 2024
|
|
|
|
# TODO Gérer
|
|
|
|
continue
|
2024-06-16 13:31:25 +00:00
|
|
|
elif com_code == "98601":
|
|
|
|
# 3 royaumes à Wallis-et-Futuna, mais un seul résultat de commune
|
|
|
|
com_code = "98611"
|
2024-06-16 06:44:32 +00:00
|
|
|
|
2024-06-14 21:58:36 +00:00
|
|
|
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é")
|
2024-06-13 10:19:02 +00:00
|
|
|
continue
|
|
|
|
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_bv = session.execute(select(ResultatsBureauVoteEuropeennes2024)
|
|
|
|
.filter_by(bv_id=bv_id)).scalar_one_or_none()
|
2024-06-13 10:19:02 +00:00
|
|
|
if not resultats_bv:
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_commune = session.execute(select(ResultatsCommuneEuropeennes2024)
|
|
|
|
.filter_by(commune_id=com_code)).scalar_one_or_none()
|
2024-06-13 10:19:02 +00:00
|
|
|
if not resultats_commune:
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_dpt = session.execute(select(ResultatsDepartementEuropeennes2024)
|
2024-06-13 10:19:02 +00:00
|
|
|
.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()
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_reg = session.execute(select(ResultatsRegionEuropeennes2024)
|
|
|
|
.filter_by(region_id=dpt.region_code)).scalar_one_or_none()
|
2024-06-13 10:19:02 +00:00
|
|
|
if not resultats_reg:
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_france = session.execute(select(ResultatsFranceEuropeennes2024)) \
|
|
|
|
.scalar_one_or_none()
|
2024-06-13 10:19:02 +00:00
|
|
|
if not resultats_france:
|
2024-06-18 12:30:14 +00:00
|
|
|
session.add(ResultatsFranceEuropeennes2024())
|
|
|
|
resultats_france = session.execute(select(ResultatsFranceEuropeennes2024)) \
|
|
|
|
.scalar_one()
|
2024-06-13 10:19:02 +00:00
|
|
|
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_reg = ResultatsRegionEuropeennes2024(region_id=str(dpt.region_code),
|
|
|
|
resultats_france_id=resultats_france.id)
|
2024-06-13 10:19:02 +00:00
|
|
|
session.add(resultats_reg)
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_reg = session.execute(select(ResultatsRegionEuropeennes2024)
|
2024-06-13 10:19:02 +00:00
|
|
|
.filter_by(region_id=dpt.region_code)).scalar_one()
|
|
|
|
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_dpt = ResultatsDepartementEuropeennes2024(dpt_id=row['Code département'],
|
|
|
|
resultats_region_id=resultats_reg.id)
|
2024-06-13 10:19:02 +00:00
|
|
|
session.add(resultats_dpt)
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_dpt = session.execute(select(ResultatsDepartementEuropeennes2024)
|
2024-06-13 10:19:02 +00:00
|
|
|
.filter_by(dpt_id=row['Code département'])).scalar_one()
|
|
|
|
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_commune = ResultatsCommuneEuropeennes2024(commune_id=com_code,
|
|
|
|
resultats_dpt_id=resultats_dpt.id)
|
2024-06-13 10:19:02 +00:00
|
|
|
session.add(resultats_commune)
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_commune = session.execute(select(ResultatsCommuneEuropeennes2024)
|
|
|
|
.filter_by(commune_id=com_code)).scalar_one()
|
2024-06-13 10:19:02 +00:00
|
|
|
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_bv = ResultatsBureauVoteEuropeennes2024(bv_id=bv_id,
|
|
|
|
resultats_commune_id=resultats_commune.id)
|
2024-06-13 10:19:02 +00:00
|
|
|
session.add(resultats_bv)
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_bv = session.execute(select(ResultatsBureauVoteEuropeennes2024)
|
|
|
|
.filter_by(bv_id=bv_id)).scalar_one()
|
2024-06-13 10:19:02 +00:00
|
|
|
|
|
|
|
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'])
|
|
|
|
|
2024-06-18 12:30:14 +00:00
|
|
|
for liste in session.execute(select(ListeEuropeennes2024)).scalars().all():
|
|
|
|
voix_liste_bv = session.execute(select(VoixListeBureauVoteEuropeennes2024)
|
2024-06-13 10:19:02 +00:00
|
|
|
.filter_by(resultats_bureau_vote_id=resultats_bv.id,
|
|
|
|
liste_id=liste.id)) \
|
|
|
|
.scalar_one_or_none()
|
|
|
|
if not voix_liste_bv:
|
2024-06-18 12:30:14 +00:00
|
|
|
voix_liste_bv = VoixListeBureauVoteEuropeennes2024(resultats_bureau_vote_id=resultats_bv.id,
|
|
|
|
liste_id=liste.id)
|
2024-06-13 10:19:02 +00:00
|
|
|
session.add(voix_liste_bv)
|
|
|
|
|
|
|
|
voix_liste_bv.voix = int(row[f"Voix {liste.id}"])
|
|
|
|
|
2024-06-13 20:49:34 +00:00
|
|
|
session.commit()
|
2024-06-13 10:19:02 +00:00
|
|
|
|
|
|
|
|
2024-06-14 21:58:36 +00:00
|
|
|
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",
|
2024-06-20 12:45:53 +00:00
|
|
|
"resultats-europeennes-2024-par-commune.csv")
|
2024-06-13 20:49:34 +00:00
|
|
|
|
2024-06-14 21:58:36 +00:00
|
|
|
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)
|
2024-06-15 18:39:42 +00:00
|
|
|
dpt_code = row['Code département']
|
|
|
|
|
2024-06-18 21:29:44 +00:00
|
|
|
if dpt_code == "ZX": # Saint-Martin/Saint-Barthélémy
|
2024-06-15 18:39:42 +00:00
|
|
|
dpt_code = "977"
|
2024-06-13 10:19:02 +00:00
|
|
|
|
2024-06-20 12:45:53 +00:00
|
|
|
if dpt_code == "977" or dpt_code == "987" or dpt_code == "988":
|
|
|
|
# Les communes de Polynésie française ne sont pas importées
|
2024-06-18 21:29:44 +00:00
|
|
|
continue
|
|
|
|
|
2024-06-16 06:44:32 +00:00
|
|
|
if com_code == "60694":
|
|
|
|
com_code = "60054" # Les Hauts-Talicans, données pas à jour
|
2024-06-16 07:30:47 +00:00
|
|
|
elif com_code in ["85165", "85212"]:
|
|
|
|
# Communes nouvelles ayant défusionné en 2024
|
|
|
|
# TODO Gérer
|
|
|
|
continue
|
2024-06-16 13:31:25 +00:00
|
|
|
elif com_code == "98601":
|
|
|
|
# 3 royaumes à Wallis-et-Futuna, mais un seul résultat de commune
|
|
|
|
com_code = "98611"
|
2024-06-16 06:44:32 +00:00
|
|
|
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_commune = session.execute(select(ResultatsCommuneEuropeennes2024)
|
|
|
|
.filter_by(commune_id=com_code)).scalar_one_or_none()
|
2024-06-14 21:58:36 +00:00
|
|
|
if not resultats_commune:
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_dpt = session.execute(select(ResultatsDepartementEuropeennes2024)
|
|
|
|
.filter_by(dpt_id=dpt_code)).scalar_one_or_none()
|
2024-06-14 21:58:36 +00:00
|
|
|
if not resultats_dpt:
|
|
|
|
dpt = session.execute(select(Departement)
|
2024-06-18 12:30:14 +00:00
|
|
|
.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()
|
2024-06-14 21:58:36 +00:00
|
|
|
if not resultats_reg:
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_france = session.execute(select(ResultatsFranceEuropeennes2024)) \
|
|
|
|
.scalar_one_or_none()
|
2024-06-14 21:58:36 +00:00
|
|
|
if not resultats_france:
|
2024-06-18 12:30:14 +00:00
|
|
|
session.add(ResultatsFranceEuropeennes2024())
|
|
|
|
resultats_france = session.execute(select(ResultatsFranceEuropeennes2024)).scalar_one()
|
2024-06-14 21:58:36 +00:00
|
|
|
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_reg = ResultatsRegionEuropeennes2024(region_id=str(dpt.region_code),
|
|
|
|
resultats_france_id=resultats_france.id)
|
2024-06-14 21:58:36 +00:00
|
|
|
session.add(resultats_reg)
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_reg = session.execute(select(ResultatsRegionEuropeennes2024)
|
2024-06-14 21:58:36 +00:00
|
|
|
.filter_by(region_id=dpt.region_code)).scalar_one()
|
2024-06-13 10:19:02 +00:00
|
|
|
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_dpt = ResultatsDepartementEuropeennes2024(dpt_id=dpt_code,
|
|
|
|
resultats_region_id=resultats_reg.id)
|
2024-06-14 21:58:36 +00:00
|
|
|
session.add(resultats_dpt)
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_dpt = session.execute(select(ResultatsDepartementEuropeennes2024)
|
2024-06-15 18:39:42 +00:00
|
|
|
.filter_by(dpt_id=dpt_code)).scalar_one()
|
2024-06-13 10:19:02 +00:00
|
|
|
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_commune = ResultatsCommuneEuropeennes2024(commune_id=com_code,
|
|
|
|
resultats_dpt_id=resultats_dpt.id)
|
2024-06-14 21:58:36 +00:00
|
|
|
session.add(resultats_commune)
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_commune = session.execute(select(ResultatsCommuneEuropeennes2024)
|
|
|
|
.filter_by(commune_id=com_code)).scalar_one()
|
2024-06-13 10:19:02 +00:00
|
|
|
|
2024-06-14 21:58:36 +00:00
|
|
|
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'])
|
2024-06-13 20:49:34 +00:00
|
|
|
|
2024-06-18 12:30:14 +00:00
|
|
|
for liste in session.execute(select(ListeEuropeennes2024)).scalars().all():
|
|
|
|
voix_liste_com = session.execute(select(VoixListeCommuneEuropeennes2024)
|
2024-06-15 17:54:49 +00:00
|
|
|
.filter_by(resultats_commune_id=resultats_commune.id,
|
|
|
|
liste_id=liste.id)) \
|
2024-06-14 21:58:36 +00:00
|
|
|
.scalar_one_or_none()
|
|
|
|
if not voix_liste_com:
|
2024-06-18 12:30:14 +00:00
|
|
|
voix_liste_com = VoixListeCommuneEuropeennes2024(resultats_commune_id=resultats_commune.id,
|
|
|
|
liste_id=liste.id)
|
2024-06-14 21:58:36 +00:00
|
|
|
session.add(voix_liste_com)
|
2024-06-13 10:19:02 +00:00
|
|
|
|
2024-06-14 21:58:36 +00:00
|
|
|
voix_liste_com.voix = int(row[f"Voix {liste.id}"])
|
2024-06-13 10:19:02 +00:00
|
|
|
|
2024-06-14 21:58:36 +00:00
|
|
|
session.commit()
|
2024-06-13 20:49:34 +00:00
|
|
|
|
2024-06-14 21:58:36 +00:00
|
|
|
|
|
|
|
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",
|
2024-06-20 12:45:53 +00:00
|
|
|
"resultats-europeennes-2024-par-circonscription.csv")
|
2024-06-14 21:58:36 +00:00
|
|
|
|
|
|
|
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']
|
2024-06-15 17:54:49 +00:00
|
|
|
dpt_code, circo_code = circo_code[:-2], circo_code[-2:]
|
2024-06-15 18:35:47 +00:00
|
|
|
|
|
|
|
if dpt_code == "ZX":
|
|
|
|
dpt_code = "977"
|
|
|
|
|
2024-06-15 18:31:30 +00:00
|
|
|
circo_id = f"{dpt_code.zfill(2)}-{circo_code.zfill(2)}"
|
2024-06-14 21:58:36 +00:00
|
|
|
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_circo = session.execute(select(ResultatsCirconscriptionEuropeennes2024)
|
|
|
|
.filter_by(circo_id=circo_id)).scalar_one_or_none()
|
2024-06-14 21:58:36 +00:00
|
|
|
if not resultats_circo:
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_dpt = session.execute(select(ResultatsDepartementEuropeennes2024)
|
|
|
|
.filter_by(dpt_id=dpt_code)).scalar_one_or_none()
|
2024-06-14 21:58:36 +00:00
|
|
|
if not resultats_dpt:
|
2024-06-15 18:35:47 +00:00
|
|
|
dpt = session.execute(select(Departement).filter_by(code_insee=f"{dpt_code.zfill(2)}")) \
|
2024-06-14 21:58:36 +00:00
|
|
|
.scalar_one()
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_reg = session.execute(select(ResultatsRegionEuropeennes2024)
|
2024-06-14 21:58:36 +00:00
|
|
|
.filter_by(region_id=dpt.region_code)) \
|
|
|
|
.scalar_one_or_none()
|
|
|
|
if not resultats_reg:
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_france = session.execute(select(ResultatsFranceEuropeennes2024)) \
|
|
|
|
.scalar_one_or_none()
|
2024-06-14 21:58:36 +00:00
|
|
|
if not resultats_france:
|
2024-06-18 12:30:14 +00:00
|
|
|
session.add(ResultatsFranceEuropeennes2024())
|
|
|
|
resultats_france = session.execute(select(ResultatsFranceEuropeennes2024)).scalar_one()
|
2024-06-14 21:58:36 +00:00
|
|
|
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_reg = ResultatsRegionEuropeennes2024(region_id=str(dpt.region_code),
|
|
|
|
resultats_france_id=resultats_france.id)
|
2024-06-14 21:58:36 +00:00
|
|
|
session.add(resultats_reg)
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_reg = session.execute(select(ResultatsRegionEuropeennes2024)
|
2024-06-14 21:58:36 +00:00
|
|
|
.filter_by(region_id=dpt.region_code)).scalar_one()
|
|
|
|
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_dpt = ResultatsDepartementEuropeennes2024(dpt_id=dpt_code,
|
|
|
|
resultats_region_id=resultats_reg.id)
|
2024-06-14 21:58:36 +00:00
|
|
|
session.add(resultats_dpt)
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_dpt = session.execute(select(ResultatsDepartementEuropeennes2024)
|
2024-06-15 18:35:47 +00:00
|
|
|
.filter_by(dpt_id=dpt_code)).scalar_one()
|
2024-06-14 21:58:36 +00:00
|
|
|
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_circo = ResultatsCirconscriptionEuropeennes2024(circo_id=circo_id,
|
|
|
|
resultats_departement_id=resultats_dpt.id)
|
2024-06-14 21:58:36 +00:00
|
|
|
session.add(resultats_circo)
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_circo = session.execute(select(ResultatsCirconscriptionEuropeennes2024)
|
|
|
|
.filter_by(circo_id=circo_id)).scalar_one()
|
2024-06-14 21:58:36 +00:00
|
|
|
|
|
|
|
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'])
|
|
|
|
|
2024-06-18 12:30:14 +00:00
|
|
|
for liste in session.execute(select(ListeEuropeennes2024)).scalars().all():
|
|
|
|
voix_liste_circo = session.execute(select(VoixListeCirconscriptionEuropeennes2024)
|
2024-06-14 21:58:36 +00:00
|
|
|
.filter_by(resultats_circonscription_id=resultats_circo.id,
|
2024-06-18 12:30:14 +00:00
|
|
|
liste_id=liste.id)).scalar_one_or_none()
|
2024-06-14 21:58:36 +00:00
|
|
|
if not voix_liste_circo:
|
2024-06-18 12:30:14 +00:00
|
|
|
voix_liste_circo = VoixListeCirconscriptionEuropeennes2024(
|
|
|
|
resultats_circonscription_id=resultats_circo.id, liste_id=liste.id)
|
2024-06-14 21:58:36 +00:00
|
|
|
session.add(voix_liste_circo)
|
|
|
|
|
|
|
|
voix_liste_circo.voix = int(row[f"Voix {liste.id}"])
|
|
|
|
|
2024-06-16 05:45:53 +00:00
|
|
|
session.commit()
|
2024-06-14 21:58:36 +00:00
|
|
|
|
|
|
|
|
|
|
|
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",
|
2024-06-20 12:45:53 +00:00
|
|
|
"resultats-europeennes-2024-par-departement.csv")
|
2024-06-14 21:58:36 +00:00
|
|
|
|
|
|
|
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']
|
|
|
|
|
2024-06-15 17:54:49 +00:00
|
|
|
if dpt_code == "ZX":
|
|
|
|
dpt_code = "977"
|
|
|
|
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_dpt = session.execute(select(ResultatsDepartementEuropeennes2024)
|
|
|
|
.filter_by(dpt_id=dpt_code)).scalar_one_or_none()
|
2024-06-14 21:58:36 +00:00
|
|
|
if not resultats_dpt:
|
|
|
|
dpt = session.execute(select(Departement)
|
2024-06-15 17:54:49 +00:00
|
|
|
.filter_by(code_insee=f"{dpt_code.zfill(2)}")) \
|
2024-06-14 21:58:36 +00:00
|
|
|
.scalar_one()
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_reg = session.execute(select(ResultatsRegionEuropeennes2024)
|
2024-06-14 21:58:36 +00:00
|
|
|
.filter_by(region_id=dpt.region_code)) \
|
|
|
|
.scalar_one_or_none()
|
|
|
|
if not resultats_reg:
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_france = session.execute(select(ResultatsFranceEuropeennes2024)).scalar_one_or_none()
|
2024-06-14 21:58:36 +00:00
|
|
|
if not resultats_france:
|
2024-06-18 12:30:14 +00:00
|
|
|
session.add(ResultatsFranceEuropeennes2024())
|
|
|
|
resultats_france = session.execute(select(ResultatsFranceEuropeennes2024)).scalar_one()
|
2024-06-14 21:58:36 +00:00
|
|
|
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_reg = ResultatsRegionEuropeennes2024(region_id=str(dpt.region_code),
|
|
|
|
resultats_france_id=resultats_france.id)
|
2024-06-14 21:58:36 +00:00
|
|
|
session.add(resultats_reg)
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_reg = session.execute(select(ResultatsRegionEuropeennes2024)
|
2024-06-14 21:58:36 +00:00
|
|
|
.filter_by(region_id=dpt.region_code)).scalar_one()
|
|
|
|
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_dpt = ResultatsDepartementEuropeennes2024(dpt_id=dpt_code,
|
|
|
|
resultats_region_id=resultats_reg.id)
|
2024-06-14 21:58:36 +00:00
|
|
|
session.add(resultats_dpt)
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_dpt = session.execute(select(ResultatsDepartementEuropeennes2024)
|
|
|
|
.filter_by(dpt_id=dpt_code)).scalar_one()
|
2024-06-14 21:58:36 +00:00
|
|
|
|
|
|
|
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'])
|
|
|
|
|
2024-06-18 12:30:14 +00:00
|
|
|
for liste in session.execute(select(ListeEuropeennes2024)).scalars().all():
|
|
|
|
voix_liste_dpt = session.execute(select(VoixListeDepartementEuropeennes2024)
|
2024-06-14 21:58:36 +00:00
|
|
|
.filter_by(resultats_departement_id=resultats_dpt.id,
|
|
|
|
liste_id=liste.id)) \
|
|
|
|
.scalar_one_or_none()
|
|
|
|
if not voix_liste_dpt:
|
2024-06-18 12:30:14 +00:00
|
|
|
voix_liste_dpt = VoixListeDepartementEuropeennes2024(resultats_departement_id=resultats_dpt.id,
|
|
|
|
liste_id=liste.id)
|
2024-06-14 21:58:36 +00:00
|
|
|
session.add(voix_liste_dpt)
|
|
|
|
|
|
|
|
voix_liste_dpt.voix = int(row[f"Voix {liste.id}"])
|
|
|
|
|
2024-06-15 17:54:49 +00:00
|
|
|
session.commit()
|
2024-06-14 21:58:36 +00:00
|
|
|
|
|
|
|
|
|
|
|
def importer_resultats_region(engine: Engine, verbose: bool = False) -> None:
|
2024-06-15 17:54:49 +00:00
|
|
|
file = get_file("https://www.data.gouv.fr/fr/datasets/r/7c3a854b-7344-4c68-b1f9-9d651b4ca823",
|
2024-06-20 12:45:53 +00:00
|
|
|
"resultats-europeennes-2024-par-region.csv")
|
2024-06-14 21:58:36 +00:00
|
|
|
|
|
|
|
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):
|
2024-06-15 17:54:49 +00:00
|
|
|
reg_code = row['Code région']
|
2024-06-14 21:58:36 +00:00
|
|
|
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_reg = session.execute(select(ResultatsRegionEuropeennes2024).filter_by(region_id=reg_code)) \
|
2024-06-14 21:58:36 +00:00
|
|
|
.scalar_one_or_none()
|
|
|
|
if not resultats_reg:
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_france = session.execute(select(ResultatsFranceEuropeennes2024)).scalar_one_or_none()
|
2024-06-14 21:58:36 +00:00
|
|
|
if not resultats_france:
|
2024-06-18 12:30:14 +00:00
|
|
|
session.add(ResultatsFranceEuropeennes2024())
|
|
|
|
resultats_france = session.execute(select(ResultatsFranceEuropeennes2024)).scalar_one()
|
2024-06-14 21:58:36 +00:00
|
|
|
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_reg = ResultatsRegionEuropeennes2024(region_id=reg_code,
|
|
|
|
resultats_france_id=resultats_france.id)
|
2024-06-14 21:58:36 +00:00
|
|
|
session.add(resultats_reg)
|
2024-06-15 17:54:49 +00:00
|
|
|
session.commit()
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_reg = session.execute(select(ResultatsRegionEuropeennes2024)
|
|
|
|
.filter_by(region_id=reg_code)).scalar_one()
|
2024-06-14 21:58:36 +00:00
|
|
|
|
|
|
|
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'])
|
|
|
|
|
2024-06-18 12:30:14 +00:00
|
|
|
for liste in session.execute(select(ListeEuropeennes2024)).scalars().all():
|
|
|
|
voix_liste_reg = session.execute(select(VoixListeRegionEuropeennes2024)
|
2024-06-14 21:58:36 +00:00
|
|
|
.filter_by(resultats_region_id=resultats_reg.id,
|
|
|
|
liste_id=liste.id)) \
|
|
|
|
.scalar_one_or_none()
|
|
|
|
if not voix_liste_reg:
|
2024-06-18 12:30:14 +00:00
|
|
|
voix_liste_reg = VoixListeRegionEuropeennes2024(resultats_region_id=resultats_reg.id,
|
|
|
|
liste_id=liste.id)
|
2024-06-14 21:58:36 +00:00
|
|
|
session.add(voix_liste_reg)
|
|
|
|
|
|
|
|
voix_liste_reg.voix = int(row[f"Voix {liste.id}"])
|
|
|
|
|
|
|
|
session.commit()
|
2024-06-13 10:19:02 +00:00
|
|
|
|
|
|
|
|
2024-06-14 21:58:36 +00:00
|
|
|
def calculer_resultats_france(engine: Engine, verbose: bool = False) -> None:
|
2024-06-13 20:49:34 +00:00
|
|
|
with Session(engine) as session:
|
2024-06-18 12:30:14 +00:00
|
|
|
resultats_france = session.execute(select(ResultatsFranceEuropeennes2024)).scalar_one()
|
2024-06-14 21:58:36 +00:00
|
|
|
|
|
|
|
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(
|
2024-06-18 12:30:14 +00:00
|
|
|
select(VoixListeFranceEuropeennes2024)
|
|
|
|
.filter_by(resultats_france_id=resultats_france.id)).scalars().all():
|
2024-06-14 21:58:36 +00:00
|
|
|
voix_liste_france.voix = 0 # Réinitialisation des voix
|
|
|
|
|
2024-06-18 12:30:14 +00:00
|
|
|
for resultats_dpt in session.execute(select(ResultatsDepartementEuropeennes2024)).scalars().all():
|
2024-06-15 17:54:49 +00:00
|
|
|
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
|
|
|
|
|
2024-06-18 12:30:14 +00:00
|
|
|
for voix_liste_dpt in session.execute(select(VoixListeDepartementEuropeennes2024)
|
2024-06-15 17:54:49 +00:00
|
|
|
.filter_by(resultats_departement_id=resultats_dpt.id)) \
|
|
|
|
.scalars().all():
|
|
|
|
if voix_liste_france := session.execute(
|
2024-06-18 12:30:14 +00:00
|
|
|
select(VoixListeFranceEuropeennes2024)
|
|
|
|
.filter_by(resultats_france_id=resultats_france.id,
|
|
|
|
liste_id=voix_liste_dpt.liste_id)).scalar_one_or_none():
|
2024-06-15 17:54:49 +00:00
|
|
|
voix_liste_france.voix += voix_liste_dpt.voix
|
|
|
|
else:
|
2024-06-18 12:30:14 +00:00
|
|
|
session.add(VoixListeFranceEuropeennes2024(resultats_france_id=resultats_france.id,
|
|
|
|
liste_id=voix_liste_dpt.liste_id,
|
|
|
|
voix=voix_liste_dpt.voix))
|
2024-06-13 20:49:34 +00:00
|
|
|
|
|
|
|
session.commit()
|
2024-06-13 10:19:02 +00:00
|
|
|
|
|
|
|
|
2024-06-13 10:21:11 +00:00
|
|
|
def run(engine: Engine, verbose: bool = False) -> None:
|
2024-06-14 21:58:36 +00:00
|
|
|
importer_resultats_region(engine, verbose)
|
2024-06-15 17:54:49 +00:00
|
|
|
importer_resultats_departement(engine, verbose)
|
2024-06-14 21:58:36 +00:00
|
|
|
calculer_resultats_france(engine, verbose)
|
2024-06-15 17:54:49 +00:00
|
|
|
importer_resultats_circo(engine, verbose)
|
|
|
|
importer_resultats_commune(engine, verbose)
|
|
|
|
importer_resultats_bv(engine, verbose)
|