src/module/Validation_Contract.py

"""
MODULE: VALIDATION CONTRACT (ETL ENGINE) - VERSION AMÉLIORÉE
=============================================================
Responsabilité : Pipeline ETL complet avec nettoyage des anomalies spécifiques hospitalières
Architecture : RAW → VALIDATED → CLEAN (3 couches)

Pipeline:
1. Extract → Chargement données brutes
2. Validate → Détection anomalies
3. Transform → Nettoyage spécifique (dates inversées, âges, doublons, etc.)
4. Load → Stockage RAW + CLEAN
5. Report → Rapport de transformation détaillé

Fonctionnalités :
- ✅ Stockage version BRUTE (pour Indicateurs_DQ)
- ✅ Stockage version CLEAN (pour EDA)
- ✅ Correction dates inversées
- ✅ Filtrage âges aberrants
- ✅ Déduplification patients
- ✅ Correction suroccupation
- ✅ Normalisation gravité
- ✅ Rapport transformation visuel
"""

import pandas as pd
import numpy as np
import streamlit as st
import plotly.express as px
import plotly.graph_objects as go
from datetime import datetime
import sys
import uuid  # <--- AJOUTER CECI EN HAUT AVEC LES AUTRES IMPORTS

# Augmentation limite sécurité Python
sys.set_int_max_str_digits(100000)


# =====================================================
# ============= CLASSES DE DONNÉES ====================
# =====================================================

