Spaces:
Running
Running
File size: 2,463 Bytes
409f286 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 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", {})
} |