| 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): |
| |
| self.EX = Namespace("http://activa.ai/ontology/") |
| self.shapes_file = os.path.join(os.path.dirname(__file__), "shapes/schema_constraints.ttl") |
| |
| |
| 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: |
| |
| subj_uri = URIRef(self.EX[t.subject.replace(" ", "_")]) |
| obj_uri = URIRef(self.EX[t.object.replace(" ", "_")]) |
| |
| |
| 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"))) |
|
|
| |
| 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: |
| |
| 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 |
|
|
| |
| if __name__ == "__main__": |
| |
| from collections import namedtuple |
| Triple = namedtuple("Triple", ["subject", "predicate", "object", "confidence"]) |
| |
| |
| 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) |