class DataExplorer:
    """Conteneur pour stocker les versions RAW et CLEAN des données"""
    
    def __init__(self):
        self.dataframes_raw = {}      # Données BRUTES (pour DQ)
        self.dataframes_clean = {}    # Données NETTOYÉES (pour EDA)
        self.metadata = {}
        self.transformation_report = {}
        self.load_timestamp = None
        
    def add_dataframe(self, name, df_raw, df_clean, metrics):
        """Ajoute les deux versions d'un dataframe"""
        self.dataframes_raw[name] = df_raw
        self.dataframes_clean[name] = df_clean
        
        # Métadonnées
        numeric_cols = df_clean.select_dtypes(include=[np.number]).columns.tolist()
        categorical_cols = df_clean.select_dtypes(include=['object']).columns.tolist()
        date_cols = df_clean.select_dtypes(include=['datetime64']).columns.tolist()
        
        # --- FIX CRITIQUE 1 : Calcul Anomalies Sécurisé (Gestion Infini) ---
        anomalies_raw = 0
        if 'dq_is_anomaly' in df_raw.columns:
            try:
                # 1. Convertir en numérique (les textes géants deviennent 'inf' ou 'NaN')
                s_anom = pd.to_numeric(df_raw['dq_is_anomaly'], errors='coerce')
                # 2. Remplacer Infini et NaN par 0 pour éviter le crash int(inf)
                s_anom = s_anom.replace([np.inf, -np.inf], 0).fillna(0)
                # 3. Somme sécurisée
                anomalies_raw = int(s_anom.sum())
                # Cap de sécurité
                if anomalies_raw > 1_000_000: anomalies_raw = 999999
            except Exception:
                anomalies_raw = 0
        
        anomalies_clean = 0
        if 'dq_is_anomaly' in df_clean.columns:
            try:
                s_anom = pd.to_numeric(df_clean['dq_is_anomaly'], errors='coerce')
                s_anom = s_anom.replace([np.inf, -np.inf], 0).fillna(0)
                anomalies_clean = int(s_anom.sum())
                if anomalies_clean > 1_000_000: anomalies_clean = 999999
            except Exception:
                anomalies_clean = 0
        # -------------------------------------------------------------------
        
        self.metadata[name] = {
            'rows_raw': len(df_raw),
            'rows_clean': len(df_clean),
            'rows_removed': len(df_raw) - len(df_clean),
            'mem_raw_mb': df_raw.memory_usage(deep=True).sum() / 1024**2,
            'mem_clean_mb': df_clean.memory_usage(deep=True).sum() / 1024**2,
            'columns': len(df_clean.columns),
            'null_cells_raw': int(df_raw.isnull().sum().sum()),
            'null_cells_clean': int(df_clean.isnull().sum().sum()),
            'anomalies_raw': anomalies_raw,
            'anomalies_clean': anomalies_clean,
            'numeric_columns': numeric_cols,
            'categorical_columns': categorical_cols,
            'datetime_columns': date_cols
        }
        
        # Rapport de transformation
        self.transformation_report[name] = metrics
    
    @property
    def dataframes(self):
        """Alias pour compatibilité avec Indicateurs_DQ (retourne version BRUTE)"""
        return self.dataframes_raw
    
    def get_clean_dataframe(self, name):
        """Récupère la version nettoyée"""
        return self.dataframes_clean.get(name)
    
    def get_raw_dataframe(self, name):
        """Récupère la version brute"""
        return self.dataframes_raw.get(name)
    
    def get_summary(self):
        """Résumé global"""
        total_rows_raw = sum(m['rows_raw'] for m in self.metadata.values())
        total_rows_clean = sum(m['rows_clean'] for m in self.metadata.values())
        total_mem = sum(m['mem_clean_mb'] for m in self.metadata.values())
        
        return {
            'tables_loaded': len(self.dataframes_clean),
            'total_rows_raw': total_rows_raw,
            'total_rows_clean': total_rows_clean,
            'total_rows_removed': total_rows_raw - total_rows_clean,
            'total_memory_mb': round(total_mem, 2),
            'timestamp': self.load_timestamp
        }
    
    def show_transformation_report(self):
        """Rapport visuel de transformation ETL"""
        st.markdown("---")
        st.markdown("### Rapport de Transformation ETL")
        
        if not self.transformation_report:
            st.warning("Aucun rapport de transformation disponible.")
            return

        # --- FIX : Génération d'un ID unique pour cet affichage ---
        # Cela permet d'afficher le rapport plusieurs fois sans conflit de clé
        unique_id = str(uuid.uuid4())[:8] 
        
        # Métriques globales
        col1, col2, col3, col4 = st.columns(4)
        
        summary = self.get_summary()
        
        col1.metric("Tables traitées", summary['tables_loaded'])
        col2.metric("Lignes BRUTES", f"{summary['total_rows_raw']:,}")
        col3.metric("Lignes CLEAN", f"{summary['total_rows_clean']:,}")
        col4.metric("Lignes supprimées", f"{summary['total_rows_removed']:,}", 
                   delta_color="inverse")
        
        # Graphiques
        col_a, col_b = st.columns(2)
        
        with col_a:
            # Anomalies avant/après
            names = list(self.metadata.keys())
            anom_raw = [self.metadata[n]['anomalies_raw'] for n in names]
            anom_clean = [self.metadata[n]['anomalies_clean'] for n in names]
            
            fig_anom = go.Figure()
            fig_anom.add_trace(go.Bar(x=names, y=anom_raw, name='BRUT', marker_color='crimson'))
            fig_anom.add_trace(go.Bar(x=names, y=anom_clean, name='CLEAN', marker_color='green'))
            fig_anom.update_layout(
                title="Anomalies : BRUT vs CLEAN",
                barmode='group',
                height=400
            )
            # UTILISATION DE L'ID UNIQUE ICI 👇
            st.plotly_chart(fig_anom, use_container_width=True, key=f"etl_chart_anom_{unique_id}")
        
        with col_b:
            # Cellules nulles avant/après
            null_raw = [self.metadata[n]['null_cells_raw'] for n in names]
            null_clean = [self.metadata[n]['null_cells_clean'] for n in names]
            
            fig_null = go.Figure()
            fig_null.add_trace(go.Bar(x=names, y=null_raw, name='BRUT', marker_color='orange'))
            fig_null.add_trace(go.Bar(x=names, y=null_clean, name='CLEAN', marker_color='blue'))
            fig_null.update_layout(
                title="Cellules Nulles : BRUT vs CLEAN",
                barmode='group',
                height=400
            )
            # UTILISATION DE L'ID UNIQUE ICI 👇
            st.plotly_chart(fig_null, use_container_width=True, key=f"etl_chart_null_{unique_id}")
        
        # Détail des transformations
        with st.expander("Détail des Transformations par Table"):
            for table_name, report in self.transformation_report.items():
                st.markdown(f"####  {table_name.upper()}")
                
                if report.get('corrections_applied'):
                    st.markdown("** Corrections appliquées :**")
                    for correction in report['corrections_applied']:
                        st.write(f"- {correction}")
                else:
                    st.info("Aucune correction nécessaire")
                
                # Stats
                meta = self.metadata[table_name]
                col1, col2, col3 = st.columns(3)
                col1.metric("Lignes supprimées", meta['rows_removed'])
                col2.metric("Anomalies corrigées", meta['anomalies_raw'] - meta['anomalies_clean'])
                col3.metric("Nulls comblés", meta['null_cells_raw'] - meta['null_cells_clean'])
                
                st.markdown("---")
        # [AJOUTER CECI À LA FIN DE LA MÉTHODE show_transformation_report]
        
        # --- EXPORT CSV DES MÉTRIQUES ---
        st.markdown("---")
        st.subheader("📥 Export du Rapport ETL")

        # 1. Préparation des données pour le CSV
        export_data = []
        for name, meta in self.metadata.items():
            # Récupération des corrections sous forme de texte
            report = self.transformation_report.get(name, {})
            corrections_txt = " | ".join(report.get('corrections_applied', []))
            
            row = {
                "Table": name,
                "Lignes Brutes": meta['rows_raw'],
                "Lignes Nettoyées": meta['rows_clean'],
                "Lignes Supprimées": meta['rows_removed'],
                "Anomalies Avant": meta['anomalies_raw'],
                "Anomalies Après": meta['anomalies_clean'],
                "Cellules Vides (Nulls)": meta['null_cells_clean'],
                "Corrections Appliquées": corrections_txt
            }
            export_data.append(row)
        
        df_export = pd.DataFrame(export_data)
        
        # 2. Bouton de téléchargement
        csv = df_export.to_csv(index=False).encode('utf-8-sig') # utf-8-sig pour Excel
        
        st.download_button(
            label="📄 Télécharger le Rapport de Transformation (CSV)",
            data=csv,
            file_name=f"rapport_etl_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv",
            mime="text/csv",
            key=f"dl_btn_etl_{unique_id}" # Utilise l'ID unique généré plus haut
        )

