151 lines
7.6 KiB
Python
151 lines
7.6 KiB
Python
|
import csv
|
||
|
|
import datetime
|
||
|
|
|
||
|
|
from sqlalchemy import Engine, select
|
||
|
|
from sqlalchemy.orm import Session
|
||
|
|
from tqdm import tqdm
|
||
|
|
|
||
|
|
from nupes.cache import get_file
|
||
|
|
from nupes.models.legislatives2022 import BlocLegislatives2022, NuanceLegislatives2022, CandidatLegislatives2022
|
||
|
|
|
||
|
|
|
||
|
|
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(BlocLegislatives2022).filter_by(id=bloc_dict["id"])).scalar_one_or_none():
|
||
|
|
bloc.nom = bloc_dict["nom"]
|
||
|
|
bloc.couleur = bloc_dict["couleur"]
|
||
|
|
else:
|
||
|
|
session.add(BlocLegislatives2022(**bloc_dict))
|
||
|
|
|
||
|
|
session.commit()
|
||
|
|
|
||
|
|
|
||
|
|
def creer_nuances(engine: Engine, verbose: bool = False) -> None:
|
||
|
|
nuances = [
|
||
|
|
{"code": "DXG", "nom": "Divers extrême gauche", "couleur": "#BB0000", "bloc_id": 1},
|
||
|
|
{"code": "RDG", "nom": "Radical de gauche", "couleur": "#FFD1DC", "bloc_id": 1},
|
||
|
|
{"code": "NUP", "nom": "Nouvelle union populaire écologique et sociale", "couleur": "#E4032E", "bloc_id": 1},
|
||
|
|
{"code": "DVG", "nom": "Divers gauche", "couleur": "#FFC0C0", "bloc_id": 1},
|
||
|
|
{"code": "ECO", "nom": "Écologistes", "couleur": "#77FF77", "bloc_id": 5},
|
||
|
|
{"code": "DIV", "nom": "Divers", "couleur": "#DCDCDC", "bloc_id": 5},
|
||
|
|
{"code": "REG", "nom": "Régionalistes", "couleur": "#DCBFA3", "bloc_id": 5},
|
||
|
|
{"code": "ENS", "nom": "Ensemble ! (Majorité présidentielle)", "couleur": "#FFEB00", "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": "REC", "nom": "Reconquête !", "couleur": "#404040", "bloc_id": 4},
|
||
|
|
{"code": "RN", "nom": "Rassemblement national", "couleur": "#0D378A", "bloc_id": 4},
|
||
|
|
{"code": "DXD", "nom": "Divers extrême droite", "couleur": "#404040", "bloc_id": 4},
|
||
|
|
]
|
||
|
|
|
||
|
|
with Session(engine) as session:
|
||
|
|
for nuance_dict in nuances:
|
||
|
|
if nuance := session.execute(select(NuanceLegislatives2022)
|
||
|
|
.filter_by(code=nuance_dict["code"])).scalar_one_or_none():
|
||
|
|
nuance.nom = nuance_dict["nom"]
|
||
|
|
nuance.couleur = nuance_dict["couleur"]
|
||
|
|
nuance.bloc_id = nuance_dict["bloc_id"]
|
||
|
|
else:
|
||
|
|
session.add(NuanceLegislatives2022(**nuance_dict))
|
||
|
|
|
||
|
|
session.commit()
|
||
|
|
|
||
|
|
|
||
|
|
def importer_candidats(engine: Engine, verbose: bool = False) -> None:
|
||
|
|
DATASET_URL = "https://www.data.gouv.fr/fr/datasets/r/c79af2f3-1733-4df3-be37-caebacbc1321"
|
||
|
|
file = get_file(DATASET_URL, "candidats_legislatives_2022.tsv")
|
||
|
|
|
||
|
|
with file.open('r', encoding="ISO-8859-1") as f:
|
||
|
|
with Session(engine) as session:
|
||
|
|
reader = csv.DictReader(f, delimiter='\t')
|
||
|
|
for line in tqdm(reader, desc="Candidat⋅es", disable=not verbose):
|
||
|
|
line: dict[str, str]
|
||
|
|
numero_panneau = int(line.get("N° panneau"))
|
||
|
|
code_dpt = line.get("Code du département")
|
||
|
|
|
||
|
|
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")
|
||
|
|
circo_id = f"{code_dpt}-{code_circo}"
|
||
|
|
|
||
|
|
nuance_id = line.get("Nuance candidat")
|
||
|
|
nuance = session.execute(select(NuanceLegislatives2022).filter_by(code=nuance_id)).scalar_one()
|
||
|
|
|
||
|
|
if candidat := session.execute(select(CandidatLegislatives2022)
|
||
|
|
.filter_by(circonscription_id=circo_id, numero=numero_panneau)) \
|
||
|
|
.scalar_one_or_none():
|
||
|
|
candidat.nuance_id = line['Nuance candidat']
|
||
|
|
candidat.bloc_id = nuance.bloc_id
|
||
|
|
candidat.nom = line['Nom candidat']
|
||
|
|
candidat.prenom = line['Prénom candidat']
|
||
|
|
candidat.sexe = CandidatLegislatives2022.Genre(line['Sexe candidat'])
|
||
|
|
candidat.date_naissance = datetime.datetime.strptime(line['Date naissance candidat'],
|
||
|
|
"%d/%m/%Y").date()
|
||
|
|
candidat.profession = line['Profession candidat']
|
||
|
|
candidat.sortant = line['Le candidat est sortant'] == "Oui"
|
||
|
|
candidat.nom_suppleance = line['Nom remplaçant']
|
||
|
|
candidat.prenom_suppleance = line['Prénom remplaçant']
|
||
|
|
candidat.sexe_suppleance = CandidatLegislatives2022.Genre(line['Sexe remplaçant'])
|
||
|
|
candidat.date_naissance_suppleance = datetime.datetime.strptime(line['Date naiss. remplaçant'],
|
||
|
|
"%d/%m/%Y").date()
|
||
|
|
candidat.sortant = line['Le remplaçant est sortant'] == "Oui"
|
||
|
|
else:
|
||
|
|
candidat = CandidatLegislatives2022(
|
||
|
|
circonscription_id=circo_id,
|
||
|
|
numero=numero_panneau,
|
||
|
|
nuance_id=nuance_id,
|
||
|
|
bloc_id=nuance.bloc_id,
|
||
|
|
nom=line['Nom candidat'],
|
||
|
|
prenom=line['Prénom candidat'],
|
||
|
|
sexe=CandidatLegislatives2022.Genre(line['Sexe candidat']).name,
|
||
|
|
date_naissance=datetime.datetime.strptime(line['Date naissance candidat'], "%d/%m/%Y").date(),
|
||
|
|
profession=line['Profession candidat'],
|
||
|
|
sortant=line['Le candidat est sortant'] == "Oui",
|
||
|
|
nom_suppleance=line['Nom remplaçant'],
|
||
|
|
prenom_suppleance=line['Prénom remplaçant'],
|
||
|
|
sexe_suppleance=CandidatLegislatives2022.Genre(line['Sexe remplaçant']).name,
|
||
|
|
date_naissance_suppleance=datetime.datetime.strptime(line['Date naiss. remplaçant'],
|
||
|
|
"%d/%m/%Y").date(),
|
||
|
|
sortant_suppleance=line['Le remplaçant est sortant'] == "Oui",
|
||
|
|
)
|
||
|
|
session.add(candidat)
|
||
|
|
|
||
|
|
session.commit()
|
||
|
|
|
||
|
|
|
||
|
|
def run(engine: Engine, verbose: bool = False) -> None:
|
||
|
|
creer_blocs(engine, verbose)
|
||
|
|
creer_nuances(engine, verbose)
|
||
|
|
importer_candidats(engine, verbose)
|