|
16 | 16 | from timeit import default_timer as timer |
17 | 17 | from typing import Optional |
18 | 18 |
|
19 | | -from rdflib import Literal, Namespace |
| 19 | +from rdflib import RDF, BNode, Literal, Namespace |
20 | 20 |
|
21 | 21 | from rocrate_validator.constants import SHACL_NS |
22 | 22 | from rocrate_validator.errors import ROCrateMetadataNotFoundError |
|
28 | 28 | RequirementCheckValidationEvent, |
29 | 29 | RequirementLevel, |
30 | 30 | SkipRequirementCheck, |
| 31 | + SourceSnippet, |
31 | 32 | ValidationContext, |
32 | 33 | ) |
33 | 34 | from rocrate_validator.requirements.shacl.models import Shape, ShapesRegistry |
34 | | -from rocrate_validator.requirements.shacl.utils import make_uris_relative, resolve_parent_shape |
| 35 | +from rocrate_validator.requirements.shacl.utils import build_node_subgraph, make_uris_relative, resolve_parent_shape |
35 | 36 | from rocrate_validator.requirements.shacl.validator import ( |
36 | 37 | SHACLValidationAlreadyProcessed, |
37 | 38 | SHACLValidationContext, |
@@ -175,6 +176,58 @@ def level(self) -> str: |
175 | 176 | def severity(self) -> str: |
176 | 177 | return self.level.severity |
177 | 178 |
|
| 179 | + def get_source_snippet(self) -> Optional[SourceSnippet]: |
| 180 | + if self._shape is None: |
| 181 | + return None |
| 182 | + try: |
| 183 | + graph = self._shape.graph |
| 184 | + # build a subgraph containing all the triples related to the shape |
| 185 | + subgraph = build_node_subgraph(graph, self._shape.node) |
| 186 | + # identify the owner of the shape |
| 187 | + owner = self._shape |
| 188 | + while getattr(owner, "parent", None) is not None: |
| 189 | + owner = owner.parent |
| 190 | + # if the shape is not a root shape, include the triples linking the owner to the shape |
| 191 | + if owner is not self._shape: |
| 192 | + shacl = Namespace(SHACL_NS) |
| 193 | + target_predicates = ( |
| 194 | + RDF.type, |
| 195 | + shacl.targetClass, |
| 196 | + shacl.targetNode, |
| 197 | + shacl.targetSubjectsOf, |
| 198 | + shacl.targetObjectsOf, |
| 199 | + shacl.target, |
| 200 | + ) |
| 201 | + for predicate in target_predicates: |
| 202 | + for triple in graph.triples((owner.node, predicate, None)): |
| 203 | + subgraph.add(triple) |
| 204 | + # follow BNode objects (e.g. sh:target referencing an inline SPARQL target) |
| 205 | + _, _, obj = triple |
| 206 | + if isinstance(obj, BNode): |
| 207 | + subgraph += build_node_subgraph(graph, obj) |
| 208 | + # link the owner to the property so the relationship is preserved in the serialization |
| 209 | + subgraph.add((owner.node, shacl.property, self._shape.node)) |
| 210 | + |
| 211 | + # copy bindings so the serialized snippet uses the same prefix declarations as the source file |
| 212 | + for prefix, namespace in graph.namespaces(): |
| 213 | + subgraph.bind(prefix, namespace, replace=True) |
| 214 | + # serialize the subgraph to Turtle format |
| 215 | + code = subgraph.serialize(format="turtle") |
| 216 | + except Exception as e: |
| 217 | + logger.debug("Unable to serialize SHACL shape for check %s: %s", self.identifier, e) |
| 218 | + return None |
| 219 | + # if the code is bytes, decode it to string |
| 220 | + if isinstance(code, bytes): |
| 221 | + code = code.decode("utf-8") |
| 222 | + # use the shape source file as the source path for the snippet if available |
| 223 | + source_path = self.requirement.path if self.requirement else None |
| 224 | + # build the source snippet for the check |
| 225 | + return SourceSnippet( |
| 226 | + language="turtle", |
| 227 | + code=code, |
| 228 | + source_path=source_path, |
| 229 | + ) |
| 230 | + |
178 | 231 | def execute_check(self, context: ValidationContext): |
179 | 232 | logger.debug("Starting check %s", self) |
180 | 233 | try: |
|
0 commit comments