178 lines
7.5 KiB
Python
178 lines
7.5 KiB
Python
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, debug: 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%2FBerlin",
|
|
"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, debug: 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%2FBerlin",
|
|
"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, debug: bool = False) -> None:
|
|
etag = requests.get(
|
|
"https://public.opendatasoft.com/api/explore/v2.1/catalog/datasets"
|
|
"/georef-france-commune?select=data_processed").json()['data_processed']
|
|
file = get_file("https://public.opendatasoft.com/api/explore/v2.1/catalog/datasets"
|
|
"/georef-france-commune/exports/geojson?lang=fr&timezone=Europe%2FBerlin",
|
|
"georef-france-commune.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_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, debug: 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%2FBerlin",
|
|
"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"
|
|
|
|
if not session.execute(select(Circonscription).filter_by(id=code_circo)).scalar_one_or_none():
|
|
session.add(Circonscription(id=code_circo, departement_code=code_commune[:2],
|
|
numero=int(code_circo[3:])))
|
|
|
|
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, debug: 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, debug: bool = False) -> None:
|
|
importer_regions(engine, debug)
|
|
importer_departements(engine, debug)
|
|
importer_communes(engine, debug)
|
|
importer_bureaux_vote(engine, debug)
|
|
importer_contours_bureaux_vote(engine, debug)
|