diff --git a/fetcher.py b/fetcher.py index abaeb88..09edfc9 100644 --- a/fetcher.py +++ b/fetcher.py @@ -1,5 +1,4 @@ -import json -from re import error +from typing import Any, cast import ovh @@ -10,13 +9,21 @@ def fetch_api(app_key: str, app_secret: str, consumer_key: str) -> list[str]: application_secret=app_secret, consumer_key=consumer_key, ) - bills = client.get("/me/bill/") + data: Any = client.get("/me/bill/") + + if data is None: + return [] + + if not isinstance(data, list) or not all(isinstance(x, str) for x in data): + raise TypeError("Réponse OVH inattendue pour /me/bill/: liste de str requise") + + bills: list[str] = cast(list[str], data) return bills def fetch_invoice_content( id: str, app_key: str, app_secret: str, consumer_key: str -) -> dict: +) -> dict[str, Any]: client = ovh.Client( endpoint="ovh-eu", application_key=app_key, @@ -24,4 +31,6 @@ def fetch_invoice_content( consumer_key=consumer_key, ) bill = client.get(f"/me/bill/{id}") + if bill is None: + raise RuntimeError(f"Facture {id} introuvable") return bill diff --git a/main.py b/main.py index d258dc8..882f4b9 100644 --- a/main.py +++ b/main.py @@ -1,14 +1,32 @@ import os +from datetime import datetime import dotenv import ovh import fetcher as ft -import datetime from urllib.request import urlretrieve dotenv.load_dotenv() -APP_KEY = os.getenv("APP_KEY") -APP_SECRET = os.getenv("APP_SECRET") -CONSUMER_KEY = os.getenv("CONSUMER_KEY") +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 + + +def indexer(ids: list[str]) -> list[str]: + ids_already_in = os.listdir(f"{PATH_OVH}/{YEAR}") + 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) + return result def get_ids() -> list[str]: @@ -35,15 +53,20 @@ def get_bill(bill_id: str) -> dict: raise RuntimeError(f"Échec récupération facture {bill_id}: {e}") from e -def get_pdf(bill: dict): +def save_pdf(bill: dict): + date = datetime.fromisoformat(bill["date"]).date() + path = f"{PATH_OVH}/{date.year}" + + if not os.path.isdir(path): + os.mkdir(path) url = bill["pdfUrl"] - date = f"{datetime.datetime.fromisoformat(bill['date']).date()}.pdf" - urlretrieve(url, date) + urlretrieve(url, f"{PATH_OVH}{bill['billId']}.pdf") if __name__ == "__main__": - ids = get_ids() - print(ids) + if not os.path.isdir(PATH_OVH): + os.mkdir(PATH_OVH) + ids = indexer(get_ids()) if ids: - bill = get_bill(ids[0]) - get_pdf(bill) + for id in ids: + save_pdf(get_bill(id))