nupes-elections/nupes/scripts/export_resultats_2024.py

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)