""" 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 []