Corrections + circonscription outre-mer et étranger

alembic/versions/d7fed76a55bd_initial.py

@@ -1,8 +1,8 @@
 """initial
 
-Revision ID: 5bf5e6526891
+Revision ID: d7fed76a55bd
 Revises: 
-Create Date: 2024-06-14 23:04:55.436991
+Create Date: 2024-06-15 00:03:00.687812
 
 """
 from typing import Sequence, Union
@@ -12,7 +12,7 @@ import sqlalchemy as sa
 
 
 # revision identifiers, used by Alembic.
-revision: str = '5bf5e6526891'
+revision: str = 'd7fed76a55bd'
 down_revision: Union[str, None] = None
 branch_labels: Union[str, Sequence[str], None] = None
 depends_on: Union[str, Sequence[str], None] = None
@@ -106,6 +106,7 @@ def upgrade() -> None:
     sa.Column('id', sa.String(length=6), nullable=False),
     sa.Column('departement_code', sa.String(length=3), nullable=False),
     sa.Column('numero', sa.Integer(), nullable=False),
+    sa.Column('geometry', sa.JSON(), nullable=False),
     sa.ForeignKeyConstraint(['departement_code'], ['departement.code_insee'], ),
     sa.PrimaryKeyConstraint('id')
     )

nupes/models/geographie.py

@@ -23,7 +23,7 @@ class Departement(Base):
 
     code_insee: Mapped[str] = mapped_column(String(3), primary_key=True)
     libelle: Mapped[str] = mapped_column(String(64), unique=True)
-    region_code: Mapped[int] = mapped_column(ForeignKey("region.code_insee"))
+    region_code: Mapped[str] = mapped_column(ForeignKey("region.code_insee"))
     geometry: Mapped[dict] = mapped_column(JSON())
 
     region: Mapped[Region] = relationship(Region, back_populates="departements")
@@ -52,6 +52,7 @@ class Circonscription(Base):
     id: Mapped[str] = mapped_column(String(6), primary_key=True)
     departement_code: Mapped[str] = mapped_column(ForeignKey("departement.code_insee"))
     numero: Mapped[int] = mapped_column(Integer())
+    geometry: Mapped[dict] = mapped_column(JSON())
 
     departement: Mapped[Departement] = relationship(Departement)
     bureaux_vote: Mapped[List["BureauVote"]] = relationship("BureauVote", back_populates="circonscription")

nupes/scripts/import_resultats_2024.py

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