scripts/register_model.py

"""
Script d'enregistrement de la version de modèle déployée.

Ce script alimente la table `model_registry` avec la version réellement
servie par l'API.

Objectif
--------
Tracer proprement :
- le nom du modèle
- la version déployée
- le stage courant
- le chemin de l'artefact
- la liste des features attendues
- les métriques offline éventuelles
- le statut actif en production

Principe
--------
Ce script doit être exécuté au moment du déploiement, après que les
artefacts soient disponibles sur disque et avant ou juste après le
démarrage applicatif.

Variables d'environnement supportées
------------------------------------
- MODEL_NAME
- MODEL_VERSION
- MODEL_STAGE
- MODEL_PATH
- TRAINING_DATA_VERSION
- MODEL_RUN_ID
- MODEL_SOURCE_PATH
- MODEL_IS_ACTIVE
- MODEL_FEATURES_PATH
- MODEL_METRICS_PATH
- MODEL_HYPERPARAMETERS_PATH
"""

from __future__ import annotations

import json
import os
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

from dotenv import load_dotenv

BASE_DIR = Path(__file__).resolve().parents[1]
if str(BASE_DIR) not in sys.path:
    sys.path.insert(0, str(BASE_DIR))

from app.core.db import SessionLocal
from app.services.monitoring_service import MonitoringService


# =============================================================================
# Chargement de l'environnement
# =============================================================================

load_dotenv()


# =============================================================================
# Configuration
# =============================================================================

ALLOWED_MODEL_STAGES = {"dev", "staging", "production", "archived"}

MODEL_NAME = os.getenv("MODEL_NAME", "credit_scoring_model")
MODEL_VERSION = os.getenv("MODEL_VERSION", "v1")
MODEL_STAGE = os.getenv("MODEL_STAGE", "production")
MODEL_PATH = os.getenv("MODEL_PATH", "artifacts/model.joblib")
TRAINING_DATA_VERSION = os.getenv("TRAINING_DATA_VERSION")
MODEL_RUN_ID = os.getenv("MODEL_RUN_ID")
MODEL_SOURCE_PATH = os.getenv("MODEL_SOURCE_PATH", MODEL_PATH)

MODEL_IS_ACTIVE = os.getenv("MODEL_IS_ACTIVE", "true").strip().lower() in {
    "1", "true", "yes", "y"
}

MODEL_FEATURES_PATH = os.getenv(
    "MODEL_FEATURES_PATH",
    "artifacts/model_features.json",
)
MODEL_METRICS_PATH = os.getenv(
    "MODEL_METRICS_PATH",
    "artifacts/metrics.json",
)
MODEL_HYPERPARAMETERS_PATH = os.getenv(
    "MODEL_HYPERPARAMETERS_PATH",
    "artifacts/hyperparameters.json",
)


# =============================================================================
# Helpers
# =============================================================================

def _utc_now() -> datetime:
    """
    Retourne l'heure actuelle en UTC.
    """
    return datetime.now(timezone.utc)


def _resolve_path(path_str: str) -> Path:
    """
    Résout un chemin absolu ou relatif au projet.
    """
    path = Path(path_str)
    if path.is_absolute():
        return path
    return BASE_DIR / path


def _read_json_file(path_str: str) -> Any | None:
    """
    Lit un fichier JSON s'il existe.

    Parameters
    ----------
    path_str : str
        Chemin du fichier JSON.

    Returns
    -------
    Any | None
        Contenu JSON ou None si absent/invalide.
    """
    path = _resolve_path(path_str)

    if not path.exists():
        return None

    try:
        with path.open("r", encoding="utf-8") as f:
            return json.load(f)
    except Exception as exc:
        print(f"[WARNING] Impossible de lire {path}: {exc}")
        return None


def _load_feature_list() -> list[str] | None:
    """
    Charge la liste des features attendues par le modèle.

    Formats supportés
    -----------------
    - JSON liste : ["f1", "f2", ...]
    - JSON dict  : {"features": ["f1", "f2", ...]}

    Returns
    -------
    list[str] | None
        Liste des features ou None.
    """
    data = _read_json_file(MODEL_FEATURES_PATH)

    if data is None:
        return None

    if isinstance(data, list):
        return [str(x) for x in data]

    if isinstance(data, dict) and isinstance(data.get("features"), list):
        return [str(x) for x in data["features"]]

    print(
        f"[WARNING] Format inattendu pour MODEL_FEATURES_PATH: {MODEL_FEATURES_PATH}"
    )
    return None


