Coverage for heritrace/utils/shacl_display.py: 97%
316 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-26 08:34 +0000
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-26 08:34 +0000
1# SPDX-FileCopyrightText: 2025 Arcangelo Massari <arcangelo.massari@unibo.it>
2#
3# SPDX-License-Identifier: ISC
5import json
6from collections import OrderedDict, defaultdict
7from collections.abc import Iterable
8from dataclasses import dataclass
9from pathlib import Path
10from typing import cast
12from flask import Flask
13from rdflib import Graph, URIRef
14from rdflib.plugins.sparql import prepareQuery
15from rdflib.plugins.sparql.sparql import Query
16from rdflib.query import Result, ResultRow
18from heritrace.sparql import select_results
19from heritrace.utils.filters import Filter
22@dataclass(slots=True)
23class ShaclProcessingContext:
24 shacl: Graph
25 display_rules: list[dict[str, object]] | None
26 app: Flask
27 processed_shapes: set[str]
30@dataclass(slots=True)
31class _ParsedRow:
32 subject_shape: str
33 entity_type: str
34 predicate: str
35 node_shape: str | None
36 has_value: str | None
37 object_class: str | None
38 min_count: int
39 max_count: int | None
40 datatype: str | None
41 optional_values: list[str]
42 or_nodes: list[str]
43 entity_key: tuple[str, str]
44 condition_entry: dict[str, object]
45 node_shapes: list[str]
48COMMON_SPARQL_QUERY = prepareQuery(
49 """
50 SELECT ?shape ?type ?predicate ?node_shape ?datatype
51 ?max_count ?min_count ?has_value ?object_class
52 ?condition_path ?condition_value ?pattern ?message
53 (GROUP_CONCAT(?optional_value; separator=",")
54 AS ?optional_values)
55 (GROUP_CONCAT(?or_node; separator=",") AS ?or_nodes)
56 WHERE {
57 ?shape sh:targetClass ?type ;
58 sh:property ?property .
59 ?property sh:path ?predicate .
60 OPTIONAL {
61 ?property sh:node ?node_shape .
62 OPTIONAL {
63 ?node_shape sh:targetClass ?object_class .
64 }
65 }
66 OPTIONAL {
67 { ?property sh:or ?orList . }
68 UNION
69 { ?property sh:qualifiedValueShape/sh:or ?orList . }
70 {
71 ?orList rdf:rest*/rdf:first ?or_constraint .
72 ?or_constraint sh:datatype ?datatype .
73 } UNION {
74 ?orList rdf:rest*/rdf:first ?or_node_shape .
75 ?or_node_shape sh:node ?or_node .
76 } UNION {
77 ?orList rdf:rest*/rdf:first ?or_constraint .
78 ?or_constraint sh:hasValue ?optional_value .
79 }
80 }
81 OPTIONAL { ?property sh:datatype ?datatype . }
82 OPTIONAL {
83 { ?property sh:maxCount ?max_count . }
84 UNION
85 { ?property sh:qualifiedMaxCount ?max_count . }
86 }
87 OPTIONAL {
88 { ?property sh:minCount ?min_count . }
89 UNION
90 { ?property sh:qualifiedMinCount ?min_count . }
91 }
92 OPTIONAL {
93 { ?property sh:hasValue ?has_value . }
94 UNION
95 { ?property sh:qualifiedValueShape/sh:hasValue ?has_value . }
96 }
97 OPTIONAL {
98 ?property sh:in ?list .
99 ?list rdf:rest*/rdf:first ?optional_value .
100 }
101 OPTIONAL {
102 ?property sh:condition ?condition_node .
103 ?condition_node sh:path ?condition_path ;
104 sh:hasValue ?condition_value .
105 }
106 OPTIONAL { ?property sh:pattern ?pattern . }
107 OPTIONAL { ?property sh:message ?message . }
108 FILTER (isURI(?predicate))
109 }
110 GROUP BY ?shape ?type ?predicate ?node_shape ?datatype
111 ?max_count ?min_count ?has_value ?object_class
112 ?condition_path ?condition_value ?pattern ?message
113""",
114 initNs={
115 "sh": "http://www.w3.org/ns/shacl#",
116 "rdf": "http://www.w3.org/1999/02/22-rdf-syntax-ns#",
117 },
118)
121def _parse_row(row: ResultRow) -> _ParsedRow:
122 subject_shape = str(row.shape)
123 entity_type = str(row.type)
124 predicate = str(row.predicate)
125 node_shape = str(row.node_shape) if row.node_shape else None
126 has_value = str(row.has_value) if row.has_value else None
127 object_class = str(row.object_class) if row.object_class else None
128 min_count = 0 if row.min_count is None else int(row.min_count)
129 max_count = None if row.max_count is None else int(row.max_count)
130 datatype = str(row.datatype) if row.datatype else None
131 optional_values = [v for v in (row.optional_values or "").split(",") if v]
132 or_nodes = [v for v in (row.or_nodes or "").split(",") if v]
134 condition_entry: dict[str, object] = {}
135 if row.condition_path and row.condition_value:
136 condition_entry["condition"] = {
137 "path": str(row.condition_path),
138 "value": str(row.condition_value),
139 }
140 if row.pattern:
141 condition_entry["pattern"] = str(row.pattern)
142 if row.message:
143 condition_entry["message"] = str(row.message)
145 node_shapes = []
146 if node_shape:
147 node_shapes.append(node_shape)
148 node_shapes.extend(or_nodes)
150 return _ParsedRow(
151 subject_shape=subject_shape,
152 entity_type=entity_type,
153 predicate=predicate,
154 node_shape=node_shape,
155 has_value=has_value,
156 object_class=object_class,
157 min_count=min_count,
158 max_count=max_count,
159 datatype=datatype,
160 optional_values=optional_values,
161 or_nodes=or_nodes,
162 entity_key=(entity_type, subject_shape),
163 condition_entry=condition_entry,
164 node_shapes=node_shapes,
165 )
168def _find_existing_field(
169 fields: list[dict[str, object]],
170 parsed: _ParsedRow,
171) -> dict[str, object] | None:
172 for field in fields:
173 if (
174 field.get("nodeShape") == parsed.node_shape
175 and field.get("nodeShapes") == parsed.node_shapes
176 and field.get("subjectShape") == parsed.subject_shape
177 and field.get("hasValue") == parsed.has_value
178 and field.get("objectClass") == parsed.object_class
179 and field.get("min") == parsed.min_count
180 and field.get("max") == parsed.max_count
181 and field.get("optionalValues") == parsed.optional_values
182 ):
183 return field
184 return None
187def _process_or_nodes(
188 ctx: ShaclProcessingContext,
189 parsed: _ParsedRow,
190 custom_filter: Filter,
191 field_info: dict[str, object],
192 depth: int,
193) -> None:
194 field_info["or"] = []
195 for node in parsed.or_nodes:
196 entity_type_or_node = get_shape_target_class(ctx.shacl, node)
197 object_class = get_object_class(ctx.shacl, node, parsed.predicate)
198 shape_display_name = custom_filter.human_readable_class(
199 (entity_type_or_node, node)
200 )
201 or_field_info: dict[str, object] = {
202 "entityType": entity_type_or_node,
203 "uri": parsed.predicate,
204 "displayName": shape_display_name,
205 "subjectShape": parsed.subject_shape,
206 "nodeShape": node,
207 "min": parsed.min_count,
208 "max": parsed.max_count,
209 "hasValue": parsed.has_value,
210 "objectClass": object_class,
211 "optionalValues": parsed.optional_values,
212 "conditions": ([parsed.condition_entry] if parsed.condition_entry else []),
213 "shouldBeDisplayed": True,
214 }
215 if node not in ctx.processed_shapes:
216 or_field_info["nestedShape"] = process_nested_shapes(
217 ctx,
218 node,
219 depth=depth + 1,
220 )
221 field_info["or"].append(or_field_info)
224def _process_single_row(
225 ctx: ShaclProcessingContext,
226 parsed: _ParsedRow,
227 custom_filter: Filter,
228 form_fields: defaultdict[tuple[str, str], dict[str, list[dict[str, object]]]],
229 depth: int,
230) -> None:
231 if parsed.predicate not in form_fields[parsed.entity_key]:
232 form_fields[parsed.entity_key][parsed.predicate] = []
234 existing_field = _find_existing_field(
235 form_fields[parsed.entity_key][parsed.predicate], parsed
236 )
238 if existing_field:
239 if parsed.datatype and str(parsed.datatype) not in cast(
240 "list", existing_field.get("datatypes", [])
241 ):
242 cast("list", existing_field.setdefault("datatypes", [])).append(
243 str(parsed.datatype)
244 )
245 if parsed.condition_entry:
246 cast("list", existing_field.setdefault("conditions", [])).append(
247 parsed.condition_entry
248 )
249 return
251 field_info: dict[str, object] = {
252 "entityType": parsed.entity_type,
253 "uri": parsed.predicate,
254 "nodeShape": parsed.node_shape,
255 "nodeShapes": parsed.node_shapes,
256 "subjectShape": parsed.subject_shape,
257 "entityKey": parsed.entity_key,
258 "datatypes": [parsed.datatype] if parsed.datatype else [],
259 "min": parsed.min_count,
260 "max": parsed.max_count,
261 "hasValue": parsed.has_value,
262 "objectClass": parsed.object_class,
263 "optionalValues": parsed.optional_values,
264 "conditions": ([parsed.condition_entry] if parsed.condition_entry else []),
265 "inputType": determine_input_type(parsed.datatype),
266 "shouldBeDisplayed": True,
267 }
269 if parsed.node_shape and parsed.node_shape not in ctx.processed_shapes:
270 field_info["nestedShape"] = process_nested_shapes(
271 ctx,
272 parsed.node_shape,
273 depth=depth + 1,
274 )
276 if parsed.or_nodes:
277 _process_or_nodes(ctx, parsed, custom_filter, field_info, depth)
279 form_fields[parsed.entity_key][parsed.predicate].append(field_info)
282def process_query_results(
283 ctx: ShaclProcessingContext,
284 results: Iterable[ResultRow],
285 depth: int = 0,
286) -> defaultdict[tuple[str, str], dict[str, list[dict[str, object]]]]:
287 form_fields: defaultdict[tuple[str, str], dict[str, list[dict[str, object]]]] = (
288 defaultdict(dict)
289 )
291 with (Path(__file__).parent / "context.json").open() as config_file:
292 context = json.load(config_file)["@context"]
294 custom_filter = Filter(context, ctx.display_rules, ctx.app.config["DATASET_DB_URL"])
296 for row in results:
297 parsed = _parse_row(row)
298 _process_single_row(ctx, parsed, custom_filter, form_fields, depth)
300 return form_fields
303def process_nested_shapes(
304 ctx: ShaclProcessingContext,
305 shape_uri: str,
306 depth: int = 0,
307) -> list[dict[str, object]]:
308 if shape_uri in ctx.processed_shapes:
309 return []
311 ctx.processed_shapes.add(shape_uri)
312 init_bindings = {"shape": URIRef(shape_uri)}
313 nested_results = execute_shacl_query(ctx.shacl, COMMON_SPARQL_QUERY, init_bindings)
314 nested_fields = []
316 temp_form_fields = process_query_results(
317 ctx,
318 select_results(nested_results),
319 depth=depth,
320 )
322 if ctx.display_rules:
323 temp_form_fields = apply_display_rules(
324 ctx.shacl, temp_form_fields, ctx.display_rules
325 )
326 temp_form_fields = order_form_fields(temp_form_fields, ctx.display_rules)
328 for entity_type in temp_form_fields:
329 for predicate in temp_form_fields[entity_type]:
330 nested_fields.extend(temp_form_fields[entity_type][predicate])
332 ctx.processed_shapes.remove(shape_uri)
333 return nested_fields
336def get_property_order(
337 entity_type: str,
338 display_rules: list[dict[str, object]] | None,
339) -> list[str | None]:
340 """
341 Recupera l'ordine delle proprietà per un tipo di entità dalle regole di
342 visualizzazione.
344 Argomenti:
345 entity_type (str): L'URI del tipo di entità.
347 Restituisce:
348 list: Una lista di URI di proprietà nell'ordine desiderato.
349 """
350 if not display_rules:
351 return []
353 for rule in display_rules:
354 if rule.get("class") == entity_type and "propertyOrder" in rule:
355 return cast("list[str | None]", rule["propertyOrder"])
356 if rule.get("class") == entity_type:
357 display_props = cast(
358 "list[dict[str, object]]", rule.get("displayProperties", [])
359 )
360 return [
361 cast("str | None", prop.get("property") or prop.get("virtual_property"))
362 for prop in display_props
363 if prop.get("property") or prop.get("virtual_property")
364 ]
365 return []
368def order_fields(
369 fields: list[dict[str, object]],
370 property_order: list[str],
371) -> list[dict[str, object]]:
372 """
373 Ordina i campi secondo l'ordine specificato delle proprietà.
375 Argomenti:
376 fields (list): Una lista di dizionari dei campi da ordinare.
377 property_order (list): Una lista di URI di proprietà nell'ordine desiderato.
379 Restituisce:
380 list: Una lista ordinata di dizionari dei campi.
381 """
382 if not fields:
383 return []
384 if not property_order:
385 return fields
387 # Create a dictionary to map predicates to their position in property_order
388 order_dict = {pred: i for i, pred in enumerate(property_order)}
390 # Sort fields based on their position in property_order
391 # Fields not in property_order will be placed at the end
392 return sorted(
393 fields,
394 key=lambda f: order_dict.get(
395 str(f.get("predicate", f.get("uri", ""))), float("inf")
396 ),
397 )
400def _find_matching_entity_keys(
401 form_fields: dict[
402 tuple[str, str],
403 dict[str, list[dict[str, object]]],
404 ],
405 entity_class: str | None,
406 entity_shape: str | None,
407) -> list[tuple[str, str]]:
408 if entity_class and entity_shape:
409 entity_key = (entity_class, entity_shape)
410 return [entity_key] if entity_key in form_fields else []
411 if entity_class:
412 return [key for key in form_fields if key[0] == entity_class]
413 if entity_shape:
414 return [key for key in form_fields if key[1] == entity_shape]
415 return []
418def _get_ordered_properties_from_rule(
419 rule: dict[str, object],
420) -> list[str | None]:
421 display_props = cast("list[dict[str, object]]", rule.get("displayProperties", []))
422 return [
423 cast(
424 "str | None",
425 prop_rule.get("property") or prop_rule.get("virtual_property"),
426 )
427 for prop_rule in display_props
428 if prop_rule.get("property") or prop_rule.get("virtual_property")
429 ]
432def _order_entity_fields(
433 form_fields: dict[
434 tuple[str, str],
435 dict[str, list[dict[str, object]]],
436 ],
437 entity_key: tuple[str, str],
438 ordered_properties: list[str | None],
439 ordered_form_fields: OrderedDict[
440 tuple[str, str],
441 OrderedDict[str, list[dict[str, object]]],
442 ],
443) -> None:
444 ordered_form_fields[entity_key] = OrderedDict()
445 for prop in ordered_properties:
446 if prop in form_fields[entity_key]:
447 ordered_form_fields[entity_key][prop] = form_fields[entity_key][prop]
448 # Aggiungi le proprietà rimanenti non specificate nell'ordine
449 for prop in form_fields[entity_key]:
450 if prop not in ordered_properties:
451 ordered_form_fields[entity_key][prop] = form_fields[entity_key][prop]
454def order_form_fields(
455 form_fields: dict[
456 tuple[str, str],
457 dict[str, list[dict[str, object]]],
458 ],
459 display_rules: list[dict[str, object]] | None,
460) -> (
461 OrderedDict[
462 tuple[str, str],
463 OrderedDict[str, list[dict[str, object]]],
464 ]
465 | dict[
466 tuple[str, str],
467 dict[str, list[dict[str, object]]],
468 ]
469):
470 """
471 Ordina i campi del form secondo le regole di visualizzazione.
473 Argomenti:
474 form_fields (dict): I campi del form con possibili modifiche dalle regole di
475 visualizzazione.
477 Restituisce:
478 OrderedDict: I campi del form ordinati.
479 """
480 ordered_form_fields = OrderedDict()
481 if not display_rules:
482 return form_fields
483 for rule in display_rules:
484 target = cast("dict[str, str]", rule.get("target", {}))
485 entity_class = target.get("class")
486 entity_shape = target.get("shape")
487 ordered_properties = _get_ordered_properties_from_rule(rule)
488 matching_keys = _find_matching_entity_keys(
489 form_fields, entity_class, entity_shape
490 )
491 for key in matching_keys:
492 _order_entity_fields(
493 form_fields, key, ordered_properties, ordered_form_fields
494 )
495 return ordered_form_fields
498def apply_display_rules(
499 shacl: Graph,
500 form_fields: dict[
501 tuple[str, str],
502 dict[str, list[dict[str, object]]],
503 ],
504 display_rules: list[dict[str, object]],
505) -> dict[
506 tuple[str, str],
507 dict[str, list[dict[str, object]]],
508]:
509 """
510 Applica le regole di visualizzazione ai campi del form.
512 Argomenti:
513 form_fields (dict): I campi del form iniziali estratti dalle shape SHACL.
515 Restituisce:
516 dict: I campi del form dopo aver applicato le regole di visualizzazione.
517 """
518 for rule in display_rules:
519 target = cast("dict[str, str]", rule.get("target", {}))
520 entity_class = target.get("class")
521 entity_shape = target.get("shape")
523 # Handle different cases based on available target information
524 # Case 1: Both class and shape are specified (exact match)
525 if entity_class and entity_shape:
526 entity_key = (entity_class, entity_shape)
527 if entity_key in form_fields:
528 apply_rule_to_entity(shacl, form_fields, entity_key, rule)
529 # Case 2: Only class is specified (apply to all matching classes)
530 elif entity_class:
531 for key in list(form_fields.keys()):
532 if key[0] == entity_class: # Check if class part of tuple matches
533 apply_rule_to_entity(shacl, form_fields, key, rule)
534 # Case 3: Only shape is specified (apply to all matching shapes)
535 elif entity_shape:
536 for key in list(form_fields.keys()):
537 if key[1] == entity_shape: # Check if shape part of tuple matches
538 apply_rule_to_entity(shacl, form_fields, key, rule)
539 return form_fields
542def apply_rule_to_entity(
543 shacl: Graph,
544 form_fields: dict[
545 tuple[str, str],
546 dict[str, list[dict[str, object]]],
547 ],
548 entity_key: tuple[str, str],
549 rule: dict[str, object],
550) -> None:
551 """
552 Apply a display rule to a specific entity key.
554 Args:
555 shacl: The SHACL graph
556 form_fields: The form fields dictionary
557 entity_key: The entity key tuple (class, shape)
558 rule: The display rule to apply
559 """
560 display_props = cast("list[dict[str, object]]", rule.get("displayProperties", []))
561 for prop in display_props:
562 prop_uri = prop.get("property") or prop.get("virtual_property")
563 if prop_uri and prop_uri in form_fields[entity_key]:
564 for field_info in form_fields[entity_key][str(prop_uri)]:
565 add_display_information(field_info, prop)
566 if "nestedShape" in field_info:
567 target = cast("dict[str, str]", rule.get("target", {}))
568 apply_display_rules_to_nested_shapes(
569 cast("list[dict[str, object]]", field_info["nestedShape"]),
570 prop,
571 target.get("shape"),
572 )
573 if "or" in field_info:
574 target = cast("dict[str, str]", rule.get("target", {}))
575 for or_field in cast("list[dict[str, object]]", field_info["or"]):
576 apply_display_rules_to_nested_shapes(
577 [or_field], field_info, target.get("shape")
578 )
579 if "intermediateRelation" in prop:
580 handle_intermediate_relation(shacl, field_info, prop)
581 if "displayRules" in prop:
582 handle_sub_display_rules(
583 shacl,
584 form_fields,
585 entity_key,
586 form_fields[entity_key][str(prop_uri)],
587 prop,
588 )
591def apply_display_rules_to_nested_shapes( # noqa: C901
592 nested_fields: list[dict[str, object]],
593 parent_prop: dict[str, object],
594 shape_uri: str | None,
595) -> list[dict[str, object]]:
596 """Apply display rules to nested shapes."""
597 if not nested_fields:
598 return []
600 # Handle case where parent_prop is not a dictionary
601 if not isinstance(parent_prop, dict):
602 return nested_fields
604 # Create a new list to avoid modifying the original
605 result_fields = []
606 for field in nested_fields:
607 # Create a copy of the field to avoid modifying the original
608 new_field = field.copy()
609 result_fields.append(new_field)
611 display_rules = cast("list[dict[str, object]]", parent_prop.get("displayRules", []))
612 for rule in display_rules:
613 if rule.get("shape") == shape_uri and "nestedDisplayRules" in rule:
614 nested_display_rules = cast(
615 "list[dict[str, object]]", rule["nestedDisplayRules"]
616 )
617 for field in result_fields:
618 for nested_rule in nested_display_rules:
619 field_key = field.get("predicate", field.get("uri"))
620 if field_key == nested_rule["property"]:
621 # Apply display properties from the rule to the field
622 for key, value in nested_rule.items():
623 if key != "property":
624 field[key] = value
625 break
627 return result_fields
630def determine_input_type(datatype: str | None) -> str:
631 """
632 Determina il tipo di input appropriato basato sul datatype XSD.
633 """
634 if not datatype:
635 return "text"
637 datatype = str(datatype)
638 datatype_to_input = {
639 "http://www.w3.org/2001/XMLSchema#string": "text",
640 "http://www.w3.org/2001/XMLSchema#integer": "number",
641 "http://www.w3.org/2001/XMLSchema#decimal": "number",
642 "http://www.w3.org/2001/XMLSchema#float": "number",
643 "http://www.w3.org/2001/XMLSchema#double": "number",
644 "http://www.w3.org/2001/XMLSchema#boolean": "checkbox",
645 "http://www.w3.org/2001/XMLSchema#date": "date",
646 "http://www.w3.org/2001/XMLSchema#time": "time",
647 "http://www.w3.org/2001/XMLSchema#dateTime": "datetime-local",
648 "http://www.w3.org/2001/XMLSchema#anyURI": "url",
649 "http://www.w3.org/2001/XMLSchema#email": "email",
650 }
651 return datatype_to_input.get(datatype, "text")
654def add_display_information(
655 field_info: dict[str, object],
656 prop: dict[str, object],
657) -> None:
658 """
659 Aggiunge informazioni di visualizzazione dal display_rules ad un campo.
661 Argomenti:
662 field_info (dict): Le informazioni del campo da aggiornare.
663 prop (dict): Le informazioni della proprietà dalle display_rules.
664 """
665 if "displayName" in prop:
666 field_info["displayName"] = prop["displayName"]
667 if "shouldBeDisplayed" in prop:
668 field_info["shouldBeDisplayed"] = prop.get("shouldBeDisplayed", True)
669 if "orderedBy" in prop:
670 field_info["orderedBy"] = prop["orderedBy"]
671 if "inputType" in prop:
672 field_info["inputType"] = prop["inputType"]
673 if "supportsSearch" in prop:
674 field_info["supportsSearch"] = prop["supportsSearch"]
675 if "minCharsForSearch" in prop:
676 field_info["minCharsForSearch"] = prop["minCharsForSearch"]
677 if "searchTarget" in prop:
678 field_info["searchTarget"] = prop["searchTarget"]
681def handle_intermediate_relation(
682 shacl: Graph,
683 field_info: dict[str, object],
684 prop: dict[str, object],
685) -> None:
686 """
687 Processa 'intermediateRelation' nelle display_rules e aggiorna il campo.
689 Argomenti:
690 field_info (dict): Le informazioni del campo da aggiornare.
691 prop (dict): Le informazioni della proprietà dalle display_rules.
692 """
693 intermediate_relation = cast("dict[str, str]", prop["intermediateRelation"])
694 target_entity_type = intermediate_relation["targetEntityType"]
695 intermediate_class = intermediate_relation["class"]
697 connecting_property_query = prepareQuery(
698 """
699 SELECT ?property
700 WHERE {
701 ?shape sh:targetClass ?intermediateClass ;
702 sh:property ?propertyShape .
703 ?propertyShape sh:path ?property ;
704 sh:node ?targetNode .
705 ?targetNode sh:targetClass ?targetClass.
706 }
707 """,
708 initNs={"sh": "http://www.w3.org/ns/shacl#"},
709 )
711 connecting_property_results = shacl.query(
712 connecting_property_query,
713 initBindings={
714 "intermediateClass": URIRef(intermediate_class),
715 "targetClass": URIRef(target_entity_type),
716 },
717 )
719 connecting_property = next(
720 (str(row.property) for row in select_results(connecting_property_results)), None
721 )
723 intermediate_properties = {}
724 target_shape = None
725 if "nestedShape" in field_info:
726 for nested_field in cast("list[dict[str, object]]", field_info["nestedShape"]):
727 if (
728 nested_field.get("uri") == connecting_property
729 and "nestedShape" in nested_field
730 ) and "nestedShape" in nested_field:
731 for target_field in cast(
732 "list[dict[str, object]]", nested_field["nestedShape"]
733 ):
734 uri = target_field.get("uri")
735 if uri:
736 if uri not in intermediate_properties:
737 intermediate_properties[uri] = []
738 intermediate_properties[uri].append(target_field)
739 if target_field.get("subjectShape"):
740 target_shape = target_field["subjectShape"]
742 field_info["intermediateRelation"] = {
743 "class": intermediate_class,
744 "targetEntityType": target_entity_type,
745 "targetShape": target_shape,
746 "connectingProperty": connecting_property,
747 "properties": intermediate_properties,
748 }
751def handle_sub_display_rules(
752 shacl: Graph,
753 form_fields: dict[
754 tuple[str, str],
755 dict[str, list[dict[str, object]]],
756 ],
757 entity_key: tuple[str, str],
758 field_info_list: list[dict[str, object]],
759 prop: dict[str, object],
760) -> None:
761 """
762 Gestisce 'displayRules' nelle display_rules, applicando la regola corretta in base
763 allo shape.
765 Argomenti:
766 form_fields (dict): I campi del form da aggiornare.
767 entity_key (tuple): La chiave dell'entità (class, shape).
768 field_info_list (list): Le informazioni del campo originale.
769 prop (dict): Le informazioni della proprietà dalle display_rules.
770 """
771 new_field_info_list = []
772 entity_class = entity_key[0] if isinstance(entity_key, tuple) else entity_key
774 for original_field in field_info_list:
775 # Trova la display rule corrispondente allo shape del campo
776 sub_display_rules = cast("list[dict[str, object]]", prop["displayRules"])
777 matching_rule = next(
778 (
779 rule
780 for rule in sub_display_rules
781 if rule["shape"] == original_field["nodeShape"]
782 ),
783 None,
784 )
786 if matching_rule:
787 new_field = {
788 "entityType": entity_class,
789 "entityKey": entity_key, # Store the tuple key
790 "objectClass": original_field.get("objectClass"),
791 "uri": prop["property"],
792 "datatype": original_field.get("datatype"),
793 "min": original_field.get("min"),
794 "max": original_field.get("max"),
795 "hasValue": original_field.get("hasValue"),
796 "nodeShape": original_field.get("nodeShape"),
797 "nodeShapes": original_field.get("nodeShapes"),
798 "subjectShape": original_field.get("subjectShape"),
799 "nestedShape": original_field.get("nestedShape"),
800 "displayName": matching_rule["displayName"],
801 "optionalValues": original_field.get("optionalValues", []),
802 "orderedBy": original_field.get("orderedBy"),
803 "or": original_field.get("or", []),
804 }
806 if "intermediateRelation" in original_field:
807 new_field["intermediateRelation"] = original_field[
808 "intermediateRelation"
809 ]
811 # Aggiungi proprietà aggiuntive dalla shape SHACL
812 if "shape" in matching_rule:
813 shape_uri = str(matching_rule["shape"])
814 additional_properties = extract_additional_properties(shacl, shape_uri)
815 if additional_properties:
816 new_field["additionalProperties"] = additional_properties
818 new_field_info_list.append(new_field)
819 else:
820 # Se non c'è una regola corrispondente, mantieni il campo originale
821 new_field_info_list.append(original_field)
823 form_fields[entity_key][str(prop["property"])] = new_field_info_list
826def get_shape_target_class(shacl: Graph, shape_uri: str) -> str | None:
827 query = prepareQuery(
828 """
829 SELECT ?targetClass
830 WHERE {
831 ?shape sh:targetClass ?targetClass .
832 }
833 """,
834 initNs={"sh": "http://www.w3.org/ns/shacl#"},
835 )
836 results = execute_shacl_query(shacl, query, {"shape": URIRef(shape_uri)})
837 for row in select_results(results):
838 return str(row.targetClass)
839 return None
842def get_object_class(shacl: Graph, shape_uri: str, predicate_uri: str) -> str | None:
843 query = prepareQuery(
844 """
845 SELECT DISTINCT ?targetClass
846 WHERE {
847 ?shape sh:property ?propertyShape .
848 ?propertyShape sh:path ?predicate .
849 {
850 # Caso 1: definizione diretta con sh:node
851 ?propertyShape sh:node ?nodeShape .
852 ?nodeShape sh:targetClass ?targetClass .
853 } UNION {
854 # Caso 2: definizione diretta con sh:class
855 ?propertyShape sh:class ?targetClass .
856 } UNION {
857 # Caso 3: definizione con sh:or che include node shapes
858 ?propertyShape sh:or ?orList .
859 ?orList rdf:rest*/rdf:first ?choice .
860 {
861 ?choice sh:node ?nodeShape .
862 ?nodeShape sh:targetClass ?targetClass .
863 } UNION {
864 ?choice sh:class ?targetClass .
865 }
866 }
867 }
868 """,
869 initNs={
870 "sh": "http://www.w3.org/ns/shacl#",
871 "rdf": "http://www.w3.org/1999/02/22-rdf-syntax-ns#",
872 },
873 )
875 results = execute_shacl_query(
876 shacl, query, {"shape": URIRef(shape_uri), "predicate": URIRef(predicate_uri)}
877 )
879 # Prendiamo il primo risultato valido
880 for row in select_results(results):
881 if row.targetClass:
882 return str(row.targetClass)
883 return None
886def extract_shacl_form_fields(
887 shacl: Graph | None,
888 display_rules: list[dict[str, object]] | None,
889 app: Flask,
890) -> (
891 dict[
892 tuple[str, str],
893 dict[str, list[dict[str, object]]],
894 ]
895 | defaultdict[
896 tuple[str, str],
897 dict[str, list[dict[str, object]]],
898 ]
899):
900 """
901 Estrae i campi del form dalle shape SHACL.
903 Args:
904 shacl: The SHACL graph
905 display_rules: The display rules configuration
906 app: Flask application instance
908 Returns:
909 defaultdict: A dictionary where the keys are tuples (class, shape) and the
910 values are dictionaries
911 of form fields with their properties.
912 """
913 if not shacl:
914 return {}
916 ctx = ShaclProcessingContext(
917 shacl=shacl,
918 display_rules=display_rules,
919 app=app,
920 processed_shapes=set(),
921 )
922 results = execute_shacl_query(shacl, COMMON_SPARQL_QUERY)
923 return process_query_results(
924 ctx,
925 select_results(results),
926 depth=0,
927 )
930def execute_shacl_query(
931 shacl: Graph,
932 query: Query,
933 init_bindings: dict[str, URIRef] | None = None,
934) -> Result:
935 """
936 Esegue una query SPARQL sul grafo SHACL con eventuali binding iniziali.
938 Args:
939 shacl (Graph): The SHACL graph on which to execute the query.
940 query (PreparedQuery): The prepared SPARQL query.
941 init_bindings (dict): Initial bindings for the query.
943 Returns:
944 Result: The query results.
945 """
946 if init_bindings:
947 return shacl.query(query, initBindings=init_bindings)
948 return shacl.query(query)
951def extract_additional_properties(shacl: Graph, shape_uri: str) -> dict[str, str]:
952 """
953 Estrae proprietà aggiuntive da una shape SHACL.
955 Argomenti:
956 shape_uri (str): L'URI della shape SHACL.
958 Restituisce:
959 dict: Un dizionario delle proprietà aggiuntive.
960 """
961 additional_properties_query = prepareQuery(
962 """
963 SELECT ?predicate ?has_value
964 WHERE {
965 ?shape a sh:NodeShape ;
966 sh:property ?property .
967 ?property sh:path ?predicate .
968 { ?property sh:hasValue ?has_value . }
969 UNION
970 { ?property sh:qualifiedValueShape/sh:hasValue ?has_value . }
971 }
972 """,
973 initNs={"sh": "http://www.w3.org/ns/shacl#"},
974 )
976 additional_properties_results = shacl.query(
977 additional_properties_query,
978 initBindings={"shape": URIRef(shape_uri)},
979 )
981 additional_properties = {}
982 for row in select_results(additional_properties_results):
983 predicate = str(row.predicate)
984 has_value = str(row.has_value)
985 additional_properties[predicate] = has_value
987 return additional_properties