Spaces:
Sleeping
Sleeping
| """ | |
| 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 | |
| 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 [] |