data_loader.py

"""
data_loader.py - Version Hybride Optimale
Module de chargement et d'exportation de données
Combine simplicité et fonctionnalités avancées
"""

import pandas as pd
import numpy as np
import json
import pickle
import sqlite3
from pathlib import Path
from typing import Union, Dict, List, Tuple, Optional, Any, Generator
from datetime import datetime
import warnings
warnings.filterwarnings('ignore')

# Imports optionnels
try:
    import sqlalchemy as sa
    from sqlalchemy import create_engine
    SQLALCHEMY_AVAILABLE = True
except ImportError:
    SQLALCHEMY_AVAILABLE = False

try:
    import yaml
    YAML_AVAILABLE = True
except ImportError:
    YAML_AVAILABLE = False


class DataLoader:
    """
    Classe pour charger et exporter des données dans différents formats
    Supporte: CSV, Excel, JSON, SQL, Pickle, TXT, Parquet, Feather, HDF5, XML
    
    Exemples:
    ---------
    >>> loader = DataLoader()
    >>> # Chargement simple
    >>> df = loader.load_data("data.csv")
    >>> # Chargement avec validation
    >>> df, report = loader.load_and_validate("data.csv", required_columns=["id", "name"])
    >>> # Chargement par chunks (gros fichiers)
    >>> for chunk in loader.load_data_chunked("big_data.csv", chunksize=10000):
    >>>     process(chunk)
    >>> # Export
    >>> loader.export_data(df, "output.xlsx")
    """
    
    def __init__(self, config: Dict = None, verbose: bool = True, max_file_size_mb: int = 1000):
        """
        Initialise le DataLoader
        
        Args:
            config: Dictionnaire de configuration (optionnel)
            verbose: Afficher les messages d'info
            max_file_size_mb: Taille maximale des fichiers en MB
        """
        self.data = None
        self.file_path = None
        self.file_type = None
        self.metadata = {}
        self.verbose = verbose
        
        # Formats supportés
        self.supported_formats = [
            '.csv', '.xlsx', '.xls', '.json', '.sql', '.db', 
            '.sqlite', '.pkl', '.pickle', '.txt', '.parquet',
            '.feather', '.h5', '.hdf5', '.xml'
        ]
        
        # Configuration par défaut
        self.config = {
            'default_encoding': 'utf-8',
            'chunk_size': 10000,
            'na_values': ['', 'NA', 'N/A', 'nan', 'NaN', 'null', 'NULL', '?', '-'],
            'csv_default_sep': ',',
            'excel_default_sheet': 0,
            'json_default_orient': 'records',
            'max_file_size_mb': max_file_size_mb,
            'parse_dates': True
        }
        
        # Mise à jour avec config utilisateur
        if config:
            self.config.update(config)
    
    def _log(self, message: str, level: str = 'INFO'):
        """Affiche un message si verbose est activé"""
        if self.verbose:
            prefix = {'INFO': '✓', 'WARNING': '⚠', 'ERROR': '✗', 'DEBUG': '→'}
            print(f"{prefix.get(level, '•')} {message}")
    
    def load_data(self, file_path: str, validate: bool = False, 
                  required_columns: List[str] = None, **kwargs) -> pd.DataFrame:
        """
        Charge les données depuis un fichier
        
        Args:
            file_path: Chemin vers le fichier
            validate: Valider les données après chargement
            required_columns: Colonnes requises (si validate=True)
            **kwargs: Arguments supplémentaires
            
        Returns:
            DataFrame contenant les données
        """
        start_time = datetime.now()
        self._log(f"Chargement de: {file_path}")
        
        self.file_path = Path(file_path)
        
        # Vérifications
        if not self.file_path.exists():
            raise FileNotFoundError(f"Le fichier {file_path} n'existe pas")
        
        # Vérifier la taille
        file_size_mb = self.file_path.stat().st_size / (1024 * 1024)
        if file_size_mb > self.config['max_file_size_mb']:
            self._log(f"Fichier volumineux: {file_size_mb:.2f} MB", 'WARNING')
            if not kwargs.get('force_load', False):
                raise ValueError(
                    f"Fichier trop gros ({file_size_mb:.2f} MB). "
                    f"Utilisez load_data_chunked() ou force_load=True"
                )
        
        self.file_type = self.file_path.suffix.lower()
        
        if self.file_type not in self.supported_formats:
            raise ValueError(f"Format non supporté: {self.file_type}")
        
        try:
            # Chargement selon le type
            if self.file_type == '.csv':
                self.data = self._load_csv(**kwargs)
            elif self.file_type in ['.xlsx', '.xls']:
                self.data = self._load_excel(**kwargs)
            elif self.file_type == '.json':
                self.data = self._load_json(**kwargs)
            elif self.file_type in ['.sql', '.db', '.sqlite']:
                self.data = self._load_sql(**kwargs)
            elif self.file_type in ['.pkl', '.pickle']:
                self.data = self._load_pickle(**kwargs)
            elif self.file_type == '.txt':
                self.data = self._load_txt(**kwargs)
            elif self.file_type == '.parquet':
                self.data = self._load_parquet(**kwargs)
            elif self.file_type == '.feather':
                self.data = self._load_feather(**kwargs)
            elif self.file_type in ['.h5', '.hdf5']:
                self.data = self._load_hdf(**kwargs)
            elif self.file_type == '.xml':
                self.data = self._load_xml(**kwargs)
            
            elapsed = (datetime.now() - start_time).total_seconds()
            self._log(f"Chargé en {elapsed:.2f}s - {self.data.shape[0]} lignes, "
                     f"{self.data.shape[1]} colonnes")
            
            # Métadonnées
            self.metadata = {
                'load_time': elapsed,
                'file_size_mb': file_size_mb,
                'timestamp': datetime.now().isoformat()
            }
            
            # Validation optionnelle
            if validate:
                is_valid, msg, _ = self.validate_data(required_columns=required_columns)
                if not is_valid:
                    self._log(f"Validation: {msg}", 'WARNING')
            
            return self.data
            
        except Exception as e:
            self._log(f"Erreur: {str(e)}", 'ERROR')
            raise
    
    def _load_csv(self, **kwargs) -> pd.DataFrame:
        """Charge un fichier CSV avec détection automatique"""
        default_params = {
            'encoding': self.config['default_encoding'],
            'sep': self.config['csv_default_sep'],
            'na_values': self.config['na_values'],
            'parse_dates': self.config['parse_dates']
        }
        default_params.update(kwargs)
        
        # Auto-détection du séparateur
        if 'sep' not in kwargs:
            detected_sep = self.auto_detect_separator(self.file_path)
            if detected_sep:
                default_params['sep'] = detected_sep
        
        try:
            return pd.read_csv(self.file_path, **default_params)
        except UnicodeDecodeError:
            for encoding in ['latin1', 'iso-8859-1', 'cp1252']:
                try:
                    default_params['encoding'] = encoding
                    return pd.read_csv(self.file_path, **default_params)
                except:
                    continue
            raise ValueError("Impossible de décoder le fichier CSV")
    
    def _load_excel(self, **kwargs) -> pd.DataFrame:
        """Charge un fichier Excel"""
        default_params = {
            'sheet_name': self.config['excel_default_sheet'],
            'na_values': self.config['na_values']
        }
        default_params.update(kwargs)
        return pd.read_excel(self.file_path, **default_params)
    
    def _load_json(self, **kwargs) -> pd.DataFrame:
        """Charge un fichier JSON"""
        default_params = {
            'orient': self.config['json_default_orient'],
            'encoding': self.config['default_encoding']
        }
        default_params.update(kwargs)
        
        try:
            return pd.read_json(self.file_path, **default_params)
        except ValueError:
            with open(self.file_path, 'r', encoding=default_params['encoding']) as f:
                data = json.load(f)
            return pd.DataFrame(data) if isinstance(data, list) else pd.DataFrame([data])
    
    def _load_sql(self, table_name: str = None, query: str = None, 
                  connection_string: str = None, **kwargs) -> pd.DataFrame:
        """Charge depuis une base SQL"""
        if self.file_type in ['.sql', '.db', '.sqlite']:
            conn = sqlite3.connect(self.file_path)
            try:
                if query:
                    return pd.read_sql_query(query, conn)
                elif table_name:
                    return pd.read_sql_table(table_name, conn)
                else:
                    cursor = conn.cursor()
                    cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
                    tables = cursor.fetchall()
                    if tables:
                        table_name = tables[0][0]
                        self._log(f"Table chargée: {table_name}")
                        return pd.read_sql_table(table_name, conn)
                    raise ValueError("Aucune table trouvée")
            finally:
                conn.close()
        
        elif connection_string and SQLALCHEMY_AVAILABLE:
            engine = create_engine(connection_string)
            with engine.connect() as conn:
                if query:
                    return pd.read_sql_query(query, conn)
                elif table_name:
                    return pd.read_sql_table(table_name, conn)
        else:
            raise ValueError("Fournissez connection_string ou installez sqlalchemy")
    
    def _load_pickle(self, **kwargs) -> pd.DataFrame:
        """Charge un fichier Pickle"""
        with open(self.file_path, 'rb') as f:
            data = pickle.load(f)
        
        if isinstance(data, pd.DataFrame):
            return data
        elif isinstance(data, (list, dict, np.ndarray)):
            return pd.DataFrame(data)
        else:
            raise ValueError(f"Type pickle non compatible: {type(data)}")
    
    def _load_txt(self, **kwargs) -> pd.DataFrame:
        """Charge un fichier texte"""
        default_params = {
            'sep': '\t',
            'encoding': self.config['default_encoding'],
            'na_values': self.config['na_values']
        }
        default_params.update(kwargs)
        
        try:
            return pd.read_csv(self.file_path, **default_params)
        except:
            for sep in [';', '|', ',', ' ']:
                try:
                    default_params['sep'] = sep
                    return pd.read_csv(self.file_path, **default_params)
                except:
                    continue
            raise ValueError("Impossible de lire le fichier texte")
    
    def _load_parquet(self, **kwargs) -> pd.DataFrame:
        """Charge un fichier Parquet"""
        return pd.read_parquet(self.file_path, **kwargs)
    
    def _load_feather(self, **kwargs) -> pd.DataFrame:
        """Charge un fichier Feather"""
        return pd.read_feather(self.file_path, **kwargs)
    
    def _load_hdf(self, **kwargs) -> pd.DataFrame:
        """Charge un fichier HDF5"""
        default_params = {'key': kwargs.get('key', 'data')}
        default_params.update(kwargs)
        
        try:
            return pd.read_hdf(self.file_path, **default_params)
        except KeyError:
            with pd.HDFStore(self.file_path, mode='r') as store:
                keys = store.keys()
                if keys:
                    return pd.read_hdf(self.file_path, key=keys[0])
            raise ValueError("Aucune clé dans HDF5")
    
    def _load_xml(self, **kwargs) -> pd.DataFrame:
        """Charge un fichier XML"""
        return pd.read_xml(self.file_path, **kwargs)
    
    def load_and_validate(self, file_path: str, required_columns: List[str] = None,
                         min_rows: int = 1, **kwargs) -> Tuple[pd.DataFrame, Dict]:
        """
        Charge et valide les données en une seule étape
        
        Returns:
            Tuple (DataFrame, rapport de validation)
        """
        df = self.load_data(file_path, **kwargs)
        is_valid, message, report = self.validate_data(df, required_columns, min_rows)
        
        if not is_valid:
            self._log(f"Validation échouée: {message}", 'WARNING')
        
        return df, report
    
    def load_data_chunked(self, file_path: str, chunksize: int = None, 
                         **kwargs) -> Generator[pd.DataFrame, None, None]:
        """
        Charge les données par morceaux (pour gros fichiers)
        
        Yields:
            Chunks de DataFrame
        """
        if chunksize is None:
            chunksize = self.config['chunk_size']
        
        self.file_path = Path(file_path)
        self.file_type = self.file_path.suffix.lower()
        
        if self.file_type not in ['.csv', '.txt']:
            raise ValueError(f"Chargement par chunks non supporté pour {self.file_type}")
        
        self._log(f"Chargement par chunks: {chunksize} lignes/chunk")
        
        chunk_iterator = pd.read_csv(
            self.file_path,
            chunksize=chunksize,
            encoding=self.config['default_encoding'],
            **kwargs
        )
        
        for i, chunk in enumerate(chunk_iterator):
            if self.verbose and i % 10 == 0:
                self._log(f"Chunk {i + 1} traité", 'DEBUG')
            yield chunk
    
    def validate_data(self, data: pd.DataFrame = None, 
                     required_columns: List[str] = None,
                     min_rows: int = 1) -> Tuple[bool, str, Dict]:
        """
        Valide les données
        
        Returns:
            Tuple (succès, message, rapport détaillé)
        """
        if data is None:
            data = self.data
        
        report = {
            'timestamp': datetime.now().isoformat(),
            'is_valid': False,
            'errors': [],
            'warnings': [],
            'summary': {}
        }
        
        if data is None:
            report['errors'].append("Aucune donnée")
            return False, "Aucune donnée", report
        
        # Résumé
        report['summary'] = {
            'rows': len(data),
            'columns': len(data.columns),
            'memory_mb': f"{data.memory_usage(deep=True).sum() / 1024**2:.2f}"
        }
        
        # Vérifications
        if len(data) < min_rows:
            report['errors'].append(f"Lignes insuffisantes ({len(data)} < {min_rows})")
        
        if required_columns:
            missing = [col for col in required_columns if col not in data.columns]
            if missing:
                report['errors'].append(f"Colonnes manquantes: {missing}")
        
        # Doublons
        duplicates = data.duplicated().sum()
        if duplicates > 0:
            report['warnings'].append(f"{duplicates} doublons détectés")
        
        # Valeurs manquantes
        total_missing = data.isnull().sum().sum()
        if total_missing > 0:
            report['warnings'].append(f"{total_missing} valeurs manquantes")
            report['summary']['missing_values'] = int(total_missing)
        
        report['is_valid'] = len(report['errors']) == 0
        message = "Valide" if report['is_valid'] else ', '.join(report['errors'])
        
        return report['is_valid'], message, report
    
    def export_data(self, data: pd.DataFrame, output_path: str, 
                   overwrite: bool = False, **kwargs) -> bool:
        """
        Exporte les données vers un fichier
        
        Returns:
            True si succès
        """
        output_path = Path(output_path)
        file_type = output_path.suffix.lower()
        
        if output_path.exists() and not overwrite:
            self._log(f"Fichier existe déjà. Utilisez overwrite=True", 'ERROR')
            return False
        
        try:
            output_path.parent.mkdir(parents=True, exist_ok=True)
            
            if file_type == '.csv':
                data.to_csv(output_path, index=False, encoding=self.config['default_encoding'], **kwargs)
            elif file_type in ['.xlsx', '.xls']:
                data.to_excel(output_path, index=False, sheet_name='Data', **kwargs)
            elif file_type == '.json':
                data.to_json(output_path, orient=self.config['json_default_orient'], 
                           indent=2, force_ascii=False, **kwargs)
            elif file_type in ['.pkl', '.pickle']:
                with open(output_path, 'wb') as f:
                    pickle.dump(data, f, protocol=pickle.HIGHEST_PROTOCOL)
            elif file_type == '.parquet':
                data.to_parquet(output_path, index=False, **kwargs)
            elif file_type == '.feather':
                data.to_feather(output_path, **kwargs)
            elif file_type in ['.h5', '.hdf5']:
                data.to_hdf(output_path, key='data', mode='w', **kwargs)
            elif file_type in ['.db', '.sqlite']:
                with sqlite3.connect(output_path) as conn:
                    data.to_sql('data', conn, if_exists='replace', index=False)
            else:
                raise ValueError(f"Format d'export non supporté: {file_type}")
            
            self._log(f"Exporté vers: {output_path}")
            return True
            
        except Exception as e:
            self._log(f"Erreur export: {str(e)}", 'ERROR')
            return False
    
    def get_data_info(self, detailed: bool = False) -> Dict:
        """Retourne des informations sur les données"""
        if self.data is None:
            return {"error": "Aucune donnée chargée"}
        
        info = {
            "shape": self.data.shape,
            "columns": list(self.data.columns),
            "dtypes": self.data.dtypes.astype(str).to_dict(),
            "missing_values": int(self.data.isnull().sum().sum()),
            "memory_mb": f"{self.data.memory_usage(deep=True).sum() / 1024**2:.2f}",
            "numeric_columns": list(self.data.select_dtypes(include=[np.number]).columns),
            "categorical_columns": list(self.data.select_dtypes(include=['object']).columns),
            "metadata": self.metadata
        }
        
        if detailed:
            info["describe"] = self.data.describe(include='all').to_dict()
            
            # Top values pour catégories
            cat_info = {}
            for col in info["categorical_columns"][:10]:  # Limiter à 10
                unique = self.data[col].nunique()
                if unique < 20:
                    cat_info[col] = {
                        "unique_count": unique,
                        "top_values": self.data[col].value_counts().head(5).to_dict()
                    }
            info["categorical_details"] = cat_info
        
        return info
    
    def preview_data(self, n_rows: int = 5) -> pd.DataFrame:
        """Aperçu des données"""
        return self.data.head(n_rows) if self.data is not None else None
    
    # Utilitaires
    def list_excel_sheets(self, file_path: str) -> List[str]:
        """Liste les feuilles Excel"""
        return pd.ExcelFile(file_path).sheet_names
    
    def list_sql_tables(self, db_path: str) -> List[str]:
        """Liste les tables SQL"""
        conn = sqlite3.connect(db_path)
        cursor = conn.cursor()
        cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
        tables = [t[0] for t in cursor.fetchall()]
        conn.close()
        return tables
    
    def auto_detect_separator(self, file_path: str, n_lines: int = 5) -> str:
        """Détecte le séparateur CSV"""
        separators = [',', ';', '\t', '|']
        max_cols = 0
        best_sep = ','
        
        with open(file_path, 'r', encoding=self.config['default_encoding']) as f:
            lines = [f.readline() for _ in range(min(n_lines, 5))]
        
        for sep in separators:
            cols = len(lines[0].split(sep))
            if cols > max_cols:
                max_cols = cols
                best_sep = sep
        
        return best_sep