# =====================================================
# ============= MOTEUR ETL AMÉLIORÉ ===================
# =====================================================

class ETL_Engine:
    """Moteur ETL avec nettoyage spécifique des anomalies hospitalières"""

    def __init__(self, gspread_client, sheet_name="bdd_mvp_faker"):
        self.client = gspread_client
        self.sheet_name = sheet_name
        
        # Schéma contractuel (types attendus)
        self.SCHEMA = {
            'patients': {
                'numeric': ['age', 'code_postal'],
                'date': ['date_naissance'],
                'categorical': ['patient_id', 'nom', 'prenom', 'sexe', 'ville', 
                               'num_carte_vitale_fictif', 'num_carte_vitale_masque', 
                               'pathologie_frequente_code'],
                'preserve_as_is': ['dq_is_anomaly', 'dq_anomaly_type', 'dq_anomaly_detail']
            },
            'sejours': {
                'numeric': ['gravite_1_5', 'duree_sejour_jours', 'consommables_estimes_score'],
                'date': ['date_entree', 'date_sortie'],
                'categorical': ['stay_id', 'patient_id', 'service_code', 'mode_entree', 
                               'type_lit', 'pathologie_code', 'scenario_jour', 
                               'equipement_principal_code'],
                'preserve_as_is': ['dq_is_anomaly', 'dq_anomaly_type', 'dq_anomaly_detail']
            },
            'ressources_jour': {
                'numeric': ['intensity', 'lits_total', 'lits_occupes', 'lits_disponibles', 
                           'taux_occupation', 'personnel_total', 'taux_absence_conges', 
                           'taux_absence_scenario', 'est_periode_conges', 'personnel_disponible', 
                           'personnel_requis', 'tension_personnel'],
                'date': ['date'],
                'categorical': ['service_code', 'scenario', 'periode_conges_label'],
                'preserve_as_is': ['dq_is_anomaly', 'dq_anomaly_type', 'dq_anomaly_detail']
            },
            'admissions_jour': {
                'numeric': ['nb_admissions', 'intensity'],
                'date': ['date'],
                'categorical': ['service_code', 'scenario']
            },
            'stock_materiel_jour': {
                'numeric': ['stock_total', 'stock_disponible', 'consommation_estimee_jour', 
                           'tension_stock'],
                'date': ['date'],
                'categorical': ['service_code', 'materiel_code', 'alerte_stock'],
                'preserve_as_is': ['dq_is_anomaly', 'dq_anomaly_type', 'dq_anomaly_detail']
            },
            'scenarios_calendrier': {
                'numeric': ['intensity'],
                'date': ['date'],
                'categorical': ['scenario']
            }
        }
        
        self.REFERENCE_TABLES = [
            'service_ref', 'pathologie_ref', 'materiel_ref', 
            'mode_entree_ref', 'type_lit_ref', 'calendrier_conges', 'hypotheses'
        ]

    def _convert_types(self, name, df):
        """Étape 1 : Conversion et Typage Strict"""
        df_typed = df.copy()
        
        # --- MODIF 1 : Normalisation des noms de colonnes ---
        # Nettoie les espaces, met en minuscule et remplace les tirets pour une détection fiable
        df_typed.columns = df_typed.columns.str.strip().str.lower().str.replace(' ', '_').str.replace('-', '_')
        
        name_normalized = name.lower().strip().replace(' ', '_').replace('-', '_')
        
        # Tables de référence : on conserve le format brut
        if name_normalized in self.REFERENCE_TABLES:
            return df_typed
        
        config = self.SCHEMA.get(name_normalized)
        if not config:
            return df_typed
        
        # 1. Conversion DATES
        for col in config.get('date', []):
            if col in df_typed.columns:
                df_typed[col] = pd.to_datetime(df_typed[col], errors='coerce')
        
        # 2. --- MODIF 2 : Conversion NUMÉRIQUES BLINDÉE ---
        # Liste des colonnes numériques à traiter
        num_cols = config.get('numeric', [])
        
        # On ajoute dynamiquement les colonnes de DQ si elles existent pour éviter le crash 'str' vs 'int'
        if 'dq_is_anomaly' in df_typed.columns:
            num_cols.append('dq_is_anomaly')

        for col in num_cols:
            if col in df_typed.columns:
                # Nettoyage des caractères parasites (espaces, virgules)
                if df_typed[col].dtype == 'object':
                    df_typed[col] = df_typed[col].astype(str).str.replace(',', '.', regex=False)
                    df_typed[col] = df_typed[col].str.replace(r'\s+', '', regex=True)
                    # Suppression des termes textuels courants
                    df_typed[col] = df_typed[col].replace(['Full', 'Erreur', 'nan', 'None', ''], np.nan)
                
                # Conversion en numérique (coerce transforme les erreurs en NaN)
                df_typed[col] = pd.to_numeric(df_typed[col], errors='coerce')
                
                # FIX CRITIQUE : Remplacement des NaN par 0 
                # C'est l'assurance vie du Dashboard pour les comparaisons mathématiques
                df_typed[col] = df_typed[col].fillna(0)

        # 3. Conversion CATÉGORIELLES
        for col in config.get('categorical', []):
            if col in df_typed.columns:
                df_typed[col] = df_typed[col].astype(str).str.strip()
                df_typed[col] = df_typed[col].replace(['nan', 'NaN', 'None', ''], np.nan)
        
        return df_typed

    def _clean_patients(self, df):
        """Nettoyage spécifique PATIENTS"""
        df_clean = df.copy()
        corrections = []
        
        # 1. Filtrer âges aberrants
        if 'age' in df_clean.columns:
            mask_age_invalid = (df_clean['age'] < 0) | (df_clean['age'] > 120)
            nb_invalid_age = mask_age_invalid.sum()
            if nb_invalid_age > 0:
                df_clean = df_clean[~mask_age_invalid].copy()
                corrections.append(f" Supprimé {nb_invalid_age} patients avec âge aberrant (<0 ou >120)")
        
        # 2. Déduplification (nom + prénom)
        if 'nom' in df_clean.columns and 'prenom' in df_clean.columns:
            nb_before = len(df_clean)
            df_clean = df_clean.drop_duplicates(subset=['nom', 'prenom'], keep='first')
            nb_removed = nb_before - len(df_clean)
            if nb_removed > 0:
                corrections.append(f" Supprimé {nb_removed} doublons (nom+prénom)")
        
        # 3. Filtrer dates de naissance futures
        if 'date_naissance' in df_clean.columns:
            mask_future = df_clean['date_naissance'] > pd.Timestamp.now()
            nb_future = mask_future.sum()
            if nb_future > 0:
                df_clean = df_clean[~mask_future].copy()
                corrections.append(f" Supprimé {nb_future} patients avec date de naissance future")
        
        return df_clean, corrections

    def _clean_sejours(self, df):
        """Nettoyage spécifique SÉJOURS"""
        df_clean = df.copy()
        corrections = []
        
        # 1. Corriger dates inversées (swap entrée/sortie)
        if 'date_entree' in df_clean.columns and 'date_sortie' in df_clean.columns:
            mask_inverted = df_clean['date_sortie'] < df_clean['date_entree']
            nb_inverted = mask_inverted.sum()
            
            if nb_inverted > 0:
                # Swap des dates
                df_clean.loc[mask_inverted, ['date_entree', 'date_sortie']] = \
                    df_clean.loc[mask_inverted, ['date_sortie', 'date_entree']].values
                
                # Recalcul durée
                if 'duree_sejour_jours' in df_clean.columns:
                    df_clean.loc[mask_inverted, 'duree_sejour_jours'] = \
                        (df_clean.loc[mask_inverted, 'date_sortie'] - 
                         df_clean.loc[mask_inverted, 'date_entree']).dt.days
                
                corrections.append(f" Inversé {nb_inverted} paires de dates (sortie < entrée)")
        
        # 2. Normaliser gravité (forcer entre 1 et 5)
        if 'gravite_1_5' in df_clean.columns:
            mask_invalid_grav = (df_clean['gravite_1_5'] < 1) | (df_clean['gravite_1_5'] > 5)
            nb_invalid_grav = mask_invalid_grav.sum()
            
            if nb_invalid_grav > 0:
                # Clip entre 1 et 5
                df_clean.loc[mask_invalid_grav, 'gravite_1_5'] = \
                    df_clean.loc[mask_invalid_grav, 'gravite_1_5'].clip(1, 5)
                corrections.append(f" Normalisé {nb_invalid_grav} gravités hors norme (1-5)")
        
        # 3. Supprimer séjours avec durée négative (même après correction dates)
        if 'duree_sejour_jours' in df_clean.columns:
            mask_negative_duration = df_clean['duree_sejour_jours'] < 0
            nb_negative = mask_negative_duration.sum()
            
            if nb_negative > 0:
                df_clean = df_clean[~mask_negative_duration].copy()
                corrections.append(f"Supprimé {nb_negative} séjours avec durée négative")
        
        return df_clean, corrections

    def _clean_ressources(self, df):
        """Nettoyage spécifique RESSOURCES"""
        df_clean = df.copy()
        corrections = []
        
        # 1. Corriger suroccupation (lits_occupes > lits_total)
        if 'lits_occupes' in df_clean.columns and 'lits_total' in df_clean.columns:
            # Conversion sécurisée
            df_clean['lits_occupes'] = pd.to_numeric(df_clean['lits_occupes'], errors='coerce').fillna(0)
            
            mask_suroccup = df_clean['lits_occupes'] > df_clean['lits_total']
            nb_suroccup = mask_suroccup.sum()
            
            if nb_suroccup > 0:
                # Plafonner à lits_total
                df_clean.loc[mask_suroccup, 'lits_occupes'] = df_clean.loc[mask_suroccup, 'lits_total']
                
                # Recalculer taux_occupation
                if 'taux_occupation' in df_clean.columns:
                    df_clean.loc[mask_suroccup, 'taux_occupation'] = 1.0
                
                corrections.append(f" Corrigé {nb_suroccup} cas de suroccupation (plafonné à capacité)")
        
        # 2. Filtrer taux d'occupation aberrants (>1.5 = 150%)
        if 'taux_occupation' in df_clean.columns:
            mask_invalid_rate = df_clean['taux_occupation'] > 1.5
            nb_invalid_rate = mask_invalid_rate.sum()
            
            if nb_invalid_rate > 0:
                df_clean = df_clean[~mask_invalid_rate].copy()
                corrections.append(f" Supprimé {nb_invalid_rate} lignes avec taux >150%")
        
        return df_clean, corrections

    def _clean_stock(self, df):
        """Nettoyage spécifique STOCK"""
        df_clean = df.copy()
        corrections = []
        
        # 1. Filtrer stocks négatifs
        if 'stock_disponible' in df_clean.columns:
            mask_negative = df_clean['stock_disponible'] < 0
            nb_negative = mask_negative.sum()
            
            if nb_negative > 0:
                df_clean.loc[mask_negative, 'stock_disponible'] = 0
                corrections.append(f"🔧 Corrigé {nb_negative} stocks négatifs (mis à 0)")
        
        return df_clean, corrections

    def _clean_dataframe(self, name, df_raw):
        """Pipeline de nettoyage complet"""
        # Étape 1 : Conversion types
        df_typed = self._convert_types(name, df_raw)
        
        # Étape 2 : Nettoyage spécifique selon la table
        name_normalized = name.lower().strip().replace(' ', '_').replace('-', '_')
        
        corrections = []
        
        if name_normalized == 'patients':
            df_clean, corr = self._clean_patients(df_typed)
            corrections.extend(corr)
        
        elif name_normalized == 'sejours':
            df_clean, corr = self._clean_sejours(df_typed)
            corrections.extend(corr)
        
        elif name_normalized == 'ressources_jour':
            df_clean, corr = self._clean_ressources(df_typed)
            corrections.extend(corr)
        
        elif name_normalized == 'stock_materiel_jour':
            df_clean, corr = self._clean_stock(df_typed)
            corrections.extend(corr)
        
        else:
            df_clean = df_typed
            corrections.append("ℹ️ Aucun nettoyage spécifique appliqué")
        
        return df_clean, corrections

    def run_etl(self):
        """Pipeline ETL complet avec stockage RAW + CLEAN"""
        data_explorer = DataExplorer()
        
        try:
            spreadsheet = self.client.open(self.sheet_name)
            worksheets = spreadsheet.worksheets()
            relevant_sheets = [ws for ws in worksheets if ws.title.lower() not in ['', 'sheet1', 'feuil1']]
            
            progress_bar = st.progress(0, text=" ETL Engine: Extraction...")
            
            for i, ws in enumerate(relevant_sheets):
                raw_data = ws.get_all_values()
                
                if len(raw_data) > 1:
                    headers = raw_data[0]
                    rows = raw_data[1:]
                    df_raw = pd.DataFrame(rows, columns=headers)
                    
                    std_name = ws.title.lower().strip().replace(' ', '_').replace('-', '_').split('.')[0]
                    
                    # TRANSFORM
                    progress_bar.progress((i + 0.5) / len(relevant_sheets), 
                                        text=f"🔧 Nettoyage: {std_name}...")
                    
                    df_clean, corrections = self._clean_dataframe(std_name, df_raw)
                    
                    # Metrics
                    metrics = {
                        'corrections_applied': corrections,
                        'rows_before': len(df_raw),
                        'rows_after': len(df_clean)
                    }
                    
                    # LOAD (RAW + CLEAN)
                    data_explorer.add_dataframe(std_name, df_raw, df_clean, metrics)
                
                progress_bar.progress((i + 1) / len(relevant_sheets))
            
            data_explorer.load_timestamp = datetime.now()
            progress_bar.empty()
            
            # Stockage session
            st.session_state.data_explorer = data_explorer
            
            # Rapport
            summary = data_explorer.get_summary()
            st.success(f"✅ ETL Terminé : {summary['total_rows_clean']:,} lignes CLEAN "
                      f"({summary['total_rows_removed']:,} supprimées)")
            
            return True, data_explorer
            
        except Exception as e:
            st.error(f"❌ Erreur ETL : {str(e)}")
            import traceback
            st.code(traceback.format_exc())
            return False, str(e)

    def get_dataframe(self, table_name, version='clean'):
        """
        Récupère un dataframe depuis le session_state
        
        Args:
            table_name: Nom de la table
            version: 'raw' ou 'clean' (défaut: 'clean')
        """
        if 'data_explorer' in st.session_state:
            if version == 'raw':
                return st.session_state.data_explorer.get_raw_dataframe(table_name)
            else:
                return st.session_state.data_explorer.get_clean_dataframe(table_name)
        return None

    def get_all_tables(self):
        """Retourne la liste de toutes les tables disponibles"""
        if 'data_explorer' in st.session_state:
            return list(st.session_state.data_explorer.dataframes_clean.keys())
        return []