nupes-elections/nupes/scripts/legislatives2024/import_candidats.py

160 lines
8.1 KiB
Python
Raw Normal View History

2024-06-20 10:17:56 +00:00
import csv
import datetime
from sqlalchemy import Engine, select
from sqlalchemy.orm import Session
from tqdm import tqdm
2024-06-20 10:17:56 +00:00
from nupes.cache import get_file
2024-06-20 14:48:34 +00:00
from nupes.models import Genre
2024-06-20 10:17:56 +00:00
from nupes.models.legislatives2024 import BlocLegislatives2024, NuanceLegislatives2024, CandidatLegislatives2024
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(BlocLegislatives2024).filter_by(id=bloc_dict["id"])).scalar_one_or_none():
bloc.nom = bloc_dict["nom"]
bloc.couleur = bloc_dict["couleur"]
else:
session.add(BlocLegislatives2024(**bloc_dict))
session.commit()
def creer_nuances(engine: Engine, verbose: bool = False) -> None:
nuances = [
{"code": "EXG", "nom": "Extrême gauche", "couleur": "#BB0000", "bloc_id": 1},
{"code": "COM", "nom": "Parti communiste français", "couleur": "#DD0000", "bloc_id": 1},
{"code": "FI", "nom": "La France insoumise", "couleur": "#CC2443", "bloc_id": 1},
{"code": "SOC", "nom": "Parti socialiste", "couleur": "#FF8080", "bloc_id": 1},
{"code": "RDG", "nom": "Parti radical de gauche", "couleur": "#FFD1DC", "bloc_id": 1},
{"code": "UG", "nom": "Union de la gauche", "couleur": "#E4032E", "bloc_id": 1},
{"code": "VEC", "nom": "Les Écologistes", "couleur": "#00C000", "bloc_id": 1},
{"code": "DVG", "nom": "Divers gauche", "couleur": "#FFC0C0", "bloc_id": 1},
{"code": "ECO", "nom": "Écologistes", "couleur": "#77FF77", "bloc_id": 5},
{"code": "REG", "nom": "Régionalistes", "couleur": "#DCBFA3", "bloc_id": 5},
{"code": "DIV", "nom": "Divers", "couleur": "#DCDCDC", "bloc_id": 5},
{"code": "REN", "nom": "Renaissance", "couleur": "#FFEB00", "bloc_id": 2},
{"code": "ENS", "nom": "Ensemble ! (Majorité présidentielle)", "couleur": "#45B3AB", "bloc_id": 2},
{"code": "MDM", "nom": "Modem", "couleur": "#FF9900", "bloc_id": 2},
{"code": "HOR", "nom": "Horizons", "couleur": "#0001B8", "bloc_id": 2},
{"code": "DVC", "nom": "Divers centre", "couleur": "#FAC577", "bloc_id": 2},
{"code": "UDI", "nom": "Union des Démocrates et Indépendants", "couleur": "#00FFFF", "bloc_id": 3},
{"code": "LR", "nom": "Les Républicains", "couleur": "#0066CC", "bloc_id": 3},
{"code": "DVD", "nom": "Divers droite", "couleur": "#26C4EC", "bloc_id": 3},
{"code": "DSV", "nom": "Droite souverainiste", "couleur": "#8040C0", "bloc_id": 4},
{"code": "RN", "nom": "Rassemblement national", "couleur": "#0D378A", "bloc_id": 4},
{"code": "REC", "nom": "Reconquête !", "couleur": "#404040", "bloc_id": 4},
{"code": "EXD", "nom": "Extrême droite", "couleur": "#404040", "bloc_id": 4},
{"code": "UXD", "nom": "Union de l'extrême droite", "couleur": "#404040", "bloc_id": 4},
2024-06-20 10:17:56 +00:00
]
with Session(engine) as session:
for nuance_dict in nuances:
if nuance := session.execute(select(NuanceLegislatives2024)
.filter_by(code=nuance_dict["code"])).scalar_one_or_none():
2024-06-20 10:17:56 +00:00
nuance.nom = nuance_dict["nom"]
nuance.couleur = nuance_dict["couleur"]
else:
session.add(NuanceLegislatives2024(**nuance_dict))
session.commit()
def importer_candidats(engine: Engine, verbose: bool = False) -> None:
DATASET_URL = "https://www.data.gouv.fr/fr/datasets/r/e02a62a1-9f4b-49d6-bb32-05dc4da8c581"
file = get_file(DATASET_URL, "candidats_legislatives_2024.csv")
with file.open('r') as f:
with Session(engine) as session:
reader = csv.DictReader(f)
for line in tqdm(reader, desc="Candidat⋅es", disable=not verbose):
line: dict[str, str]
numero_panneau = int(line.get("Numéro de panneau"))
code_dpt = line.get("Code département").zfill(2)
match code_dpt:
case "ZA":
code_dpt = "971"
case "ZB":
code_dpt = "972"
case "ZC":
code_dpt = "973"
case "ZD":
code_dpt = "974"
case "ZS":
code_dpt = "975"
case "ZM":
code_dpt = "976"
case "ZX":
code_dpt = "977"
case "ZW":
code_dpt = "986"
case "ZP":
code_dpt = "987"
case "ZN":
code_dpt = "988"
code_circo = line.get("Code circonscription")[-2:]
circo_id = f"{code_dpt}-{code_circo}"
nuance_id = line.get("Code nuance")
nuance = session.execute(select(NuanceLegislatives2024).filter_by(code=nuance_id)).scalar_one()
if candidat := session.execute(select(CandidatLegislatives2024)
.filter_by(circonscription_id=circo_id, numero=numero_panneau)) \
.scalar_one_or_none():
candidat.nuance_id = nuance_id
candidat.bloc_id = nuance.bloc_id
candidat.nom = line['Nom du candidat']
candidat.prenom = line['Prénom du candidat']
2024-06-20 14:48:34 +00:00
candidat.sexe = Genre(line['Sexe du candidat'])
candidat.date_naissance = datetime.datetime.strptime(line['Date de naissance du candidat'],
"%d/%m/%Y").date()
candidat.profession = line['Profession']
candidat.sortant = line['Sortant'] == "OUI"
candidat.nom_suppleance = line['Nom remplaçant']
candidat.prenom_suppleance = line['Prénom remplaçant']
2024-06-20 14:48:34 +00:00
candidat.sexe_suppleance = Genre(line['Sexe remplaçant'])
candidat.date_naissance_suppleance = datetime.datetime.strptime(
line['Date de naissance remplaçant'], "%d/%m/%Y").date()
candidat.sortant = line['Sortant remplaçant'] == "OUI"
else:
candidat = CandidatLegislatives2024(
circonscription_id=circo_id,
numero=numero_panneau,
nuance_id=nuance_id,
bloc_id=nuance.bloc_id,
nom=line['Nom du candidat'],
prenom=line['Prénom du candidat'],
2024-06-20 14:48:34 +00:00
sexe=Genre(line['Sexe du candidat']).name,
date_naissance=datetime.datetime.strptime(line['Date de naissance du candidat'],
"%d/%m/%Y").date(),
profession=line['Profession'],
sortant=line['Sortant'] == "OUI",
nom_suppleance=line['Nom remplaçant'],
prenom_suppleance=line['Prénom remplaçant'],
2024-06-20 14:48:34 +00:00
sexe_suppleance=Genre(line['Sexe remplaçant']).name,
date_naissance_suppleance=datetime.datetime.strptime(line['Date de naissance remplaçant'],
"%d/%m/%Y").date(),
sortant_suppleance=line['Sortant remplaçant'] == "OUI",
)
session.add(candidat)
session.commit()
2024-06-20 10:17:56 +00:00
def run(engine: Engine, verbose: bool = False) -> None:
creer_blocs(engine, verbose)
creer_nuances(engine, verbose)
importer_candidats(engine, verbose)