scripts/seed_from_csv.py

from __future__ import annotations

import os
import sys
import logging
from pathlib import Path

import pandas as pd
from dotenv import load_dotenv, find_dotenv
from sqlalchemy import create_engine, text, bindparam

load_dotenv(find_dotenv())

logging.basicConfig(level=logging.INFO, format="%(levelname)s | %(message)s")
logger = logging.getLogger("technova.seed")

# Utilitaire pour convertir des valeurs oui/non en 1/0
def yesno_to_int(x):
    if x is None:
        return None
    if isinstance(x, bool):
        return int(x)
    if isinstance(x, (int, float)) and x in (0, 1):
        return int(x)

    s = str(x).strip().lower()
    if s in {"oui", "y", "yes", "1", "true", "vrai"}:
        return 1
    if s in {"non", "n", "no", "0", "false", "faux"}:
        return 0
    return None

# Normalisation basique des colonnes (trim, etc.)
def norm_cols(df: pd.DataFrame) -> pd.DataFrame:
    df = df.copy()
    df.columns = [c.strip() for c in df.columns]
    return df

# Utilitaire pour pick une valeur parmi plusieurs clés possibles (utile pour gérer les variations de noms de colonnes)
def pick(r: dict, *keys, default=None):
    for k in keys:
        if k in r and r.get(k) is not None:
            return r.get(k)
    return default

