import json 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 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 tqdm(features, desc="Régions", disable=not verbose): 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 tqdm(features, desc="Départements", disable=not verbose): 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_circonscriptions(engine: Engine, verbose: bool = False) -> None: file = get_file("https://www.data.gouv.fr/fr/datasets/r/67c0f382-dc8d-4d1f-8a76-1162c53b9dfe", "circonscriptions-legislatives-p10.geojson") with file.open('r') as f: features = json.load(f)['features'] with Session(engine) as session: for feature in tqdm(features, desc="Circonscriptions", disable=not verbose): circo_dict = feature['properties'] code_circo = circo_dict['codeCirconscription'] code_dpt = circo_dict['codeDepartement'] 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" numero_circo = int(code_circo[len(code_dpt):]) circo_id = f"{code_dpt}-{numero_circo:02d}" if not session.execute(select(Departement).filter_by(code_insee=code_dpt)).scalar_one_or_none(): print("Département non trouvé avec le code", code_dpt) continue if circo := session.execute(select(Circonscription).filter_by(id=circo_id)).scalar_one_or_none(): circo.departement_code = code_dpt circo.numero = numero_circo circo.geometry = feature['geometry'] else: circo = Circonscription(id=circo_id, departement_code=code_dpt, numero=numero_circo, geometry=feature['geometry']) session.add(circo) 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 tqdm(features, desc="Communes", disable=not verbose): 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: file = get_file("https://files.data.gouv.fr/reu/contours-france-entiere-latest-v2.geojson", "contours-france-entiere-latest-v2.geojson") 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): 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}" 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']) 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.geometry = feature['geometry'] 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']) session.add(bv) session.commit() def run(engine: Engine, verbose: bool = False) -> None: importer_regions(engine, verbose) importer_departements(engine, verbose) importer_circonscriptions(engine, verbose) importer_communes(engine, verbose) importer_bureaux_vote(engine, verbose)