mirror of
https://github.com/Fare-spec/get_ovh_bills.git
synced 2025-12-07 10:20:36 +00:00
add a couple of things
This commit is contained in:
17
fetcher.py
17
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
|
||||
|
||||
45
main.py
45
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))
|
||||
|
||||
Reference in New Issue
Block a user