From afd27a811a9a117d56e23f200664a9523ed198a7 Mon Sep 17 00:00:00 2001 From: Spectre Date: Fri, 5 Sep 2025 11:08:40 +0200 Subject: [PATCH] Add logging and doc --- fetcher.py | 8 ++++ main.py | 126 +++++++++++++++++++++++++++++++++++++++++------------ 2 files changed, 107 insertions(+), 27 deletions(-) diff --git a/fetcher.py b/fetcher.py index 09edfc9..2cadd01 100644 --- a/fetcher.py +++ b/fetcher.py @@ -1,5 +1,8 @@ from typing import Any, cast import ovh +import logging + +logger = logging.getLogger("ovh_factures.fetcher") def fetch_api(app_key: str, app_secret: str, consumer_key: str) -> list[str]: @@ -12,12 +15,15 @@ def fetch_api(app_key: str, app_secret: str, consumer_key: str) -> list[str]: data: Any = client.get("/me/bill/") if data is None: + logger.warning("Réponse vide pour /me/bill/") return [] if not isinstance(data, list) or not all(isinstance(x, str) for x in data): + logger.error("Réponse inattendue pour /me/bill/: %r", data) raise TypeError("Réponse OVH inattendue pour /me/bill/: liste de str requise") bills: list[str] = cast(list[str], data) + logger.info("%d factures détectées dans /me/bill/", len(bills)) return bills @@ -32,5 +38,7 @@ def fetch_invoice_content( ) bill = client.get(f"/me/bill/{id}") if bill is None: + logger.error("Facture %s introuvable", id) raise RuntimeError(f"Facture {id} introuvable") + logger.debug("Facture %s récupérée avec succès", id) return bill diff --git a/main.py b/main.py index 882f4b9..df71a78 100644 --- a/main.py +++ b/main.py @@ -4,44 +4,100 @@ import dotenv import ovh import fetcher as ft from urllib.request import urlretrieve +import logging +from logging.handlers import RotatingFileHandler +# --- Configuration du logging --- +logging.addLevelName(logging.DEBUG, "DÉBOGAGE") +logging.addLevelName(logging.INFO, "INFO") +logging.addLevelName(logging.WARNING, "AVERTISSEMENT") +logging.addLevelName(logging.ERROR, "ERREUR") +logging.addLevelName(logging.CRITICAL, "CRITIQUE") + +logger = logging.getLogger("ovh_factures") +logger.setLevel(logging.INFO) +formatter = logging.Formatter( + fmt="%(asctime)s | %(levelname)s | %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", +) +# Console +ch = logging.StreamHandler() +ch.setFormatter(formatter) +logger.addHandler(ch) +# Fichier +fh = RotatingFileHandler( + "ovh_factures.log", maxBytes=5_000_000, backupCount=3, encoding="utf-8" +) +fh.setFormatter(formatter) +logger.addHandler(fh) + +# Chargement des variables d'environnement (.env) dotenv.load_dotenv() APP_KEY = os.environ["APP_KEY"] APP_SECRET = os.environ["APP_SECRET"] CONSUMER_KEY = os.environ["CONSUMER_KEY"] PATH_OVH = os.environ["OVH_PATH"] -YEAR = datetime.now().year +YEAR = datetime.now().year # Année courante (int) def indexer(ids: list[str]) -> list[str]: - ids_already_in = os.listdir(f"{PATH_OVH}/{YEAR}") + """ + Parcourt le répertoire de l'année courante et compare les factures déjà présentes + avec la liste d'IDs renvoyée par OVH. Ne conserve que les factures absentes + ET datées de l'année courante. + """ + logger.info("Indexation des factures pour l'année %s", YEAR) + target_dir = f"{PATH_OVH}{YEAR}" + try: + ids_already_in = os.listdir(target_dir) + except FileNotFoundError: + logger.warning("Dossier %s inexistant, aucune facture locale", target_dir) + ids_already_in = [] + missing = [x for x in ids if f"{x}.pdf" not in ids_already_in] - result = [] - for x in missing: - date_str = ft.fetch_invoice_content( - x, - app_secret=APP_SECRET, - app_key=APP_KEY, - consumer_key=CONSUMER_KEY, - )["date"] - if datetime.fromisoformat(date_str).year >= int(YEAR): - result.append(x) + logger.info("%d factures absentes détectées", len(missing)) + + result: list[str] = [] + for bill_id in missing: + try: + meta = ft.fetch_invoice_content( + bill_id, + app_key=APP_KEY, + app_secret=APP_SECRET, + consumer_key=CONSUMER_KEY, + ) + except Exception as e: + logger.error("Impossible de récupérer la méta pour %s : %s", bill_id, e) + continue + bill_year = datetime.fromisoformat(meta["date"]).year + if bill_year == YEAR: + result.append(bill_id) + + logger.info("%d factures retenues pour téléchargement", len(result)) return result def get_ids() -> list[str]: + """ + Interroge l’API OVH et renvoie la liste des IDs de toutes les factures. + """ + logger.info("Récupération de la liste des factures via API OVH") try: - ids = ft.fetch_api( + return ft.fetch_api( app_key=APP_KEY, app_secret=APP_SECRET, consumer_key=CONSUMER_KEY, ) - return ids except ovh.exceptions.APIError as e: - raise RuntimeError(f"Échec récupération IDs factures: {e}") from e + logger.error("Échec récupération des IDs de factures : %s", e) + raise RuntimeError(f"Échec de la récupération des IDs de factures : {e}") from e def get_bill(bill_id: str) -> dict: + """ + Récupère, via l’API OVH, les informations détaillées d’une facture (JSON). + """ + logger.debug("Récupération de la facture %s", bill_id) try: return ft.fetch_invoice_content( bill_id, @@ -50,23 +106,39 @@ def get_bill(bill_id: str) -> dict: consumer_key=CONSUMER_KEY, ) except ovh.exceptions.APIError as e: - raise RuntimeError(f"Échec récupération facture {bill_id}: {e}") from e + logger.error("Échec récupération de la facture %s : %s", bill_id, e) + raise RuntimeError( + f"Échec de la récupération de la facture {bill_id} : {e}" + ) from e -def save_pdf(bill: dict): +def save_pdf(bill: dict) -> None: + """ + Télécharge le PDF d’une facture dans un sous-dossier par année. + Noms de fichiers : .pdf + """ date = datetime.fromisoformat(bill["date"]).date() - path = f"{PATH_OVH}/{date.year}" + path = f"{PATH_OVH}{date.year}/" + + os.makedirs(path, exist_ok=True) - if not os.path.isdir(path): - os.mkdir(path) url = bill["pdfUrl"] - urlretrieve(url, f"{PATH_OVH}{bill['billId']}.pdf") + dest = f"{path}{bill['billId']}.pdf" + try: + urlretrieve(url, dest) + logger.info("Facture %s sauvegardée dans %s", bill["billId"], dest) + except Exception as e: + logger.error("Impossible de télécharger la facture %s : %s", bill["billId"], e) + raise if __name__ == "__main__": - if not os.path.isdir(PATH_OVH): - os.mkdir(PATH_OVH) - ids = indexer(get_ids()) - if ids: - for id in ids: - save_pdf(get_bill(id)) + logger.info("Démarrage du traitement des factures OVH pour %s", YEAR) + os.makedirs(f"{PATH_OVH}{YEAR}", exist_ok=True) + + ids_candidats = indexer(get_ids()) + + for bill_id in ids_candidats: + save_pdf(get_bill(bill_id)) + + logger.info("Traitement terminé : %d factures téléchargées", len(ids_candidats))