2024-07-02 12:43:48 +00:00
|
|
|
import csv
|
|
|
|
|
|
|
|
|
|
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
|
|
|
|
|
from nupes.models.legislatives2024 import *
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def importer_resultats_bv(engine: Engine, verbose: bool = False) -> None:
|
|
|
|
|
tours = [(1, "https://www.data.gouv.fr/fr/datasets/r/6813fb28-7ec0-42ff-a528-2bc3d82d7dcd")]
|
|
|
|
|
|
|
|
|
|
with Session(engine) as session:
|
|
|
|
|
candidats = {(candidat.circonscription_id, candidat.numero): candidat
|
|
|
|
|
for candidat in session.execute(select(CandidatLegislatives2024)).scalars().all()}
|
2024-07-02 19:54:02 +00:00
|
|
|
bv_ids = {(bv.commune_code, bv.code_bureau): bv
|
2024-07-02 12:43:48 +00:00
|
|
|
for bv in session.execute(select(BureauVote)).scalars().all()}
|
|
|
|
|
|
|
|
|
|
for tour, file_url in tours:
|
|
|
|
|
file = get_file(file_url, f"resultats-legislatives-2024-t{tour}-par-bureau-de-vote.csv")
|
|
|
|
|
|
|
|
|
|
with file.open('r') as f:
|
|
|
|
|
reader = csv.DictReader(f, delimiter=';')
|
|
|
|
|
with Session(engine) as session:
|
|
|
|
|
for row in tqdm(reader, desc=f"Bureau de vote tour {tour}", disable=not verbose):
|
|
|
|
|
dpt_code = row['Code département']
|
|
|
|
|
|
|
|
|
|
match dpt_code:
|
|
|
|
|
case "ZX":
|
|
|
|
|
dpt_code = "977"
|
|
|
|
|
|
|
|
|
|
if dpt_code == "987" or dpt_code == "988":
|
|
|
|
|
# Les communes de Polynésie française ne sont pas importées
|
|
|
|
|
continue
|
|
|
|
|
|
2024-07-02 15:14:28 +00:00
|
|
|
com_code = row['Code commune']
|
2024-07-02 12:43:48 +00:00
|
|
|
|
|
|
|
|
if com_code == "98601":
|
|
|
|
|
# 3 royaumes à Wallis-et-Futuna, mais un seul résultat de commune
|
|
|
|
|
com_code = "98611"
|
|
|
|
|
|
|
|
|
|
if com_code.startswith("987"): # Les communes de Polynésie française ne sont pas importées
|
|
|
|
|
continue
|
|
|
|
|
|
|
|
|
|
bv_code = row['Code BV']
|
|
|
|
|
|
|
|
|
|
if (com_code, bv_code) not in bv_ids:
|
|
|
|
|
print(f"Bureau de vote {com_code}_{bv_code} non trouvé")
|
|
|
|
|
continue
|
2024-07-02 19:54:02 +00:00
|
|
|
bv = bv_ids[(com_code, bv_code)]
|
|
|
|
|
bv_id = bv.id
|
|
|
|
|
circo_id = bv.circo_code
|
2024-07-02 12:43:48 +00:00
|
|
|
|
|
|
|
|
resultats_bv = session.execute(select(ResultatsBureauVoteLegislatives2024)
|
|
|
|
|
.filter_by(bv_id=bv_id)).scalar_one_or_none()
|
|
|
|
|
if not resultats_bv:
|
|
|
|
|
resultats_commune = session.execute(select(ResultatsCommuneLegislatives2024)
|
|
|
|
|
.filter_by(commune_id=com_code)).scalar_one_or_none()
|
|
|
|
|
if not resultats_commune:
|
|
|
|
|
resultats_dpt = session.execute(select(ResultatsDepartementLegislatives2024)
|
|
|
|
|
.filter_by(dpt_id=dpt_code)) \
|
|
|
|
|
.scalar_one_or_none()
|
|
|
|
|
if not resultats_dpt:
|
|
|
|
|
dpt = session.execute(select(Departement).filter_by(code_insee=dpt_code)) \
|
|
|
|
|
.scalar_one()
|
|
|
|
|
resultats_reg = session.execute(
|
|
|
|
|
select(ResultatsRegionLegislatives2024).filter_by(region_id=dpt.region_code)) \
|
|
|
|
|
.scalar_one_or_none()
|
|
|
|
|
if not resultats_reg:
|
|
|
|
|
resultats_france = session.execute(select(ResultatsFranceLegislatives2024)) \
|
|
|
|
|
.scalar_one_or_none()
|
|
|
|
|
if not resultats_france:
|
|
|
|
|
session.add(ResultatsFranceLegislatives2024())
|
|
|
|
|
resultats_france = session.execute(select(ResultatsFranceLegislatives2024)) \
|
|
|
|
|
.scalar_one()
|
|
|
|
|
|
|
|
|
|
resultats_reg = ResultatsRegionLegislatives2024(
|
|
|
|
|
region_id=str(dpt.region_code), resultats_france_id=resultats_france.id)
|
|
|
|
|
session.add(resultats_reg)
|
|
|
|
|
resultats_reg = session.execute(select(ResultatsRegionLegislatives2024)
|
|
|
|
|
.filter_by(region_id=dpt.region_code)).scalar_one()
|
|
|
|
|
|
|
|
|
|
resultats_dpt = ResultatsDepartementLegislatives2024(
|
|
|
|
|
dpt_id=dpt_code, resultats_region_id=resultats_reg.id)
|
|
|
|
|
session.add(resultats_dpt)
|
|
|
|
|
resultats_dpt = session.execute(select(ResultatsDepartementLegislatives2024)
|
|
|
|
|
.filter_by(dpt_id=dpt_code)).scalar_one()
|
|
|
|
|
|
|
|
|
|
resultats_commune = ResultatsCommuneLegislatives2024(commune_id=com_code,
|
|
|
|
|
resultats_dpt_id=resultats_dpt.id)
|
|
|
|
|
session.add(resultats_commune)
|
|
|
|
|
resultats_commune = session.execute(select(ResultatsCommuneLegislatives2024)
|
|
|
|
|
.filter_by(commune_id=com_code)).scalar_one()
|
|
|
|
|
|
|
|
|
|
resultats_circo = session.execute(select(ResultatsCirconscriptionLegislatives2024)
|
|
|
|
|
.filter_by(circo_id=circo_id)).scalar_one_or_none()
|
|
|
|
|
|
|
|
|
|
resultats_bv = ResultatsBureauVoteLegislatives2024(bv_id=bv_id,
|
|
|
|
|
resultats_commune_id=resultats_commune.id,
|
|
|
|
|
resultats_circo_id=resultats_circo.id)
|
|
|
|
|
session.add(resultats_bv)
|
|
|
|
|
resultats_bv = session.execute(select(ResultatsBureauVoteLegislatives2024)
|
|
|
|
|
.filter_by(bv_id=bv_id)).scalar_one()
|
|
|
|
|
|
|
|
|
|
if tour == 1:
|
|
|
|
|
resultats_bv.inscrits_t1 = int(row['Inscrits'])
|
|
|
|
|
resultats_bv.votants_t1 = int(row['Votants'])
|
|
|
|
|
resultats_bv.abstentions_t1 = int(row['Abstentions'])
|
2024-07-02 20:18:08 +00:00
|
|
|
resultats_bv.exprimes_t1 = int(row['Exprimés'])
|
2024-07-02 12:43:48 +00:00
|
|
|
resultats_bv.blancs_t1 = int(row['Blancs'])
|
|
|
|
|
resultats_bv.nuls_t1 = int(row['Nuls'])
|
|
|
|
|
elif tour == 2:
|
|
|
|
|
resultats_bv.inscrits_t2 = int(row['Inscrits'])
|
|
|
|
|
resultats_bv.votants_t2 = int(row['Votants'])
|
|
|
|
|
resultats_bv.abstentions_t2 = int(row['Abstentions'])
|
2024-07-02 20:18:08 +00:00
|
|
|
resultats_bv.exprimes_t2 = int(row['Exprimés'])
|
2024-07-02 12:43:48 +00:00
|
|
|
resultats_bv.blancs_t2 = int(row['Blancs'])
|
|
|
|
|
resultats_bv.nuls_t2 = int(row['Nuls'])
|
|
|
|
|
|
|
|
|
|
for i in range(1, 23):
|
|
|
|
|
if f'Nuance candidat {i}' not in row:
|
|
|
|
|
# On a épuisé toutes les nuances
|
|
|
|
|
break
|
|
|
|
|
|
|
|
|
|
numero = row[f'Numéro de panneau {i}']
|
|
|
|
|
if not numero:
|
|
|
|
|
continue
|
|
|
|
|
|
|
|
|
|
voix = int(row[f'Voix {i}'])
|
|
|
|
|
|
|
|
|
|
candidat = candidats[(circo_id, int(numero))]
|
|
|
|
|
|
|
|
|
|
voix_candidat_bv = session.execute(select(VoixBureauVoteLegislatives2024)
|
|
|
|
|
.filter_by(resultats_bureau_vote_id=resultats_bv.id,
|
|
|
|
|
candidat_id=candidat.id)) \
|
|
|
|
|
.scalar_one_or_none()
|
|
|
|
|
if not voix_candidat_bv:
|
|
|
|
|
voix_candidat_bv = VoixBureauVoteLegislatives2024(
|
|
|
|
|
resultats_bureau_vote_id=resultats_bv.id, candidat_id=candidat.id)
|
|
|
|
|
session.add(voix_candidat_bv)
|
|
|
|
|
|
|
|
|
|
if tour == 1:
|
|
|
|
|
voix_candidat_bv.voix_t1 = voix
|
|
|
|
|
elif tour == 2:
|
|
|
|
|
voix_candidat_bv.voix_t2 = voix
|
|
|
|
|
|
|
|
|
|
session.commit()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def importer_resultats_commune(engine: Engine, verbose: bool = False) -> None:
|
|
|
|
|
tours = [(1, "https://www.data.gouv.fr/fr/datasets/r/bd32fcd3-53df-47ac-bf1d-8d8003fe23a1")]
|
|
|
|
|
|
|
|
|
|
with Session(engine) as session:
|
|
|
|
|
for resultats_commune in session.execute(select(ResultatsCommuneLegislatives2024)).scalars().all():
|
|
|
|
|
resultats_commune.inscrits_t1 = 0
|
|
|
|
|
resultats_commune.votants_t1 = 0
|
|
|
|
|
resultats_commune.abstentions_t1 = 0
|
|
|
|
|
resultats_commune.exprimes_t1 = 0
|
|
|
|
|
resultats_commune.blancs_t1 = 0
|
|
|
|
|
resultats_commune.nuls_t1 = 0
|
|
|
|
|
resultats_commune.inscrits_t2 = 0
|
|
|
|
|
resultats_commune.votants_t2 = 0
|
|
|
|
|
resultats_commune.abstentions_t2 = 0
|
|
|
|
|
resultats_commune.exprimes_t2 = 0
|
|
|
|
|
resultats_commune.blancs_t2 = 0
|
|
|
|
|
resultats_commune.nuls_t2 = 0
|
|
|
|
|
|
|
|
|
|
for voix_nuance in resultats_commune.voix:
|
|
|
|
|
voix_nuance.voix_t1 = 0
|
|
|
|
|
voix_nuance.voix_t2 = 0
|
|
|
|
|
|
|
|
|
|
session.commit()
|
|
|
|
|
|
|
|
|
|
for tour, file_url in tours:
|
|
|
|
|
file = get_file(file_url,
|
|
|
|
|
f"resultats-legislatives-2024-t{tour}-par-commune.csv")
|
|
|
|
|
|
|
|
|
|
with file.open('r') as f:
|
|
|
|
|
reader = csv.DictReader(f, delimiter=';')
|
|
|
|
|
with Session(engine) as session:
|
|
|
|
|
for row in tqdm(reader, desc=f"Commune tour {tour}", disable=not verbose):
|
|
|
|
|
dpt_code = row['Code département']
|
|
|
|
|
|
|
|
|
|
match dpt_code:
|
|
|
|
|
case "ZX":
|
|
|
|
|
dpt_code = "977"
|
|
|
|
|
|
|
|
|
|
if dpt_code == "987" or dpt_code == "988":
|
|
|
|
|
# Les communes de Polynésie française ne sont pas importées
|
|
|
|
|
continue
|
|
|
|
|
|
2024-07-02 15:14:28 +00:00
|
|
|
com_code = row['Code commune']
|
2024-07-02 12:43:48 +00:00
|
|
|
|
|
|
|
|
if com_code == "98601":
|
|
|
|
|
# 3 royaumes à Wallis-et-Futuna, mais un seul résultat de commune
|
|
|
|
|
com_code = "98611"
|
|
|
|
|
|
|
|
|
|
resultats_commune = session.execute(select(ResultatsCommuneLegislatives2024)
|
|
|
|
|
.filter_by(commune_id=com_code)).scalar_one_or_none()
|
|
|
|
|
if not resultats_commune:
|
|
|
|
|
resultats_dpt = session.execute(select(ResultatsDepartementLegislatives2024)
|
|
|
|
|
.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(ResultatsRegionLegislatives2024)
|
|
|
|
|
.filter_by(region_id=dpt.region_code)).scalar_one_or_none()
|
|
|
|
|
if not resultats_reg:
|
|
|
|
|
resultats_france = session.execute(select(ResultatsFranceLegislatives2024)) \
|
|
|
|
|
.scalar_one_or_none()
|
|
|
|
|
if not resultats_france:
|
|
|
|
|
session.add(ResultatsFranceLegislatives2024())
|
|
|
|
|
resultats_france = session.execute(select(ResultatsFranceLegislatives2024)) \
|
|
|
|
|
.scalar_one()
|
|
|
|
|
|
|
|
|
|
resultats_reg = ResultatsRegionLegislatives2024(region_id=str(dpt.region_code),
|
|
|
|
|
resultats_france_id=resultats_france.id)
|
|
|
|
|
session.add(resultats_reg)
|
|
|
|
|
resultats_reg = session.execute(select(ResultatsRegionLegislatives2024)
|
|
|
|
|
.filter_by(region_id=dpt.region_code)).scalar_one()
|
|
|
|
|
|
|
|
|
|
resultats_dpt = ResultatsDepartementLegislatives2024(dpt_id=dpt_code,
|
|
|
|
|
resultats_region_id=resultats_reg.id)
|
|
|
|
|
session.add(resultats_dpt)
|
|
|
|
|
resultats_dpt = session.execute(select(ResultatsDepartementLegislatives2024)
|
|
|
|
|
.filter_by(dpt_id=dpt_code)).scalar_one()
|
|
|
|
|
|
|
|
|
|
resultats_commune = ResultatsCommuneLegislatives2024(commune_id=com_code,
|
|
|
|
|
resultats_dpt_id=resultats_dpt.id)
|
|
|
|
|
session.add(resultats_commune)
|
|
|
|
|
resultats_commune = session.execute(select(ResultatsCommuneLegislatives2024)
|
|
|
|
|
.filter_by(commune_id=com_code)).scalar_one()
|
|
|
|
|
|
|
|
|
|
if tour == 1:
|
|
|
|
|
resultats_commune.inscrits_t1 += int(row['Inscrits'])
|
|
|
|
|
resultats_commune.votants_t1 += int(row['Votants'])
|
|
|
|
|
resultats_commune.abstentions_t1 += int(row['Abstentions'])
|
|
|
|
|
resultats_commune.exprimes_t1 += int(row['Exprimés'])
|
|
|
|
|
resultats_commune.blancs_t1 += int(row['Blancs'])
|
|
|
|
|
resultats_commune.nuls_t1 += int(row['Nuls'])
|
|
|
|
|
elif tour == 2:
|
|
|
|
|
resultats_commune.inscrits_t2 += int(row['Inscrits'])
|
|
|
|
|
resultats_commune.votants_t2 += int(row['Votants'])
|
|
|
|
|
resultats_commune.abstentions_t2 += int(row['Abstentions'])
|
|
|
|
|
resultats_commune.exprimes_t2 += int(row['Exprimés'])
|
|
|
|
|
resultats_commune.blancs_t2 += int(row['Blancs'])
|
|
|
|
|
resultats_commune.nuls_t2 += int(row['Nuls'])
|
|
|
|
|
|
|
|
|
|
for i in range(1, 23):
|
|
|
|
|
if f'Nuance candidat {i}' not in row:
|
|
|
|
|
# On a épuisé toutes les nuances
|
|
|
|
|
break
|
|
|
|
|
|
|
|
|
|
nuance_code = row[f'Nuance candidat {i}']
|
|
|
|
|
if not nuance_code:
|
|
|
|
|
continue
|
|
|
|
|
|
|
|
|
|
voix = int(row[f'Voix {i}'])
|
|
|
|
|
|
|
|
|
|
voix_nuance_commune = session.execute(select(VoixCommuneLegislatives2024)
|
|
|
|
|
.filter_by(resultats_commune_id=resultats_commune.id,
|
|
|
|
|
nuance_id=nuance_code)) \
|
|
|
|
|
.scalar_one_or_none()
|
|
|
|
|
if not voix_nuance_commune:
|
|
|
|
|
voix_nuance_commune = VoixCommuneLegislatives2024(resultats_commune_id=resultats_commune.id,
|
|
|
|
|
nuance_id=nuance_code,
|
|
|
|
|
voix_t1=0, voix_t2=0)
|
|
|
|
|
session.add(voix_nuance_commune)
|
|
|
|
|
|
|
|
|
|
if tour == 1:
|
|
|
|
|
voix_nuance_commune.voix_t1 += voix
|
|
|
|
|
elif tour == 2:
|
|
|
|
|
voix_nuance_commune.voix_t2 += voix
|
|
|
|
|
|
|
|
|
|
session.commit()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def importer_resultats_circo(engine: Engine, verbose: bool = False) -> None:
|
|
|
|
|
tours = [(1, "https://www.data.gouv.fr/fr/datasets/r/5163f2e3-1362-4c35-89a0-1934bb74f2d9")]
|
|
|
|
|
|
|
|
|
|
for tour, file_url in tours:
|
|
|
|
|
file = get_file(file_url, f"resultats-legislatives-2024-t{tour}-par-circonscription.csv")
|
|
|
|
|
|
|
|
|
|
with file.open('r') as f:
|
|
|
|
|
reader = csv.DictReader(f, delimiter=';')
|
|
|
|
|
with Session(engine) as session:
|
|
|
|
|
for row in tqdm(reader, desc=f"Circonscription tour {tour}", disable=not verbose):
|
|
|
|
|
dpt_code = row['Code département']
|
|
|
|
|
circo_code = row['Code circonscription législative'][-2:]
|
|
|
|
|
|
|
|
|
|
match dpt_code:
|
|
|
|
|
case "ZX":
|
|
|
|
|
dpt_code = "977"
|
|
|
|
|
|
|
|
|
|
circo_id = f"{dpt_code.zfill(2)}-{circo_code.zfill(2)}"
|
|
|
|
|
|
|
|
|
|
resultats_circo = session.execute(select(ResultatsCirconscriptionLegislatives2024)
|
|
|
|
|
.filter_by(circo_id=circo_id)).scalar_one_or_none()
|
|
|
|
|
if not resultats_circo:
|
|
|
|
|
resultats_dpt = session.execute(select(ResultatsDepartementLegislatives2024)
|
|
|
|
|
.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(ResultatsRegionLegislatives2024)
|
|
|
|
|
.filter_by(region_id=dpt.region_code)) \
|
|
|
|
|
.scalar_one_or_none()
|
|
|
|
|
if not resultats_reg:
|
|
|
|
|
resultats_france = session.execute(select(ResultatsFranceLegislatives2024)) \
|
|
|
|
|
.scalar_one_or_none()
|
|
|
|
|
if not resultats_france:
|
|
|
|
|
session.add(ResultatsFranceLegislatives2024())
|
|
|
|
|
resultats_france = session.execute(select(ResultatsFranceLegislatives2024)).scalar_one()
|
|
|
|
|
|
|
|
|
|
resultats_reg = ResultatsRegionLegislatives2024(region_id=str(dpt.region_code),
|
|
|
|
|
resultats_france_id=resultats_france.id)
|
|
|
|
|
session.add(resultats_reg)
|
|
|
|
|
resultats_reg = session.execute(select(ResultatsRegionLegislatives2024)
|
|
|
|
|
.filter_by(region_id=dpt.region_code)).scalar_one()
|
|
|
|
|
|
|
|
|
|
resultats_dpt = ResultatsDepartementLegislatives2024(dpt_id=dpt_code,
|
|
|
|
|
resultats_region_id=resultats_reg.id)
|
|
|
|
|
session.add(resultats_dpt)
|
|
|
|
|
resultats_dpt = session.execute(select(ResultatsDepartementLegislatives2024)
|
|
|
|
|
.filter_by(dpt_id=dpt_code)).scalar_one()
|
|
|
|
|
|
|
|
|
|
resultats_circo = ResultatsCirconscriptionLegislatives2024(
|
|
|
|
|
circo_id=circo_id, resultats_departement_id=resultats_dpt.id)
|
|
|
|
|
session.add(resultats_circo)
|
|
|
|
|
resultats_circo = session.execute(select(ResultatsCirconscriptionLegislatives2024)
|
|
|
|
|
.filter_by(circo_id=circo_id)).scalar_one()
|
|
|
|
|
|
|
|
|
|
if tour == 1:
|
|
|
|
|
resultats_circo.inscrits_t1 = int(row['Inscrits'])
|
|
|
|
|
resultats_circo.votants_t1 = int(row['Votants'])
|
|
|
|
|
resultats_circo.abstentions_t1 = int(row['Abstentions'])
|
|
|
|
|
resultats_circo.exprimes_t1 = int(row['Exprimés'])
|
|
|
|
|
resultats_circo.blancs_t1 = int(row['Blancs'])
|
|
|
|
|
resultats_circo.nuls_t1 = int(row['Nuls'])
|
|
|
|
|
elif tour == 2:
|
|
|
|
|
resultats_circo.inscrits_t2 = int(row['Inscrits'])
|
|
|
|
|
resultats_circo.votants_t2 = int(row['Votants'])
|
|
|
|
|
resultats_circo.abstentions_t2 = int(row['Abstentions'])
|
|
|
|
|
resultats_circo.exprimes_t2 = int(row['Exprimés'])
|
|
|
|
|
resultats_circo.blancs_t2 = int(row['Blancs'])
|
|
|
|
|
resultats_circo.nuls_t2 = int(row['Nuls'])
|
|
|
|
|
|
|
|
|
|
for i in range(1, 23):
|
|
|
|
|
if f'Nuance candidat {i}' not in row:
|
|
|
|
|
# On a épuisé toutes les nuances
|
|
|
|
|
break
|
|
|
|
|
|
|
|
|
|
numero = row[f'Numéro de panneau {i}']
|
|
|
|
|
if not numero:
|
|
|
|
|
continue
|
|
|
|
|
|
|
|
|
|
voix = int(row[f'Voix {i}'])
|
|
|
|
|
|
|
|
|
|
candidat = session.execute(select(CandidatLegislatives2024)
|
|
|
|
|
.filter_by(circonscription_id=circo_id, numero=numero)) \
|
|
|
|
|
.scalar_one_or_none()
|
|
|
|
|
|
|
|
|
|
voix_candidat_circo = session.execute(select(VoixCirconscriptionLegislatives2024)
|
|
|
|
|
.filter_by(resultats_circonscription_id=resultats_circo.id,
|
|
|
|
|
candidat_id=candidat.id)) \
|
|
|
|
|
.scalar_one_or_none()
|
|
|
|
|
if not voix_candidat_circo:
|
|
|
|
|
voix_candidat_circo = VoixCirconscriptionLegislatives2024(
|
|
|
|
|
resultats_circonscription_id=resultats_circo.id, candidat_id=candidat.id)
|
|
|
|
|
session.add(voix_candidat_circo)
|
|
|
|
|
|
|
|
|
|
if tour == 1:
|
|
|
|
|
voix_candidat_circo.voix_t1 = voix
|
|
|
|
|
elif tour == 2:
|
|
|
|
|
voix_candidat_circo.voix_t2 = voix
|
|
|
|
|
|
|
|
|
|
session.commit()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def importer_resultats_departement(engine: Engine, verbose: bool = False) -> None:
|
|
|
|
|
tours = [(1, "https://www.data.gouv.fr/fr/datasets/r/78c708c5-5bc5-438d-8379-f432beae3f2b")]
|
|
|
|
|
|
|
|
|
|
for tour, file_url in tours:
|
|
|
|
|
file = get_file(file_url, f"resultats-legislatives-2024-t{tour}-par-departement.csv")
|
|
|
|
|
|
|
|
|
|
with file.open('r') as f:
|
|
|
|
|
reader = csv.DictReader(f, delimiter=';')
|
|
|
|
|
with Session(engine) as session:
|
|
|
|
|
for row in tqdm(reader, desc=f"Département tour {tour}", disable=not verbose):
|
|
|
|
|
dpt_code = row['Code département']
|
|
|
|
|
|
|
|
|
|
match dpt_code:
|
|
|
|
|
case "ZX":
|
|
|
|
|
dpt_code = "977"
|
|
|
|
|
|
|
|
|
|
resultats_dpt = session.execute(select(ResultatsDepartementLegislatives2024)
|
|
|
|
|
.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(ResultatsRegionLegislatives2024)
|
|
|
|
|
.filter_by(region_id=dpt.region_code)) \
|
|
|
|
|
.scalar_one_or_none()
|
|
|
|
|
if not resultats_reg:
|
|
|
|
|
resultats_france = session.execute(select(ResultatsFranceLegislatives2024)) \
|
|
|
|
|
.scalar_one_or_none()
|
|
|
|
|
if not resultats_france:
|
|
|
|
|
session.add(ResultatsFranceLegislatives2024())
|
|
|
|
|
resultats_france = session.execute(select(ResultatsFranceLegislatives2024)).scalar_one()
|
|
|
|
|
|
|
|
|
|
resultats_reg = ResultatsRegionLegislatives2024(region_id=str(dpt.region_code),
|
|
|
|
|
resultats_france_id=resultats_france.id)
|
|
|
|
|
session.add(resultats_reg)
|
|
|
|
|
resultats_reg = session.execute(select(ResultatsRegionLegislatives2024)
|
|
|
|
|
.filter_by(region_id=dpt.region_code)).scalar_one()
|
|
|
|
|
|
|
|
|
|
resultats_dpt = ResultatsDepartementLegislatives2024(dpt_id=dpt_code,
|
|
|
|
|
resultats_region_id=resultats_reg.id)
|
|
|
|
|
session.add(resultats_dpt)
|
|
|
|
|
resultats_dpt = session.execute(select(ResultatsDepartementLegislatives2024)
|
|
|
|
|
.filter_by(dpt_id=dpt_code)).scalar_one()
|
|
|
|
|
|
|
|
|
|
if tour == 1:
|
|
|
|
|
resultats_dpt.inscrits_t1 = int(row['Inscrits'])
|
|
|
|
|
resultats_dpt.votants_t1 = int(row['Votants'])
|
|
|
|
|
resultats_dpt.abstentions_t1 = int(row['Abstentions'])
|
|
|
|
|
resultats_dpt.exprimes_t1 = int(row['Exprimés'])
|
|
|
|
|
resultats_dpt.blancs_t1 = int(row['Blancs'])
|
|
|
|
|
resultats_dpt.nuls_t1 = int(row['Nuls'])
|
|
|
|
|
elif tour == 2:
|
|
|
|
|
resultats_dpt.inscrits_t2 = int(row['Inscrits'])
|
|
|
|
|
resultats_dpt.votants_t2 = int(row['Votants'])
|
|
|
|
|
resultats_dpt.abstentions_t2 = int(row['Abstentions'])
|
|
|
|
|
resultats_dpt.exprimes_t2 = int(row['Exprimés'])
|
|
|
|
|
resultats_dpt.blancs_t2 = int(row['Blancs'])
|
|
|
|
|
resultats_dpt.nuls_t2 = int(row['Nuls'])
|
|
|
|
|
|
|
|
|
|
for i in range(1, 23):
|
|
|
|
|
if f'Nuance candidat {i}' not in row:
|
|
|
|
|
# On a épuisé toutes les nuances
|
|
|
|
|
break
|
|
|
|
|
|
|
|
|
|
nuance_code = row[f'Nuance candidat {i}']
|
|
|
|
|
if not nuance_code:
|
|
|
|
|
continue
|
|
|
|
|
|
|
|
|
|
voix = int(row[f'Voix {i}'])
|
|
|
|
|
|
|
|
|
|
voix_nuance_dpt = session.execute(select(VoixDepartementLegislatives2024)
|
|
|
|
|
.filter_by(resultats_departement_id=resultats_dpt.id,
|
|
|
|
|
nuance_id=nuance_code)) \
|
|
|
|
|
.scalar_one_or_none()
|
|
|
|
|
if not voix_nuance_dpt:
|
|
|
|
|
voix_nuance_dpt = VoixDepartementLegislatives2024(resultats_departement_id=resultats_dpt.id,
|
|
|
|
|
nuance_id=nuance_code)
|
|
|
|
|
session.add(voix_nuance_dpt)
|
|
|
|
|
|
|
|
|
|
if tour == 1:
|
|
|
|
|
voix_nuance_dpt.voix_t1 = voix
|
|
|
|
|
elif tour == 2:
|
|
|
|
|
voix_nuance_dpt.voix_t2 = voix
|
|
|
|
|
|
|
|
|
|
session.commit()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def importer_resultats_region(engine: Engine, verbose: bool = False) -> None:
|
|
|
|
|
tours = [(1, "https://www.data.gouv.fr/fr/datasets/r/f69ffab7-fe37-494e-ad6d-a7cfc80ddc1f")]
|
|
|
|
|
|
|
|
|
|
for tour, file_url in tours:
|
|
|
|
|
file = get_file(file_url, f"resultats-legislatives-2024-t{tour}-par-region.csv")
|
|
|
|
|
|
|
|
|
|
with file.open('r') as f:
|
|
|
|
|
reader = csv.DictReader(f, delimiter=';')
|
|
|
|
|
|
|
|
|
|
with Session(engine) as session:
|
|
|
|
|
for row in tqdm(reader, desc=f"Région tour {tour}", disable=not verbose):
|
|
|
|
|
reg_code = row['Code région']
|
|
|
|
|
|
|
|
|
|
resultats_reg = session.execute(select(ResultatsRegionLegislatives2024)
|
|
|
|
|
.filter_by(region_id=reg_code)).scalar_one_or_none()
|
|
|
|
|
if not resultats_reg:
|
|
|
|
|
resultats_france = session.execute(select(ResultatsFranceLegislatives2024)).scalar_one_or_none()
|
|
|
|
|
if not resultats_france:
|
|
|
|
|
session.add(ResultatsFranceLegislatives2024())
|
|
|
|
|
resultats_france = session.execute(select(ResultatsFranceLegislatives2024)).scalar_one()
|
|
|
|
|
|
|
|
|
|
resultats_reg = ResultatsRegionLegislatives2024(region_id=reg_code,
|
|
|
|
|
resultats_france_id=resultats_france.id)
|
|
|
|
|
session.add(resultats_reg)
|
|
|
|
|
session.commit()
|
|
|
|
|
resultats_reg = session.execute(select(ResultatsRegionLegislatives2024)
|
|
|
|
|
.filter_by(region_id=reg_code)).scalar_one()
|
|
|
|
|
|
|
|
|
|
if tour == 1:
|
|
|
|
|
resultats_reg.inscrits_t1 = int(row['Inscrits'])
|
|
|
|
|
resultats_reg.votants_t1 = int(row['Votants'])
|
|
|
|
|
resultats_reg.abstentions_t1 = int(row['Abstentions'])
|
|
|
|
|
resultats_reg.exprimes_t1 = int(row['Exprimés'])
|
|
|
|
|
resultats_reg.blancs_t1 = int(row['Blancs'])
|
|
|
|
|
resultats_reg.nuls_t1 = int(row['Nuls'])
|
|
|
|
|
elif tour == 2:
|
|
|
|
|
resultats_reg.inscrits_t2 = int(row['Inscrits'])
|
|
|
|
|
resultats_reg.votants_t2 = int(row['Votants'])
|
|
|
|
|
resultats_reg.abstentions_t2 = int(row['Abstentions'])
|
|
|
|
|
resultats_reg.exprimes_t2 = int(row['Exprimés'])
|
|
|
|
|
resultats_reg.blancs_t2 = int(row['Blancs'])
|
|
|
|
|
resultats_reg.nuls_t2 = int(row['Nuls'])
|
|
|
|
|
|
|
|
|
|
for i in range(1, 23):
|
|
|
|
|
if f'Nuance candidat {i}' not in row:
|
|
|
|
|
# On a épuisé toutes les nuances
|
|
|
|
|
break
|
|
|
|
|
|
|
|
|
|
nuance_code = row[f'Nuance candidat {i}']
|
|
|
|
|
if not nuance_code:
|
|
|
|
|
continue
|
|
|
|
|
|
|
|
|
|
voix = int(row[f'Voix {i}'])
|
|
|
|
|
|
|
|
|
|
voix_nuance_reg = session.execute(select(VoixRegionLegislatives2024)
|
|
|
|
|
.filter_by(resultats_region_id=resultats_reg.id,
|
|
|
|
|
nuance_id=nuance_code)) \
|
|
|
|
|
.scalar_one_or_none()
|
|
|
|
|
if not voix_nuance_reg:
|
|
|
|
|
voix_nuance_reg = VoixRegionLegislatives2024(resultats_region_id=resultats_reg.id,
|
|
|
|
|
nuance_id=nuance_code)
|
|
|
|
|
session.add(voix_nuance_reg)
|
|
|
|
|
|
|
|
|
|
if tour == 1:
|
|
|
|
|
voix_nuance_reg.voix_t1 = voix
|
|
|
|
|
elif tour == 2:
|
|
|
|
|
voix_nuance_reg.voix_t2 = voix
|
|
|
|
|
|
|
|
|
|
session.commit()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def importer_resultats_france(engine: Engine, verbose: bool = False) -> None:
|
|
|
|
|
tours = [(1, "https://www.data.gouv.fr/fr/datasets/r/386fd5ac-e7f1-4e0f-8929-12d2c5391081")]
|
|
|
|
|
for tour, file_url in tours:
|
|
|
|
|
file = get_file(file_url, f"resultats-legislatives-2024-t{tour}-france-entiere.csv")
|
|
|
|
|
|
|
|
|
|
with file.open('r') as f:
|
|
|
|
|
reader = csv.DictReader(f, delimiter=';')
|
|
|
|
|
|
|
|
|
|
with Session(engine) as session:
|
|
|
|
|
for row in tqdm(reader, desc=f"France Entière tour {tour}", disable=not verbose):
|
|
|
|
|
resultats_france = session.execute(select(ResultatsFranceLegislatives2024)).scalar_one_or_none()
|
|
|
|
|
|
|
|
|
|
if not resultats_france:
|
|
|
|
|
session.add(ResultatsFranceLegislatives2024())
|
|
|
|
|
session.commit()
|
|
|
|
|
resultats_france = session.execute(select(ResultatsFranceLegislatives2024)).scalar_one()
|
|
|
|
|
|
|
|
|
|
if tour == 1:
|
|
|
|
|
resultats_france.inscrits_t1 = int(row['Inscrits'])
|
|
|
|
|
resultats_france.votants_t1 = int(row['Votants'])
|
|
|
|
|
resultats_france.abstentions_t1 = int(row['Abstentions'])
|
|
|
|
|
resultats_france.exprimes_t1 = int(row['Votants'])
|
|
|
|
|
resultats_france.blancs_t1 = int(row['Blancs'])
|
|
|
|
|
resultats_france.nuls_t1 = int(row['Nuls'])
|
|
|
|
|
elif tour == 2:
|
|
|
|
|
resultats_france.inscrits_t2 = int(row['Inscrits'])
|
|
|
|
|
resultats_france.votants_t2 = int(row['Votants'])
|
|
|
|
|
resultats_france.abstentions_t2 = int(row['Abstentions'])
|
|
|
|
|
resultats_france.exprimes_t2 = int(row['Votants'])
|
|
|
|
|
resultats_france.blancs_t2 = int(row['Blancs'])
|
|
|
|
|
resultats_france.nuls_t2 = int(row['Nuls'])
|
|
|
|
|
|
|
|
|
|
for i in range(1, 23):
|
|
|
|
|
if f'Nuance candidat {i}' not in row:
|
|
|
|
|
# On a épuisé toutes les nuances
|
|
|
|
|
break
|
|
|
|
|
|
|
|
|
|
nuance_code = row[f'Nuance candidat {i}']
|
|
|
|
|
voix = int(row[f'Voix {i}'])
|
|
|
|
|
|
|
|
|
|
voix_nuance_france = session.execute(select(VoixFranceLegislatives2024)
|
|
|
|
|
.filter_by(resultats_france_id=resultats_france.id,
|
|
|
|
|
nuance_id=nuance_code)) \
|
|
|
|
|
.scalar_one_or_none()
|
|
|
|
|
if not voix_nuance_france:
|
|
|
|
|
voix_nuance_france = VoixFranceLegislatives2024(resultats_france_id=resultats_france.id,
|
|
|
|
|
nuance_id=nuance_code)
|
|
|
|
|
session.add(voix_nuance_france)
|
|
|
|
|
|
|
|
|
|
if tour == 1:
|
|
|
|
|
voix_nuance_france.voix_t1 = voix
|
|
|
|
|
elif tour == 2:
|
|
|
|
|
voix_nuance_france.voix_t2 = voix
|
|
|
|
|
|
|
|
|
|
session.commit()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def run(engine: Engine, verbose: bool = False) -> None:
|
|
|
|
|
importer_resultats_france(engine, verbose)
|
|
|
|
|
importer_resultats_region(engine, verbose)
|
|
|
|
|
importer_resultats_departement(engine, verbose)
|
|
|
|
|
importer_resultats_circo(engine, verbose)
|
2024-07-02 20:18:08 +00:00
|
|
|
# importer_resultats_commune(engine, verbose)
|
2024-07-02 12:43:48 +00:00
|
|
|
importer_resultats_bv(engine, verbose)
|
