Update src/Function/Validation_Contract.py

src/Function/Validation_Contract.py

@@ -2,105 +2,341 @@
 MODULE: VALIDATION CONTRACT (ETL ENGINE)
 ========================================
 Responsabilité : Centraliser le schéma, extraire et nettoyer les données.
+Pipeline: Extract → Transform → Validate → Load
 """
 import pandas as pd
 import numpy as np
 import streamlit as st
+from datetime import datetime
+
+
+class DataExplorer:
+    """Conteneur pour stocker les dataframes et métadonnées d'exploration"""
+    
+    def __init__(self):
+        self.dataframes = {}
+        self.metadata = {}
+        self.load_timestamp = None
+        
+    def add_dataframe(self, name, df, rows_before=None, rows_after=None):
+        """Ajoute un dataframe avec ses métadonnées"""
+        self.dataframes[name] = df
+        self.metadata[name] = {
+            'rows': len(df),
+            'columns': len(df.columns),
+            'rows_before_cleaning': rows_before,
+            'rows_after_cleaning': rows_after,
+            'memory_usage': df.memory_usage(deep=True).sum() / 1024**2,  # MB
+            'dtypes': df.dtypes.to_dict(),
+            'null_counts': df.isnull().sum().to_dict(),
+            'numeric_columns': df.select_dtypes(include=[np.number]).columns.tolist(),
+            'categorical_columns': df.select_dtypes(include=['object']).columns.tolist(),
+            'datetime_columns': df.select_dtypes(include=['datetime64']).columns.tolist()
+        }
+    
+    def get_summary(self):
+        """Retourne un résumé global des données chargées"""
+        total_rows = sum(meta['rows'] for meta in self.metadata.values())
+        total_memory = sum(meta['memory_usage'] for meta in self.metadata.values())
+        
+        return {
+            'tables_loaded': len(self.dataframes),
+            'total_rows': total_rows,
+            'total_memory_mb': round(total_memory, 2),
+            'timestamp': self.load_timestamp
+        }
+
 
 class ETL_Engine:
-    """Moteur ETL garantissant le Contrat de Données"""
+    """Moteur ETL garantissant le Contrat de Données pour l'EDA"""
 
     def __init__(self, gspread_client, sheet_name="bdd_mvp_faker"):
         self.client = gspread_client
         self.sheet_name = sheet_name
         
         # --- SCHEMA CONTRACTUEL ---
-        # Définit ce qui DOIT être un nombre ou une date
+        # Définit précisément ce qui DOIT être un nombre ou une date
+        # Basé sur l'analyse réelle du fichier bdd_mvp_faker.xlsx
         self.SCHEMA = {
-            'ressources': {
-                'numeric': ['lits_total', 'lits_occupes', 'taux_occupation', 'personnel_total', 'tension_personnel'],
-                'date': ['date']
+            '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']
             },