# Fonction utilitaire
def create_sample_datasets() -> Tuple[pd.DataFrame, pd.DataFrame]:
    """Crée des datasets d'exemple"""
    # Dataset Iris
    iris = pd.DataFrame({
        'sepal_length': np.random.uniform(4.5, 7.5, 150),
        'sepal_width': np.random.uniform(2.0, 4.5, 150),
        'petal_length': np.random.uniform(1.0, 7.0, 150),
        'petal_width': np.random.uniform(0.1, 2.5, 150),
        'species': np.random.choice(['setosa', 'versicolor', 'virginica'], 150)
    })
    
    # Dataset Housing
    housing = pd.DataFrame({
        'surface': np.random.uniform(50, 200, 200),
        'rooms': np.random.randint(1, 6, 200),
        'age': np.random.randint(0, 50, 200),
        'distance': np.random.uniform(1, 30, 200),
        'price': np.random.uniform(100000, 500000, 200)
    })
    
    return iris, housing


# Exemple d'utilisation
if __name__ == "__main__":
    loader = DataLoader(verbose=True)
    
    # Créer des exemples
    print("\n📦 Création de datasets d'exemple...")
    iris, housing = create_sample_datasets()
    loader.export_data(iris, "iris_sample.csv")
    loader.export_data(housing, "housing_sample.xlsx")
    
    # Charger et valider
    print("\n📊 Chargement avec validation...")
    df, report = loader.load_and_validate("iris_sample.csv", 
                                          required_columns=['species'])
    
    print("\n📈 Informations:")
    print(loader.get_data_info())
    
    print("\n👁️ Aperçu:")
    print(loader.preview_data())