ynerant/nupes-elections
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Corrections + circonscription outre-mer et étranger

Browse Source
This commit is contained in:
Emmy D'Anello 2024-06-15 19:54:49 +02:00
parent d913379421
commit 032d589a34
Signed by: ynerant
GPG Key ID: 3A75C55819C8CF85
3 changed files with 101 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
nupes/models/geographie.py
View File

@ -23,7 +23,7 @@ class Departement(Base):
code_insee: Mapped[str] = mapped_column(String(3), primary_key=True) code_insee: Mapped[str] = mapped_column(String(3), primary_key=True)
libelle: Mapped[str] = mapped_column(String(64), unique=True) libelle: Mapped[str] = mapped_column(String(64), unique=True)
region_code: Mapped[int] = mapped_column(ForeignKey("region.code_insee")) region_code: Mapped[str] = mapped_column(ForeignKey("region.code_insee"))
geometry: Mapped[dict] = mapped_column(JSON()) geometry: Mapped[dict] = mapped_column(JSON())
region: Mapped[Region] = relationship(Region, back_populates="departements") region: Mapped[Region] = relationship(Region, back_populates="departements")

63
nupes/scripts/import_geographie.py
View File

File diff suppressed because one or more lines are too long

76
nupes/scripts/import_resultats_2024.py
View File

