get_ovh_bills/main.py

import os
import mail as ml
from datetime import date, datetime
import dotenv
import ovh
import fetcher as ft
from urllib.request import urlretrieve
import logging
from logging.handlers import RotatingFileHandler
import traceback
import sqlite3

# --- 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"]
DB_PATH = os.environ["DB_PATH"]
EMAIL = os.environ["EMAIL"]
EMAIL_PASSWORD = os.environ["EMAIL_PASSWORD"]
SMTP_MAIL_ADDRESS = os.environ["SMTP_MAIL_ADDRESS"]
SMTP_PORT = os.environ["SMTP_PORT"]
EMAIL_TO = os.environ["EMAIL_TO"]
YEAR = datetime.now().year  # Année courante (int)


def get_conn():
    """
    Ouvre une connexion SQLite vers DB_PATH, crée la table 'bills' si nécessaire, puis retourne la connexion.
    """
    try:
        logger.debug("Ouverture de la connexion SQLite vers %s", DB_PATH)
        conn = sqlite3.connect(DB_PATH)
        logger.debug("Connexion établie, vérification/creation de la table 'bills'")
        conn.execute("""
        CREATE TABLE IF NOT EXISTS bills (
            bill_id TEXT PRIMARY KEY,
            bill_year INT
        )""")
        conn.commit()
        logger.info("Base SQLite initialisée et table 'bills' disponible")
        return conn
    except Exception as e:
        logger.exception("Erreur lors de l'initialisation de la base SQLite: %s", e)
        raise


def send_error_mail(error_msg):
    try:
        ml.send_email(
            subject="[OVH_FACTURES] ERREUR",
            content=f"<pre>{error_msg}</pre>",
            email_from=EMAIL,
            email_password=EMAIL_PASSWORD,
            smpt_port=SMTP_PORT,
            smtp_mail_address=SMTP_MAIL_ADDRESS,
            email_to=EMAIL_TO,
        )
    except Exception:
        pass


def add_entries_to_db(entries: list[tuple[str, int]], conn):
    """
    Insère en lot des paires (bill_id, bill_year) dans la table 'bills' avec gestion de conflit sur bill_id.
    """
    try:
        logger.debug("Insertion batch dans 'bills': %d entrées", len(entries))
        conn.executemany(
            """
            INSERT INTO bills (bill_id, bill_year)
            VALUES (?, ?)
            ON CONFLICT(bill_id) DO NOTHING
            """,
            entries,
        )
        conn.commit()
        logger.info("Insertion batch dans 'bills' validée")
    except Exception as e:
        logger.exception("Échec d'insertion batch dans 'bills': %s", e)
        send_error_mail(traceback.format_exc())
        raise


def get_entries_from_db(conn) -> set[str]:
    """
    Récupère l'ensemble des bill_id présents dans la table 'bills' et les retourne sous forme de set[str].
    """
    try:
        logger.debug("Sélection des bill_id depuis 'bills'")
        cursor = conn.execute("SELECT bill_id FROM bills")
        rows = cursor.fetchall()
        logger.info("Sélection terminée: %d bill_id récupérés", len(rows))
        return {row[0] for row in rows}
    except Exception as e:
        logger.exception("Échec de lecture des bill_id depuis 'bills': %s", e)
        send_error_mail(traceback.format_exc())
        raise


def compare_db_to_data(db_data: set[str], data: list[str]) -> list[str]:
    """
    Compare une collection d'identifiants 'data' à l'ensemble 'db_data' et retourne la liste des éléments absents de 'db_data'.
    """
    missings_current_year = list()
    for bill_id in data:
        if bill_id not in db_data:
            missings_current_year.append(bill_id)
    return missings_current_year


def indexer(ids: list[str]) -> list[str]:
    """
    Parcourt le répertoire de l'année courante, filtre les factures déjà présentes localement, conserve les factures absentes datées de l'année courante, et enregistre en base celles qui appartiennent à une autre année.
    """
    conn = get_conn()
    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 = compare_db_to_data(
        get_entries_from_db(conn), [x for x in ids if f"{x}.pdf" not in ids_already_in]
    )
    logger.info("%d factures absentes détectées", len(missing))

    result: list[str] = []
    not_valid_year: list[tuple[str, int]] = list()
    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 le json pour %s : %s", bill_id, e)

            send_error_mail(traceback.format_exc())
            continue
        bill_year = datetime.fromisoformat(meta["date"]).year
        if bill_year == YEAR:
            result.append(bill_id)
        else:
            not_valid_year.append((bill_id, bill_year))

    add_entries_to_db(not_valid_year, conn)
    logger.info(f"Ajouter {len(not_valid_year)} entrées a la base de donnée")
    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:
        return ft.fetch_api(
            app_key=APP_KEY,
            app_secret=APP_SECRET,
            consumer_key=CONSUMER_KEY,
        )
    except ovh.exceptions.APIError as e:
        logger.error("Échec récupération des IDs de factures : %s", e)

        send_error_mail(traceback.format_exc())
        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,
            app_key=APP_KEY,
            app_secret=APP_SECRET,
            consumer_key=CONSUMER_KEY,
        )
    except ovh.exceptions.APIError as e:
        logger.error("Échec récupération de la facture %s : %s", bill_id, e)

        send_error_mail(traceback.format_exc())
        raise RuntimeError(
            f"Échec de la récupération de la facture {bill_id} : {e}"
        ) from e


def save_pdf(bill: dict) -> None:
    """
    Télécharge le PDF d’une facture dans un sous-dossier par année.
    Noms de fichiers : <billId>.pdf
    """
    date = datetime.fromisoformat(bill["date"]).date()
    path = f"{PATH_OVH}{date.year}/"

    os.makedirs(path, exist_ok=True)

    url = bill["pdfUrl"]
    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)
        send_error_mail(traceback.format_exc())
        raise


if __name__ == "__main__":
    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())
    bills_json = []
    bills_str = []
    for bill_id in ids_candidats:
        bills_json.append((bill_id, get_bill(bill_id)))
    # pdf enregistrement.
    if len(bills_json) > 0:
        for bill_json in bills_json:
            save_pdf(bill_json[1])
            date = datetime.fromisoformat(bill_json[1]["date"]).date()

            bills_str.append(
                (
                    bill_json[0],
                    f"{date}",
                )
            )
        content = ml.construct_html(bills_str)
        ml.send_email(
            "Reçu de facture(s)",
            content,
            email_from=EMAIL,
            email_password=EMAIL_PASSWORD,
            smpt_port=SMTP_PORT,
            smtp_mail_address=SMTP_MAIL_ADDRESS,
            email_to=EMAIL_TO,
        )
    logger.info("Traitement terminé : %d factures téléchargées", len(ids_candidats))