File size: 3,355 Bytes
a968971 | 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 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 | import os
from rdflib import Graph, Literal, RDF, URIRef, Namespace
from rdflib.namespace import SKOS, XSD
from pyshacl import validate
class SemanticValidator:
def __init__(self):
# Definiamo i namespace
self.EX = Namespace("http://activa.ai/ontology/")
self.shapes_file = os.path.join(os.path.dirname(__file__), "shapes/schema_constraints.ttl")
# Carica le shapes se il file esiste, altrimenti usa grafo vuoto
if os.path.exists(self.shapes_file):
self.shacl_graph = Graph()
self.shacl_graph.parse(self.shapes_file, format="turtle")
print("🛡️ SHACL Constraints caricati.")
else:
print("⚠️ File SHACL non trovato. Validazione disabilitata.")
self.shacl_graph = None
def _json_to_rdf(self, triples):
"""Converte le triple JSON (Pydantic) in un grafo RDFLib in memoria."""
g = Graph()
g.bind("skos", SKOS)
g.bind("ex", self.EX)
for t in triples:
# Creiamo URI sanitizzati
subj_uri = URIRef(self.EX[t.subject.replace(" ", "_")])
obj_uri = URIRef(self.EX[t.object.replace(" ", "_")])
# Aggiungiamo il tipo Concept
g.add((subj_uri, RDF.type, SKOS.Concept))
g.add((subj_uri, SKOS.prefLabel, Literal(t.subject, lang="it")))
g.add((obj_uri, RDF.type, SKOS.Concept))
g.add((obj_uri, SKOS.prefLabel, Literal(t.object, lang="it")))
# Mappiamo il predicato (se è standard o custom)
if t.predicate == "skos:related" or t.predicate == "related":
pred = SKOS.related
elif t.predicate == "skos:broader" or t.predicate == "broader":
pred = SKOS.broader
else:
# Fallback su namespace custom per predicati non standard (es. situato_in)
pred = self.EX[t.predicate]
g.add((subj_uri, pred, obj_uri))
return g
def validate_batch(self, triples):
"""
Esegue la validazione SHACL sulle triple.
Ritorna (is_valid, report_text, rdf_graph)
"""
if not self.shacl_graph:
return True, "No Constraints", None
data_graph = self._json_to_rdf(triples)
print("🔍 Esecuzione Validazione SHACL...")
conforms, report_graph, report_text = validate(
data_graph,
shacl_graph=self.shacl_graph,
inference='rdfs',
serialize_report_graph=True
)
return conforms, report_text, data_graph
# --- TEST DEL MODULO ---
if __name__ == "__main__":
# Simuliamo triple dall'LLM
from collections import namedtuple
Triple = namedtuple("Triple", ["subject", "predicate", "object", "confidence"])
# Caso Test: Una tripla valida e una (potenzialmente) invalida
mock_triples = [
Triple("Basilica San Marco", "situato_in", "Venezia", 0.9),
Triple("Venezia", "skos:related", "Laguna", 0.95)
]
validator = SemanticValidator()
is_valid, report, _ = validator.validate_batch(mock_triples)
if is_valid:
print("✅ Dati Conformi allo Schema SHACL.")
else:
print("❌ Violazione dei vincoli rilevata!")
print(report) |