nupes-elections/nupes/scripts/import_geographie.py

import json
from datetime import datetime

import requests
from sqlalchemy import Engine, select
from sqlalchemy.orm import Session

from nupes.cache import get_file
from nupes.models.geographie import BureauVote, Circonscription, Commune, Departement, Region


def importer_regions(engine: Engine, verbose: bool = False) -> None:
    etag = requests.get(
        "https://public.opendatasoft.com/api/explore/v2.1/catalog/datasets"
        "/georef-france-region?select=data_processed").json()['data_processed']
    file = get_file("https://public.opendatasoft.com/api/explore/v2.1/catalog/datasets"
                    "/georef-france-region/exports/geojson?lang=fr&timezone=Europe%2FParis",
                    "georef-france-region.geojson", etag)

    with file.open('r') as f:
        features = json.load(f)['features']

    with Session(engine) as session:
        for feature in features:
            region_dict = feature['properties']
            code_region = region_dict['reg_code'][0]
            nom_region = region_dict['reg_name'][0]

            if region := session.execute(select(Region).filter_by(code_insee=code_region)).scalar_one_or_none():
                region.libelle = nom_region
                region.geometry = feature['geometry']
            else:
                region = Region(code_insee=code_region, libelle=nom_region, geometry=feature['geometry'])
                session.add(region)

        session.commit()


def importer_departements(engine: Engine, verbose: bool = False) -> None:
    etag = requests.get(
        "https://public.opendatasoft.com/api/explore/v2.1/catalog/datasets"
        "/georef-france-departement?select=data_processed").json()['data_processed']
    file = get_file("https://public.opendatasoft.com/api/explore/v2.1/catalog/datasets"
                    "/georef-france-departement/exports/geojson?lang=fr&timezone=Europe%2FParis",
                    "georef-france-departement.geojson", etag)


    with file.open('r') as f:
        features = json.load(f)['features']

    with Session(engine) as session:
        for feature in features:
            dpt_dict = feature['properties']
            code_dpt = dpt_dict['dep_code'][0]
            nom_dpt = dpt_dict['dep_name'][0]

            if dpt := session.execute(select(Departement).filter_by(code_insee=code_dpt)).scalar_one_or_none():
                dpt.libelle = nom_dpt
                dpt.region_code = dpt_dict['reg_code'][0]
                dpt.geometry = feature['geometry']
            else:
                dpt = Departement(code_insee=code_dpt, libelle=nom_dpt, region_code=dpt_dict['reg_code'][0],
                                  geometry=feature['geometry'])
                session.add(dpt)

        session.commit()


def importer_communes(engine: Engine, verbose: bool = False) -> None:
    etag = requests.get(
        "https://public.opendatasoft.com/api/explore/v2.1/catalog/datasets"
        "/georef-france-commune-arrondissement-municipal?select=data_processed").json()['data_processed']
    file = get_file("https://public.opendatasoft.com/api/explore/v2.1/catalog/datasets"
                    "/georef-france-commune-arrondissement-municipal/exports/geojson?lang=fr&timezone=Europe%2FParis",
                    "georef-france-commune-arrondissement-municipal.geojson", etag)

    with file.open('r') as f:
        features = json.load(f)['features']

    with Session(engine) as session:
        for feature in features:
            commune_dict = feature['properties']
            code_commune = commune_dict['com_arm_code'][0]
            nom_commune = commune_dict['com_name'][0]

            if commune := session.execute(select(Commune).filter_by(code_insee=code_commune)).scalar_one_or_none():
                commune.libelle = nom_commune
                commune.departement_code = commune_dict['dep_code'][0]
                commune.geometry = feature['geometry']
            else:
                commune = Commune(code_insee=code_commune, libelle=nom_commune,
                                  departement_code=commune_dict['dep_code'][0], geometry=feature['geometry'])
                session.add(commune)

        session.commit()


def importer_bureaux_vote(engine: Engine, verbose: bool = False) -> None:
    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 features:
            bv_dict = feature['properties']
            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['com_name'])
                continue

            if not session.execute(select(Circonscription).filter_by(id=code_circo)).scalar_one_or_none():
                session.add(Circonscription(id=code_circo, departement_code=dpt_code, numero=numero_circo))

            if bv := session.execute(select(BureauVote).filter_by(id=bv_id)).scalar_one_or_none():
                bv.commune_code = code_commune
                bv.code_bureau = code_bv
                bv.circo_code = code_circo
                bv.libelle = bv_libelle
                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=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)