Script export données législatives 2024
This commit is contained in:
		
							
								
								
									
										494
									
								
								nupes/scripts/legislatives2024/export_resultats.py
									
									
									
									
									
										Normal file
									
								
							
							
						
						
									
										494
									
								
								nupes/scripts/legislatives2024/export_resultats.py
									
									
									
									
									
										Normal file
									
								
							@@ -0,0 +1,494 @@
 | 
				
			|||||||
 | 
					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 BureauVote
 | 
				
			||||||
 | 
					from nupes.models.legislatives2024 import *
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					DATA_DIR = Path(__file__).parent.parent.parent.parent / 'data'
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					def exporter_nuances(engine: Engine, verbose: bool = False) -> None:
 | 
				
			||||||
 | 
					    with Session(engine) as session:
 | 
				
			||||||
 | 
					        blocs = session.execute(select(BlocLegislatives2024)).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" / "legislatives" / "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(NuanceLegislatives2024)).scalars().all()
 | 
				
			||||||
 | 
					        nuances_json = []
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					        for nuance in nuances:
 | 
				
			||||||
 | 
					            nuance_json = {
 | 
				
			||||||
 | 
					                'code': nuance.code,
 | 
				
			||||||
 | 
					                'nom': nuance.nom,
 | 
				
			||||||
 | 
					                'couleur': nuance.couleur,
 | 
				
			||||||
 | 
					                "bloc": nuance.bloc.nom,
 | 
				
			||||||
 | 
					            }
 | 
				
			||||||
 | 
					            nuances_json.append(nuance_json)
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					        file = DATA_DIR / "resultats" / "legislatives" / "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)
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					def exporter_candidats(engine: Engine, verbose: bool = False) -> None:
 | 
				
			||||||
 | 
					    with Session(engine) as session:
 | 
				
			||||||
 | 
					        circonscriptions = session.execute(select(Circonscription)).scalars().all()
 | 
				
			||||||
 | 
					        for circonscription in tqdm(circonscriptions, desc="Candidat⋅es", disable=not verbose):
 | 
				
			||||||
 | 
					            candidats = session.execute(select(CandidatLegislatives2024)
 | 
				
			||||||
 | 
					                                        .filter_by(circonscription_id=circonscription.id)).scalars().all()
 | 
				
			||||||
 | 
					            candidats_json = []
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					            for candidat in candidats:
 | 
				
			||||||
 | 
					                candidat_json = {
 | 
				
			||||||
 | 
					                    'numero': candidat.numero,
 | 
				
			||||||
 | 
					                    'nuance': candidat.nuance_id,
 | 
				
			||||||
 | 
					                    'bloc': candidat.nuance.bloc.nom,
 | 
				
			||||||
 | 
					                    'nom': candidat.nom,
 | 
				
			||||||
 | 
					                    'prenom': candidat.prenom,
 | 
				
			||||||
 | 
					                    'nom_suppleance': candidat.nom_suppleance,
 | 
				
			||||||
 | 
					                    'prenom_suppleance': candidat.prenom_suppleance,
 | 
				
			||||||
 | 
					                }
 | 
				
			||||||
 | 
					                candidats_json.append(candidat_json)
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					            file = DATA_DIR / "resultats" / "legislatives" / "2024" / "candidats" / f"{circonscription.id}.json"
 | 
				
			||||||
 | 
					            if not file.parent.is_dir():
 | 
				
			||||||
 | 
					                file.parent.mkdir(parents=True)
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					            with file.open('w') as f:
 | 
				
			||||||
 | 
					                json.dump(candidats_json, f)
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					def exporter_resultats_france(engine: Engine, verbose: bool = False) -> None:
 | 
				
			||||||
 | 
					    with Session(engine) as session:
 | 
				
			||||||
 | 
					        resultats_france = session.execute(select(ResultatsFranceLegislatives2024)).scalar_one_or_none()
 | 
				
			||||||
 | 
					        if not resultats_france:
 | 
				
			||||||
 | 
					            resultats_france = ResultatsFranceLegislatives2024()
 | 
				
			||||||
 | 
					            session.add(resultats_france)
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					        resultats_dict = {
 | 
				
			||||||
 | 
					            'zone': {
 | 
				
			||||||
 | 
					                'type': "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()],
 | 
				
			||||||
 | 
					                'geometry': {},
 | 
				
			||||||
 | 
					            },
 | 
				
			||||||
 | 
					            'tour1': {
 | 
				
			||||||
 | 
					                'inscrits': resultats_france.inscrits_t1,
 | 
				
			||||||
 | 
					                'votants': resultats_france.votants_t1,
 | 
				
			||||||
 | 
					                'abstentions': resultats_france.abstentions_t1,
 | 
				
			||||||
 | 
					                'exprimes': resultats_france.exprimes_t1,
 | 
				
			||||||
 | 
					                'blancs': resultats_france.blancs_t1,
 | 
				
			||||||
 | 
					                'nuls': resultats_france.nuls_t1,
 | 
				
			||||||
 | 
					            },
 | 
				
			||||||
 | 
					            'tour2': {
 | 
				
			||||||
 | 
					                'inscrits': resultats_france.inscrits_t2,
 | 
				
			||||||
 | 
					                'votants': resultats_france.votants_t2,
 | 
				
			||||||
 | 
					                'abstentions': resultats_france.abstentions_t2,
 | 
				
			||||||
 | 
					                'exprimes': resultats_france.exprimes_t2,
 | 
				
			||||||
 | 
					                'blancs': resultats_france.blancs_t2,
 | 
				
			||||||
 | 
					                'nuls': resultats_france.nuls_t2,
 | 
				
			||||||
 | 
					            },
 | 
				
			||||||
 | 
					        }
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					        resultats_t1, resultats_t2 = {}, {}
 | 
				
			||||||
 | 
					        resultats_dict['tour1']['voix'] = resultats_t1
 | 
				
			||||||
 | 
					        resultats_dict['tour2']['voix'] = resultats_t2
 | 
				
			||||||
 | 
					        for voix_nuance in resultats_france.voix:
 | 
				
			||||||
 | 
					            resultats_t1[voix_nuance.nuance_id] = voix_nuance.voix_t1
 | 
				
			||||||
 | 
					            resultats_t2[voix_nuance.nuance_id] = voix_nuance.voix_t2
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					        file = DATA_DIR / "resultats" / "legislatives" / "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 = {
 | 
				
			||||||
 | 
					                'type': "region",
 | 
				
			||||||
 | 
					                '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()
 | 
				
			||||||
 | 
					                ],
 | 
				
			||||||
 | 
					                'geometry': region.geometry,
 | 
				
			||||||
 | 
					            }
 | 
				
			||||||
 | 
					            regions_json.append(region_json)
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					            resultats_region = session.execute(select(ResultatsRegionLegislatives2024)
 | 
				
			||||||
 | 
					                                               .filter_by(region_id=region.code_insee)).scalar_one_or_none()
 | 
				
			||||||
 | 
					            if not resultats_region:
 | 
				
			||||||
 | 
					                resultats_france = session.execute(select(ResultatsFranceLegislatives2024)).scalar_one()
 | 
				
			||||||
 | 
					                resultats_region = ResultatsRegionLegislatives2024(region_id=region.code_insee,
 | 
				
			||||||
 | 
					                                                                   resultats_france_id=resultats_france.id)
 | 
				
			||||||
 | 
					                session.add(resultats_region)
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					            resultats_dict = {
 | 
				
			||||||
 | 
					                'zone': region_json,
 | 
				
			||||||
 | 
					                'tour1': {
 | 
				
			||||||
 | 
					                    'inscrits': resultats_region.inscrits_t1,
 | 
				
			||||||
 | 
					                    'votants': resultats_region.votants_t1,
 | 
				
			||||||
 | 
					                    'abstentions': resultats_region.abstentions_t1,
 | 
				
			||||||
 | 
					                    'exprimes': resultats_region.exprimes_t1,
 | 
				
			||||||
 | 
					                    'blancs': resultats_region.blancs_t1,
 | 
				
			||||||
 | 
					                    'nuls': resultats_region.nuls_t1,
 | 
				
			||||||
 | 
					                },
 | 
				
			||||||
 | 
					                'tour2': {
 | 
				
			||||||
 | 
					                    'inscrits': resultats_region.inscrits_t2,
 | 
				
			||||||
 | 
					                    'votants': resultats_region.votants_t2,
 | 
				
			||||||
 | 
					                    'abstentions': resultats_region.abstentions_t2,
 | 
				
			||||||
 | 
					                    'exprimes': resultats_region.exprimes_t2,
 | 
				
			||||||
 | 
					                    'blancs': resultats_region.blancs_t2,
 | 
				
			||||||
 | 
					                    'nuls': resultats_region.nuls_t2,
 | 
				
			||||||
 | 
					                },
 | 
				
			||||||
 | 
					            }
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					            resultats_t1, resultats_t2 = {}, {}
 | 
				
			||||||
 | 
					            resultats_dict['tour1']['voix'] = resultats_t1
 | 
				
			||||||
 | 
					            resultats_dict['tour2']['voix'] = resultats_t2
 | 
				
			||||||
 | 
					            for voix_nuance in resultats_region.voix:
 | 
				
			||||||
 | 
					                resultats_t1[voix_nuance.nuance_id] = voix_nuance.voix_t1
 | 
				
			||||||
 | 
					                resultats_t2[voix_nuance.nuance_id] = voix_nuance.voix_t2
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					            file = DATA_DIR / "resultats" / "legislatives" / "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" / "legislatives" / "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 = {
 | 
				
			||||||
 | 
					                'type': "departement",
 | 
				
			||||||
 | 
					                '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],
 | 
				
			||||||
 | 
					                'geometry': departement.geometry,
 | 
				
			||||||
 | 
					            }
 | 
				
			||||||
 | 
					            departements_json.append(departement_json)
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					            resultats_departement = session.execute(
 | 
				
			||||||
 | 
					                select(ResultatsDepartementLegislatives2024)
 | 
				
			||||||
 | 
					                .filter_by(dpt_id=departement.code_insee)).scalar_one_or_none()
 | 
				
			||||||
 | 
					            if not resultats_departement:
 | 
				
			||||||
 | 
					                resultats_region = session.execute(select(ResultatsRegionLegislatives2024)
 | 
				
			||||||
 | 
					                                                   .filter_by(region_id=departement.region_code)).scalar_one()
 | 
				
			||||||
 | 
					                resultats_departement = ResultatsDepartementLegislatives2024(dpt_id=departement.code_insee,
 | 
				
			||||||
 | 
					                                                                             resultats_region_id=resultats_region.id)
 | 
				
			||||||
 | 
					                session.add(resultats_departement)
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					            resultats_dict = {
 | 
				
			||||||
 | 
					                'zone': departement_json,
 | 
				
			||||||
 | 
					                'tour1': {
 | 
				
			||||||
 | 
					                    'inscrits': resultats_departement.inscrits_t1,
 | 
				
			||||||
 | 
					                    'votants': resultats_departement.votants_t1,
 | 
				
			||||||
 | 
					                    'abstentions': resultats_departement.abstentions_t1,
 | 
				
			||||||
 | 
					                    'exprimes': resultats_departement.exprimes_t1,
 | 
				
			||||||
 | 
					                    'blancs': resultats_departement.blancs_t1,
 | 
				
			||||||
 | 
					                    'nuls': resultats_departement.nuls_t1,
 | 
				
			||||||
 | 
					                },
 | 
				
			||||||
 | 
					                'tour2': {
 | 
				
			||||||
 | 
					                    'inscrits': resultats_departement.inscrits_t2,
 | 
				
			||||||
 | 
					                    'votants': resultats_departement.votants_t2,
 | 
				
			||||||
 | 
					                    'abstentions': resultats_departement.abstentions_t2,
 | 
				
			||||||
 | 
					                    'exprimes': resultats_departement.exprimes_t2,
 | 
				
			||||||
 | 
					                    'blancs': resultats_departement.blancs_t2,
 | 
				
			||||||
 | 
					                    'nuls': resultats_departement.nuls_t2,
 | 
				
			||||||
 | 
					                },
 | 
				
			||||||
 | 
					            }
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					            resultats_t1, resultats_t2 = {}, {}
 | 
				
			||||||
 | 
					            resultats_dict['tour1']['voix'] = resultats_t1
 | 
				
			||||||
 | 
					            resultats_dict['tour2']['voix'] = resultats_t2
 | 
				
			||||||
 | 
					            for voix_nuance in resultats_departement.voix:
 | 
				
			||||||
 | 
					                resultats_t1[voix_nuance.nuance_id] = voix_nuance.voix_t1
 | 
				
			||||||
 | 
					                resultats_t2[voix_nuance.nuance_id] = voix_nuance.voix_t2
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					            file = DATA_DIR / "resultats" / "legislatives" / "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" / "legislatives" / "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 = {
 | 
				
			||||||
 | 
					                'type': "circonscription",
 | 
				
			||||||
 | 
					                'id': circonscription.id,
 | 
				
			||||||
 | 
					                'departement': circonscription.departement_code,
 | 
				
			||||||
 | 
					                'numero': circonscription.numero,
 | 
				
			||||||
 | 
					                'bureaux_vote': [bv.id for bv in circonscription.bureaux_vote],
 | 
				
			||||||
 | 
					                'geometry': circonscription.geometry,
 | 
				
			||||||
 | 
					            }
 | 
				
			||||||
 | 
					            circonscriptions_json.append(circonscription_json)
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					            resultats_circonscription = session.execute(
 | 
				
			||||||
 | 
					                select(ResultatsCirconscriptionLegislatives2024)
 | 
				
			||||||
 | 
					                .filter_by(circo_id=circonscription.id)).scalar_one_or_none()
 | 
				
			||||||
 | 
					            if not resultats_circonscription:
 | 
				
			||||||
 | 
					                resultats_departement = session.execute(select(ResultatsDepartementLegislatives2024)
 | 
				
			||||||
 | 
					                                                        .filter_by(dpt_id=circonscription.departement_code)) \
 | 
				
			||||||
 | 
					                    .scalar_one()
 | 
				
			||||||
 | 
					                resultats_circonscription = ResultatsCirconscriptionLegislatives2024(
 | 
				
			||||||
 | 
					                    circo_id=circonscription.id, resultats_departement_id=resultats_departement.id)
 | 
				
			||||||
 | 
					                session.add(resultats_circonscription)
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					            resultats_dict = {
 | 
				
			||||||
 | 
					                'zone': circonscription_json,
 | 
				
			||||||
 | 
					                'tour1': {
 | 
				
			||||||
 | 
					                    'inscrits': resultats_circonscription.inscrits_t1,
 | 
				
			||||||
 | 
					                    'votants': resultats_circonscription.votants_t1,
 | 
				
			||||||
 | 
					                    'abstentions': resultats_circonscription.abstentions_t1,
 | 
				
			||||||
 | 
					                    'exprimes': resultats_circonscription.exprimes_t1,
 | 
				
			||||||
 | 
					                    'blancs': resultats_circonscription.blancs_t1,
 | 
				
			||||||
 | 
					                    'nuls': resultats_circonscription.nuls_t1,
 | 
				
			||||||
 | 
					                },
 | 
				
			||||||
 | 
					                'tour2': {
 | 
				
			||||||
 | 
					                    'inscrits': resultats_circonscription.inscrits_t2,
 | 
				
			||||||
 | 
					                    'votants': resultats_circonscription.votants_t2,
 | 
				
			||||||
 | 
					                    'abstentions': resultats_circonscription.abstentions_t2,
 | 
				
			||||||
 | 
					                    'exprimes': resultats_circonscription.exprimes_t2,
 | 
				
			||||||
 | 
					                    'blancs': resultats_circonscription.blancs_t2,
 | 
				
			||||||
 | 
					                    'nuls': resultats_circonscription.nuls_t2,
 | 
				
			||||||
 | 
					                },
 | 
				
			||||||
 | 
					            }
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					            resultats_t1, resultats_t2 = {}, {}
 | 
				
			||||||
 | 
					            resultats_dict['tour1']['voix'] = resultats_t1
 | 
				
			||||||
 | 
					            resultats_dict['tour2']['voix'] = resultats_t2
 | 
				
			||||||
 | 
					            for voix_candidat in resultats_circonscription.voix:
 | 
				
			||||||
 | 
					                resultats_t1[voix_candidat.candidat.numero] = voix_candidat.voix_t1
 | 
				
			||||||
 | 
					                resultats_t2[voix_candidat.candidat.numero] = voix_candidat.voix_t2
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					            file = DATA_DIR / "resultats" / "legislatives" / "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" / "legislatives" / "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 = {
 | 
				
			||||||
 | 
					                'type': "commune",
 | 
				
			||||||
 | 
					                'code_insee': commune.code_insee,
 | 
				
			||||||
 | 
					                'nom': commune.libelle,
 | 
				
			||||||
 | 
					                'departement': commune.departement_code,
 | 
				
			||||||
 | 
					                'bureaux_vote': [bv.id for bv in commune.bureaux_vote],
 | 
				
			||||||
 | 
					                'geometry': commune.geometry,
 | 
				
			||||||
 | 
					            }
 | 
				
			||||||
 | 
					            communes_json.append(commune_json)
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					            resultats_commune = session.execute(
 | 
				
			||||||
 | 
					                select(ResultatsCommuneLegislatives2024).filter_by(commune_id=commune.code_insee)).scalar_one_or_none()
 | 
				
			||||||
 | 
					            if not resultats_commune:
 | 
				
			||||||
 | 
					                resultats_departement = session.execute(select(ResultatsDepartementLegislatives2024)
 | 
				
			||||||
 | 
					                                                        .filter_by(dpt_id=commune.departement_code)).scalar_one()
 | 
				
			||||||
 | 
					                resultats_commune = ResultatsCommuneLegislatives2024(commune_id=commune.code_insee,
 | 
				
			||||||
 | 
					                                                                     resultats_dpt_id=resultats_departement.id)
 | 
				
			||||||
 | 
					                session.add(resultats_commune)
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					            resultats_dict = {
 | 
				
			||||||
 | 
					                'zone': commune_json,
 | 
				
			||||||
 | 
					                'tour1': {
 | 
				
			||||||
 | 
					                    'inscrits': resultats_commune.inscrits_t1,
 | 
				
			||||||
 | 
					                    'votants': resultats_commune.votants_t1,
 | 
				
			||||||
 | 
					                    'abstentions': resultats_commune.abstentions_t1,
 | 
				
			||||||
 | 
					                    'exprimes': resultats_commune.exprimes_t1,
 | 
				
			||||||
 | 
					                    'blancs': resultats_commune.blancs_t1,
 | 
				
			||||||
 | 
					                    'nuls': resultats_commune.nuls_t1,
 | 
				
			||||||
 | 
					                },
 | 
				
			||||||
 | 
					                'tour2': {
 | 
				
			||||||
 | 
					                    'inscrits': resultats_commune.inscrits_t2,
 | 
				
			||||||
 | 
					                    'votants': resultats_commune.votants_t2,
 | 
				
			||||||
 | 
					                    'abstentions': resultats_commune.abstentions_t2,
 | 
				
			||||||
 | 
					                    'exprimes': resultats_commune.exprimes_t2,
 | 
				
			||||||
 | 
					                    'blancs': resultats_commune.blancs_t2,
 | 
				
			||||||
 | 
					                    'nuls': resultats_commune.nuls_t2,
 | 
				
			||||||
 | 
					                },
 | 
				
			||||||
 | 
					            }
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					            resultats_t1, resultats_t2 = {}, {}
 | 
				
			||||||
 | 
					            resultats_dict['tour1']['voix'] = resultats_t1
 | 
				
			||||||
 | 
					            resultats_dict['tour2']['voix'] = resultats_t2
 | 
				
			||||||
 | 
					            for voix_nuance in resultats_commune.voix:
 | 
				
			||||||
 | 
					                resultats_t1[voix_nuance.nuance_id] = voix_nuance.voix_t1
 | 
				
			||||||
 | 
					                resultats_t2[voix_nuance.nuance_id] = voix_nuance.voix_t2
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					            file = DATA_DIR / "resultats" / "legislatives" / "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" / "legislatives" / "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 = {
 | 
				
			||||||
 | 
					                'type': "bureau_vote",
 | 
				
			||||||
 | 
					                'id': bureau_vote.id,
 | 
				
			||||||
 | 
					                'libelle': bureau_vote.libelle,
 | 
				
			||||||
 | 
					                'commune': bureau_vote.commune_code,
 | 
				
			||||||
 | 
					                'circonscription': bureau_vote.circo_code,
 | 
				
			||||||
 | 
					                'geometry': bureau_vote.geometry,
 | 
				
			||||||
 | 
					            }
 | 
				
			||||||
 | 
					            bureaux_vote_json.append(bureau_vote_json)
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					            resultats_bureau_vote = session.execute(
 | 
				
			||||||
 | 
					                select(ResultatsBureauVoteLegislatives2024).filter_by(bv_id=bureau_vote.id)).scalar_one_or_none()
 | 
				
			||||||
 | 
					            if not resultats_bureau_vote:
 | 
				
			||||||
 | 
					                resultats_commune = session.execute(select(ResultatsCommuneLegislatives2024)
 | 
				
			||||||
 | 
					                                                    .filter_by(commune_id=bureau_vote.commune_code)).scalar_one()
 | 
				
			||||||
 | 
					                resultats_bureau_vote = ResultatsBureauVoteLegislatives2024(bv_id=bureau_vote.id,
 | 
				
			||||||
 | 
					                                                                            resultats_commune_id=resultats_commune.id)
 | 
				
			||||||
 | 
					                session.add(resultats_bureau_vote)
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					            resultats_dict = {
 | 
				
			||||||
 | 
					                'zone': bureau_vote_json,
 | 
				
			||||||
 | 
					                'tour1': {
 | 
				
			||||||
 | 
					                    'inscrits': resultats_bureau_vote.inscrits_t1,
 | 
				
			||||||
 | 
					                    'votants': resultats_bureau_vote.votants_t1,
 | 
				
			||||||
 | 
					                    'abstentions': resultats_bureau_vote.abstentions_t1,
 | 
				
			||||||
 | 
					                    'exprimes': resultats_bureau_vote.exprimes_t1,
 | 
				
			||||||
 | 
					                    'blancs': resultats_bureau_vote.blancs_t1,
 | 
				
			||||||
 | 
					                    'nuls': resultats_bureau_vote.nuls_t1,
 | 
				
			||||||
 | 
					                },
 | 
				
			||||||
 | 
					                'tour2': {
 | 
				
			||||||
 | 
					                    'inscrits': resultats_bureau_vote.inscrits_t2,
 | 
				
			||||||
 | 
					                    'votants': resultats_bureau_vote.votants_t2,
 | 
				
			||||||
 | 
					                    'abstentions': resultats_bureau_vote.abstentions_t2,
 | 
				
			||||||
 | 
					                    'exprimes': resultats_bureau_vote.exprimes_t2,
 | 
				
			||||||
 | 
					                    'blancs': resultats_bureau_vote.blancs_t2,
 | 
				
			||||||
 | 
					                    'nuls': resultats_bureau_vote.nuls_t2,
 | 
				
			||||||
 | 
					                },
 | 
				
			||||||
 | 
					            }
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					            resultats_t1, resultats_t2 = {}, {}
 | 
				
			||||||
 | 
					            resultats_dict['tour1']['voix'] = resultats_t1
 | 
				
			||||||
 | 
					            resultats_dict['tour2']['voix'] = resultats_t2
 | 
				
			||||||
 | 
					            for voix_candidat in resultats_bureau_vote.voix:
 | 
				
			||||||
 | 
					                resultats_t1[voix_candidat.candidat.numero] = voix_candidat.voix_t1
 | 
				
			||||||
 | 
					                resultats_t2[voix_candidat.candidat.numero] = voix_candidat.voix_t2
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					            file = DATA_DIR / "resultats" / "legislatives" / "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" / "legislatives" / "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_nuances(engine, verbose)
 | 
				
			||||||
 | 
					    exporter_candidats(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)
 | 
				
			||||||
		Reference in New Issue
	
	Block a user