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

Compare commits

base: ynerant:a85739159fea3bad8f5611f21e3ea49ac2ddfe61
Branches Tags
ynerant:main
..
compare: ynerant:fbbe34518eaaa8bd5ef81fee0d3106eada5947c7
Branches Tags
ynerant:main

No commits in common. "a85739159fea3bad8f5611f21e3ea49ac2ddfe61" and "fbbe34518eaaa8bd5ef81fee0d3106eada5947c7" have entirely different histories.
a85739159f ... fbbe34518e

6 changed files with 95 additions and 386 deletions
Show Stats Expand all files Collapse all files

32
alembic/versions/5bf5e6526891_initial.py → alembic/versions/a2c743418d84_initial.py
View File

@ -1,8 +1,8 @@
"""initial
Revision ID: 5bf5e6526891
Revision ID: a2c743418d84
Revises:
Create Date: 2024-06-14 23:04:55.436991
Create Date: 2024-06-09 23:51:19.572272
"""
from typing import Sequence, Union
@ -12,7 +12,7 @@ import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = '5bf5e6526891'
revision: str = 'a2c743418d84'
down_revision: Union[str, None] = None
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
@ -159,20 +159,6 @@ def upgrade() -> None:
sa.ForeignKeyConstraint(['commune_code'], ['commune.code_insee'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_table('resultats2024_circonscription',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('circo_id', sa.String(length=6), nullable=False),
sa.Column('resultats_departement_id', sa.Integer(), nullable=False),
sa.Column('inscrits', sa.Integer(), nullable=False),
sa.Column('votants', sa.Integer(), nullable=False),
sa.Column('abstentions', sa.Integer(), nullable=False),
sa.Column('exprimes', sa.Integer(), nullable=False),
sa.Column('blancs', sa.Integer(), nullable=False),
sa.Column('nuls', sa.Integer(), nullable=False),
sa.ForeignKeyConstraint(['circo_id'], ['circonscription.id'], ),
sa.ForeignKeyConstraint(['resultats_departement_id'], ['resultats2024_departement.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_table('resultats2024_commune',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('commune_id', sa.String(length=5), nullable=False),
@ -199,7 +185,6 @@ def upgrade() -> None:
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('bv_id', sa.String(length=16), nullable=False),
sa.Column('resultats_commune_id', sa.Integer(), nullable=False),
sa.Column('resultats_circo_id', sa.Integer(), nullable=True),
sa.Column('inscrits', sa.Integer(), nullable=False),
sa.Column('votants', sa.Integer(), nullable=False),
sa.Column('abstentions', sa.Integer(), nullable=False),
@ -207,18 +192,9 @@ def upgrade() -> None:
sa.Column('blancs', sa.Integer(), nullable=False),
sa.Column('nuls', sa.Integer(), nullable=False),
sa.ForeignKeyConstraint(['bv_id'], ['bureau_vote.id'], ),
sa.ForeignKeyConstraint(['resultats_circo_id'], ['resultats2024_circonscription.id'], ),
sa.ForeignKeyConstraint(['resultats_commune_id'], ['resultats2024_commune.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_table('voix2024_circonscription',
sa.Column('liste_id', sa.Integer(), nullable=False),
sa.Column('resultats_circonscription_id', sa.Integer(), nullable=False),
sa.Column('voix', sa.Integer(), nullable=False),
sa.ForeignKeyConstraint(['liste_id'], ['liste2024.id'], ),
sa.ForeignKeyConstraint(['resultats_circonscription_id'], ['resultats2024_circonscription.id'], ),
sa.PrimaryKeyConstraint('liste_id', 'resultats_circonscription_id')
)
op.create_table('voix2024_commune',
sa.Column('liste_id', sa.Integer(), nullable=False),
sa.Column('resultats_commune_id', sa.Integer(), nullable=False),
@ -242,11 +218,9 @@ def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.drop_table('voix2024_bureau_vote')
op.drop_table('voix2024_commune')
op.drop_table('voix2024_circonscription')
op.drop_table('resultats2024_bureau_vote')
op.drop_table('voix2024_departement')
op.drop_table('resultats2024_commune')
op.drop_table('resultats2024_circonscription')
op.drop_table('bureau_vote')
op.drop_table('voix2024_region')
op.drop_table('voix2024_france')

6
nupes/models/__init__.py
View File

@ -1,7 +1,3 @@
from .base import Base
from .geographie import Region, Departement, Commune, Circonscription, BureauVote
from .europeennes2024 import Bloc as Bloc2024, Nuance as Nuance2024, Liste as Liste2024, Candidat as Candidat2024, \
ResultatsFrance as ResultatsFrance2024, ResultatsRegion as ResultatsRegion2024, ResultatsDepartement as ResultatsDepartement2024, \
ResultatsCommune as ResultatsCommune2024, ResultatsBureauVote as ResultatsBureauVote2024, \
VoixListeFrance as VoixListeFrance2024, VoixListeRegion as VoixListeRegion2024, VoixListeDepartement as VoixListeDepartement2024, \
VoixListeCommune as VoixListeCommune2024, VoixListeBureauVote as VoixListeBureauVote2024
from .europeennes2024 import Bloc as Bloc2024, Nuance as Nuance2024, Liste as Liste2024, Candidat as Candidat2024

44
nupes/models/europeennes2024.py
View File

@ -45,12 +45,10 @@ class Liste(Base):
resultats_nationaux: Mapped[List["VoixListeFrance"]] = relationship("VoixListeFrance", back_populates="liste")
resultats_par_region: Mapped[List["VoixListeRegion"]] = relationship("VoixListeRegion", back_populates="liste")
resultats_par_departement: Mapped[List["VoixListeDepartement"]] = relationship("VoixListeDepartement",
back_populates="liste")
resultats_par_circonscription: Mapped[List["VoixListeCirconscription"]] = relationship(
"VoixListeCirconscription", back_populates="liste")
back_populates="liste")
resultats_par_commune: Mapped[List["VoixListeCommune"]] = relationship("VoixListeCommune", back_populates="liste")
resultats_par_bureau_vote: Mapped[List["VoixListeBureauVote"]] = relationship("VoixListeBureauVote",
back_populates="liste")
back_populates="liste")
class Candidat(Base):
@ -137,33 +135,10 @@ class ResultatsDepartement(Base):
resultats_region = relationship(ResultatsRegion, back_populates="resultats_departements")
resultats_communes: Mapped[List["ResultatsCommune"]] = relationship("ResultatsCommune",
back_populates="resultats_departement")
resultats_circonscriptions: Mapped[List["ResultatsCirconscription"]] = relationship(
"ResultatsCirconscription", back_populates="resultats_departement")
voix_listes: Mapped[List["VoixListeDepartement"]] = relationship("VoixListeDepartement",
back_populates="resultats_departement")
class ResultatsCirconscription(Base):
__tablename__ = "resultats2024_circonscription"
id: Mapped[int] = mapped_column(primary_key=True)
circo_id: Mapped[str] = mapped_column(ForeignKey("circonscription.id"))
resultats_departement_id: Mapped[int] = mapped_column(ForeignKey("resultats2024_departement.id"))
inscrits: Mapped[int] = mapped_column(Integer(), default=0)
votants: Mapped[int] = mapped_column(Integer(), default=0)
abstentions: Mapped[int] = mapped_column(Integer(), default=0)
exprimes: Mapped[int] = mapped_column(Integer(), default=0)
blancs: Mapped[int] = mapped_column(Integer(), default=0)
nuls: Mapped[int] = mapped_column(Integer(), default=0)
circonscription = relationship("Circonscription", back_populates="resultats2024")
resultats_departement = relationship(ResultatsDepartement, back_populates="resultats_circonscriptions")
resultats_bureaux_vote: Mapped[List["ResultatsBureauVote"]] = relationship(
"ResultatsBureauVote", back_populates="resultats_circonscription")
voix_listes: Mapped[List["VoixListeCirconscription"]] = relationship("VoixListeCirconscription",
back_populates="resultats_circonscription")
class ResultatsCommune(Base):
__tablename__ = "resultats2024_commune"
@ -190,7 +165,6 @@ class ResultatsBureauVote(Base):
id: Mapped[int] = mapped_column(primary_key=True)
bv_id: Mapped[str] = mapped_column(ForeignKey("bureau_vote.id"))
resultats_commune_id: Mapped[int] = mapped_column(ForeignKey("resultats2024_commune.id"))
resultats_circo_id: Mapped[int] = mapped_column(ForeignKey("resultats2024_circonscription.id"), nullable=True)
inscrits: Mapped[int] = mapped_column(Integer(), default=0)
votants: Mapped[int] = mapped_column(Integer(), default=0)
abstentions: Mapped[int] = mapped_column(Integer(), default=0)
@ -200,7 +174,6 @@ class ResultatsBureauVote(Base):
bureau_vote = relationship("BureauVote", back_populates="resultats2024")
resultats_commune = relationship(ResultatsCommune, back_populates="resultats_bureaux_vote")
resultats_circonscription = relationship(ResultatsCirconscription, back_populates="resultats_bureaux_vote")
voix_listes: Mapped[List["VoixListeBureauVote"]] = relationship("VoixListeBureauVote",
back_populates="resultats_bureau_vote")
@ -239,19 +212,6 @@ class VoixListeDepartement(Base):
back_populates="voix_listes")
class VoixListeCirconscription(Base):
__tablename__ = "voix2024_circonscription"
liste_id: Mapped[int] = mapped_column(ForeignKey("liste2024.id"), primary_key=True)
resultats_circonscription_id: Mapped[int] = mapped_column(ForeignKey("resultats2024_circonscription.id"),
primary_key=True)
voix: Mapped[int] = mapped_column(Integer(), default=0)
liste: Mapped[Liste] = relationship(Liste, back_populates="resultats_par_circonscription")
resultats_circonscription: Mapped[ResultatsCirconscription] = relationship(ResultatsCirconscription,
back_populates="voix_listes")
class VoixListeCommune(Base):
__tablename__ = "voix2024_commune"

2
nupes/models/geographie.py
View File

@ -56,8 +56,6 @@ class Circonscription(Base):
departement: Mapped[Departement] = relationship(Departement)
bureaux_vote: Mapped[List["BureauVote"]] = relationship("BureauVote", back_populates="circonscription")
resultats2024 = relationship("ResultatsCirconscription", back_populates="circonscription")
class BureauVote(Base):
__tablename__ = "bureau_vote"

81
nupes/scripts/import_geographie.py
View File

@ -4,7 +4,6 @@ from datetime import datetime
import requests
from sqlalchemy import Engine, select
from sqlalchemy.orm import Session
from tqdm import tqdm
from nupes.cache import get_file
from nupes.models.geographie import BureauVote, Circonscription, Commune, Departement, Region
@ -22,7 +21,7 @@ def importer_regions(engine: Engine, verbose: bool = False) -> None:
features = json.load(f)['features']
with Session(engine) as session:
for feature in tqdm(features, desc="Régions", disable=not verbose):
for feature in features:
region_dict = feature['properties']
code_region = region_dict['reg_code'][0]
nom_region = region_dict['reg_name'][0]
@ -50,7 +49,7 @@ def importer_departements(engine: Engine, verbose: bool = False) -> None:
features = json.load(f)['features']
with Session(engine) as session:
for feature in tqdm(features, desc="Départements", disable=not verbose):
for feature in features:
dpt_dict = feature['properties']
code_dpt = dpt_dict['dep_code'][0]
nom_dpt = dpt_dict['dep_name'][0]
@ -79,7 +78,7 @@ def importer_communes(engine: Engine, verbose: bool = False) -> None:
features = json.load(f)['features']
with Session(engine) as session:
for feature in tqdm(features, desc="Communes", disable=not verbose):
for feature in features:
commune_dict = feature['properties']
code_commune = commune_dict['com_arm_code'][0]
nom_commune = commune_dict['com_name'][0]
@ -97,25 +96,40 @@ def importer_communes(engine: Engine, verbose: bool = False) -> None:
def importer_bureaux_vote(engine: Engine, verbose: bool = False) -> None:
file = get_file("https://files.data.gouv.fr/reu/contours-france-entiere-latest-v2.geojson",
"contours-france-entiere-latest-v2.geojson")
etag = requests.get(
"https://public.opendatasoft.com/api/explore/v2.1/catalog/datasets"
"/elections-france-bureau-vote-2022?select=data_processed").json()['data_processed']
file = get_file("https://public.opendatasoft.com/api/explore/v2.1/catalog/datasets"
"/elections-france-bureau-vote-2022/exports/geojson?lang=fr&timezone=Europe%2FParis",
"elections-france-bureau-vote-2022.geojson", etag)
with file.open('r') as f:
features = json.load(f)['features']
with Session(engine) as session:
for feature in tqdm(features, desc="Bureaux de vote", disable=not verbose):
for feature in features:
bv_dict = feature['properties']
code_commune = bv_dict['id_bv'].split('_')[0]
code_bv = bv_dict['numeroBureauVote']
dpt_code = bv_dict['codeDepartement']
numero_circo = int(bv_dict['codeCirconscription'][len(dpt_code):])
code_circo = f"{dpt_code}-{numero_circo:02d}"
bv_id = bv_dict['id_bv'].split()[0]
bv_libelle = f"Bureau {code_bv}"
code_commune = bv_dict['com_code']
if not code_commune:
print(feature)
continue
code_commune = code_commune.split('/')[0]
code_bv = bv_dict['code'] or "0"
code_circo = bv_dict['circonscription_code']
bv_id = f"{code_commune}_{code_bv}"
bv_libelle = bv_dict['libelle'] or "Bureau unique"
dpt_code, numero_circo = code_circo.split('-')
numero_circo = int(numero_circo)
if dpt_code == "987" or dpt_code == "988":
# Les communes de la Polynésie française et de Nouvelle-Calédonie ne sont pas disponibles,
# on les crée à la volée
if not session.execute(select(Commune).filter_by(code_insee=code_commune)).scalar_one_or_none():
session.add(Commune(code_insee=code_commune, libelle=bv_dict['com_name'], departement_code=dpt_code,
geometry={}))
if not session.execute(select(Commune).filter_by(code_insee=code_commune)).scalar_one_or_none():
print("Commune non trouvée avec le code", code_commune, "et le nom", bv_dict['nomCommune'])
print("Commune non trouvée avec le code", code_commune, "et le nom", bv_dict['com_name'])
continue
if not session.execute(select(Circonscription).filter_by(id=code_circo)).scalar_one_or_none():
@ -126,17 +140,50 @@ def importer_bureaux_vote(engine: Engine, verbose: bool = False) -> None:
bv.code_bureau = code_bv
bv.circo_code = code_circo
bv.libelle = bv_libelle
bv.geometry = feature['geometry']
bv.adresse = bv_dict['adresse']
else:
bv = BureauVote(id=bv_id, commune_code=code_commune, code_bureau=code_bv, circo_code=code_circo,
libelle=bv_libelle, adresse="", geometry=feature['geometry'])
libelle=bv_libelle, adresse=bv_dict['adresse'],
geometry={})
session.add(bv)
session.commit()
def importer_contours_bureaux_vote(engine: Engine, verbose: bool = False) -> None:
file = get_file("https://www.data.gouv.fr/fr/datasets/r/f98165a7-7c37-4705-a181-bcfc943edc73",
"contours-bureaux-vote.geojson")
with file.open('r') as f:
features = json.load(f)['features']
with Session(engine) as session:
for feature in features:
bv_id: str = feature['properties']['id_bv']
com_code, bv_code = bv_id.split('_')
bv_code = bv_code.replace("-", " ").replace(".", " ").strip()
while len(bv_code) >= 2 and bv_code[0] == '0':
bv_code = bv_code[1:]
while " " in bv_code:
bv_code = bv_code.replace(" ", " ")
bv_id = f"{com_code}_{bv_code}"
if bv := session.execute(select(BureauVote).filter_by(id=bv_id)).scalar_one_or_none():
bv.geometry = feature['geometry']
else:
results = session.execute(select(BureauVote).filter_by(commune_code=com_code)).scalars().all()
if len(results) == 1:
bv = results[0]
bv.geometry = feature['geometry']
else:
print(f"Bureau de vote {bv_id} non trouvé")
session.commit()
def run(engine: Engine, verbose: bool = False) -> None:
importer_regions(engine, verbose)
importer_departements(engine, verbose)
importer_communes(engine, verbose)
importer_bureaux_vote(engine, verbose)
importer_contours_bureaux_vote(engine, verbose)

316
nupes/scripts/import_resultats_2024.py
View File

@ -6,28 +6,25 @@ from tqdm import tqdm
from nupes.cache import get_file
from nupes.models import BureauVote, Departement
from nupes.models.europeennes2024 import ResultatsBureauVote, Liste, ResultatsCommune, \
ResultatsDepartement, ResultatsRegion, ResultatsFrance, \
VoixListeBureauVote, VoixListeCommune, VoixListeDepartement, VoixListeRegion, VoixListeFrance, \
ResultatsCirconscription, VoixListeCirconscription
from nupes.models.europeennes2024 import ResultatsBureauVote, Liste, VoixListeBureauVote, ResultatsCommune, \
ResultatsDepartement, ResultatsRegion, ResultatsFrance
def importer_resultats_bv(engine: Engine, verbose: bool = False) -> None:
file = get_file("https://www.data.gouv.fr/fr/datasets/r/cc1883d9-1265-4365-b754-fb6aef22d82e",
"resultats-europeennes-par-bureau-de-vote.csv")
file = get_file("https://www.data.gouv.fr/fr/datasets/r/937bb638-a487-40cd-9a0b-610d539a4207",
"resultats-temporaires-par-bureau-de-vote.csv")
with file.open('r') as f:
reader = DictReader(f, delimiter=';')
next(f) # On saute la première ligne
reader = DictReader(f)
with Session(engine) as session:
for row in tqdm(reader, desc="Bureau de vote", disable=not verbose):
com_code = str(row['Code commune']).zfill(5)
bv_code = row['Code BV']
bv_id = f"{com_code}_{bv_code}"
if bv := session.execute(select(BureauVote).filter_by(commune_code=com_code, code_bureau=bv_code)) \
.scalar_one_or_none():
bv_id = bv.id
else:
print(f"Bureau de vote {com_code}_{bv_code} non trouvé")
if not session.execute(select(BureauVote).filter_by(id=bv_id)).scalar_one_or_none():
print(f"Bureau de vote {bv_id} non trouvé")
continue
resultats_bv = session.execute(select(ResultatsBureauVote).filter_by(bv_id=bv_id)).scalar_one_or_none()
@ -90,296 +87,33 @@ def importer_resultats_bv(engine: Engine, verbose: bool = False) -> None:
voix_liste_bv.voix = int(row[f"Voix {liste.id}"])
session.commit()
session.commit()
def importer_resultats_commune(engine: Engine, verbose: bool = False) -> None:
file = get_file("https://www.data.gouv.fr/fr/datasets/r/6a782ef9-8ad6-4e66-832d-338b1041a42d",
"resultats-europeennes-par-commune.csv")
with file.open('r') as f:
reader = DictReader(f, delimiter=';')
with Session(engine) as session:
for row in tqdm(reader, desc="Commune", disable=not verbose):
com_code = str(row['Code commune']).zfill(5)
resultats_commune = session.execute(select(ResultatsCommune).filter_by(commune_id=com_code)) \
.scalar_one_or_none()
if not resultats_commune:
resultats_dpt = session.execute(select(ResultatsDepartement)
.filter_by(dpt_id=row['Code département'])) \
.scalar_one_or_none()
if not resultats_dpt:
dpt = session.execute(select(Departement)
.filter_by(code_insee=f"{row['Code département'].zfill(2)}")) \
.scalar_one()
resultats_reg = session.execute(select(ResultatsRegion)
.filter_by(region_id=dpt.region_code)) \
.scalar_one_or_none()
if not resultats_reg:
resultats_france = session.execute(select(ResultatsFrance)).scalar_one_or_none()
if not resultats_france:
session.add(ResultatsFrance())
resultats_france = session.execute(select(ResultatsFrance)).scalar_one()
resultats_reg = ResultatsRegion(region_id=str(dpt.region_code),
resultats_france_id=resultats_france.id)
session.add(resultats_reg)
resultats_reg = session.execute(select(ResultatsRegion)
.filter_by(region_id=dpt.region_code)).scalar_one()
resultats_dpt = ResultatsDepartement(dpt_id=row['Code département'],
resultats_region_id=resultats_reg.id)
session.add(resultats_dpt)
resultats_dpt = session.execute(select(ResultatsDepartement)
.filter_by(dpt_id=row['Code département'])).scalar_one()
resultats_commune = ResultatsCommune(commune_id=com_code, resultats_dpt_id=resultats_dpt.id)
session.add(resultats_commune)
resultats_commune = session.execute(select(ResultatsCommune).filter_by(commune_id=com_code)) \
.scalar_one()
resultats_commune.inscrits = int(row['Inscrits'])
resultats_commune.votants = int(row['Votants'])
resultats_commune.abstentions = int(row['Abstentions'])
resultats_commune.exprimes = int(row['Exprimés'])
resultats_commune.blancs = int(row['Blancs'])
resultats_commune.nuls = int(row['Nuls'])
for liste in session.execute(select(Liste)).scalars().all():
voix_liste_com = session.execute(select(VoixListeCommune)
.filter_by(resultats_commune_id=resultats_commune.id,
liste_id=liste.id)) \
.scalar_one_or_none()
if not voix_liste_com:
voix_liste_com = VoixListeCommune(resultats_commune_id=resultats_commune.id, liste_id=liste.id)
session.add(voix_liste_com)
voix_liste_com.voix = int(row[f"Voix {liste.id}"])
session.commit()
def calculer_resultats_commune(engine: Engine, verbose: bool = False) -> None:
pass
def importer_resultats_circo(engine: Engine, verbose: bool = False) -> None:
file = get_file("https://www.data.gouv.fr/fr/datasets/r/ee37cbef-3d2a-4efe-a395-530b85a63028",
"resultats-europeennes-par-circonscription.csv")
with file.open('r') as f:
reader = DictReader(f, delimiter=';')
with Session(engine) as session:
for row in tqdm(reader, desc="Circonscription", disable=not verbose):
circo_code = row['Code circonscription législative']
dpt_code, circo_code = circo_code[:]
resultats_circo = session.execute(select(ResultatsCirconscription).filter_by(circo_id=circo_code)) \
.scalar_one_or_none()
if not resultats_circo:
resultats_dpt = session.execute(select(ResultatsDepartement)
.filter_by(dpt_id=row['Code département'])) \
.scalar_one_or_none()
if not resultats_dpt:
dpt = session.execute(select(Departement)
.filter_by(code_insee=f"{row['Code département'].zfill(2)}")) \
.scalar_one()
resultats_reg = session.execute(select(ResultatsRegion)
.filter_by(region_id=dpt.region_code)) \
.scalar_one_or_none()
if not resultats_reg:
resultats_france = session.execute(select(ResultatsFrance)).scalar_one_or_none()
if not resultats_france:
session.add(ResultatsFrance())
resultats_france = session.execute(select(ResultatsFrance)).scalar_one()
resultats_reg = ResultatsRegion(region_id=str(dpt.region_code),
resultats_france_id=resultats_france.id)
session.add(resultats_reg)
resultats_reg = session.execute(select(ResultatsRegion)
.filter_by(region_id=dpt.region_code)).scalar_one()
resultats_dpt = ResultatsDepartement(dpt_id=row['Code département'],
resultats_region_id=resultats_reg.id)
session.add(resultats_dpt)
resultats_dpt = session.execute(select(ResultatsDepartement)
.filter_by(dpt_id=row['Code département'])).scalar_one()
resultats_circo = ResultatsCirconscription(circo_id=circo_code,
resultats_departement_id=resultats_dpt.id)
session.add(resultats_circo)
resultats_circo = session.execute(select(ResultatsCirconscription).filter_by(circo_id=circo_code)) \
.scalar_one()
resultats_circo.inscrits = int(row['Inscrits'])
resultats_circo.votants = int(row['Votants'])
resultats_circo.abstentions = int(row['Abstentions'])
resultats_circo.exprimes = int(row['Exprimés'])
resultats_circo.blancs = int(row['Blancs'])
resultats_circo.nuls = int(row['Nuls'])
for liste in session.execute(select(Liste)).scalars().all():
voix_liste_circo = session.execute(select(VoixListeCirconscription)
.filter_by(resultats_circonscription_id=resultats_circo.id,
liste_id=liste.id)) \
.scalar_one_or_none()
if not voix_liste_circo:
voix_liste_circo = VoixListeCirconscription(resultats_circonscription_id=resultats_circo.id,
liste_id=liste.id)
session.add(voix_liste_circo)
voix_liste_circo.voix = int(row[f"Voix {liste.id}"])
session.commit()
def calculer_resultats_circo(engine: Engine, verbose: bool = False) -> None:
pass
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",
"resultats-europeennes-par-circonscription.csv")
with file.open('r') as f:
reader = DictReader(f, delimiter=';')
with Session(engine) as session:
for row in tqdm(reader, desc="Département", disable=not verbose):
dpt_code = row['Code département']
resultats_dpt = session.execute(select(ResultatsDepartement).filter_by(dpt_id=dpt_code)) \
.scalar_one_or_none()
if not resultats_dpt:
dpt = session.execute(select(Departement)
.filter_by(code_insee=f"{row['Code département'].zfill(2)}")) \
.scalar_one()
resultats_reg = session.execute(select(ResultatsRegion)
.filter_by(region_id=dpt.region_code)) \
.scalar_one_or_none()
if not resultats_reg:
resultats_france = session.execute(select(ResultatsFrance)).scalar_one_or_none()
if not resultats_france:
session.add(ResultatsFrance())
resultats_france = session.execute(select(ResultatsFrance)).scalar_one()
resultats_reg = ResultatsRegion(region_id=str(dpt.region_code),
resultats_france_id=resultats_france.id)
session.add(resultats_reg)
resultats_reg = session.execute(select(ResultatsRegion)
.filter_by(region_id=dpt.region_code)).scalar_one()
resultats_dpt = ResultatsDepartement(dpt_id=dpt_code,
resultats_region_id=resultats_reg.id)
session.add(resultats_dpt)
resultats_dpt = session.execute(select(ResultatsDepartement).filter_by(dpt_id=dpt_code)) \
.scalar_one()
resultats_dpt.inscrits = int(row['Inscrits'])
resultats_dpt.votants = int(row['Votants'])
resultats_dpt.abstentions = int(row['Abstentions'])
resultats_dpt.exprimes = int(row['Exprimés'])
resultats_dpt.blancs = int(row['Blancs'])
resultats_dpt.nuls = int(row['Nuls'])
for liste in session.execute(select(Liste)).scalars().all():
voix_liste_dpt = session.execute(select(VoixListeDepartement)
.filter_by(resultats_departement_id=resultats_dpt.id,
liste_id=liste.id)) \
.scalar_one_or_none()
if not voix_liste_dpt:
voix_liste_dpt = VoixListeDepartement(resultats_departement_id=resultats_dpt.id,
liste_id=liste.id)
session.add(voix_liste_dpt)
voix_liste_dpt.voix = int(row[f"Voix {liste.id}"])
session.commit()
def calculer_resultats_departement(engine: Engine, verbose: bool = False) -> None:
pass
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",
"resultats-europeennes-par-commune.csv")
with file.open('r') as f:
reader = DictReader(f, delimiter=';')
with Session(engine) as session:
for row in tqdm(reader, desc="Région", disable=not verbose):
dpt_code = row['Code département']
resultats_reg = session.execute(select(ResultatsRegion).filter_by(region_id=dpt_code)) \
.scalar_one_or_none()
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()
if not resultats_france:
session.add(ResultatsFrance())
resultats_france = session.execute(select(ResultatsFrance)).scalar_one()
resultats_reg = ResultatsRegion(region_id=str(dpt.region_code),
resultats_france_id=resultats_france.id)
session.add(resultats_reg)
resultats_reg = session.execute(select(ResultatsRegion).filter_by(region_id=dpt_code)) \
.scalar_one()
resultats_reg.inscrits = int(row['Inscrits'])
resultats_reg.votants = int(row['Votants'])
resultats_reg.abstentions = int(row['Abstentions'])
resultats_reg.exprimes = int(row['Exprimés'])
resultats_reg.blancs = int(row['Blancs'])
resultats_reg.nuls = int(row['Nuls'])
for liste in session.execute(select(Liste)).scalars().all():
voix_liste_reg = session.execute(select(VoixListeRegion)
.filter_by(resultats_region_id=resultats_reg.id,
liste_id=liste.id)) \
.scalar_one_or_none()
if not voix_liste_reg:
voix_liste_reg = VoixListeRegion(resultats_region_id=resultats_reg.id, liste_id=liste.id)
session.add(voix_liste_reg)
voix_liste_reg.voix = int(row[f"Voix {liste.id}"])
session.commit()
def calculer_resultats_region(engine: Engine, verbose: bool = False) -> None:
pass
def calculer_resultats_france(engine: Engine, verbose: bool = False) -> None:
with Session(engine) as session:
resultats_france = session.execute(select(ResultatsFrance)).scalar_one()
resultats_france.inscrits = 0
resultats_france.votants = 0
resultats_france.abstentions = 0
resultats_france.exprimes = 0
resultats_france.blancs = 0
resultats_france.nuls = 0
for voix_liste_france in session.execute(
select(VoixListeFrance).filter_by(resultats_france_id=resultats_france.id)).scalars().all():
voix_liste_france.voix = 0 # Réinitialisation des voix
for resultats_reg in session.execute(select(ResultatsRegion)).scalars().all():
resultats_france.inscrits += resultats_reg.inscrits
resultats_france.votants += resultats_reg.votants
resultats_france.abstentions += resultats_reg.abstentions
resultats_france.exprimes += resultats_reg.exprimes
resultats_france.blancs += resultats_reg.blancs
resultats_france.nuls += resultats_reg.nuls
for voix_liste_reg in session.execute(select(VoixListeRegion)
.filter_by(resultats_region_id=resultats_reg.id)).scalars().all():
voix_liste_france = session.execute(select(VoixListeFrance)
.filter_by(resultats_france_id=resultats_france.id,
liste_id=voix_liste_reg.liste_id)) \
.scalar_one_or_none()
if not voix_liste_france:
voix_liste_france = VoixListeFrance(resultats_france_id=resultats_france.id,
liste_id=voix_liste_reg.liste_id)
session.add(voix_liste_france)
voix_liste_france.voix += voix_liste_reg.voix
session.commit()
def calculer_resultats_national(engine: Engine, verbose: bool = False) -> None:
pass
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)
calculer_resultats_france(engine, verbose)
calculer_resultats_commune(engine, verbose)
# calculer_resultats_circo(engine, verbose)
calculer_resultats_departement(engine, verbose)
calculer_resultats_region(engine, verbose)
calculer_resultats_national(engine, verbose)