# Point d'entrée du script
def main(
    sirh_csv: str = "data/extrait_sirh.csv",
    eval_csv: str = "data/extrait_eval.csv",
    sondage_csv: str = "data/extrait_sondage.csv",
):
    refresh = "--refresh" in sys.argv

    db_url = os.getenv("DATABASE_URL")
    if not db_url:
        raise RuntimeError("DATABASE_URL manquant dans .env")

    engine = create_engine(db_url, pool_pre_ping=True)

    base = Path(__file__).resolve().parents[1]
    sirh_path = (base / sirh_csv).resolve()
    eval_path = (base / eval_csv).resolve()
    sondage_path = (base / sondage_csv).resolve()

    for p in (sirh_path, eval_path, sondage_path):
        if not p.exists():
            raise RuntimeError(f"Fichier introuvable: {p}")

    df_sirh = norm_cols(pd.read_csv(sirh_path))
    df_eval = norm_cols(pd.read_csv(eval_path))
    df_sond = norm_cols(pd.read_csv(sondage_path))

    if "id_employee" not in df_sirh.columns:
        raise RuntimeError("extrait_sirh.csv doit contenir la colonne id_employee")

    if not (len(df_sirh) == len(df_eval) == len(df_sond)):
        raise RuntimeError("Les 3 CSV doivent avoir le même nombre de lignes (jointure par index).")

    df = pd.concat([df_sirh, df_eval, df_sond], axis=1)

    # Remplace NaN -> None
    df = df.where(pd.notnull(df), None)
    # Optionnel mais utile: transforme "" / "   " -> None
    df = df.replace(r"^\s*$", None, regex=True)

    # SQL (SCHEMA raw.*)
    upsert_employees = text(
        """
        INSERT INTO raw.employees (
            employee_external_id, age, genre, statut_marital, ayant_enfants,
            niveau_education, domaine_etude, departement, poste, distance_domicile_travail
        )
        VALUES (
            :employee_external_id, :age, :genre, :statut_marital, :ayant_enfants,
            :niveau_education, :domaine_etude, :departement, :poste, :distance_domicile_travail
        )
        ON CONFLICT (employee_external_id) DO UPDATE SET
            age = EXCLUDED.age,
            genre = EXCLUDED.genre,
            statut_marital = EXCLUDED.statut_marital,
            ayant_enfants = EXCLUDED.ayant_enfants,
            niveau_education = EXCLUDED.niveau_education,
            domaine_etude = EXCLUDED.domaine_etude,
            departement = EXCLUDED.departement,
            poste = EXCLUDED.poste,
            distance_domicile_travail = EXCLUDED.distance_domicile_travail
        """
    )

    select_emp_map = (
        text(
            """
            SELECT id, employee_external_id
            FROM raw.employees
            WHERE employee_external_id IN :ext_ids
            """
        )
        .bindparams(bindparam("ext_ids", expanding=True))
    )

    insert_snapshot = text(
        """
        INSERT INTO raw.employee_snapshots (
            employee_id,
            nombre_experiences_precedentes,
            nombre_heures_travaillees,
            annee_experience_totale,
            annees_dans_l_entreprise,
            annees_dans_le_poste_actuel,
            annees_sous_responsable_actuel,
            niveau_hierarchique_poste,
            revenu_mensuel,
            augmentation_salaire_precedente,
            heures_supplementaires,
            nombre_participation_pee,
            nb_formations_suivies,
            nombre_employee_sous_responsabilite,
            frequence_deplacement,
            annees_depuis_la_derniere_promotion
        )
        VALUES (
            :employee_id,
            :nombre_experiences_precedentes,
            :nombre_heures_travaillees,
            :annee_experience_totale,
            :annees_dans_l_entreprise,
            :annees_dans_le_poste_actuel,
            :annees_sous_responsable_actuel,
            :niveau_hierarchique_poste,
            :revenu_mensuel,
            :augmentation_salaire_precedente,
            :heures_supplementaires,
            :nombre_participation_pee,
            :nb_formations_suivies,
            :nombre_employee_sous_responsabilite,
            :frequence_deplacement,
            :annees_depuis_la_derniere_promotion
        )
        """
    )

    insert_survey = text(
        """
        INSERT INTO raw.surveys (
            employee_id,
            code_sondage,
            eval_number,
            note_evaluation_precedente,
            note_evaluation_actuelle,
            satisfaction_employee_environnement,
            satisfaction_employee_nature_travail,
            satisfaction_employee_equipe,
            satisfaction_employee_equilibre_pro_perso
        )
        VALUES (
            :employee_id,
            :code_sondage,
            :eval_number,
            :note_evaluation_precedente,
            :note_evaluation_actuelle,
            :satisfaction_employee_environnement,
            :satisfaction_employee_nature_travail,
            :satisfaction_employee_equipe,
            :satisfaction_employee_equilibre_pro_perso
        )
        """
    )

    insert_gt = text(
        """
        INSERT INTO raw.ground_truth (employee_id, date_event, a_quitte_l_entreprise)
        VALUES (:employee_id, now(), :a_quitte_l_entreprise)
        """
    )

    # Build records
    employees_records = []
    for r in df.to_dict(orient="records"):
        employees_records.append(
            {
                "employee_external_id": int(r["id_employee"]),
                "age": r.get("age"),
                "genre": r.get("genre"),
                "statut_marital": r.get("statut_marital"),
                "ayant_enfants": r.get("ayant_enfants"),
                "niveau_education": r.get("niveau_education"),
                "domaine_etude": r.get("domaine_etude"),
                "departement": r.get("departement"),
                "poste": r.get("poste"),
                "distance_domicile_travail": r.get("distance_domicile_travail")
            }
        )

    ext_ids = df["id_employee"].astype(int).unique().tolist()

    # RUN
    with engine.begin() as conn:
        if refresh:
            logger.info("Mode --refresh: purge raw tables (snapshots/surveys/ground_truth)")
            conn.execute(text("TRUNCATE TABLE raw.employee_snapshots RESTART IDENTITY CASCADE;"))
            conn.execute(text("TRUNCATE TABLE raw.surveys RESTART IDENTITY CASCADE;"))
            conn.execute(text("TRUNCATE TABLE raw.ground_truth RESTART IDENTITY CASCADE;"))
            # raw.employees: on garde l'upsert

        # 1) upsert employees
        conn.execute(upsert_employees, employees_records)

        # 2) map ext -> id
        emp_rows = conn.execute(select_emp_map, {"ext_ids": ext_ids}).mappings().all()
        ext_to_id = {int(r["employee_external_id"]): int(r["id"]) for r in emp_rows}

        snapshots_records = []
        surveys_records = []
        gt_records = []

        for r in df.to_dict(orient="records"):
            ext = int(r["id_employee"])
            employee_id = ext_to_id.get(ext)
            if not employee_id:
                continue

            snapshots_records.append(
                {
                    "employee_id": employee_id,
                    "nombre_experiences_precedentes": pick(r, "nombre_experiences_precedentes"),
                    "nombre_heures_travaillees": pick(r, "nombre_heures_travaillees", "nombre_heures_travailless"),
                    "annee_experience_totale": pick(r, "annee_experience_totale"),
                    "annees_dans_l_entreprise": pick(r, "annees_dans_l_entreprise"),
                    "annees_dans_le_poste_actuel": pick(r, "annees_dans_le_poste_actuel"),
                    "annees_sous_responsable_actuel": pick(r, "annees_sous_responsable_actuel", "annes_sous_responsable_actuel"),
                    "niveau_hierarchique_poste": pick(r, "niveau_hierarchique_poste"),
                    "revenu_mensuel": pick(r, "revenu_mensuel"),
                    "augmentation_salaire_precedente": pick(r, "augmentation_salaire_precedente", "augementation_salaire_precedente"),
                    "heures_supplementaires": pick(r, "heures_supplementaires", "heure_supplementaires"),
                    "nombre_participation_pee": pick(r, "nombre_participation_pee"),
                    "nb_formations_suivies": pick(r, "nb_formations_suivies"),
                    "nombre_employee_sous_responsabilite": pick(r, "nombre_employee_sous_responsabilite"),
                    "frequence_deplacement": pick(r, "frequence_deplacement"),
                    "annees_depuis_la_derniere_promotion": pick(r, "annees_depuis_la_derniere_promotion")
                }
            )

            surveys_records.append(
                {
                    "employee_id": employee_id,
                    "code_sondage": pick(r, "code_sondage"),
                    "eval_number": pick(r, "eval_number"),
                    "note_evaluation_precedente": pick(r, "note_evaluation_precedente"),
                    "note_evaluation_actuelle": pick(r, "note_evaluation_actuelle"),
                    "satisfaction_employee_environnement": pick(r, "satisfaction_employee_environnement"),
                    "satisfaction_employee_nature_travail": pick(r, "satisfaction_employee_nature_travail"),
                    "satisfaction_employee_equipe": pick(r, "satisfaction_employee_equipe"),
                    "satisfaction_employee_equilibre_pro_perso": pick(r, "satisfaction_employee_equilibre_pro_perso")
                }
            )

            gt_records.append(
                {
                    "employee_id": employee_id,
                    "a_quitte_l_entreprise": yesno_to_int(pick(r, "a_quitte_l_entreprise"))
                }
            )

        if snapshots_records:
            conn.execute(insert_snapshot, snapshots_records)
        if surveys_records:
            conn.execute(insert_survey, surveys_records)
        if gt_records:
            conn.execute(insert_gt, gt_records)

    print("Seed terminé depuis les 3 CSV")
    print(f"   employees: {len(employees_records)} (upsert)")
    print(f"   snapshots: {len(snapshots_records)}")
    print(f"   surveys:   {len(surveys_records)}")
    print(f"   ground_truth: {len(gt_records)}")

if __name__ == "__main__":
    main()