@ -148,8 +148,8 @@ def importer_resultats_commune(engine: Engine, verbose: bool = False) -> None:
for liste in session.execute(select(Liste)).scalars().all(): for liste in session.execute(select(Liste)).scalars().all():
voix_liste_com = session.execute(select(VoixListeCommune) voix_liste_com = session.execute(select(VoixListeCommune)
.filter_by(resultats_commune_id=resultats_commune.id, .filter_by(resultats_commune_id=resultats_commune.id,
liste_id=liste.id)) \ liste_id=liste.id)) \
.scalar_one_or_none() .scalar_one_or_none()
if not voix_liste_com: if not voix_liste_com:
voix_liste_com = VoixListeCommune(resultats_commune_id=resultats_commune.id, liste_id=liste.id) voix_liste_com = VoixListeCommune(resultats_commune_id=resultats_commune.id, liste_id=liste.id)
@ -169,7 +169,7 @@ def importer_resultats_circo(engine: Engine, verbose: bool = False) -> None:
with Session(engine) as session: with Session(engine) as session:
for row in tqdm(reader, desc="Circonscription", disable=not verbose): for row in tqdm(reader, desc="Circonscription", disable=not verbose):
circo_code = row['Code circonscription législative'] circo_code = row['Code circonscription législative']
dpt_code, circo_code = circo_code[:] dpt_code, circo_code = circo_code[:-2], circo_code[-2:]
resultats_circo = session.execute(select(ResultatsCirconscription).filter_by(circo_id=circo_code)) \ resultats_circo = session.execute(select(ResultatsCirconscription).filter_by(circo_id=circo_code)) \
.scalar_one_or_none() .scalar_one_or_none()
@ -232,7 +232,7 @@ def importer_resultats_circo(engine: Engine, verbose: bool = False) -> None:
def importer_resultats_departement(engine: Engine, verbose: bool = False) -> None: def importer_resultats_departement(engine: Engine, verbose: bool = False) -> None:
file = get_file("https://www.data.gouv.fr/fr/datasets/r/b77cc4da-644f-4323-b6f7-ae6fe9b33f86", file = get_file("https://www.data.gouv.fr/fr/datasets/r/b77cc4da-644f-4323-b6f7-ae6fe9b33f86",
"resultats-europeennes-par-circonscription.csv") "resultats-europeennes-par-departement.csv")
with file.open('r') as f: with file.open('r') as f:
reader = DictReader(f, delimiter=';') reader = DictReader(f, delimiter=';')
@ -240,11 +240,15 @@ def importer_resultats_departement(engine: Engine, verbose: bool = False) -> Non
for row in tqdm(reader, desc="Département", disable=not verbose): for row in tqdm(reader, desc="Département", disable=not verbose):
dpt_code = row['Code département'] dpt_code = row['Code département']
if dpt_code == "ZX":
dpt_code = "977"
resultats_dpt = session.execute(select(ResultatsDepartement).filter_by(dpt_id=dpt_code)) \ resultats_dpt = session.execute(select(ResultatsDepartement).filter_by(dpt_id=dpt_code)) \
.scalar_one_or_none() .scalar_one_or_none()
if not resultats_dpt: if not resultats_dpt:
print(f"Département {dpt_code} non trouvé")
dpt = session.execute(select(Departement) dpt = session.execute(select(Departement)
.filter_by(code_insee=f"{row['Code département'].zfill(2)}")) \ .filter_by(code_insee=f"{dpt_code.zfill(2)}")) \
.scalar_one() .scalar_one()
resultats_reg = session.execute(select(ResultatsRegion) resultats_reg = session.execute(select(ResultatsRegion)
.filter_by(region_id=dpt.region_code)) \ .filter_by(region_id=dpt.region_code)) \
@ -286,34 +290,32 @@ def importer_resultats_departement(engine: Engine, verbose: bool = False) -> Non
voix_liste_dpt.voix = int(row[f"Voix {liste.id}"]) voix_liste_dpt.voix = int(row[f"Voix {liste.id}"])
session.commit() session.commit()
def importer_resultats_region(engine: Engine, verbose: bool = False) -> None: def importer_resultats_region(engine: Engine, verbose: bool = False) -> None:
file = get_file("https://www.data.gouv.fr/fr/datasets/r/6a782ef9-8ad6-4e66-832d-338b1041a42d", file = get_file("https://www.data.gouv.fr/fr/datasets/r/7c3a854b-7344-4c68-b1f9-9d651b4ca823",
"resultats-europeennes-par-commune.csv") "resultats-europeennes-par-region.csv")
with file.open('r') as f: with file.open('r') as f:
reader = DictReader(f, delimiter=';') reader = DictReader(f, delimiter=';')
with Session(engine) as session: with Session(engine) as session:
for row in tqdm(reader, desc="Région", disable=not verbose): for row in tqdm(reader, desc="Région", disable=not verbose):
dpt_code = row['Code département'] reg_code = row['Code région']
resultats_reg = session.execute(select(ResultatsRegion).filter_by(region_id=dpt_code)) \ resultats_reg = session.execute(select(ResultatsRegion).filter_by(region_id=reg_code)) \
.scalar_one_or_none() .scalar_one_or_none()
if not resultats_reg: if not resultats_reg:
dpt = session.execute(select(Departement)
.filter_by(code_insee=f"{row['Code département'].zfill(2)}")) \
.scalar_one()
resultats_france = session.execute(select(ResultatsFrance)).scalar_one_or_none() resultats_france = session.execute(select(ResultatsFrance)).scalar_one_or_none()
if not resultats_france: if not resultats_france:
session.add(ResultatsFrance()) session.add(ResultatsFrance())
resultats_france = session.execute(select(ResultatsFrance)).scalar_one() resultats_france = session.execute(select(ResultatsFrance)).scalar_one()
resultats_reg = ResultatsRegion(region_id=str(dpt.region_code), resultats_reg = ResultatsRegion(region_id=reg_code,
resultats_france_id=resultats_france.id) resultats_france_id=resultats_france.id)
session.add(resultats_reg) session.add(resultats_reg)
resultats_reg = session.execute(select(ResultatsRegion).filter_by(region_id=dpt_code)) \ session.commit()
resultats_reg = session.execute(select(ResultatsRegion).filter_by(region_id=reg_code)) \
.scalar_one() .scalar_one()
resultats_reg.inscrits = int(row['Inscrits']) resultats_reg.inscrits = int(row['Inscrits'])
@ -352,34 +354,32 @@ def calculer_resultats_france(engine: Engine, verbose: bool = False) -> None:
select(VoixListeFrance).filter_by(resultats_france_id=resultats_france.id)).scalars().all(): select(VoixListeFrance).filter_by(resultats_france_id=resultats_france.id)).scalars().all():
voix_liste_france.voix = 0 # Réinitialisation des voix voix_liste_france.voix = 0 # Réinitialisation des voix
for resultats_reg in session.execute(select(ResultatsRegion)).scalars().all(): for resultats_dpt in session.execute(select(ResultatsDepartement)).scalars().all():
resultats_france.inscrits += resultats_reg.inscrits resultats_france.inscrits += resultats_dpt.inscrits
resultats_france.votants += resultats_reg.votants resultats_france.votants += resultats_dpt.votants
resultats_france.abstentions += resultats_reg.abstentions resultats_france.abstentions += resultats_dpt.abstentions
resultats_france.exprimes += resultats_reg.exprimes resultats_france.exprimes += resultats_dpt.exprimes
resultats_france.blancs += resultats_reg.blancs resultats_france.blancs += resultats_dpt.blancs
resultats_france.nuls += resultats_reg.nuls resultats_france.nuls += resultats_dpt.nuls
for voix_liste_reg in session.execute(select(VoixListeRegion) for voix_liste_dpt in session.execute(select(VoixListeDepartement)
.filter_by(resultats_region_id=resultats_reg.id)).scalars().all(): .filter_by(resultats_departement_id=resultats_dpt.id)) \
voix_liste_france = session.execute(select(VoixListeFrance) .scalars().all():
.filter_by(resultats_france_id=resultats_france.id, if voix_liste_france := session.execute(
liste_id=voix_liste_reg.liste_id)) \ select(VoixListeFrance).filter_by(resultats_france_id=resultats_france.id,
.scalar_one_or_none() liste_id=voix_liste_dpt.liste_id)).scalar_one_or_none():
if not voix_liste_france: voix_liste_france.voix += voix_liste_dpt.voix
voix_liste_france = VoixListeFrance(resultats_france_id=resultats_france.id, else:
liste_id=voix_liste_reg.liste_id) session.add(VoixListeFrance(resultats_france_id=resultats_france.id,
session.add(voix_liste_france) liste_id=voix_liste_dpt.liste_id, voix=voix_liste_dpt.voix))
voix_liste_france.voix += voix_liste_reg.voix
session.commit() session.commit()
def run(engine: Engine, verbose: bool = False) -> None: def run(engine: Engine, verbose: bool = False) -> None:
importer_resultats_bv(engine, verbose)
importer_resultats_commune(engine, verbose)
importer_resultats_circo(engine, verbose)
importer_resultats_departement(engine, verbose)
importer_resultats_region(engine, verbose) importer_resultats_region(engine, verbose)
importer_resultats_departement(engine, verbose)
calculer_resultats_france(engine, verbose) calculer_resultats_france(engine, verbose)
importer_resultats_circo(engine, verbose)
importer_resultats_commune(engine, verbose)
importer_resultats_bv(engine, verbose)