Introduction : Pourquoi automatiser vos flux CSV avec l'IA ?
Chaque entreprise manipule aujourd'hui des montagnes de données CSV non structurées : exports CRM, fichiers logs, rapports Excel bruts. Le problème ? Ces données nécessitent un nettoyage fastidieux avant toute utilisation. classification automatique, extraction d'entités, détection d'anomalies — autant de tâches qui prennent des heures en traitement manuel.
HolySheep AI propose une alternative aux API OpenAI et Anthropic avec une latence inférieure à 50 millisecondes et des tarifs jusqu'à 85% inférieurs. Découvrez comment construire un pipeline ETL complet qui автоматически nettoie, transforme et enrichit vos fichiers CSV grâce à l'intelligence artificielle.
Architecture du pipeline Tardis ETL
Notre architecture s'articule autour de 4 modules distincts :
- TardisExtractor : Lecture et parsing des fichiers CSV avec gestion des caractères spéciaux
- TardisCleaner : Normalisation, dédoublonnage, gestion des valeurs nulles
- TardisTransformer : Enrichissement via API HolySheep AI (classification, extraction)
- TardisLoader : Insertion en base de données avec gestion des transactions
Installation et dépendances
# Installation des dépendances
pip install pandas openai python-dotenv psycopg2-binary
pip install sqlalchemy pydantic httpx
Configuration de l'environnement
cat > .env << 'EOF'
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
HOLYSHEEP_BASE_URL=https://api.holysheep.ai/v1
DATABASE_URL=postgresql://user:pass@localhost:5432/tardis_db
EOFModule 1 : Extracteur CSV haute performance
Notre extracteur gère les fichiers volumineux par chunking pour éviter les problèmes de mémoire. Il supporte l'encodage UTF-8, Latin-1 et détecte automatiquement le délimiteur.
import pandas as pd
import chardet
from pathlib import Path
from typing import Iterator, Optional
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class TardisExtractor:
"""Extracteur CSV avec détection automatique d'encodage et chunking."""
def __init__(self, chunk_size: int = 10000):
self.chunk_size = chunk_size
def detect_encoding(self, file_path: str) -> str:
"""Détecte l'encodage du fichier."""
with open(file_path, 'rb') as f:
raw_data = f.read(100000)
result = chardet.detect(raw_data)
logger.info(f"Encodage détecté: {result}")
return result['encoding'] or 'utf-8'
def extract(self, file_path: str,
delimiter: Optional[str] = None,
encoding: Optional[str] = None) -> Iterator[pd.DataFrame]:
"""
Extrait les données CSV par chunks pour les fichiers volumineux.
Args:
file_path: Chemin vers le fichier CSV
delimiter: Délimiteur (détection auto si None)
encoding: Encodage (détection auto si None)
Yields:
DataFrame par chunks de taille chunk_size
"""
if encoding is None:
encoding = self.detect_encoding(file_path)
# Lecture en chunks pour éviter les MemoryError
logger.info(f"Début extraction: {file_path}")
try:
for chunk_idx, chunk in enumerate(pd.read_csv(
file_path,
chunksize=self.chunk_size,
encoding=encoding,
delimiter=delimiter,
on_bad_lines='skip', # Ignore les lignes corrompues
low_memory=False
)):
logger.info(f"Chunk {chunk_idx}: {len(chunk)} lignes traitées")
yield chunk
except Exception as e:
logger.error(f"Erreur extraction: {e}")
raise
Utilisation
extractor = TardisExtractor(chunk_size=5000)
for chunk in extractor.extract("data/clients_export.csv"):
# Traitement de chaque chunk
print(f"Traitement de {len(chunk)} lignes...")Module 2 : Nettoyeur de données intelligent
Le module de nettoyage applique des règles métier : suppression des doublons, normalisation des chaînes, conversion des types, gestion des valeurs manquantes avec stratégie d'imputation.
import pandas as pd
import numpy as np
from pydantic import BaseModel, Field, validator
from typing import Dict, Any, List, Optional
from datetime import datetime
import re
class TardisCleaner:
"""Nettoyeur de données avec validation et transformation."""
def __init__(self,
drop_duplicates: bool = True,
strip_whitespace: bool = True,
lowercase_columns: bool = True):
self.drop_duplicates = drop_duplicates
self.strip_whitespace = strip_whitespace
self.lowercase_columns = lowercase_columns
def clean(self, df: pd.DataFrame) -> pd.DataFrame:
"""Applique toutes les règles de nettoyage."""
df = df.copy()
# Normalisation des noms de colonnes
if self.lowercase_columns:
df.columns = [col.lower().strip().replace(' ', '_')
for col in df.columns]
# Suppression des espaces superflus
if self.strip_whitespace:
df = df.apply(lambda x: x.str.strip() if x.dtype == "object" else x)
# Suppression des doublons
if self.drop_duplicates:
before = len(df)
df = df.drop_duplicates()
logger.info(f"Doublons supprimés: {before - len(df)}")
# Conversion des types dates
date_columns = [col for col in df.columns if 'date' in col.lower()]
for col in date_columns:
df[col] = pd.to_datetime(df[col], errors='coerce')
# Remplacement des valeurs nulles vides par NaN
df = df.replace(r'^\s*$', np.nan, regex=True)
return df
def impute_numeric(self, df: pd.DataFrame,
strategy: str = 'median') -> pd.DataFrame:
"""Impute les valeurs numériques manquantes."""
numeric_cols = df.select_dtypes(include=[np.number]).columns
for col in numeric_cols:
if df[col].isna().sum() > 0:
if strategy == 'median':
df[col].fillna(df[col].median(), inplace=True)
elif strategy == 'mean':
df[col].fillna(df[col].mean(), inplace=True)
elif strategy == 'zero':
df[col].fillna(0, inplace=True)
return df
def clean_text_field(self, text: str) -> str:
"""Nettoie un champ texte individuel."""
if pd.isna(text):
return ""
text = str(text).strip()
text = re.sub(r'\s+', ' ', text) # Esp