Update src/Analytics/Pipeline_Builder.py

src/Analytics/Pipeline_Builder.py

@@ -0,0 +1,83 @@
+"""
+MODULE: PIPELINE BUILDER (DATA REFINERY)
+========================================
+Responsabilité : 
+1. Standardiser et Nettoyer les données (Dates, Nombres, Strings).
+2. Appliquer les règles métier AVANT l'ingestion RDF.
+3. Garantir les "Contrats de Données" (Data Contracts).
+"""
+import re
+import pandas as pd
+from datetime import datetime
+
+class PipelineBuilder:
+    def __init__(self):
+        self.transformations_log = []
+
+    def enforce_contract(self, value, target_type):
+        """
+        Point d'entrée unique pour transformer une valeur brute 
+        en valeur conforme au schéma cible (T-Box).
+        """
+        if pd.isna(value) or str(value).strip().lower() in ['nan', 'none', '', 'null']:
+            return None
+
+        val_str = str(value).strip()
+
+        # --- ROUTAGE SELON LE TYPE CIBLE ---
+        if target_type == "xsd:decimal" or target_type == "xsd:float":
+            return self._clean_currency(val_str)
+        
+        elif target_type == "xsd:integer":
+            return self._clean_integer(val_str)
+        
+        elif target_type == "xsd:date":
+            return self._standardize_date(val_str)
+        
+        elif target_type == "xsd:boolean":
+            return self._to_boolean(val_str)
+            
+        # Par défaut : Nettoyage de texte standard
+        return self._clean_text(val_str)
+
+    # --- LOGIQUE DE TRANSFORMATION (Unitaires) ---
+
+    def _clean_currency(self, val):
+        """Transforme '20 000 FCFA' en 20000.0"""
+        try:
+            # Retire tout ce qui n'est pas chiffre, point ou virgule
+            clean = re.sub(r'[^\d,.-]', '', val)
+            clean = clean.replace(',', '.') # Standardisation US
+            return float(clean)
+        except:
+            return 0.0 # Ou None selon la rigueur souhaitée
+
+    def _clean_integer(self, val):
+        """Transforme '12 mois' en 12"""
+        try:
+            val_float = self._clean_currency(val)
+            return int(val_float)
+        except:
+            return 0
+
+    def _standardize_date(self, val):
+        """Transforme n'importe quel format en YYYY-MM-DD (ISO 8601)"""
+        # Formats supportés : DD/MM/YYYY, DD-MM-YYYY, YYYY-MM-DD
+        patterns = [
+            (r'(\d{2})[-/](\d{2})[-/](\d{4})', lambda m: f"{m.group(3)}-{m.group(2)}-{m.group(1)}"), # 29-01-2026 -> 2026-01-29
+            (r'(\d{4})[-/](\d{2})[-/](\d{2})', lambda m: f"{m.group(1)}-{m.group(2)}-{m.group(3)}")  # Déjà ISO
+        ]
+        
+        for pat, converter in patterns:
+            match = re.search(pat, val)
+            if match:
+                return converter(match)
+        
+        return val # Retourne brut si échec (pour logguer l'erreur plus tard)
+
+    def _to_boolean(self, val):
+        return val.lower() in ['true', '1', 'oui', 'yes', 'vrai', 'active']
+
+    def _clean_text(self, val):
+        """Retire les espaces superflus et caractères invisibles"""
+        return re.sub(r'\s+', ' ', val).strip()