Coverage for heritrace/utils/shacl_utils.py: 96%
221 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-26 08:34 +0000
1# SPDX-FileCopyrightText: 2024-2025 Arcangelo Massari <arcangelo.massari@unibo.it>
2#
3# SPDX-License-Identifier: ISC
5from collections.abc import Iterable
6from weakref import WeakKeyDictionary
8from flask import Flask
9from rdflib import RDF, Graph
10from SPARQLWrapper import JSON
12from heritrace.extensions import get_form_fields, get_shacl_graph, get_sparql
13from heritrace.sparql import get_sparql_bindings, select_results
14from heritrace.utils.display_rules_utils import get_class_priority
15from heritrace.utils.filters import format_uri_as_readable
16from heritrace.utils.shacl_display import (
17 ShaclProcessingContext,
18 apply_display_rules,
19 extract_shacl_form_fields,
20 order_form_fields,
21 process_nested_shapes,
22)
23from heritrace.utils.virtual_properties import get_virtual_properties_for_entity
25_class_shapes_cache: WeakKeyDictionary[Graph, dict[str, list[str]]] = (
26 WeakKeyDictionary()
27)
28_shape_properties_cache: WeakKeyDictionary[Graph, dict[str, set[str]]] = (
29 WeakKeyDictionary()
30)
31_hasvalue_constraints_cache: WeakKeyDictionary[
32 Graph, dict[str, list[tuple[str, str]]]
33] = WeakKeyDictionary()
36def get_form_fields_from_shacl(
37 shacl: Graph | None, display_rules: list[dict] | None, app: Flask
38) -> dict:
39 """
40 Analyze SHACL shapes to extract form fields for each entity type.
42 Args:
43 shacl: The SHACL graph
44 display_rules: The display rules configuration
45 app: Flask application instance
47 Returns:
48 OrderedDict: A dictionary where the keys are tuples (class, shape) and the
49 values are dictionaries
50 of form fields with their properties.
51 """
52 if not shacl:
53 return {}
55 # Step 1: Get the initial form fields from SHACL shapes
56 form_fields = extract_shacl_form_fields(shacl, display_rules, app=app)
58 # Step 2: Process nested shapes for each field
59 processed_shapes: set[str] = set()
60 ctx = ShaclProcessingContext(
61 shacl=shacl,
62 display_rules=display_rules,
63 app=app,
64 processed_shapes=processed_shapes,
65 )
66 for entity_key in form_fields:
67 for predicate in form_fields[entity_key]:
68 for field_info in form_fields[entity_key][predicate]:
69 if field_info.get("nodeShape"):
70 field_info["nestedShape"] = process_nested_shapes(
71 ctx,
72 str(field_info["nodeShape"]),
73 )
75 # Step 3: Apply display rules to the form fields
76 if display_rules:
77 form_fields = apply_display_rules(shacl, form_fields, display_rules)
79 # Step 3.5: Ensure all form fields have displayName, using fallback for those
80 # without display rules
81 ensure_display_names(form_fields)
83 # Step 4: Add virtual properties to form_fields
84 enhanced_form_fields = add_virtual_properties_to_form_fields_internal(form_fields)
86 # Step 5: Order form fields (including virtual properties)
87 return order_form_fields(enhanced_form_fields, display_rules)
90def _apply_field_overrides(shape_data: dict, override: dict) -> dict:
91 nested_field = shape_data.copy()
92 if "shouldBeDisplayed" in override:
93 nested_field["shouldBeDisplayed"] = override["shouldBeDisplayed"]
94 if "displayName" in override:
95 nested_field["displayName"] = override["displayName"]
96 if "value" in override:
97 nested_field["hasValue"] = override["value"]
98 nested_field["nestedShape"] = []
99 return nested_field
102def _build_nested_shape_entry(vp: dict, enhanced_form_fields: dict) -> list[dict]:
103 implementation = vp.get("implementedVia", {})
104 target = implementation.get("target", {})
105 intermediate_class = target.get("class")
106 specific_shape = target.get("shape")
108 if not specific_shape and intermediate_class:
109 specific_shape = determine_shape_for_classes([intermediate_class])
111 intermediate_entity_key = find_matching_form_field(
112 class_uri=intermediate_class,
113 shape_uri=specific_shape,
114 form_fields=enhanced_form_fields,
115 )
117 nested_shape_list: list[dict] = []
118 if not intermediate_entity_key:
119 return nested_shape_list
121 nested_shape_data = enhanced_form_fields.get(intermediate_entity_key, {})
122 field_overrides = implementation.get("fieldOverrides", {})
124 for nested_prop_uri, nested_details_list in nested_shape_data.items():
125 for nested_details in nested_details_list:
126 if nested_prop_uri in field_overrides:
127 nested_field = _apply_field_overrides(
128 nested_details, field_overrides[nested_prop_uri]
129 )
130 else:
131 nested_field = nested_details.copy()
133 if nested_field.get("shouldBeDisplayed", True):
134 nested_shape_list.append(nested_field)
136 return nested_shape_list
139def add_virtual_properties_to_form_fields_internal(form_fields: dict) -> dict:
140 enhanced_form_fields = form_fields.copy() if form_fields else {}
142 for entity_key in enhanced_form_fields:
143 entity_class, entity_shape = entity_key
145 virtual_properties = get_virtual_properties_for_entity(
146 entity_class, entity_shape
147 )
149 if not virtual_properties:
150 continue
152 for display_name, prop_config in virtual_properties:
153 if not prop_config.get("shouldBeDisplayed", True):
154 continue
156 nested_shape_list = _build_nested_shape_entry(
157 prop_config, enhanced_form_fields
158 )
160 virtual_form_field = {
161 "displayName": prop_config.get("displayName", display_name),
162 "uri": display_name,
163 "is_virtual": True,
164 "min": 0,
165 "max": None,
166 "datatypes": [],
167 "optionalValues": [],
168 "orderedBy": None,
169 "nodeShape": None,
170 "subjectClass": None,
171 "subjectShape": None,
172 "objectClass": None,
173 "entityType": None,
174 "nestedShape": nested_shape_list,
175 "or": None,
176 }
178 enhanced_form_fields[entity_key][display_name] = [virtual_form_field]
180 return enhanced_form_fields
183def _get_shapes_for_class(shacl_graph: Graph, class_uri: str) -> list[str]:
184 per_graph = _class_shapes_cache.setdefault(shacl_graph, {})
185 if class_uri not in per_graph:
186 query_string = f"""
187 SELECT DISTINCT ?shape WHERE {{
188 ?shape <http://www.w3.org/ns/shacl#targetClass> <{class_uri}> .
189 }}
190 """
192 results = shacl_graph.query(query_string)
193 per_graph[class_uri] = [str(row.shape) for row in select_results(results)]
194 return per_graph[class_uri]
197def determine_shape_for_classes(class_list: list[str]) -> str | None:
198 """
199 Determine the most appropriate SHACL shape for a list of class URIs.
201 Args:
202 class_list: List of class URIs to find shapes for
204 Returns:
205 The most appropriate shape URI based on priority, or None if no shapes are found
206 """
207 shacl_graph = get_shacl_graph()
208 if not shacl_graph:
209 return None
211 all_shacl_shapes = []
213 for class_uri in class_list:
214 shapes = _get_shapes_for_class(shacl_graph, class_uri)
215 all_shacl_shapes.extend((class_uri, shape) for shape in shapes)
217 return _find_highest_priority_shape(all_shacl_shapes)
220def determine_shape_for_entity_triples(entity_triples: Iterable) -> str | None:
221 """
222 Determine the most appropriate SHACL shape for an entity based on its triples.
224 Uses a multi-criteria scoring system to distinguish between shapes:
225 1. sh:hasValue constraint matches (highest priority)
226 2. Property matching - number of shape properties present in entity
227 3. Class priority - predefined priority ordering
229 Args:
230 entity_triples: List of triples (subject, predicate, object) for the entity
232 Returns:
233 The most appropriate shape URI, or None if no shapes are found
234 """
235 shacl_graph = get_shacl_graph()
236 if not shacl_graph:
237 return None
239 entity_classes = []
240 entity_properties = set()
242 for _subject, predicate, obj in entity_triples:
243 if str(predicate) == str(RDF.type):
244 entity_classes.append(str(obj))
245 entity_properties.add(str(predicate))
247 if not entity_classes:
248 return None
250 candidate_shapes = []
252 for class_uri in entity_classes:
253 shapes = _get_shapes_for_class(shacl_graph, class_uri)
254 candidate_shapes.extend((class_uri, shape) for shape in shapes)
256 if not candidate_shapes:
257 return None
259 if len(candidate_shapes) == 1:
260 return candidate_shapes[0][1]
262 shape_scores = {}
264 for class_uri, shape_uri in candidate_shapes:
265 shape_properties = _get_shape_properties(shacl_graph, shape_uri)
266 property_matches = len(entity_properties.intersection(shape_properties))
268 hasvalue_matches = _check_hasvalue_constraints(
269 shacl_graph, shape_uri, entity_triples
270 )
272 entity_key = (class_uri, shape_uri)
273 priority = get_class_priority(entity_key)
275 # Combined score: (hasvalue_matches, property_matches, -priority)
276 # hasValue matches are most important, then property matches, then priority
277 combined_score = (hasvalue_matches, property_matches, -priority)
278 shape_scores[shape_uri] = combined_score
280 return max(shape_scores.keys(), key=lambda s: shape_scores[s])
283def _find_highest_priority_shape(
284 class_shape_pairs: list[tuple[str, str]],
285) -> str | None:
286 """
287 Helper function to find the shape with the highest priority from a list of
288 (class_uri, shape) pairs.
290 Args:
291 class_shape_pairs: List of tuples (class_uri, shape)
293 Returns:
294 The shape with the highest priority, or None if the list is empty
295 """
296 highest_priority = float("inf")
297 highest_priority_shape = None
299 for class_uri, shape in class_shape_pairs:
300 entity_key = (class_uri, shape)
301 priority = get_class_priority(entity_key)
302 if priority < highest_priority:
303 highest_priority = priority
304 highest_priority_shape = shape
306 return highest_priority_shape
309def _get_shape_properties(shacl_graph: Graph, shape_uri: str) -> set:
310 """
311 Extract all properties defined in a SHACL shape.
313 Args:
314 shacl_graph: The SHACL graph
315 shape_uri: URI of the shape to analyze
317 Returns:
318 Set of property URIs defined in the shape
319 """
320 per_graph = _shape_properties_cache.setdefault(shacl_graph, {})
321 if shape_uri not in per_graph:
322 query_string = f"""
323 PREFIX sh: <http://www.w3.org/ns/shacl#>
324 SELECT DISTINCT ?property WHERE {{
325 <{shape_uri}> sh:property ?propertyShape .
326 ?propertyShape sh:path ?property .
327 }}
328 """
330 results = shacl_graph.query(query_string)
331 per_graph[shape_uri] = {str(row.property) for row in select_results(results)}
333 return per_graph[shape_uri]
336def _get_hasvalue_constraints(
337 shacl_graph: Graph, shape_uri: str
338) -> list[tuple[str, str]]:
339 per_graph = _hasvalue_constraints_cache.setdefault(shacl_graph, {})
340 if shape_uri not in per_graph:
341 query_string = f"""
342 PREFIX sh: <http://www.w3.org/ns/shacl#>
343 SELECT DISTINCT ?property ?value WHERE {{
344 <{shape_uri}> sh:property ?propertyShape .
345 ?propertyShape sh:path ?property .
346 {{ ?propertyShape sh:hasValue ?value . }}
347 UNION
348 {{ ?propertyShape sh:qualifiedValueShape/sh:hasValue ?value . }}
349 }}
350 """
352 results = shacl_graph.query(query_string)
353 per_graph[shape_uri] = [
354 (str(row.property), str(row.value)) for row in select_results(results)
355 ]
357 return per_graph[shape_uri]
360def _check_hasvalue_constraints(
361 shacl_graph: Graph, shape_uri: str, entity_triples: Iterable
362) -> int:
363 """
364 Check how many sh:hasValue constraints the entity satisfies for a given shape.
366 Args:
367 shacl_graph: The SHACL graph
368 shape_uri: URI of the shape to check
369 entity_triples: List of triples (subject, predicate, object) for the entity
371 Returns:
372 Number of hasValue constraints satisfied by the entity
373 """
374 constraints = _get_hasvalue_constraints(shacl_graph, shape_uri)
376 if not constraints:
377 return 0
379 # Create a set of (predicate, object) pairs from entity triples
380 entity_property_values = set()
381 for _, predicate, obj in entity_triples:
382 entity_property_values.add((str(predicate), str(obj)))
384 # Count how many constraints are satisfied
385 satisfied_constraints = 0
386 for property_uri, required_value in constraints:
387 if (property_uri, required_value) in entity_property_values:
388 satisfied_constraints += 1
390 return satisfied_constraints
393def ensure_display_names(form_fields: dict) -> None:
394 """
395 Ensures all form fields have a displayName, using URI formatting as fallback.
397 Args:
398 form_fields: Dictionary of form fields to process
399 """
400 for predicates in form_fields.values():
401 for predicate_uri, details_list in predicates.items():
402 for field_info in details_list:
403 # Only add displayName if not already present
404 if not field_info.get("displayName"):
405 field_info["displayName"] = format_uri_as_readable(predicate_uri)
408def find_matching_form_field(
409 class_uri: str | None = None,
410 shape_uri: str | None = None,
411 form_fields: dict | None = None,
412) -> tuple[str, str] | None:
413 """
414 Find the most appropriate form field configuration for a given class and/or shape.
415 At least one of class_uri or shape_uri must be provided.
417 Args:
418 class_uri: Optional URI of the class
419 shape_uri: Optional URI of the shape
420 form_fields: Optional dictionary of form fields to search in, defaults to global
421 form_fields
423 Returns:
424 The matching form field key (class_uri, shape_uri) or None if no match is found
425 """
426 if not form_fields:
427 form_fields = get_form_fields()
429 if not form_fields:
430 return None
432 class_match = None
433 shape_match = None
435 for field_key in form_fields:
436 field_class_uri = field_key[0]
437 field_shape_uri = field_key[1]
439 # Case 1: Both class and shape match (exact match)
440 if (
441 class_uri
442 and shape_uri
443 and field_class_uri == str(class_uri)
444 and field_shape_uri == str(shape_uri)
445 ):
446 return field_key
448 # Case 2: Only class matches (and form field has no shape constraint)
449 if class_uri and field_class_uri == str(class_uri) and field_shape_uri is None:
450 class_match = field_key
452 # Case 3: Only shape matches (and form field has no class constraint)
453 elif (
454 shape_uri and field_shape_uri == str(shape_uri) and field_class_uri is None
455 ):
456 shape_match = field_key
458 # Case 4: Only class matches (even if form field has a shape)
459 elif class_uri and field_class_uri == str(class_uri) and not class_match:
460 class_match = field_key
462 # Return the best match based on specificity
463 # Shape rules typically have higher specificity, so prefer them
464 if shape_match:
465 return shape_match
466 if class_match:
467 return class_match
469 return None
472def _find_entity_position_in_order_map(entity_uri: str, order_map: dict) -> int | None:
473 """
474 Helper function to find entity position in an order map.
476 This function handles the case where there might be multiple independent ordered
477 chains
478 within the same predicate relationship. Each chain has its own starting element and
479 follows a linked-list structure where each entity points to the next one.
481 Args:
482 entity_uri: URI of the entity to find position for
483 order_map: Dictionary mapping entities to their next entity in sequence.
484 Key = current entity URI, Value = next entity URI (or None for last
485 element)
486 Example: {'entity1': 'entity2', 'entity2': 'entity3', 'entity3':
487 None,
488 'entity4': 'entity5', 'entity5': None}
489 This represents two chains: [entity1 -> entity2 -> entity3] and
490 [entity4 -> entity5]
492 Returns:
493 1-based position in the sequence, or None if not found
494 """
495 # Find all starting elements of ordered chains.
496 # A start element is one that appears as a key in the order_map but never as a
497 # value,
498 # meaning no other entity points to it (it's the head of a chain).
499 start_elements = set(order_map.keys()) - {
500 v for v in order_map.values() if v is not None
501 }
503 if not start_elements:
504 # No valid starting points found - this shouldn't happen in well-formed data
505 return None
507 # Since there can be multiple independent ordered chains, we need to check each one
508 # to find which chain contains our target entity
509 for start_element in start_elements:
510 # Build the complete sequence for this chain by following the linked-list
511 # structure
512 sequence = []
513 current_element = start_element
515 # Follow the chain from start to end
516 while current_element in order_map:
517 sequence.append(current_element)
518 # Move to the next element in the chain (or None if we've reached the end)
519 current_element = order_map[current_element]
521 # Check if our target entity is in this particular chain
522 try:
523 # If found, return its 1-based position within this chain
524 return (
525 sequence.index(entity_uri) + 1
526 ) # Convert from 0-based to 1-based indexing
527 except ValueError:
528 # Entity not found in this chain, try the next one
529 continue
531 # Entity was not found in any of the ordered chains
532 return None
535def get_entity_position_in_sequence(
536 entity_uri: str,
537 subject_uri: str,
538 predicate_uri: str,
539 order_property: str,
540 snapshot: Graph | None = None,
541) -> int | None:
542 """
543 Get the position of an entity in an ordered sequence.
545 Args:
546 entity_uri: URI of the entity to find position for
547 subject_uri: URI of the subject that has the ordered property
548 predicate_uri: URI of the ordered predicate
549 order_property: URI of the property that defines the ordering
550 snapshot: Optional graph snapshot for historical queries
552 Returns:
553 1-based position in the sequence, or None if not found
554 """
555 order_query = f"""
556 SELECT ?orderedEntity (COALESCE(?next, "NONE") AS ?nextValue)
557 WHERE {{
558 <{subject_uri}> <{predicate_uri}> ?orderedEntity.
559 OPTIONAL {{
560 ?orderedEntity <{order_property}> ?next.
561 }}
562 }}
563 """
565 if snapshot:
566 order_results = list(select_results(snapshot.query(order_query)))
568 order_map = {}
569 for res in order_results:
570 ordered_entity = str(res[0])
571 next_value = str(res[1])
572 order_map[ordered_entity] = None if next_value == "NONE" else next_value
574 return _find_entity_position_in_order_map(entity_uri, order_map)
575 sparql = get_sparql()
576 sparql.setQuery(order_query)
577 sparql.setReturnFormat(JSON)
578 order_results = get_sparql_bindings(sparql.query().convert())
580 order_map = {}
581 for res in order_results:
582 ordered_entity = res["orderedEntity"]["value"]
583 next_value = res["nextValue"]["value"]
584 order_map[ordered_entity] = None if next_value == "NONE" else next_value
586 return _find_entity_position_in_order_map(entity_uri, order_map)