-            'stock': {
-                'numeric': ['stock_total', 'stock_disponible', 'tension_stock'],
-                'date': ['date']
+            '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': {
+            'admissions_jour': {
                 'numeric': ['nb_admissions', 'intensity'],
-                'date': ['date']
+                'date': ['date'],
+                'categorical': ['service_code', 'scenario']
             },
-            'patients': {
-                'numeric': ['age'],
-                'date': ['date_naissance']
+            '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']
             },
-            'sejours': {
-                'numeric': ['duree_sejour_jours', 'gravite_1_5'],
-                'date': ['date_entree', 'date_sortie']
+            'scenarios_calendrier': {
+                'numeric': ['intensity'],
+                'date': ['date'],
+                'categorical': ['scenario']
             }
         }
+        
+        # Tables de référence (pas de nettoyage, juste chargement)
+        self.REFERENCE_TABLES = [
+            'service_ref', 'pathologie_ref', 'materiel_ref', 
+            'mode_entree_ref', 'type_lit_ref', 'calendrier_conges', 'hypotheses'
+        ]
 
     def _clean_dataframe(self, name, df_raw):
-        """Transform: Applique le schéma et nettoie les types"""
+        """
+        Transform: Applique le schéma et nettoie les types
+        
+        Args:
+            name: Nom de la feuille/table
+            df_raw: DataFrame brut
+            
+        Returns:
+            df_clean: DataFrame nettoyé selon le schéma
+        """
         df = df_raw.copy()
         
-        # Identification du type de table basé sur le nom
-        config_type = next((key for key in self.SCHEMA if key in name), None)
+        # Normalisation du nom
+        name_normalized = name.lower().strip().replace(' ', '_').replace('-', '_')
         
-        if config_type:
-            config = self.SCHEMA[config_type]
-            
-            # 1. Dates : Coercition stricte (les erreurs deviennent NaT)
-            for col in config.get('date', []):
+        # Si c'est une table de référence, on la retourne telle quelle
+        if name_normalized in self.REFERENCE_TABLES:
+            return df
+        
+        # Recherche de la configuration correspondante
+        config = self.SCHEMA.get(name_normalized)
+        
+        if not config:
+            # Si pas de config, nettoyage générique minimal
+            return self._generic_cleaning(df)
+        
+        # --- 1. CONVERSION DES DATES ---
+        for col in config.get('date', []):
+            if col in df.columns:
+                # Conversion stricte : les erreurs deviennent NaT
+                df[col] = pd.to_datetime(df[col], errors='coerce')
+        
+        # --- 2. CONVERSION DES NUMÉRIQUES ---
+        for col in config.get('numeric', []):
+            if col in df.columns:
+                # Gestion des formats français (virgules) et texte
+                if df[col].dtype == 'object':
+                    # Remplacer virgules par points
+                    df[col] = df[col].astype(str).str.replace(',', '.').str.replace(' ', '')
+                    # Gérer les cas comme "Full", "Erreur", etc.
+                    df[col] = df[col].replace(['Full', 'Erreur', 'full', 'erreur', ''], np.nan)
+                
+                # Conversion numérique stricte
+                df[col] = pd.to_numeric(df[col], errors='coerce')
+        
+        # --- 3. CATÉGORIELLES : Nettoyage basique ---
+        for col in config.get('categorical', []):
+            if col in df.columns and df[col].dtype == 'object':
+                # Strip des espaces, uniformisation
+                df[col] = df[col].astype(str).str.strip()
+                # Remplacer 'nan' string par vrai NaN
+                df[col] = df[col].replace(['nan', 'NaN', 'None', ''], np.nan)
+        
+        # --- 4. COLONNES DQ : Préserver telles quelles ---
+        # Les colonnes dq_* ne sont pas modifiées car elles contiennent
+        # les informations d'anomalies du dataset
+        
+        return df
+
+    def _generic_cleaning(self, df):
+        """Nettoyage générique pour les tables sans schéma défini"""
+        df_clean = df.copy()
+        
+        # Détection et conversion automatique des dates
+        for col in df_clean.columns:
+            if 'date' in col.lower():
+                df_clean[col] = pd.to_datetime(df_clean[col], errors='coerce')
+        
+        # Nettoyage des strings
+        for col in df_clean.select_dtypes(include=['object']).columns:
+            df_clean[col] = df_clean[col].astype(str).str.strip()
+            df_clean[col] = df_clean[col].replace(['nan', 'NaN', 'None', ''], np.nan)
+        
+        return df_clean
+
+    def _validate_dataframe(self, name, df):
+        """
+        Validate: Vérifie la cohérence post-nettoyage
+        
+        Returns:
+            dict: Rapport de validation avec warnings si nécessaire
+        """
+        validation_report = {
+            'table': name,
+            'valid': True,
+            'warnings': [],
+            'errors': []
+        }
+        
+        # Vérification 1: Table vide
+        if len(df) == 0:
+            validation_report['errors'].append(f"Table {name} est vide après nettoyage")
+            validation_report['valid'] = False
+            return validation_report
+        
+        # Vérification 2: Perte massive de données (>50% de NaN sur colonnes critiques)
+        name_normalized = name.lower().strip().replace(' ', '_')
+        config = self.SCHEMA.get(name_normalized)
+        
+        if config:
+            # Vérifier les colonnes numériques critiques
+            for col in config.get('numeric', []):
                 if col in df.columns:
-                    df[col] = pd.to_datetime(df[col], errors='coerce')
+                    null_pct = df[col].isnull().sum() / len(df) * 100
+                    if null_pct > 50:
+                        validation_report['warnings'].append(
+                            f"Colonne {col}: {null_pct:.1f}% de valeurs manquantes"
+                        )
             
-            # 2. Nombres : Coercition stricte (les 'Full', 'Erreur' deviennent NaN)
-            for col in config.get('numeric', []):
+            # Vérifier les colonnes dates
+            for col in config.get('date', []):
                 if col in df.columns:
-                    # Gestion des virgules françaises
-                    if df[col].dtype == 'object':
-                        df[col] = df[col].astype(str).str.replace(',', '.')
-                    df[col] = pd.to_numeric(df[col], errors='coerce')
-                    
-        # Nettoyage générique dates si non spécifié
-        else:
-            for col in df.columns:
-                if 'date' in col.lower():
-                    df[col] = pd.to_datetime(df[col], errors='coerce')
+                    null_pct = df[col].isnull().sum() / len(df) * 100
+                    if null_pct > 50:
+                        validation_report['warnings'].append(
+                            f"Colonne date {col}: {null_pct:.1f}% de valeurs manquantes"
+                        )
         
-        return df
+        return validation_report
 
     def run_etl(self):
-        """Pipeline complet : Extract -> Transform -> Load"""
-        data_store = {}
+        """
+        Pipeline complet : Extract → Transform → Validate → Load
+        
+        Returns:
+            tuple: (success: bool, result: DataExplorer ou str d'erreur)
+        """
+        data_explorer = DataExplorer()
+        validation_reports = []
+        
         try:
+            # EXTRACT: Connexion au spreadsheet
             spreadsheet = self.client.open(self.sheet_name)
             worksheets = spreadsheet.worksheets()
             
-            # Barre de progression interne au module
-            progress_bar = st.progress(0, text="ETL Engine: Extraction des données...")
+            # Filtrer les feuilles pertinentes pour l'EDA
+            # On exclut les feuilles vides ou non pertinentes
+            relevant_sheets = [ws for ws in worksheets 
+                             if ws.title.lower() not in ['', 'sheet1', 'feuil1']]
             
-            for i, ws in enumerate(worksheets):
+            # Barre de progression
+            progress_bar = st.progress(0, text="🔄 ETL Engine: Extraction des données...")
+            
+            for i, ws in enumerate(relevant_sheets):
                 # EXTRACT
                 raw_data = ws.get_all_values()
+                
+                # Vérifier qu'il y a au moins des headers + 1 ligne
                 if len(raw_data) > 1:
                     headers = raw_data[0]
                     rows = raw_data[1:]
                     df_raw = pd.DataFrame(rows, columns=headers)
                     
+                    rows_before = len(df_raw)
+                    
                     # Nom standardisé
-                    std_name = ws.title.lower().strip().replace(' ', '_').replace('-', '_').split('.')[0]
+                    std_name = ws.title.lower().strip().replace(' ', '_').replace('-', '_')
                     
                     # TRANSFORM
+                    progress_bar.progress((i + 0.5) / len(relevant_sheets), 
+                                        text=f"🔧 Nettoyage: {std_name}...")
                     df_clean = self._clean_dataframe(std_name, df_raw)
                     
-                    # LOAD
-                    data_store[std_name] = df_clean
+                    rows_after = len(df_clean)
                     
-                progress_bar.progress((i + 1) / len(worksheets))
+                    # VALIDATE
+                    validation = self._validate_dataframe(std_name, df_clean)
+                    validation_reports.append(validation)
+                    
+                    # LOAD dans le DataExplorer
+                    data_explorer.add_dataframe(std_name, df_clean, rows_before, rows_after)
+                
+                progress_bar.progress((i + 1) / len(relevant_sheets))
+            
+            # Timestamp du chargement
+            data_explorer.load_timestamp = datetime.now()
             
             progress_bar.empty()
-            return True, data_store
+            
+            # Afficher les warnings de validation
+            has_warnings = any(report['warnings'] for report in validation_reports)
+            has_errors = any(report['errors'] for report in validation_reports)
+            
+            if has_errors:
+                st.error("⚠️ Erreurs détectées lors du nettoyage ETL")
+                for report in validation_reports:
+                    if report['errors']:
+                        for error in report['errors']:
+                            st.error(f"  • {error}")
+            
+            if has_warnings:
+                with st.expander("⚠️ Avertissements ETL", expanded=False):
+                    for report in validation_reports:
+                        if report['warnings']:
+                            st.warning(f"**Table: {report['table']}**")
+                            for warning in report['warnings']:
+                                st.write(f"  • {warning}")
+            
+            # Stocker dans session_state pour réutilisation
+            st.session_state.data_explorer = data_explorer
+            
+            # Message de succès
+            summary = data_explorer.get_summary()
+            st.success(f"✅ ETL terminé : {summary['tables_loaded']} tables chargées "
+                      f"({summary['total_rows']:,} lignes, {summary['total_memory_mb']} MB)")
+            
+            return True, data_explorer
             
         except Exception as e:
-            return False, str(e)
+            st.error(f"❌ Erreur ETL : {str(e)}")
+            return False, str(e)
+
+    def get_dataframe(self, table_name):
+        """
+        Récupère un dataframe spécifique depuis le session_state
+        
+        Args:
+            table_name: Nom de la table
+            
+        Returns:
+            DataFrame ou None
+        """
+        if 'data_explorer' in st.session_state:
+            return st.session_state.data_explorer.dataframes.get(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.keys())
+        return []