src/core/validation_engine.py

"""
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", {})
        }