diff --git a/.gitignore b/.gitignore index 56e2b66..d8ecbd6 100644 --- a/.gitignore +++ b/.gitignore @@ -6,6 +6,7 @@ __pycache__ *.sqlite3 *.sqlite3-journal +cache/ data/ migrations/versions/* diff --git a/main.py b/main.py index 198857a..8306bc4 100755 --- a/main.py +++ b/main.py @@ -6,7 +6,7 @@ import os from dotenv import load_dotenv from sqlalchemy import create_engine -from nupes.scripts import import_candidats_2024, import_geographie, import_resultats_web_2024 +from nupes.scripts import import_candidats_2024, import_geographie, import_resultats_web_2024, export_resultats_2024 def parse_args(): @@ -16,7 +16,7 @@ def parse_args(): "régionales", "départementales", "municipales"], default="européennes") parser.add_argument('--year', '-y', type=int, help="Année de l'élection", default=2024) parser.add_argument('action', help="Action à réaliser", - choices=["import_candidats", "import_geographie", "import_resultats"]) + choices=["import_candidats", "import_geographie", "import_resultats", "export_resultats"]) parser.add_argument('--debug', '-d', action='store_true', help="Mode debug") return parser.parse_args() @@ -38,6 +38,9 @@ def main(): case "import_resultats": print(f"Import des résultats pour les élections {args.type} {args.year}") import_resultats_web_2024.run(engine) + case "export_resultats": + print(f"Export des résultats pour les élections {args.type} {args.year}") + export_resultats_2024.run(engine) case _: print(f"Action {args.action} non reconnue") diff --git a/nupes/data.py b/nupes/cache.py similarity index 86% rename from nupes/data.py rename to nupes/cache.py index cfd4ad7..12c4351 100644 --- a/nupes/data.py +++ b/nupes/cache.py @@ -3,14 +3,14 @@ from pathlib import Path import requests -DATA_DIR = Path(__file__).parent / 'data' +CACHE_DIR = Path(__file__).parent / 'cache' def get_file(url, filename, force_etag: str = ""): - if not DATA_DIR.is_dir(): - DATA_DIR.mkdir(parents=True) + if not CACHE_DIR.is_dir(): + CACHE_DIR.mkdir(parents=True) - file_dir = DATA_DIR / filename + file_dir = CACHE_DIR / filename if not file_dir.is_dir(): file_dir.mkdir(parents=True) diff --git a/nupes/scripts/export_resultats_2024.py b/nupes/scripts/export_resultats_2024.py new file mode 100644 index 0000000..1a24616 --- /dev/null +++ b/nupes/scripts/export_resultats_2024.py @@ -0,0 +1,255 @@ +import json +from pathlib import Path + +from sqlalchemy import Engine, select +from sqlalchemy.orm import Session + +from nupes.models.geographie import Region, Departement, Commune +from nupes.models.europeennes2024 import Bloc, Nuance, Liste, \ + ResultatsFrance, ResultatsRegion, ResultatsDepartement, ResultatsCommune + +DATA_DIR = Path(__file__).parent.parent / 'data' + + +def exporter_listes(engine: Engine) -> None: + with Session(engine) as session: + blocs = session.execute(select(Bloc)).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" / "europeennes2024" / "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(Nuance)).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" / "europeennes2024" / "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(Liste)).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" / "europeennes2024" / "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) -> None: + with Session(engine) as session: + resultats_france = session.execute(select(ResultatsFrance)).scalar_one_or_none() + if not resultats_france: + resultats_france = ResultatsFrance() + session.add(resultats_france) + + resultats_dict = { + "inscrits": resultats_france.inscrits, + "votants": resultats_france.votants, + "exprimes": resultats_france.exprimes, + "blancs": resultats_france.blancs, + "nuls": resultats_france.nuls, + 'geometry': {}, + } + + resultats_listes = {} + resultats_dict['voix_listes'] = resultats_listes + for voix_liste in resultats_france.voix_listes: + resultats_listes[voix_liste.liste.numero] = voix_liste.voix + + file = DATA_DIR / "resultats" / "europeennes2024" / "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) -> None: + with Session(engine) as session: + regions = session.execute(select(Region)).scalars().all() + regions_json = [] + + for region in regions: + region_json = {'code_insee': region.code_insee, 'nom': region.libelle, + 'departements': [dpt.code_insee for dpt in region.departements]} + regions_json.append(region_json) + + resultats_region = session.execute(select(ResultatsRegion).filter_by(region_id=region.code_insee)) \ + .scalar_one_or_none() + if not resultats_region: + resultats_france = session.execute(select(ResultatsFrance)).scalar_one() + resultats_region = ResultatsRegion(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, + "exprimes": resultats_region.exprimes, + "blancs": resultats_region.blancs, + "nuls": resultats_region.nuls, + 'region': region_json, + 'geometry': region.geometry, + } + + resultats_listes = {} + resultats_dict['voix_listes'] = resultats_listes + for voix_liste in resultats_region.voix_listes: + resultats_listes[voix_liste.liste.numero] = voix_liste.voix + + file = DATA_DIR / "resultats" / "europeennes2024" / "regions" / 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" / "europeennes2024" / "regions" / "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) -> None: + with Session(engine) as session: + departements = session.execute(select(Departement)).scalars().all() + departements_json = [] + + for departement in departements: + departement_json = {'code_insee': departement.code_insee, 'nom': departement.libelle, + 'region': departement.region_code, + 'communes': [commune.code_insee for commune in departement.communes]} + departements_json.append(departement_json) + + resultats_departement = session.execute( + select(ResultatsDepartement).filter_by(dpt_id=departement.code_insee)).scalar_one_or_none() + if not resultats_departement: + resultats_region = session.execute(select(ResultatsRegion) + .filter_by(region_id=departement.region_code)).scalar_one() + resultats_departement = ResultatsDepartement(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, + "exprimes": resultats_departement.exprimes, + "blancs": resultats_departement.blancs, + "nuls": resultats_departement.nuls, + 'departement': departement_json, + 'geometry': departement.geometry, + } + + resultats_listes = {} + resultats_dict['voix_listes'] = resultats_listes + for voix_liste in resultats_departement.voix_listes: + resultats_listes[voix_liste.liste.numero] = voix_liste.voix + + file = DATA_DIR / "resultats" / "europeennes2024" / "departements" / 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" / "europeennes2024" / "departements" / "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_communes(engine: Engine) -> None: + with Session(engine) as session: + communes = session.execute(select(Commune)).scalars().all() + communes_json = [] + + for commune in communes: + 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(ResultatsCommune).filter_by(commune_id=commune.code_insee)).scalar_one_or_none() + if not resultats_commune: + resultats_departement = session.execute(select(ResultatsDepartement) + .filter_by(dpt_id=commune.departement_code)).scalar_one() + resultats_commune = ResultatsCommune(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, + "exprimes": resultats_commune.exprimes, + "blancs": resultats_commune.blancs, + "nuls": resultats_commune.nuls, + 'commune': commune_json, + 'geometry': commune.geometry, + } + + resultats_listes = {} + resultats_dict['voix_listes'] = resultats_listes + for voix_liste in resultats_commune.voix_listes: + resultats_listes[voix_liste.liste.numero] = voix_liste.voix + + file = DATA_DIR / "resultats" / "europeennes2024" / "communes" / 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" / "europeennes2024" / "communes" / "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 run(engine: Engine) -> None: + exporter_listes(engine) + exporter_resultats_france(engine) + exporter_resultats_regions(engine) + exporter_resultats_departements(engine) + exporter_resultats_communes(engine) diff --git a/nupes/scripts/import_candidats_2024.py b/nupes/scripts/import_candidats_2024.py index f7806fd..e7a2c53 100644 --- a/nupes/scripts/import_candidats_2024.py +++ b/nupes/scripts/import_candidats_2024.py @@ -4,7 +4,7 @@ import datetime from sqlalchemy import Engine, select from sqlalchemy.orm import Session -from nupes.data import get_file +from nupes.cache import get_file from nupes.models.europeennes2024 import Bloc, Nuance, Liste, Candidat diff --git a/nupes/scripts/import_geographie.py b/nupes/scripts/import_geographie.py index ea09bb8..0818f89 100644 --- a/nupes/scripts/import_geographie.py +++ b/nupes/scripts/import_geographie.py @@ -5,7 +5,7 @@ import requests from sqlalchemy import Engine, select from sqlalchemy.orm import Session -from nupes.data import get_file +from nupes.cache import get_file from nupes.models.geographie import BureauVote, Circonscription, Commune, Departement, Region diff --git a/nupes/scripts/import_resultats_web_2024.py b/nupes/scripts/import_resultats_web_2024.py index bab2d7d..69d54a1 100644 --- a/nupes/scripts/import_resultats_web_2024.py +++ b/nupes/scripts/import_resultats_web_2024.py @@ -5,7 +5,7 @@ from sqlalchemy import Engine, select from sqlalchemy.orm import Session from tqdm import tqdm -from nupes.data import get_file +from nupes.cache import get_file from nupes.models.europeennes2024 import ResultatsFrance, ResultatsRegion, ResultatsDepartement, ResultatsCommune, \ VoixListeFrance, VoixListeRegion, VoixListeDepartement, VoixListeCommune, Liste from nupes.models.geographie import Region, Departement, Commune @@ -187,7 +187,9 @@ def importer_resultats_communes(engine: Engine) -> None: # Pour ces collectivités d'outre-mer, qui ne sont pas dans des régions, # on ne les regroupe pas dans des régions reg_path = "" - if com_code in ["16355", "25282", "55050", "55189", "69152"]: + if com_code in ["08294", "16355", "18131", "25282", "25549", "35112", "49321", "55039", "55050", "55139", + "55189", "55239", "55307", "64541", "69152", "75056", "85041", "85271", "86231", "95282", + "98611", "98612", "98613"]: # Aucun⋅e habitant⋅e donc pas de bureau de vote # Ou alors la commune a récemment fusionné continue @@ -293,5 +295,3 @@ def run(engine: Engine) -> None: importer_resultats_regions(engine) importer_resultats_departements(engine) importer_resultats_communes(engine) - - analyser_resultats("")