Create validation_engine.py

src/core/validation_engine.py

@@ -0,0 +1,60 @@
+"""
+MODULE: VALIDATION ENGINE
+=========================
+Responsabilité : Vérifier la qualité des données post-ingestion.
+"""
+import pandas as pd
+
+class ValidationEngine:
+    def __init__(self, rdf_store, ontology_manager):
+        self.rdf = rdf_store
+        self.onto = ontology_manager
+        self.report = {}
+
+    def run_validation(self):
+        """Exécute une batterie de tests sur le graphe"""
+        print("🔍 [VALIDATION] Démarrage de l'audit des données...")
+        
+        real_schema = self.rdf.get_full_schema()
+        theoretical_classes = self.onto.structure["classes"]
+        
+        # 1. Validation des Volumes
+        self.report["volumes"] = real_schema.get("classes", {})
+        
+        # 2. Validation de la Couverture (Propriétés Critiques)
+        # On vérifie si les propriétés définies dans l'Ontology existent vraiment
+        self.report["missing_props"] = {}
+        
+        # Pour faire simple et rapide : on check un échantillon
+        for cls in list(theoretical_classes)[:5]: # Check les 5 premières classes majeures
+            expected_props = self.onto.structure["properties"].get(cls, [])
+            if not expected_props: continue
+            
+            # Requête SPARQL pour vérifier si ces props sont utilisées
+            for prop in expected_props:
+                # Est-ce que au moins 1 entité de cette classe a cette prop ?
+                q = f"""
+                ASK {{
+                    ?s a vortex:{cls} .
+                    ?s vortex:{prop} ?o .
+                }}
+                """
+                exists = bool(self.rdf.g.query(f"PREFIX vortex: <http://vortex.ai/ontology#> {q}"))
+                if not exists:
+                    if cls not in self.report["missing_props"]: self.report["missing_props"][cls] = []
+                    self.report["missing_props"][cls].append(prop)
+
+        return self.report
+
+    def generate_metrics_dashboard(self):
+        """Retourne des KPIs pour l'interface Streamlit"""
+        total_triples = len(self.rdf.g)
+        total_entities = sum(self.report.get("volumes", {}).values())
+        missing_count = sum([len(v) for v in self.report.get("missing_props", {}).values()])
+        
+        return {
+            "triples": total_triples,
+            "entities": total_entities,
+            "coverage_score": max(0, 100 - (missing_count * 2)), # Score arbitraire
+            "alerts": self.report.get("missing_props", {})
+        }