RDOO/rdoo.py

184 lines
5.1 KiB
Python
Raw Blame History

This file contains invisible Unicode characters

This file contains invisible Unicode characters that are indistinguishable to humans but may be processed differently by a computer. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env python3
# Copyright 2021 Olav63, SebF
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""
Module principal : 
- récupération de données par appel à Overpass
- géocodage inverse
- export des données en JSON pour utilisation avec umap
- sauvegarde des données en ods
"""
import time
import os
import argparse
import logging
import sys
from osm_vc63 import errors
from osm_vc63.utils import Utils
OVERPASS_URL = "http://overpass-api.de/api/interpreter"
GEO_API_URL = "https://api-adresse.data.gouv.fr"
# nombre maxi de retries quand echec API
MAX_RETRY = 4
# delai en secondes entre les tentatives
RETRY_DELAY = 120
def init_argparse() -> argparse.ArgumentParser:
"""Définition des arguments possibles."""
parser = argparse.ArgumentParser(
usage="%(prog)s [OPTIONS] ...",
description="Exporte les données de la Cyclosphère d'une zone géographique.",
)
parser.add_argument(
"-z",
"--zone",
type=int,
help="Choisir la zone géographique à inspecter.",
default=7406,
)
parser.add_argument(
"-gi",
"--geocoding-inverse",
dest="geocoding_inverse",
action="store_true",
help="Activer le géocodage inverse",
)
parser.add_argument(
"-ngi",
"--no-geocoding-inverse",
dest="geocoding_inverse",
action="store_false",
help="Désactiver le géocodage inverse",
)
parser.set_defaults(geocoding_inverse=True)
parser.add_argument(
"-l", "--log-level", type=str, help="Définir le niveau de log.", default="INFO"
)
parser.add_argument(
"-r",
"--dossier-resultats",
type=str,
help="Définir le dossier d'archive",
default="resultats/",
)
parser.add_argument(
"-a",
"--dossier-archive",
type=str,
help="Définir le dossier d'archive",
default="resultats/archives/",
)
parser.add_argument(
"-na",
"--no-archivage",
dest="archivage",
action="store_false",
help="Désactiver l'archivage'",
)
parser.set_defaults(archivage=True)
return parser
def main():
"""Routine principale"""
parser = init_argparse()
args = parser.parse_args()
logging.basicConfig(
format="%(asctime)s [%(levelname)s] %(message)s",
level=getattr(logging, args.log_level.upper()),
handlers=[logging.FileHandler("rdoo.log"), logging.StreamHandler(sys.stdout)],
)
utils = Utils(OVERPASS_URL, GEO_API_URL, args.dossier_resultats)
if args.archivage:
utils.archivage(args.dossier_archive)
logging.info("##### Nouvelle récupération ######")
# l'id de l'area se calcule en ajoutant 3600000000 au numéro de l'objet OSM
aire_de_recherche = str(3_600_000_000 + args.zone)
for req in utils.json_reqs:
for nb_essai in range(MAX_RETRY): # on tente max_retry fois
try:
logging.info(f"# Requête en cours : {req}")
# appel overpass
data = utils.run_overpass_query(
utils.json_reqs[req]["overpass"], aire_de_recherche
)
nb_resultats = len(data["elements"])
logging.info(f"{nb_resultats} résultats")
if nb_resultats > 0:
if args.geocoding_inverse:
# géocodage inverse
data = utils.geocodage_csv(data)
# traduction
for key in utils.traductions:
data = utils.traduction(key, utils.traductions[key], data)
# Sauvegarde
os.makedirs(args.dossier_resultats, exist_ok=True)
export_json = utils.nettoyage_json_pour_umap(
data, utils.json_reqs[req]["champs"]
)
utils.save_as_json(export_json, req)
utils.save_as_ods(utils.json_reqs[req]["champs"], data, req)
# doucement sur overpass
time.sleep(30)
break
except errors.ApiError:
if nb_essai >= MAX_RETRY - 1:
logging.error("Trop d'erreurs d'API - abandon")
exit()
logging.error(f"Erreur API - on retente dans {RETRY_DELAY}s")
except:
logging.error("Exception", stack_info=True, exc_info=True)
time.sleep(RETRY_DELAY)
logging.info("##### Terminé #####")
if __name__ == "__main__":
main()