def _load_metrics() -> dict[str, Any] | None:
    """
    Charge les métriques offline du modèle.

    Returns
    -------
    dict[str, Any] | None
        Dictionnaire de métriques ou None.
    """
    data = _read_json_file(MODEL_METRICS_PATH)

    if data is None:
        return None

    if isinstance(data, dict):
        return data

    print(
        f"[WARNING] Format inattendu pour MODEL_METRICS_PATH: {MODEL_METRICS_PATH}"
    )
    return None


def _load_hyperparameters() -> dict[str, Any] | None:
    """
    Charge les hyperparamètres du modèle.

    Returns
    -------
    dict[str, Any] | None
        Dictionnaire d'hyperparamètres ou None.
    """
    data = _read_json_file(MODEL_HYPERPARAMETERS_PATH)

    if data is None:
        return None

    if isinstance(data, dict):
        return data

    print(
        "[WARNING] Format inattendu pour MODEL_HYPERPARAMETERS_PATH: "
        f"{MODEL_HYPERPARAMETERS_PATH}"
    )
    return None


def _validate_required_files() -> None:
    """
    Vérifie les artefacts indispensables.

    Raises
    ------
    FileNotFoundError
        Si le fichier modèle principal est absent.
    """
    model_path = _resolve_path(MODEL_PATH)

    if not model_path.exists():
        raise FileNotFoundError(f"Artefact modèle introuvable: {model_path}")


def _validate_configuration() -> None:
    """
    Vérifie la cohérence minimale de la configuration.

    Raises
    ------
    ValueError
        Si une variable essentielle est invalide.
    """
    if MODEL_STAGE not in ALLOWED_MODEL_STAGES:
        raise ValueError(
            f"MODEL_STAGE invalide: {MODEL_STAGE}. "
            f"Valeurs autorisées: {sorted(ALLOWED_MODEL_STAGES)}"
        )

    if not MODEL_NAME.strip():
        raise ValueError("MODEL_NAME est vide.")

    if not MODEL_VERSION.strip():
        raise ValueError("MODEL_VERSION est vide.")


# =============================================================================
# Main
# =============================================================================

def main() -> None:
    """
    Enregistre la version actuellement déployée dans model_registry.
    """
    print("=" * 80)
    print("ENREGISTREMENT DU MODÈLE DÉPLOYÉ")
    print("=" * 80)

    _validate_configuration()
    _validate_required_files()

    feature_list = _load_feature_list()
    metrics = _load_metrics()
    hyperparameters = _load_hyperparameters()

    print(f"MODEL_NAME              : {MODEL_NAME}")
    print(f"MODEL_VERSION           : {MODEL_VERSION}")
    print(f"MODEL_STAGE             : {MODEL_STAGE}")
    print(f"MODEL_PATH              : {_resolve_path(MODEL_PATH)}")
    print(f"MODEL_SOURCE_PATH       : {MODEL_SOURCE_PATH}")
    print(f"TRAINING_DATA_VERSION   : {TRAINING_DATA_VERSION}")
    print(f"MODEL_RUN_ID            : {MODEL_RUN_ID}")
    print(f"MODEL_IS_ACTIVE         : {MODEL_IS_ACTIVE}")
    print(f"NB FEATURES             : {len(feature_list) if feature_list else 0}")
    print(f"METRICS DISPONIBLES     : {list(metrics.keys()) if metrics else []}")

    db = SessionLocal()

    try:
        service = MonitoringService(db)

        result = service.register_model_version(
            model_name=MODEL_NAME,
            model_version=MODEL_VERSION,
            stage=MODEL_STAGE,
            run_id=MODEL_RUN_ID,
            source_path=MODEL_SOURCE_PATH,
            training_data_version=TRAINING_DATA_VERSION,
            feature_list=feature_list,
            hyperparameters=hyperparameters,
            metrics=metrics,
            deployed_at=_utc_now(),
            is_active=MODEL_IS_ACTIVE,
        )

        db.commit()

        print("[OK] Modèle enregistré avec succès")
        print(result)

    except Exception as exc:
        db.rollback()
        print("[ERROR] Échec de l'enregistrement du modèle")
        print(f"Type    : {type(exc).__name__}")
        print(f"Message : {exc}")
        raise

    finally:
        db.close()


if __name__ == "__main__":
    main()