Compare commits

..

No commits in common. "7aa92d28d49798f42190e97167a35c73154a781d" and "df85902b5bccdbf43099b400f1a60149248142bc" have entirely different histories.

11 changed files with 9 additions and 1505 deletions

View File

@ -1,259 +0,0 @@
"""ajout modèles législatives
Revision ID: 99948a33112b
Revises: c0443e979a90
Create Date: 2024-06-20 12:47:52.732224
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = '99948a33112b'
down_revision: Union[str, None] = 'c0443e979a90'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('legislatives_2022_bloc',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('nom', sa.String(length=32), nullable=False),
sa.Column('couleur', sa.String(length=7), nullable=False),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('nom')
)
op.create_table('legislatives_2022_resultats_france',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('inscrits_t1', sa.Integer(), nullable=False),
sa.Column('votants_t1', sa.Integer(), nullable=False),
sa.Column('abstentions_t1', sa.Integer(), nullable=False),
sa.Column('exprimes_t1', sa.Integer(), nullable=False),
sa.Column('blancs_t1', sa.Integer(), nullable=False),
sa.Column('nuls_t1', sa.Integer(), nullable=False),
sa.Column('inscrits_t2', sa.Integer(), nullable=False),
sa.Column('votants_t2', sa.Integer(), nullable=False),
sa.Column('abstentions_t2', sa.Integer(), nullable=False),
sa.Column('exprimes_t2', sa.Integer(), nullable=False),
sa.Column('blancs_t2', sa.Integer(), nullable=False),
sa.Column('nuls_t2', sa.Integer(), nullable=False),
sa.PrimaryKeyConstraint('id')
)
op.create_table('legislatives_2022_nuance',
sa.Column('code', sa.String(length=8), nullable=False),
sa.Column('nom', sa.String(length=64), nullable=False),
sa.Column('couleur', sa.String(length=7), nullable=False),
sa.Column('bloc_id', sa.Integer(), nullable=False),
sa.ForeignKeyConstraint(['bloc_id'], ['legislatives_2022_bloc.id'], ),
sa.PrimaryKeyConstraint('code'),
sa.UniqueConstraint('nom')
)
op.create_table('legislatives_2022_resultats_region',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('region_id', sa.String(length=3), nullable=False),
sa.Column('resultats_france_id', sa.Integer(), nullable=False),
sa.Column('inscrits_t1', sa.Integer(), nullable=False),
sa.Column('votants_t1', sa.Integer(), nullable=False),
sa.Column('abstentions_t1', sa.Integer(), nullable=False),
sa.Column('exprimes_t1', sa.Integer(), nullable=False),
sa.Column('blancs_t1', sa.Integer(), nullable=False),
sa.Column('nuls_t1', sa.Integer(), nullable=False),
sa.Column('inscrits_t2', sa.Integer(), nullable=False),
sa.Column('votants_t2', sa.Integer(), nullable=False),
sa.Column('abstentions_t2', sa.Integer(), nullable=False),
sa.Column('exprimes_t2', sa.Integer(), nullable=False),
sa.Column('blancs_t2', sa.Integer(), nullable=False),
sa.Column('nuls_t2', sa.Integer(), nullable=False),
sa.ForeignKeyConstraint(['region_id'], ['region.code_insee'], ),
sa.ForeignKeyConstraint(['resultats_france_id'], ['legislatives_2022_resultats_france.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_table('legislatives_2022_resultats_departement',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('dpt_id', sa.String(length=3), nullable=False),
sa.Column('resultats_region_id', sa.Integer(), nullable=True),
sa.Column('inscrits_t1', sa.Integer(), nullable=False),
sa.Column('votants_t1', sa.Integer(), nullable=False),
sa.Column('abstentions_t1', sa.Integer(), nullable=False),
sa.Column('exprimes_t1', sa.Integer(), nullable=False),
sa.Column('blancs_t1', sa.Integer(), nullable=False),
sa.Column('nuls_t1', sa.Integer(), nullable=False),
sa.Column('inscrits_t2', sa.Integer(), nullable=False),
sa.Column('votants_t2', sa.Integer(), nullable=False),
sa.Column('abstentions_t2', sa.Integer(), nullable=False),
sa.Column('exprimes_t2', sa.Integer(), nullable=False),
sa.Column('blancs_t2', sa.Integer(), nullable=False),
sa.Column('nuls_t2', sa.Integer(), nullable=False),
sa.ForeignKeyConstraint(['dpt_id'], ['departement.code_insee'], ),
sa.ForeignKeyConstraint(['resultats_region_id'], ['legislatives_2022_resultats_region.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_table('legislatives_2022_voix_france',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('nuance_id', sa.String(length=8), nullable=False),
sa.Column('resultats_france_id', sa.Integer(), nullable=False),
sa.Column('voix_t1', sa.Integer(), nullable=False),
sa.Column('voix_t2', sa.Integer(), nullable=False),
sa.ForeignKeyConstraint(['nuance_id'], ['legislatives_2022_nuance.code'], ),
sa.ForeignKeyConstraint(['resultats_france_id'], ['legislatives_2022_resultats_france.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_table('legislatives_2022_voix_region',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('nuance_id', sa.String(length=8), nullable=False),
sa.Column('resultats_region_id', sa.Integer(), nullable=False),
sa.Column('voix_t1', sa.Integer(), nullable=False),
sa.Column('voix_t2', sa.Integer(), nullable=False),
sa.ForeignKeyConstraint(['nuance_id'], ['legislatives_2022_nuance.code'], ),
sa.ForeignKeyConstraint(['resultats_region_id'], ['legislatives_2022_resultats_region.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_table('legislatives_2022_candidat',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('circonscription_id', sa.String(length=6), nullable=False),
sa.Column('numero', sa.Integer(), nullable=False),
sa.Column('nuance_id', sa.String(length=8), nullable=False),
sa.Column('bloc_id', sa.Integer(), nullable=False),
sa.Column('nom', sa.String(length=256), nullable=False),
sa.Column('prenom', sa.String(length=256), nullable=False),
sa.Column('sexe', sa.Enum('MASCULIN', 'FEMININ', name='genre'), nullable=False),
sa.Column('date_naissance', sa.Date(), nullable=False),
sa.Column('profession', sa.String(length=256), nullable=False),
sa.Column('sortant', sa.Boolean(), nullable=False),
sa.Column('nom_suppleance', sa.String(length=256), nullable=False),
sa.Column('prenom_suppleance', sa.String(length=256), nullable=False),
sa.Column('sexe_suppleance', sa.Enum('MASCULIN', 'FEMININ', name='genre'), nullable=False),
sa.Column('date_naissance_suppleance', sa.Date(), nullable=False),
sa.Column('sortant_suppleance', sa.Boolean(), nullable=False),
sa.ForeignKeyConstraint(['bloc_id'], ['legislatives_2022_bloc.id'], ),
sa.ForeignKeyConstraint(['circonscription_id'], ['circonscription.id'], ),
sa.ForeignKeyConstraint(['nuance_id'], ['legislatives_2022_nuance.code'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_table('legislatives_2022_resultats_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_t1', sa.Integer(), nullable=False),
sa.Column('votants_t1', sa.Integer(), nullable=False),
sa.Column('abstentions_t1', sa.Integer(), nullable=False),
sa.Column('exprimes_t1', sa.Integer(), nullable=False),
sa.Column('blancs_t1', sa.Integer(), nullable=False),
sa.Column('nuls_t1', sa.Integer(), nullable=False),
sa.Column('inscrits_t2', sa.Integer(), nullable=False),
sa.Column('votants_t2', sa.Integer(), nullable=False),
sa.Column('abstentions_t2', sa.Integer(), nullable=False),
sa.Column('exprimes_t2', sa.Integer(), nullable=False),
sa.Column('blancs_t2', sa.Integer(), nullable=False),
sa.Column('nuls_t2', sa.Integer(), nullable=False),
sa.ForeignKeyConstraint(['circo_id'], ['circonscription.id'], ),
sa.ForeignKeyConstraint(['resultats_departement_id'], ['legislatives_2022_resultats_departement.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_table('legislatives_2022_resultats_commune',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('commune_id', sa.String(length=5), nullable=False),
sa.Column('resultats_dpt_id', sa.Integer(), nullable=False),
sa.Column('inscrits_t1', sa.Integer(), nullable=False),
sa.Column('votants_t1', sa.Integer(), nullable=False),
sa.Column('abstentions_t1', sa.Integer(), nullable=False),
sa.Column('exprimes_t1', sa.Integer(), nullable=False),
sa.Column('blancs_t1', sa.Integer(), nullable=False),
sa.Column('nuls_t1', sa.Integer(), nullable=False),
sa.Column('inscrits_t2', sa.Integer(), nullable=False),
sa.Column('votants_t2', sa.Integer(), nullable=False),
sa.Column('abstentions_t2', sa.Integer(), nullable=False),
sa.Column('exprimes_t2', sa.Integer(), nullable=False),
sa.Column('blancs_t2', sa.Integer(), nullable=False),
sa.Column('nuls_t2', sa.Integer(), nullable=False),
sa.ForeignKeyConstraint(['commune_id'], ['commune.code_insee'], ),
sa.ForeignKeyConstraint(['resultats_dpt_id'], ['legislatives_2022_resultats_departement.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_table('legislatives_2022_voix_departement',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('nuance_id', sa.String(length=8), nullable=False),
sa.Column('resultats_departement_id', sa.Integer(), nullable=False),
sa.Column('voix_t1', sa.Integer(), nullable=False),
sa.Column('voix_t2', sa.Integer(), nullable=False),
sa.ForeignKeyConstraint(['nuance_id'], ['legislatives_2022_nuance.code'], ),
sa.ForeignKeyConstraint(['resultats_departement_id'], ['legislatives_2022_resultats_departement.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_table('legislatives_2022_resultats_bureau_vote',
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_t1', sa.Integer(), nullable=False),
sa.Column('votants_t1', sa.Integer(), nullable=False),
sa.Column('abstentions_t1', sa.Integer(), nullable=False),
sa.Column('exprimes_t1', sa.Integer(), nullable=False),
sa.Column('blancs_t1', sa.Integer(), nullable=False),
sa.Column('nuls_t1', sa.Integer(), nullable=False),
sa.Column('inscrits_t2', sa.Integer(), nullable=False),
sa.Column('votants_t2', sa.Integer(), nullable=False),
sa.Column('abstentions_t2', sa.Integer(), nullable=False),
sa.Column('exprimes_t2', sa.Integer(), nullable=False),
sa.Column('blancs_t2', sa.Integer(), nullable=False),
sa.Column('nuls_t2', sa.Integer(), nullable=False),
sa.ForeignKeyConstraint(['bv_id'], ['bureau_vote.id'], ),
sa.ForeignKeyConstraint(['resultats_circo_id'], ['legislatives_2022_resultats_circonscription.id'], ),
sa.ForeignKeyConstraint(['resultats_commune_id'], ['legislatives_2022_resultats_commune.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_table('legislatives_2022_voix_circonscription',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('candidat_id', sa.Integer(), nullable=False),
sa.Column('resultats_circonscription_id', sa.Integer(), nullable=False),
sa.Column('voix_t1', sa.Integer(), nullable=False),
sa.Column('voix_t2', sa.Integer(), nullable=False),
sa.ForeignKeyConstraint(['candidat_id'], ['legislatives_2022_candidat.id'], ),
sa.ForeignKeyConstraint(['resultats_circonscription_id'], ['legislatives_2022_resultats_circonscription.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_table('legislatives_2022_voix_commune',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('nuance_id', sa.String(length=8), nullable=False),
sa.Column('resultats_commune_id', sa.Integer(), nullable=False),
sa.Column('voix_t1', sa.Integer(), nullable=False),
sa.Column('voix_t2', sa.Integer(), nullable=False),
sa.ForeignKeyConstraint(['nuance_id'], ['legislatives_2022_nuance.code'], ),
sa.ForeignKeyConstraint(['resultats_commune_id'], ['legislatives_2022_resultats_commune.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_table('legislatives_2022_voix_bureau_vote',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('candidat_id', sa.Integer(), nullable=False),
sa.Column('resultats_bureau_vote_id', sa.Integer(), nullable=False),
sa.Column('voix_t1', sa.Integer(), nullable=False),
sa.Column('voix_t2', sa.Integer(), nullable=False),
sa.ForeignKeyConstraint(['candidat_id'], ['legislatives_2022_candidat.id'], ),
sa.ForeignKeyConstraint(['resultats_bureau_vote_id'], ['legislatives_2022_resultats_bureau_vote.id'], ),
sa.PrimaryKeyConstraint('id')
)
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.drop_table('legislatives_2022_voix_bureau_vote')
op.drop_table('legislatives_2022_voix_commune')
op.drop_table('legislatives_2022_voix_circonscription')
op.drop_table('legislatives_2022_resultats_bureau_vote')
op.drop_table('legislatives_2022_voix_departement')
op.drop_table('legislatives_2022_resultats_commune')
op.drop_table('legislatives_2022_resultats_circonscription')
op.drop_table('legislatives_2022_candidat')
op.drop_table('legislatives_2022_voix_region')
op.drop_table('legislatives_2022_voix_france')
op.drop_table('legislatives_2022_resultats_departement')
op.drop_table('legislatives_2022_resultats_region')
op.drop_table('legislatives_2022_nuance')
op.drop_table('legislatives_2022_resultats_france')
op.drop_table('legislatives_2022_bloc')
# ### end Alembic commands ###

View File

@ -7,7 +7,7 @@ from dotenv import load_dotenv
from sqlalchemy import create_engine from sqlalchemy import create_engine
from nupes.scripts import import_geographie from nupes.scripts import import_geographie
from nupes.scripts import legislatives2022, europeennes2024, legislatives2024 from nupes.scripts import europeennes2024
def parse_args(): def parse_args():
@ -15,8 +15,7 @@ def parse_args():
parser.add_argument('--type', '-t', type=str, help="Type d'élection", parser.add_argument('--type', '-t', type=str, help="Type d'élection",
choices=["présidentielle", "législatives", "sénatoriales", "européennes", choices=["présidentielle", "législatives", "sénatoriales", "européennes",
"régionales", "départementales", "municipales"], default="européennes") "régionales", "départementales", "municipales"], default="européennes")
parser.add_argument('--year', '-y', type=int, help="Année de l'élection", parser.add_argument('--year', '-y', type=int, help="Année de l'élection", default=2024)
choices=[2022, 2024], default=2024)
parser.add_argument('action', help="Action à réaliser", parser.add_argument('action', help="Action à réaliser",
choices=["import_candidats", "import_geographie", "import_resultats", choices=["import_candidats", "import_geographie", "import_resultats",
"import_resultats_web", "export_resultats"]) "import_resultats_web", "export_resultats"])
@ -34,12 +33,8 @@ def main():
verbose = args.verbose verbose = args.verbose
match args.type, args.year: match args.type, args.year:
case "législatives", 2022:
election_module = legislatives2022
case "européennes", 2024: case "européennes", 2024:
election_module = europeennes2024 election_module = europeennes2024
case "législatives", 2024:
election_module = legislatives2024
case _: case _:
print(f"Type d'élection {args.type} {args.year} non reconnu") print(f"Type d'élection {args.type} {args.year} non reconnu")
return return

View File

@ -1,4 +1,3 @@
from .base import Base from .base import Base
from .geographie import * from .geographie import *
from .legislatives2022 import *
from .europeennes2024 import * from .europeennes2024 import *

View File

@ -15,7 +15,6 @@ class Region(Base):
departements: Mapped[List["Departement"]] = relationship("Departement", back_populates="region") departements: Mapped[List["Departement"]] = relationship("Departement", back_populates="region")
resultats_legislatives_2022 = relationship("ResultatsRegionLegislatives2022", back_populates="region")
resultats_europeennes_2024 = relationship("ResultatsRegionEuropeennes2024", back_populates="region") resultats_europeennes_2024 = relationship("ResultatsRegionEuropeennes2024", back_populates="region")
@ -31,7 +30,6 @@ class Departement(Base):
circonscriptions: Mapped[List["Circonscription"]] = relationship("Circonscription", back_populates="departement") circonscriptions: Mapped[List["Circonscription"]] = relationship("Circonscription", back_populates="departement")
communes: Mapped[List["Commune"]] = relationship("Commune", back_populates="departement") communes: Mapped[List["Commune"]] = relationship("Commune", back_populates="departement")
resultats_legislatives_2022 = relationship("ResultatsDepartementLegislatives2022", back_populates="departement")
resultats_europeennes_2024 = relationship("ResultatsDepartementEuropeennes2024", back_populates="departement") resultats_europeennes_2024 = relationship("ResultatsDepartementEuropeennes2024", back_populates="departement")
@ -46,7 +44,6 @@ class Commune(Base):
departement: Mapped[Departement] = relationship(Departement, back_populates="communes") departement: Mapped[Departement] = relationship(Departement, back_populates="communes")
bureaux_vote: Mapped[List["BureauVote"]] = relationship("BureauVote", back_populates="commune") bureaux_vote: Mapped[List["BureauVote"]] = relationship("BureauVote", back_populates="commune")
resultats_legislatives_2022 = relationship("ResultatsCommuneLegislatives2022", back_populates="commune")
resultats_europeennes_2024 = relationship("ResultatsCommuneEuropeennes2024", back_populates="commune") resultats_europeennes_2024 = relationship("ResultatsCommuneEuropeennes2024", back_populates="commune")
@ -61,9 +58,6 @@ class Circonscription(Base):
departement: Mapped[Departement] = relationship(Departement, back_populates="circonscriptions") departement: Mapped[Departement] = relationship(Departement, back_populates="circonscriptions")
bureaux_vote: Mapped[List["BureauVote"]] = relationship("BureauVote", back_populates="circonscription") bureaux_vote: Mapped[List["BureauVote"]] = relationship("BureauVote", back_populates="circonscription")
candidats_legislatives_2022 = relationship("CandidatLegislatives2022", back_populates="circonscription")
resultats_legislatives_2022 = relationship("ResultatsCirconscriptionLegislatives2022",
back_populates="circonscription")
resultats_europeennes_2024 = relationship("ResultatsCirconscriptionEuropeennes2024", resultats_europeennes_2024 = relationship("ResultatsCirconscriptionEuropeennes2024",
back_populates="circonscription") back_populates="circonscription")
@ -82,5 +76,4 @@ class BureauVote(Base):
commune: Mapped[Commune] = relationship(Commune, back_populates="bureaux_vote") commune: Mapped[Commune] = relationship(Commune, back_populates="bureaux_vote")
circonscription: Mapped[Circonscription] = relationship(Circonscription, back_populates="bureaux_vote") circonscription: Mapped[Circonscription] = relationship(Circonscription, back_populates="bureaux_vote")
resultats_legislatives_2022 = relationship("ResultatsBureauVoteLegislatives2022", back_populates="bureau_vote")
resultats_europeennes_2024 = relationship("ResultatsBureauVoteEuropeennes2024", back_populates="bureau_vote") resultats_europeennes_2024 = relationship("ResultatsBureauVoteEuropeennes2024", back_populates="bureau_vote")

View File

@ -1,333 +0,0 @@
import enum
from datetime import date
from typing import List
from sqlalchemy import Boolean, Date, Enum, ForeignKey, Integer, String
from sqlalchemy.orm import mapped_column, Mapped, relationship
from nupes.models import Base, Region, Departement, Commune, Circonscription
class BlocLegislatives2022(Base):
__tablename__ = "legislatives_2022_bloc"
id: Mapped[int] = mapped_column(primary_key=True)
nom: Mapped[str] = mapped_column(String(32), unique=True)
couleur: Mapped[str] = mapped_column(String(7))
candidats: Mapped[List["CandidatLegislatives2022"]] = relationship("CandidatLegislatives2022",
back_populates="bloc")
nuances: Mapped[List["NuanceLegislatives2022"]] = relationship(
"NuanceLegislatives2022", back_populates="bloc")
class NuanceLegislatives2022(Base):
__tablename__ = "legislatives_2022_nuance"
code: Mapped[str] = mapped_column(String(8), primary_key=True)
nom: Mapped[str] = mapped_column(String(64), unique=True)
couleur: Mapped[str] = mapped_column(String(7))
bloc_id: Mapped[int] = mapped_column(ForeignKey("legislatives_2022_bloc.id"))
bloc: Mapped[BlocLegislatives2022] = relationship(BlocLegislatives2022, back_populates="nuances")
candidats: Mapped[List["CandidatLegislatives2022"]] = relationship(
"CandidatLegislatives2022", back_populates="nuance")
resultats_nationaux: Mapped[List["VoixFranceLegislatives2022"]] = relationship(
"VoixFranceLegislatives2022", back_populates="nuance")
resultats_par_region: Mapped[List["VoixRegionLegislatives2022"]] = relationship(
"VoixRegionLegislatives2022", back_populates="nuance")
resultats_par_departement: Mapped[List["VoixDepartementLegislatives2022"]] = relationship(
"VoixDepartementLegislatives2022", back_populates="nuance")
resultats_par_commune: Mapped[List["VoixCommuneLegislatives2022"]] = relationship(
"VoixCommuneLegislatives2022", back_populates="nuance")
class CandidatLegislatives2022(Base):
class Genre(enum.Enum):
MASCULIN = "M"
FEMININ = "F"
__tablename__ = "legislatives_2022_candidat"
id: Mapped[int] = mapped_column(primary_key=True)
circonscription_id: Mapped[str] = mapped_column(ForeignKey("circonscription.id"))
numero: Mapped[int] = mapped_column(Integer())
nuance_id: Mapped[str] = mapped_column(ForeignKey("legislatives_2022_nuance.code"))
bloc_id: Mapped[int] = mapped_column(ForeignKey("legislatives_2022_bloc.id"))
nom: Mapped[str] = mapped_column(String(256))
prenom: Mapped[str] = mapped_column(String(256))
sexe: Mapped[str] = mapped_column(Enum(Genre))
date_naissance: Mapped[date] = mapped_column(Date())
profession: Mapped[str] = mapped_column(String(256))
sortant: Mapped[bool] = mapped_column(Boolean())
nom_suppleance: Mapped[str] = mapped_column(String(256))
prenom_suppleance: Mapped[str] = mapped_column(String(256))
sexe_suppleance: Mapped[str] = mapped_column(Enum(Genre))
date_naissance_suppleance: Mapped[date] = mapped_column(Date())
sortant_suppleance: Mapped[bool] = mapped_column(Boolean())
circonscription: Mapped[Circonscription] = relationship(
"Circonscription", back_populates="candidats_legislatives_2022")
nuance: Mapped[NuanceLegislatives2022] = relationship(NuanceLegislatives2022, back_populates="candidats")
bloc: Mapped[BlocLegislatives2022] = relationship(BlocLegislatives2022, back_populates="candidats")
resultats_par_circonscription: Mapped[List["VoixCirconscriptionLegislatives2022"]] = relationship(
"VoixCirconscriptionLegislatives2022", back_populates="candidat")
resultats_par_bureau_vote: Mapped[List["VoixBureauVoteLegislatives2022"]] = relationship(
"VoixBureauVoteLegislatives2022", back_populates="candidat")
class ResultatsFranceLegislatives2022(Base):
__tablename__ = "legislatives_2022_resultats_france"
id: Mapped[int] = mapped_column(primary_key=True)
inscrits_t1: Mapped[int] = mapped_column(Integer(), default=0)
votants_t1: Mapped[int] = mapped_column(Integer(), default=0)
abstentions_t1: Mapped[int] = mapped_column(Integer(), default=0)
exprimes_t1: Mapped[int] = mapped_column(Integer(), default=0)
blancs_t1: Mapped[int] = mapped_column(Integer(), default=0)
nuls_t1: Mapped[int] = mapped_column(Integer(), default=0)
inscrits_t2: Mapped[int] = mapped_column(Integer(), default=0)
votants_t2: Mapped[int] = mapped_column(Integer(), default=0)
abstentions_t2: Mapped[int] = mapped_column(Integer(), default=0)
exprimes_t2: Mapped[int] = mapped_column(Integer(), default=0)
blancs_t2: Mapped[int] = mapped_column(Integer(), default=0)
nuls_t2: Mapped[int] = mapped_column(Integer(), default=0)
resultats_regions: Mapped[List["ResultatsRegionLegislatives2022"]] = relationship(
"ResultatsRegionLegislatives2022", back_populates="resultats_france")
voix: Mapped[List["VoixFranceLegislatives2022"]] = relationship(
"VoixFranceLegislatives2022", back_populates="resultats_france")
class ResultatsRegionLegislatives2022(Base):
__tablename__ = "legislatives_2022_resultats_region"
id: Mapped[int] = mapped_column(primary_key=True)
region_id: Mapped[str] = mapped_column(ForeignKey("region.code_insee"))
resultats_france_id: Mapped[int] = mapped_column(ForeignKey("legislatives_2022_resultats_france.id"))
inscrits_t1: Mapped[int] = mapped_column(Integer(), default=0)
votants_t1: Mapped[int] = mapped_column(Integer(), default=0)
abstentions_t1: Mapped[int] = mapped_column(Integer(), default=0)
exprimes_t1: Mapped[int] = mapped_column(Integer(), default=0)
blancs_t1: Mapped[int] = mapped_column(Integer(), default=0)
nuls_t1: Mapped[int] = mapped_column(Integer(), default=0)
inscrits_t2: Mapped[int] = mapped_column(Integer(), default=0)
votants_t2: Mapped[int] = mapped_column(Integer(), default=0)
abstentions_t2: Mapped[int] = mapped_column(Integer(), default=0)
exprimes_t2: Mapped[int] = mapped_column(Integer(), default=0)
blancs_t2: Mapped[int] = mapped_column(Integer(), default=0)
nuls_t2: Mapped[int] = mapped_column(Integer(), default=0)
region = relationship(Region, back_populates="resultats_legislatives_2022")
resultats_france = relationship(ResultatsFranceLegislatives2022, back_populates="resultats_regions")
resultats_departements: Mapped[List["ResultatsDepartementLegislatives2022"]] = relationship(
"ResultatsDepartementLegislatives2022", back_populates="resultats_region")
voix: Mapped[List["VoixRegionLegislatives2022"]] = relationship(
"VoixRegionLegislatives2022", back_populates="resultats_region")
class ResultatsDepartementLegislatives2022(Base):
__tablename__ = "legislatives_2022_resultats_departement"
id: Mapped[int] = mapped_column(primary_key=True)
dpt_id: Mapped[str] = mapped_column(ForeignKey("departement.code_insee"))
resultats_region_id: Mapped[int] = mapped_column(ForeignKey("legislatives_2022_resultats_region.id"), nullable=True)
inscrits_t1: Mapped[int] = mapped_column(Integer(), default=0)
votants_t1: Mapped[int] = mapped_column(Integer(), default=0)
abstentions_t1: Mapped[int] = mapped_column(Integer(), default=0)
exprimes_t1: Mapped[int] = mapped_column(Integer(), default=0)
blancs_t1: Mapped[int] = mapped_column(Integer(), default=0)
nuls_t1: Mapped[int] = mapped_column(Integer(), default=0)
inscrits_t2: Mapped[int] = mapped_column(Integer(), default=0)
votants_t2: Mapped[int] = mapped_column(Integer(), default=0)
abstentions_t2: Mapped[int] = mapped_column(Integer(), default=0)
exprimes_t2: Mapped[int] = mapped_column(Integer(), default=0)
blancs_t2: Mapped[int] = mapped_column(Integer(), default=0)
nuls_t2: Mapped[int] = mapped_column(Integer(), default=0)
departement = relationship(Departement, back_populates="resultats_legislatives_2022")
resultats_region = relationship(ResultatsRegionLegislatives2022, back_populates="resultats_departements")
resultats_communes: Mapped[List["ResultatsCommuneLegislatives2022"]] = relationship(
"ResultatsCommuneLegislatives2022", back_populates="resultats_departement")
resultats_circonscriptions: Mapped[List["ResultatsCirconscriptionLegislatives2022"]] = relationship(
"ResultatsCirconscriptionLegislatives2022", back_populates="resultats_departement")
voix: Mapped[List["VoixDepartementLegislatives2022"]] = relationship(
"VoixDepartementLegislatives2022", back_populates="resultats_departement")
class ResultatsCirconscriptionLegislatives2022(Base):
__tablename__ = "legislatives_2022_resultats_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("legislatives_2022_resultats_departement.id"))
inscrits_t1: Mapped[int] = mapped_column(Integer(), default=0)
votants_t1: Mapped[int] = mapped_column(Integer(), default=0)
abstentions_t1: Mapped[int] = mapped_column(Integer(), default=0)
exprimes_t1: Mapped[int] = mapped_column(Integer(), default=0)
blancs_t1: Mapped[int] = mapped_column(Integer(), default=0)
nuls_t1: Mapped[int] = mapped_column(Integer(), default=0)
inscrits_t2: Mapped[int] = mapped_column(Integer(), default=0)
votants_t2: Mapped[int] = mapped_column(Integer(), default=0)
abstentions_t2: Mapped[int] = mapped_column(Integer(), default=0)
exprimes_t2: Mapped[int] = mapped_column(Integer(), default=0)
blancs_t2: Mapped[int] = mapped_column(Integer(), default=0)
nuls_t2: Mapped[int] = mapped_column(Integer(), default=0)
circonscription = relationship("Circonscription", back_populates="resultats_legislatives_2022")
resultats_departement = relationship(ResultatsDepartementLegislatives2022,
back_populates="resultats_circonscriptions")
resultats_bureaux_vote: Mapped[List["ResultatsBureauVoteLegislatives2022"]] = relationship(
"ResultatsBureauVoteLegislatives2022", back_populates="resultats_circonscription")
voix: Mapped[List["VoixCirconscriptionLegislatives2022"]] = relationship(
"VoixCirconscriptionLegislatives2022", back_populates="resultats_circonscription")
class ResultatsCommuneLegislatives2022(Base):
__tablename__ = "legislatives_2022_resultats_commune"
id: Mapped[int] = mapped_column(primary_key=True)
commune_id: Mapped[str] = mapped_column(ForeignKey("commune.code_insee"))
resultats_dpt_id: Mapped[int] = mapped_column(ForeignKey("legislatives_2022_resultats_departement.id"))
inscrits_t1: Mapped[int] = mapped_column(Integer(), default=0)
votants_t1: Mapped[int] = mapped_column(Integer(), default=0)
abstentions_t1: Mapped[int] = mapped_column(Integer(), default=0)
exprimes_t1: Mapped[int] = mapped_column(Integer(), default=0)
blancs_t1: Mapped[int] = mapped_column(Integer(), default=0)
nuls_t1: Mapped[int] = mapped_column(Integer(), default=0)
inscrits_t2: Mapped[int] = mapped_column(Integer(), default=0)
votants_t2: Mapped[int] = mapped_column(Integer(), default=0)
abstentions_t2: Mapped[int] = mapped_column(Integer(), default=0)
exprimes_t2: Mapped[int] = mapped_column(Integer(), default=0)
blancs_t2: Mapped[int] = mapped_column(Integer(), default=0)
nuls_t2: Mapped[int] = mapped_column(Integer(), default=0)
commune = relationship(Commune, back_populates="resultats_legislatives_2022")
resultats_departement = relationship(ResultatsDepartementLegislatives2022, back_populates="resultats_communes")
resultats_bureaux_vote: Mapped[List["ResultatsBureauVoteLegislatives2022"]] = relationship(
"ResultatsBureauVoteLegislatives2022", back_populates="resultats_commune")
voix: Mapped[List["VoixCommuneLegislatives2022"]] = relationship(
"VoixCommuneLegislatives2022", back_populates="resultats_commune")
class ResultatsBureauVoteLegislatives2022(Base):
__tablename__ = "legislatives_2022_resultats_bureau_vote"
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("legislatives_2022_resultats_commune.id"))
resultats_circo_id: Mapped[int] = mapped_column(ForeignKey("legislatives_2022_resultats_circonscription.id"),
nullable=True)
inscrits_t1: Mapped[int] = mapped_column(Integer(), default=0)
votants_t1: Mapped[int] = mapped_column(Integer(), default=0)
abstentions_t1: Mapped[int] = mapped_column(Integer(), default=0)
exprimes_t1: Mapped[int] = mapped_column(Integer(), default=0)
blancs_t1: Mapped[int] = mapped_column(Integer(), default=0)
nuls_t1: Mapped[int] = mapped_column(Integer(), default=0)
inscrits_t2: Mapped[int] = mapped_column(Integer(), default=0)
votants_t2: Mapped[int] = mapped_column(Integer(), default=0)
abstentions_t2: Mapped[int] = mapped_column(Integer(), default=0)
exprimes_t2: Mapped[int] = mapped_column(Integer(), default=0)
blancs_t2: Mapped[int] = mapped_column(Integer(), default=0)
nuls_t2: Mapped[int] = mapped_column(Integer(), default=0)
bureau_vote = relationship("BureauVote", back_populates="resultats_legislatives_2022")
resultats_commune = relationship(ResultatsCommuneLegislatives2022, back_populates="resultats_bureaux_vote")
resultats_circonscription = relationship(ResultatsCirconscriptionLegislatives2022,
back_populates="resultats_bureaux_vote")
voix: Mapped[List["VoixBureauVoteLegislatives2022"]] = relationship(
"VoixBureauVoteLegislatives2022", back_populates="resultats_bureau_vote")
class VoixFranceLegislatives2022(Base):
__tablename__ = "legislatives_2022_voix_france"
id: Mapped[int] = mapped_column(primary_key=True)
nuance_id: Mapped[str] = mapped_column(ForeignKey("legislatives_2022_nuance.code"))
resultats_france_id: Mapped[int] = mapped_column(ForeignKey("legislatives_2022_resultats_france.id"))
voix_t1: Mapped[int] = mapped_column(Integer(), default=0)
voix_t2: Mapped[int] = mapped_column(Integer(), default=0)
nuance: Mapped[NuanceLegislatives2022] = relationship(
NuanceLegislatives2022, back_populates="resultats_nationaux")
resultats_france: Mapped[ResultatsFranceLegislatives2022] = relationship(
ResultatsFranceLegislatives2022, back_populates="voix")
class VoixRegionLegislatives2022(Base):
__tablename__ = "legislatives_2022_voix_region"
id: Mapped[int] = mapped_column(primary_key=True)
nuance_id: Mapped[str] = mapped_column(ForeignKey("legislatives_2022_nuance.code"))
resultats_region_id: Mapped[int] = mapped_column(ForeignKey("legislatives_2022_resultats_region.id"))
voix_t1: Mapped[int] = mapped_column(Integer(), default=0)
voix_t2: Mapped[int] = mapped_column(Integer(), default=0)
nuance: Mapped[NuanceLegislatives2022] = relationship(
NuanceLegislatives2022, back_populates="resultats_par_region")
resultats_region: Mapped[ResultatsRegionLegislatives2022] = relationship(
ResultatsRegionLegislatives2022, back_populates="voix")
class VoixDepartementLegislatives2022(Base):
__tablename__ = "legislatives_2022_voix_departement"
id: Mapped[int] = mapped_column(primary_key=True)
nuance_id: Mapped[str] = mapped_column(ForeignKey("legislatives_2022_nuance.code"))
resultats_departement_id: Mapped[int] = mapped_column(ForeignKey("legislatives_2022_resultats_departement.id"))
voix_t1: Mapped[int] = mapped_column(Integer(), default=0)
voix_t2: Mapped[int] = mapped_column(Integer(), default=0)
nuance: Mapped[NuanceLegislatives2022] = relationship(
NuanceLegislatives2022, back_populates="resultats_par_departement")
resultats_departement: Mapped[ResultatsDepartementLegislatives2022] = relationship(
ResultatsDepartementLegislatives2022, back_populates="voix")
class VoixCirconscriptionLegislatives2022(Base):
__tablename__ = "legislatives_2022_voix_circonscription"
id: Mapped[int] = mapped_column(primary_key=True)
candidat_id: Mapped[int] = mapped_column(ForeignKey("legislatives_2022_candidat.id"))
resultats_circonscription_id: Mapped[int] = mapped_column(
ForeignKey("legislatives_2022_resultats_circonscription.id"))
voix_t1: Mapped[int] = mapped_column(Integer(), default=0)
voix_t2: Mapped[int] = mapped_column(Integer(), default=0)
candidat: Mapped[CandidatLegislatives2022] = relationship(
CandidatLegislatives2022, back_populates="resultats_par_circonscription")
resultats_circonscription: Mapped[ResultatsCirconscriptionLegislatives2022] = relationship(
ResultatsCirconscriptionLegislatives2022, back_populates="voix")
class VoixCommuneLegislatives2022(Base):
__tablename__ = "legislatives_2022_voix_commune"
id: Mapped[int] = mapped_column(primary_key=True)
nuance_id: Mapped[str] = mapped_column(ForeignKey("legislatives_2022_nuance.code"))
resultats_commune_id: Mapped[int] = mapped_column(ForeignKey("legislatives_2022_resultats_commune.id"))
voix_t1: Mapped[int] = mapped_column(Integer(), default=0)
voix_t2: Mapped[int] = mapped_column(Integer(), default=0)
nuance: Mapped[NuanceLegislatives2022] = relationship(
NuanceLegislatives2022, back_populates="resultats_par_commune")
resultats_commune: Mapped[ResultatsCommuneLegislatives2022] = relationship(
ResultatsCommuneLegislatives2022, back_populates="voix")
class VoixBureauVoteLegislatives2022(Base):
__tablename__ = "legislatives_2022_voix_bureau_vote"
id: Mapped[int] = mapped_column(primary_key=True)
candidat_id: Mapped[int] = mapped_column(ForeignKey("legislatives_2022_candidat.id"))
resultats_bureau_vote_id: Mapped[int] = mapped_column(ForeignKey("legislatives_2022_resultats_bureau_vote.id"))
voix_t1: Mapped[int] = mapped_column(Integer(), default=0)
voix_t2: Mapped[int] = mapped_column(Integer(), default=0)
candidat: Mapped[CandidatLegislatives2022] = relationship(CandidatLegislatives2022,
back_populates="resultats_par_bureau_vote")
resultats_bureau_vote: Mapped[ResultatsBureauVoteLegislatives2022] = relationship(
ResultatsBureauVoteLegislatives2022, back_populates="voix")

View File

@ -11,7 +11,7 @@ from nupes.models.europeennes2024 import *
def importer_resultats_bv(engine: Engine, verbose: bool = False) -> None: 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", file = get_file("https://www.data.gouv.fr/fr/datasets/r/cc1883d9-1265-4365-b754-fb6aef22d82e",
"resultats-europeennes-2024-par-bureau-de-vote.csv") "resultats-europeennes-par-bureau-de-vote.csv")
with file.open('r') as f: with file.open('r') as f:
reader = DictReader(f, delimiter=';') reader = DictReader(f, delimiter=';')
@ -20,8 +20,7 @@ def importer_resultats_bv(engine: Engine, verbose: bool = False) -> None:
com_code = str(row['Code commune']).zfill(5) com_code = str(row['Code commune']).zfill(5)
bv_code = row['Code BV'] bv_code = row['Code BV']
if com_code.startswith("987") or com_code.startswith("988"): if com_code.startswith("987"): # Les communes de Polynésie française ne sont pas importées
# Les communes de Polynésie française ne sont pas importées
continue continue
if com_code == "60694": if com_code == "60694":
@ -112,7 +111,7 @@ def importer_resultats_bv(engine: Engine, verbose: bool = False) -> None:
def importer_resultats_commune(engine: Engine, verbose: bool = False) -> None: 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", file = get_file("https://www.data.gouv.fr/fr/datasets/r/6a782ef9-8ad6-4e66-832d-338b1041a42d",
"resultats-europeennes-2024-par-commune.csv") "resultats-europeennes-par-commune.csv")
with file.open('r') as f: with file.open('r') as f:
reader = DictReader(f, delimiter=';') reader = DictReader(f, delimiter=';')
@ -124,8 +123,7 @@ def importer_resultats_commune(engine: Engine, verbose: bool = False) -> None:
if dpt_code == "ZX": # Saint-Martin/Saint-Barthélémy if dpt_code == "ZX": # Saint-Martin/Saint-Barthélémy
dpt_code = "977" dpt_code = "977"
if dpt_code == "977" or dpt_code == "987" or dpt_code == "988": if dpt_code == "987": # Les communes de Polynésie française ne sont pas importées
# Les communes de Polynésie française ne sont pas importées
continue continue
if com_code == "60694": if com_code == "60694":
@ -197,7 +195,7 @@ def importer_resultats_commune(engine: Engine, verbose: bool = False) -> None:
def importer_resultats_circo(engine: Engine, verbose: bool = False) -> None: 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", file = get_file("https://www.data.gouv.fr/fr/datasets/r/ee37cbef-3d2a-4efe-a395-530b85a63028",
"resultats-europeennes-2024-par-circonscription.csv") "resultats-europeennes-par-circonscription.csv")
with file.open('r') as f: with file.open('r') as f:
reader = DictReader(f, delimiter=';') reader = DictReader(f, delimiter=';')
@ -270,7 +268,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-2024-par-departement.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=';')
@ -332,7 +330,7 @@ def importer_resultats_departement(engine: Engine, verbose: bool = False) -> Non
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/7c3a854b-7344-4c68-b1f9-9d651b4ca823", file = get_file("https://www.data.gouv.fr/fr/datasets/r/7c3a854b-7344-4c68-b1f9-9d651b4ca823",
"resultats-europeennes-2024-par-region.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=';')

View File

@ -1 +0,0 @@
from . import import_candidats, import_resultats

View File

@ -1,150 +0,0 @@
import csv
import datetime
from sqlalchemy import Engine, select
from sqlalchemy.orm import Session
from tqdm import tqdm
from nupes.cache import get_file
from nupes.models.legislatives2022 import BlocLegislatives2022, NuanceLegislatives2022, CandidatLegislatives2022
def creer_blocs(engine: Engine, verbose: bool = False) -> None:
blocs = [
{"id": 1, "nom": "Gauche", "couleur": "#BB1840"},
{"id": 2, "nom": "Droite libérale", "couleur": "#FFEB00"},
{"id": 3, "nom": "Droite conservatrice", "couleur": "#0066CC"},
{"id": 4, "nom": "Extrême droite", "couleur": "#404040"},
{"id": 5, "nom": "Autres", "couleur": "#DCBFA3"}
]
with Session(engine) as session:
for bloc_dict in blocs:
if bloc := session.execute(select(BlocLegislatives2022).filter_by(id=bloc_dict["id"])).scalar_one_or_none():
bloc.nom = bloc_dict["nom"]
bloc.couleur = bloc_dict["couleur"]
else:
session.add(BlocLegislatives2022(**bloc_dict))
session.commit()
def creer_nuances(engine: Engine, verbose: bool = False) -> None:
nuances = [
{"code": "DXG", "nom": "Divers extrême gauche", "couleur": "#BB0000", "bloc_id": 1},
{"code": "RDG", "nom": "Radical de gauche", "couleur": "#FFD1DC", "bloc_id": 1},
{"code": "NUP", "nom": "Nouvelle union populaire écologique et sociale", "couleur": "#E4032E", "bloc_id": 1},
{"code": "DVG", "nom": "Divers gauche", "couleur": "#FFC0C0", "bloc_id": 1},
{"code": "ECO", "nom": "Écologistes", "couleur": "#77FF77", "bloc_id": 5},
{"code": "DIV", "nom": "Divers", "couleur": "#DCDCDC", "bloc_id": 5},
{"code": "REG", "nom": "Régionalistes", "couleur": "#DCBFA3", "bloc_id": 5},
{"code": "ENS", "nom": "Ensemble ! (Majorité présidentielle)", "couleur": "#FFEB00", "bloc_id": 2},
{"code": "DVC", "nom": "Divers centre", "couleur": "#FAC577", "bloc_id": 2},
{"code": "UDI", "nom": "Union des Démocrates et Indépendants", "couleur": "#00FFFF", "bloc_id": 3},
{"code": "LR", "nom": "Les Républicains", "couleur": "#0066CC", "bloc_id": 3},
{"code": "DVD", "nom": "Divers droite", "couleur": "#26C4EC", "bloc_id": 3},
{"code": "DSV", "nom": "Droite souverainiste", "couleur": "#8040C0", "bloc_id": 4},
{"code": "REC", "nom": "Reconquête !", "couleur": "#404040", "bloc_id": 4},
{"code": "RN", "nom": "Rassemblement national", "couleur": "#0D378A", "bloc_id": 4},
{"code": "DXD", "nom": "Divers extrême droite", "couleur": "#404040", "bloc_id": 4},
]
with Session(engine) as session:
for nuance_dict in nuances:
if nuance := session.execute(select(NuanceLegislatives2022)
.filter_by(code=nuance_dict["code"])).scalar_one_or_none():
nuance.nom = nuance_dict["nom"]
nuance.couleur = nuance_dict["couleur"]
nuance.bloc_id = nuance_dict["bloc_id"]
else:
session.add(NuanceLegislatives2022(**nuance_dict))
session.commit()
def importer_candidats(engine: Engine, verbose: bool = False) -> None:
DATASET_URL = "https://www.data.gouv.fr/fr/datasets/r/c79af2f3-1733-4df3-be37-caebacbc1321"
file = get_file(DATASET_URL, "candidats_legislatives_2022.tsv")
with file.open('r', encoding="ISO-8859-1") as f:
with Session(engine) as session:
reader = csv.DictReader(f, delimiter='\t')
for line in tqdm(reader, desc="Candidat⋅es", disable=not verbose):
line: dict[str, str]
numero_panneau = int(line.get("N° panneau"))
code_dpt = line.get("Code du département")
match code_dpt:
case "ZA":
code_dpt = "971"
case "ZB":
code_dpt = "972"
case "ZC":
code_dpt = "973"
case "ZD":
code_dpt = "974"
case "ZS":
code_dpt = "975"
case "ZM":
code_dpt = "976"
case "ZX":
code_dpt = "977"
case "ZW":
code_dpt = "986"
case "ZP":
code_dpt = "987"
case "ZN":
code_dpt = "988"
code_circo = line.get("Code circonscription")
circo_id = f"{code_dpt}-{code_circo}"
nuance_id = line.get("Nuance candidat")
nuance = session.execute(select(NuanceLegislatives2022).filter_by(code=nuance_id)).scalar_one()
if candidat := session.execute(select(CandidatLegislatives2022)
.filter_by(circonscription_id=circo_id, numero=numero_panneau)) \
.scalar_one_or_none():
candidat.nuance_id = line['Nuance candidat']
candidat.bloc_id = nuance.bloc_id
candidat.nom = line['Nom candidat']
candidat.prenom = line['Prénom candidat']
candidat.sexe = CandidatLegislatives2022.Genre(line['Sexe candidat'])
candidat.date_naissance = datetime.datetime.strptime(line['Date naissance candidat'],
"%d/%m/%Y").date()
candidat.profession = line['Profession candidat']
candidat.sortant = line['Le candidat est sortant'] == "Oui"
candidat.nom_suppleance = line['Nom remplaçant']
candidat.prenom_suppleance = line['Prénom remplaçant']
candidat.sexe_suppleance = CandidatLegislatives2022.Genre(line['Sexe remplaçant'])
candidat.date_naissance_suppleance = datetime.datetime.strptime(line['Date naiss. remplaçant'],
"%d/%m/%Y").date()
candidat.sortant = line['Le remplaçant est sortant'] == "Oui"
else:
candidat = CandidatLegislatives2022(
circonscription_id=circo_id,
numero=numero_panneau,
nuance_id=nuance_id,
bloc_id=nuance.bloc_id,
nom=line['Nom candidat'],
prenom=line['Prénom candidat'],
sexe=CandidatLegislatives2022.Genre(line['Sexe candidat']).name,
date_naissance=datetime.datetime.strptime(line['Date naissance candidat'], "%d/%m/%Y").date(),
profession=line['Profession candidat'],
sortant=line['Le candidat est sortant'] == "Oui",
nom_suppleance=line['Nom remplaçant'],
prenom_suppleance=line['Prénom remplaçant'],
sexe_suppleance=CandidatLegislatives2022.Genre(line['Sexe remplaçant']).name,
date_naissance_suppleance=datetime.datetime.strptime(line['Date naiss. remplaçant'],
"%d/%m/%Y").date(),
sortant_suppleance=line['Le remplaçant est sortant'] == "Oui",
)
session.add(candidat)
session.commit()
def run(engine: Engine, verbose: bool = False) -> None:
creer_blocs(engine, verbose)
creer_nuances(engine, verbose)
importer_candidats(engine, verbose)

View File

@ -1,665 +0,0 @@
import csv
from sqlalchemy import Engine, select
from sqlalchemy.orm import Session
from tqdm import tqdm
from nupes.cache import get_file
from nupes.models import BureauVote
from nupes.models.legislatives2022 import *
def importer_resultats_bv(engine: Engine, verbose: bool = False) -> None:
tours = [(1, "https://www.data.gouv.fr/fr/datasets/r/a1f73b85-8194-44f4-a2b7-c343edb47d32"),
(2, "https://www.data.gouv.fr/fr/datasets/r/96ffddda-59b4-41b8-a6a3-dfe1adb7fa36")]
for tour, file_url in tours:
file = get_file("https://www.data.gouv.fr/fr/datasets/r/a1f73b85-8194-44f4-a2b7-c343edb47d32",
"resultats-legislatives-2022-t1-par-bureau-de-vote.csv")
with file.open('r', encoding="ISO-8859-1") as f:
reader = csv.reader(f, delimiter=';')
next(reader)
with Session(engine) as session:
for row in tqdm(reader, desc=f"Bureau de vote tour {tour}", disable=not verbose):
dpt_code = row[0]
match dpt_code:
case "ZA":
dpt_code = "971"
case "ZB":
dpt_code = "972"
case "ZC":
dpt_code = "973"
case "ZD":
dpt_code = "974"
case "ZS":
dpt_code = "975"
case "ZM":
dpt_code = "976"
case "ZX":
dpt_code = "977"
case "ZW":
dpt_code = "986"
case "ZP":
dpt_code = "987"
case "ZN":
dpt_code = "988"
if dpt_code == "987" or dpt_code == "988":
# Les communes de Polynésie française ne sont pas importées
continue
com_code = str(dpt_code + row[4][(1 if len(dpt_code) == 3 else 0):]).zfill(5)
if com_code == "98601":
# 3 royaumes à Wallis-et-Futuna, mais un seul résultat de commune
com_code = "98611"
if com_code.startswith("987"): # Les communes de Polynésie française ne sont pas importées
continue
bv_code = row[6]
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é")
continue
circo_code = row[2]
circo_id = f"{dpt_code.zfill(2)}-{circo_code.zfill(2)}"
resultats_bv = session.execute(select(ResultatsBureauVoteLegislatives2022)
.filter_by(bv_id=bv_id)).scalar_one_or_none()
if not resultats_bv:
resultats_commune = session.execute(select(ResultatsCommuneLegislatives2022)
.filter_by(commune_id=com_code)).scalar_one_or_none()
if not resultats_commune:
resultats_dpt = session.execute(select(ResultatsDepartementLegislatives2022)
.filter_by(dpt_id=dpt_code)) \
.scalar_one_or_none()
if not resultats_dpt:
dpt = session.execute(select(Departement).filter_by(code_insee=dpt_code)) \
.scalar_one()
resultats_reg = session.execute(
select(ResultatsRegionLegislatives2022).filter_by(region_id=dpt.region_code)) \
.scalar_one_or_none()
if not resultats_reg:
resultats_france = session.execute(select(ResultatsFranceLegislatives2022)) \
.scalar_one_or_none()
if not resultats_france:
session.add(ResultatsFranceLegislatives2022())
resultats_france = session.execute(select(ResultatsFranceLegislatives2022)) \
.scalar_one()
resultats_reg = ResultatsRegionLegislatives2022(
region_id=str(dpt.region_code), resultats_france_id=resultats_france.id)
session.add(resultats_reg)
resultats_reg = session.execute(select(ResultatsRegionLegislatives2022)
.filter_by(region_id=dpt.region_code)).scalar_one()
resultats_dpt = ResultatsDepartementLegislatives2022(
dpt_id=dpt_code, resultats_region_id=resultats_reg.id)
session.add(resultats_dpt)
resultats_dpt = session.execute(select(ResultatsDepartementLegislatives2022)
.filter_by(dpt_id=dpt_code)).scalar_one()
resultats_commune = ResultatsCommuneLegislatives2022(commune_id=com_code,
resultats_dpt_id=resultats_dpt.id)
session.add(resultats_commune)
resultats_commune = session.execute(select(ResultatsCommuneLegislatives2022)
.filter_by(commune_id=com_code)).scalar_one()
resultats_circo = session.execute(select(ResultatsCirconscriptionLegislatives2022)
.filter_by(circo_id=circo_id)).scalar_one_or_none()
resultats_bv = ResultatsBureauVoteLegislatives2022(bv_id=bv_id,
resultats_commune_id=resultats_commune.id,
resultats_circo_id=resultats_circo.id)
session.add(resultats_bv)
resultats_bv = session.execute(select(ResultatsBureauVoteLegislatives2022)
.filter_by(bv_id=bv_id)).scalar_one()
if tour == 1:
resultats_bv.inscrits_t1 = int(row[5])
resultats_bv.votants_t1 = int(row[8])
resultats_bv.abstentions_t1 = int(row[6])
resultats_bv.exprimes_t1 = int(row[16])
resultats_bv.blancs_t1 = int(row[10])
resultats_bv.nuls_t1 = int(row[13])
elif tour == 2:
resultats_bv.inscrits_t2 = int(row[5])
resultats_bv.votants_t2 = int(row[8])
resultats_bv.abstentions_t2 = int(row[6])
resultats_bv.exprimes_t2 = int(row[16])
resultats_bv.blancs_t2 = int(row[10])
resultats_bv.nuls_t2 = int(row[13])
row_voix = row[19:]
for i in range(len(row_voix) // 9):
bloc_voix = row_voix[i * 9:(i + 1) * 9]
numero = bloc_voix[0]
voix = int(bloc_voix[5])
candidat = session.execute(select(CandidatLegislatives2022)
.filter_by(circonscription_id=circo_id, numero=numero)) \
.scalar_one_or_none()
voix_candidat_bv = session.execute(select(VoixBureauVoteLegislatives2022)
.filter_by(resultats_bureau_vote_id=resultats_bv.id,
candidat_id=candidat.id)) \
.scalar_one_or_none()
if not voix_candidat_bv:
voix_candidat_bv = VoixBureauVoteLegislatives2022(
resultats_bureau_vote_id=resultats_bv.id, candidat_id=candidat.id)
session.add(voix_candidat_bv)
if tour == 1:
voix_candidat_bv.voix_t1 = voix
elif tour == 2:
voix_candidat_bv.voix_t2 = voix
session.commit()
def importer_resultats_commune(engine: Engine, verbose: bool = False) -> None:
tours = [(1, "https://www.data.gouv.fr/fr/datasets/r/a9a82bcc-304e-491f-a4f0-c06575113745"),
(2, "https://www.data.gouv.fr/fr/datasets/r/e79b2e51-0841-4266-b626-57cb271c7a35")]
with Session(engine) as session:
for resultats_commune in session.execute(select(ResultatsCommuneLegislatives2022)).scalars().all():
resultats_commune.inscrits_t1 = 0
resultats_commune.votants_t1 = 0
resultats_commune.abstentions_t1 = 0
resultats_commune.exprimes_t1 = 0
resultats_commune.blancs_t1 = 0
resultats_commune.nuls_t1 = 0
resultats_commune.inscrits_t2 = 0
resultats_commune.votants_t2 = 0
resultats_commune.abstentions_t2 = 0
resultats_commune.exprimes_t2 = 0
resultats_commune.blancs_t2 = 0
resultats_commune.nuls_t2 = 0
for voix_nuance in resultats_commune.voix:
voix_nuance.voix_t1 = 0
voix_nuance.voix_t2 = 0
session.commit()
for tour, file_url in tours:
file = get_file(file_url,
f"resultats-legislatives-2022-t{tour}-par-commune.csv")
with file.open('r', encoding="ISO-8859-1") as f:
reader = csv.reader(f, delimiter=';')
next(reader)
with Session(engine) as session:
for row in tqdm(reader, desc=f"Commune tour {tour}", disable=not verbose):
dpt_code = row[0]
match dpt_code:
case "ZA":
dpt_code = "971"
case "ZB":
dpt_code = "972"
case "ZC":
dpt_code = "973"
case "ZD":
dpt_code = "974"
case "ZS":
dpt_code = "975"
case "ZM":
dpt_code = "976"
case "ZX":
dpt_code = "977"
case "ZW":
dpt_code = "986"
case "ZP":
dpt_code = "987"
case "ZN":
dpt_code = "988"
if dpt_code == "987" or dpt_code == "988":
# Les communes de Polynésie française ne sont pas importées
continue
com_code = str(dpt_code + row[4][(1 if len(dpt_code) == 3 else 0):]).zfill(5)
if com_code == "98601":
# 3 royaumes à Wallis-et-Futuna, mais un seul résultat de commune
com_code = "98611"
resultats_commune = session.execute(select(ResultatsCommuneLegislatives2022)
.filter_by(commune_id=com_code)).scalar_one_or_none()
if not resultats_commune:
resultats_dpt = session.execute(select(ResultatsDepartementLegislatives2022)
.filter_by(dpt_id=dpt_code)).scalar_one_or_none()
if not resultats_dpt:
dpt = session.execute(select(Departement)
.filter_by(code_insee=f"{dpt_code.zfill(2)}")).scalar_one()
resultats_reg = session.execute(select(ResultatsRegionLegislatives2022)
.filter_by(region_id=dpt.region_code)).scalar_one_or_none()
if not resultats_reg:
resultats_france = session.execute(select(ResultatsFranceLegislatives2022)) \
.scalar_one_or_none()
if not resultats_france:
session.add(ResultatsFranceLegislatives2022())
resultats_france = session.execute(select(ResultatsFranceLegislatives2022)) \
.scalar_one()
resultats_reg = ResultatsRegionLegislatives2022(region_id=str(dpt.region_code),
resultats_france_id=resultats_france.id)
session.add(resultats_reg)
resultats_reg = session.execute(select(ResultatsRegionLegislatives2022)
.filter_by(region_id=dpt.region_code)).scalar_one()
resultats_dpt = ResultatsDepartementLegislatives2022(dpt_id=dpt_code,
resultats_region_id=resultats_reg.id)
session.add(resultats_dpt)
resultats_dpt = session.execute(select(ResultatsDepartementLegislatives2022)
.filter_by(dpt_id=dpt_code)).scalar_one()
resultats_commune = ResultatsCommuneLegislatives2022(commune_id=com_code,
resultats_dpt_id=resultats_dpt.id)
session.add(resultats_commune)
resultats_commune = session.execute(select(ResultatsCommuneLegislatives2022)
.filter_by(commune_id=com_code)).scalar_one()
if tour == 1:
resultats_commune.inscrits_t1 = int(row[7])
resultats_commune.votants_t1 = int(row[10])
resultats_commune.abstentions_t1 = int(row[8])
resultats_commune.exprimes_t1 = int(row[18])
resultats_commune.blancs_t1 = int(row[12])
resultats_commune.nuls_t1 = int(row[15])
elif tour == 2:
resultats_commune.inscrits_t2 = int(row[7])
resultats_commune.votants_t2 = int(row[10])
resultats_commune.abstentions_t2 = int(row[8])
resultats_commune.exprimes_t2 = int(row[18])
resultats_commune.blancs_t2 = int(row[12])
resultats_commune.nuls_t2 = int(row[15])
row_voix = row[21:]
for i in range(len(row_voix) // 8):
bloc_voix = row_voix[i * 8:(i + 1) * 8]
nuance_code = bloc_voix[4]
voix = int(bloc_voix[5])
voix_nuance_commune = session.execute(select(VoixCommuneLegislatives2022)
.filter_by(resultats_commune_id=resultats_commune.id,
nuance_id=nuance_code)) \
.scalar_one_or_none()
if not voix_nuance_commune:
voix_nuance_commune = VoixCommuneLegislatives2022(resultats_commune_id=resultats_commune.id,
nuance_id=nuance_code)
session.add(voix_nuance_commune)
if tour == 1:
voix_nuance_commune.voix_t1 += voix
elif tour == 2:
voix_nuance_commune.voix_t2 += voix
session.commit()
def importer_resultats_circo(engine: Engine, verbose: bool = False) -> None:
tours = [(1, "https://www.data.gouv.fr/fr/datasets/r/114b13e8-7ff9-437f-9ec8-7a29258a80e3"),
(2, "https://www.data.gouv.fr/fr/datasets/r/436b9eaa-64e1-46bd-ad1d-9638ebbd6caf")]
for tour, file_url in tours:
file = get_file(file_url, f"resultats-legislatives-2022-t{tour}-par-circonscription.csv")
with file.open('r', encoding="ISO-8859-1") as f:
reader = csv.reader(f, delimiter=';')
next(reader)
with Session(engine) as session:
for row in tqdm(reader, desc=f"Circonscription tour {tour}", disable=not verbose):
dpt_code = row[0]
circo_code = row[2]
match dpt_code:
case "ZA":
dpt_code = "971"
case "ZB":
dpt_code = "972"
case "ZC":
dpt_code = "973"
case "ZD":
dpt_code = "974"
case "ZS":
dpt_code = "975"
case "ZM":
dpt_code = "976"
case "ZX":
dpt_code = "977"
case "ZW":
dpt_code = "986"
case "ZP":
dpt_code = "987"
case "ZN":
dpt_code = "988"
circo_id = f"{dpt_code.zfill(2)}-{circo_code.zfill(2)}"
resultats_circo = session.execute(select(ResultatsCirconscriptionLegislatives2022)
.filter_by(circo_id=circo_id)).scalar_one_or_none()
if not resultats_circo:
resultats_dpt = session.execute(select(ResultatsDepartementLegislatives2022)
.filter_by(dpt_id=dpt_code)).scalar_one_or_none()
if not resultats_dpt:
dpt = session.execute(select(Departement).filter_by(code_insee=f"{dpt_code.zfill(2)}")) \
.scalar_one()
resultats_reg = session.execute(select(ResultatsRegionLegislatives2022)
.filter_by(region_id=dpt.region_code)) \
.scalar_one_or_none()
if not resultats_reg:
resultats_france = session.execute(select(ResultatsFranceLegislatives2022)) \
.scalar_one_or_none()
if not resultats_france:
session.add(ResultatsFranceLegislatives2022())
resultats_france = session.execute(select(ResultatsFranceLegislatives2022)).scalar_one()
resultats_reg = ResultatsRegionLegislatives2022(region_id=str(dpt.region_code),
resultats_france_id=resultats_france.id)
session.add(resultats_reg)
resultats_reg = session.execute(select(ResultatsRegionLegislatives2022)
.filter_by(region_id=dpt.region_code)).scalar_one()
resultats_dpt = ResultatsDepartementLegislatives2022(dpt_id=dpt_code,
resultats_region_id=resultats_reg.id)
session.add(resultats_dpt)
resultats_dpt = session.execute(select(ResultatsDepartementLegislatives2022)
.filter_by(dpt_id=dpt_code)).scalar_one()
resultats_circo = ResultatsCirconscriptionLegislatives2022(
circo_id=circo_id, resultats_departement_id=resultats_dpt.id)
session.add(resultats_circo)
resultats_circo = session.execute(select(ResultatsCirconscriptionLegislatives2022)
.filter_by(circo_id=circo_id)).scalar_one()
if tour == 1:
resultats_circo.inscrits_t1 = int(row[5])
resultats_circo.votants_t1 = int(row[8])
resultats_circo.abstentions_t1 = int(row[6])
resultats_circo.exprimes_t1 = int(row[16])
resultats_circo.blancs_t1 = int(row[10])
resultats_circo.nuls_t1 = int(row[13])
elif tour == 2:
resultats_circo.inscrits_t2 = int(row[5])
resultats_circo.votants_t2 = int(row[8])
resultats_circo.abstentions_t2 = int(row[6])
resultats_circo.exprimes_t2 = int(row[16])
resultats_circo.blancs_t2 = int(row[10])
resultats_circo.nuls_t2 = int(row[13])
row_voix = row[19:]
for i in range(len(row_voix) // 9):
bloc_voix = row_voix[i * 9:(i + 1) * 9]
numero = bloc_voix[0]
voix = int(bloc_voix[5])
candidat = session.execute(select(CandidatLegislatives2022)
.filter_by(circonscription_id=circo_id, numero=numero)) \
.scalar_one_or_none()
voix_candidat_circo = session.execute(select(VoixCirconscriptionLegislatives2022)
.filter_by(resultats_circonscription_id=resultats_circo.id,
candidat_id=candidat.id)) \
.scalar_one_or_none()
if not voix_candidat_circo:
voix_candidat_circo = VoixCirconscriptionLegislatives2022(
resultats_circonscription_id=resultats_circo.id, candidat_id=candidat.id)
session.add(voix_candidat_circo)
if tour == 1:
voix_candidat_circo.voix_t1 = voix
elif tour == 2:
voix_candidat_circo.voix_t2 = voix
session.commit()
def importer_resultats_departement(engine: Engine, verbose: bool = False) -> None:
tours = [(1, "https://www.data.gouv.fr/fr/datasets/r/4acba699-e512-4dc5-87c7-ba74db773633"),
(2, "https://www.data.gouv.fr/fr/datasets/r/bc76bd72-cf65-45b2-b651-7716dc3c5520")]
for tour, file_url in tours:
file = get_file(file_url, f"resultats-legislatives-2022-t{tour}-par-departement.csv")
with file.open('r', encoding="ISO-8859-1") as f:
reader = csv.reader(f, delimiter=';')
next(reader)
with Session(engine) as session:
for row in tqdm(reader, desc=f"Département tour {tour}", disable=not verbose):
dpt_code = row[0]
match dpt_code:
case "ZA":
dpt_code = "971"
case "ZB":
dpt_code = "972"
case "ZC":
dpt_code = "973"
case "ZD":
dpt_code = "974"
case "ZS":
dpt_code = "975"
case "ZM":
dpt_code = "976"
case "ZX":
dpt_code = "977"
case "ZW":
dpt_code = "986"
case "ZP":
dpt_code = "987"
case "ZN":
dpt_code = "988"
resultats_dpt = session.execute(select(ResultatsDepartementLegislatives2022)
.filter_by(dpt_id=dpt_code)).scalar_one_or_none()
if not resultats_dpt:
dpt = session.execute(select(Departement)
.filter_by(code_insee=f"{dpt_code.zfill(2)}")) \
.scalar_one()
resultats_reg = session.execute(select(ResultatsRegionLegislatives2022)
.filter_by(region_id=dpt.region_code)) \
.scalar_one_or_none()
if not resultats_reg:
resultats_france = session.execute(select(ResultatsFranceLegislatives2022)) \
.scalar_one_or_none()
if not resultats_france:
session.add(ResultatsFranceLegislatives2022())
resultats_france = session.execute(select(ResultatsFranceLegislatives2022)).scalar_one()
resultats_reg = ResultatsRegionLegislatives2022(region_id=str(dpt.region_code),
resultats_france_id=resultats_france.id)
session.add(resultats_reg)
resultats_reg = session.execute(select(ResultatsRegionLegislatives2022)
.filter_by(region_id=dpt.region_code)).scalar_one()
resultats_dpt = ResultatsDepartementLegislatives2022(dpt_id=dpt_code,
resultats_region_id=resultats_reg.id)
session.add(resultats_dpt)
resultats_dpt = session.execute(select(ResultatsDepartementLegislatives2022)
.filter_by(dpt_id=dpt_code)).scalar_one()
if tour == 1:
resultats_dpt.inscrits_t1 = int(row[3])
resultats_dpt.votants_t1 = int(row[6])
resultats_dpt.abstentions_t1 = int(row[4])
resultats_dpt.exprimes_t1 = int(row[14])
resultats_dpt.blancs_t1 = int(row[8])
resultats_dpt.nuls_t1 = int(row[11])
elif tour == 2:
resultats_dpt.inscrits_t2 = int(row[3])
resultats_dpt.votants_t2 = int(row[6])
resultats_dpt.abstentions_t2 = int(row[4])
resultats_dpt.exprimes_t2 = int(row[14])
resultats_dpt.blancs_t2 = int(row[8])
resultats_dpt.nuls_t2 = int(row[11])
row_voix = row[17:]
for i in range(len(row_voix) // 5):
bloc_voix = row_voix[i * 5:(i + 1) * 5]
nuance_code = bloc_voix[0]
voix = int(bloc_voix[1])
voix_nuance_dpt = session.execute(select(VoixDepartementLegislatives2022)
.filter_by(resultats_departement_id=resultats_dpt.id,
nuance_id=nuance_code)) \
.scalar_one_or_none()
if not voix_nuance_dpt:
voix_nuance_dpt = VoixDepartementLegislatives2022(resultats_departement_id=resultats_dpt.id,
nuance_id=nuance_code)
session.add(voix_nuance_dpt)
if tour == 1:
voix_nuance_dpt.voix_t1 = voix
elif tour == 2:
voix_nuance_dpt.voix_t2 = voix
session.commit()
def importer_resultats_region(engine: Engine, verbose: bool = False) -> None:
tours = [(1, "https://www.data.gouv.fr/fr/datasets/r/f79576c9-9ddc-43f9-8d99-f0856550054e"),
(2, "https://www.data.gouv.fr/fr/datasets/r/fdba421a-d4e9-4f38-9753-982a7c065911")]
for tour, file_url in tours:
file = get_file(file_url, f"resultats-legislatives-2022-t{tour}-par-region.csv")
with file.open('r', encoding="ISO-8859-1") as f:
reader = csv.reader(f, delimiter=';')
next(reader)
with Session(engine) as session:
for row in tqdm(reader, desc=f"Région tour {tour}", disable=not verbose):
reg_code = row[0]
resultats_reg = session.execute(select(ResultatsRegionLegislatives2022)
.filter_by(region_id=reg_code)).scalar_one_or_none()
if not resultats_reg:
resultats_france = session.execute(select(ResultatsFranceLegislatives2022)).scalar_one_or_none()
if not resultats_france:
session.add(ResultatsFranceLegislatives2022())
resultats_france = session.execute(select(ResultatsFranceLegislatives2022)).scalar_one()
resultats_reg = ResultatsRegionLegislatives2022(region_id=reg_code,
resultats_france_id=resultats_france.id)
session.add(resultats_reg)
session.commit()
resultats_reg = session.execute(select(ResultatsRegionLegislatives2022)
.filter_by(region_id=reg_code)).scalar_one()
if tour == 1:
resultats_reg.inscrits_t1 = int(row[3])
resultats_reg.votants_t1 = int(row[6])
resultats_reg.abstentions_t1 = int(row[4])
resultats_reg.exprimes_t1 = int(row[14])
resultats_reg.blancs_t1 = int(row[8])
resultats_reg.nuls_t1 = int(row[11])
elif tour == 2:
resultats_reg.inscrits_t2 = int(row[3])
resultats_reg.votants_t2 = int(row[6])
resultats_reg.abstentions_t2 = int(row[4])
resultats_reg.exprimes_t2 = int(row[14])
resultats_reg.blancs_t2 = int(row[8])
resultats_reg.nuls_t2 = int(row[11])
row_voix = row[17:]
for i in range(len(row_voix) // 5):
bloc_voix = row_voix[i * 5:(i + 1) * 5]
nuance_code = bloc_voix[0]
voix = int(bloc_voix[1])
voix_nuance_reg = session.execute(select(VoixRegionLegislatives2022)
.filter_by(resultats_region_id=resultats_reg.id,
nuance_id=nuance_code)) \
.scalar_one_or_none()
if not voix_nuance_reg:
voix_nuance_reg = VoixRegionLegislatives2022(resultats_region_id=resultats_reg.id,
nuance_id=nuance_code)
session.add(voix_nuance_reg)
if tour == 1:
voix_nuance_reg.voix_t1 = voix
elif tour == 2:
voix_nuance_reg.voix_t2 = voix
session.commit()
def importer_resultats_france(engine: Engine, verbose: bool = False) -> None:
tours = [(1, "https://www.data.gouv.fr/fr/datasets/r/cbe72f2c-2c56-4251-89cb-213de8bfbea0"),
(2, "https://www.data.gouv.fr/fr/datasets/r/6639bb39-13ec-49dd-ab4d-1f8227b86c01")]
for tour, file_url in tours:
file = get_file(file_url, f"resultats-legislatives-2022-t{tour}-france-entiere.csv")
with file.open('r', encoding="ISO-8859-1") as f:
reader = csv.reader(f, delimiter=';')
next(reader)
with Session(engine) as session:
for row in tqdm(reader, desc=f"France Entière tour {tour}", disable=not verbose):
resultats_france = session.execute(select(ResultatsFranceLegislatives2022)).scalar_one_or_none()
if not resultats_france:
session.add(ResultatsFranceLegislatives2022())
session.commit()
resultats_france = session.execute(select(ResultatsFranceLegislatives2022)).scalar_one()
if tour == 1:
resultats_france.inscrits_t1 = int(row[3])
resultats_france.votants_t1 = int(row[6])
resultats_france.abstentions_t1 = int(row[4])
resultats_france.exprimes_t1 = int(row[14])
resultats_france.blancs_t1 = int(row[8])
resultats_france.nuls_t1 = int(row[11])
elif tour == 2:
resultats_france.inscrits_t2 = int(row[3])
resultats_france.votants_t2 = int(row[6])
resultats_france.abstentions_t2 = int(row[4])
resultats_france.exprimes_t2 = int(row[14])
resultats_france.blancs_t2 = int(row[8])
resultats_france.nuls_t2 = int(row[11])
row_voix = row[17:]
for i in range(len(row_voix) // 5):
bloc_voix = row_voix[i * 5:(i + 1) * 5]
nuance_code = bloc_voix[0]
voix = int(bloc_voix[1])
voix_nuance_france = session.execute(select(VoixFranceLegislatives2022)
.filter_by(resultats_france_id=resultats_france.id,
nuance_id=nuance_code)) \
.scalar_one_or_none()
if not voix_nuance_france:
voix_nuance_france = VoixFranceLegislatives2022(resultats_france_id=resultats_france.id,
nuance_id=nuance_code)
session.add(voix_nuance_france)
if tour == 1:
voix_nuance_france.voix_t1 = voix
elif tour == 2:
voix_nuance_france.voix_t2 = voix
session.commit()
def run(engine: Engine, verbose: bool = False) -> None:
importer_resultats_france(engine, verbose)
importer_resultats_region(engine, verbose)
importer_resultats_departement(engine, verbose)
importer_resultats_circo(engine, verbose)
importer_resultats_commune(engine, verbose)
importer_resultats_bv(engine, verbose)

View File

@ -1,73 +0,0 @@
import csv
import datetime
from sqlalchemy import Engine, select
from sqlalchemy.orm import Session
from nupes.cache import get_file
from nupes.models.legislatives2024 import BlocLegislatives2024, NuanceLegislatives2024, CandidatLegislatives2024
def creer_blocs(engine: Engine, verbose: bool = False) -> None:
blocs = [
{"id": 1, "nom": "Gauche", "couleur": "#BB1840"},
{"id": 2, "nom": "Droite libérale", "couleur": "#FFEB00"},
{"id": 3, "nom": "Droite conservatrice", "couleur": "#0066CC"},
{"id": 4, "nom": "Extrême droite", "couleur": "#404040"},
{"id": 5, "nom": "Autres", "couleur": "#DCBFA3"}
]
with Session(engine) as session:
for bloc_dict in blocs:
if bloc := session.execute(select(BlocLegislatives2024).filter_by(id=bloc_dict["id"])).scalar_one_or_none():
bloc.nom = bloc_dict["nom"]
bloc.couleur = bloc_dict["couleur"]
else:
session.add(BlocLegislatives2024(**bloc_dict))
session.commit()
def creer_nuances(engine: Engine, verbose: bool = False) -> None:
nuances = [
{"code": "EXG", "nom": "Extrême gauche", "couleur": "#BB0000", "bloc": 1},
{"code": "COM", "nom": "Parti communiste français", "couleur": "#DD0000", "bloc": 1},
{"code": "FI", "nom": "La France insoumise", "couleur": "#CC2443", "bloc": 1},
{"code": "SOC", "nom": "Parti socialiste", "couleur": "#FF8080", "bloc": 1},
{"code": "UG", "nom": "Union de la gauche", "couleur": "#E4032E", "bloc": 1},
{"code": "PRG", "nom": "Parti radical de gauche", "couleur": "#FFD1DC", "bloc": 1},
{"code": "VEC", "nom": "Les Écologistes", "couleur": "#00C000", "bloc": 1},
{"code": "DVG", "nom": "Divers gauche", "couleur": "#FFC0C0", "bloc": 1},
{"code": "ECO", "nom": "Écologistes", "couleur": "#77FF77", "bloc": 5},
{"code": "REG", "nom": "Régionalistes", "couleur": "#DCBFA3", "bloc": 5},
{"code": "DIV", "nom": "Divers", "couleur": "#DCDCDC", "bloc": 5},
{"code": "REN", "nom": "Renaissance", "couleur": "#FFEB00", "bloc": 2},
{"code": "ENS", "nom": "Ensemble ! (Majorité présidentielle)", "couleur": "#45B3AB", "bloc": 2},
{"code": "MDM", "nom": "Modem", "couleur": "#FF9900", "bloc": 2},
{"code": "HOR", "nom": "Horizons", "couleur": "#0001B8", "bloc": 2},
{"code": "DVC", "nom": "Divers centre", "couleur": "#FAC577", "bloc": 2},
{"code": "UDI", "nom": "Union des Démocrates et Indépendants", "couleur": "#00FFFF", "bloc": 3},
{"code": "LR", "nom": "Les Républicains", "couleur": "#0066CC", "bloc": 3},
{"code": "DVD", "nom": "Divers droite", "couleur": "#26C4EC", "bloc": 3},
{"code": "DSV", "nom": "Droite souverainiste", "couleur": "#8040C0", "bloc": 4},
{"code": "RN", "nom": "Rassemblement national", "couleur": "#0D378A", "bloc": 4},
{"code": "REC", "nom": "Reconquête !", "couleur": "#404040", "bloc": 4},
{"code": "EXD", "nom": "Extrême droite", "couleur": "#404040", "bloc": 4},
{"code": "UXD", "nom": "Union de l'extrême droite", "couleur": "#404040", "bloc": 4},
]
with Session(engine) as session:
for nuance_dict in nuances:
if nuance := session.execute(select(NuanceLegislatives2024)
.filter_by(code=nuance_dict["code"])).scalar_one_or_none():
nuance.nom = nuance_dict["nom"]
nuance.couleur = nuance_dict["couleur"]
else:
session.add(NuanceLegislatives2024(**nuance_dict))
session.commit()
def run(engine: Engine, verbose: bool = False) -> None:
creer_blocs(engine, verbose)
creer_nuances(engine, verbose)