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

1# SPDX-FileCopyrightText: 2025 Arcangelo Massari <arcangelo.massari@unibo.it> 

2# 

3# SPDX-License-Identifier: ISC 

4 

5import json 

6from collections import OrderedDict, defaultdict 

7from collections.abc import Iterable 

8from dataclasses import dataclass 

9from pathlib import Path 

10from typing import cast 

11 

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 

17 

18from heritrace.sparql import select_results 

19from heritrace.utils.filters import Filter 

20 

21 

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] 

28 

29 

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] 

46 

47 

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) 

119 

120 

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] 

133 

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) 

144 

145 node_shapes = [] 

146 if node_shape: 

147 node_shapes.append(node_shape) 

148 node_shapes.extend(or_nodes) 

149 

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 ) 

166 

167 

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 

185 

186 

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) 

222 

223 

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] = [] 

233 

234 existing_field = _find_existing_field( 

235 form_fields[parsed.entity_key][parsed.predicate], parsed 

236 ) 

237 

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 

250 

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 } 

268 

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 ) 

275 

276 if parsed.or_nodes: 

277 _process_or_nodes(ctx, parsed, custom_filter, field_info, depth) 

278 

279 form_fields[parsed.entity_key][parsed.predicate].append(field_info) 

280 

281 

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 ) 

290 

291 with (Path(__file__).parent / "context.json").open() as config_file: 

292 context = json.load(config_file)["@context"] 

293 

294 custom_filter = Filter(context, ctx.display_rules, ctx.app.config["DATASET_DB_URL"]) 

295 

296 for row in results: 

297 parsed = _parse_row(row) 

298 _process_single_row(ctx, parsed, custom_filter, form_fields, depth) 

299 

300 return form_fields 

301 

302 

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 [] 

310 

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 = [] 

315 

316 temp_form_fields = process_query_results( 

317 ctx, 

318 select_results(nested_results), 

319 depth=depth, 

320 ) 

321 

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) 

327 

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]) 

331 

332 ctx.processed_shapes.remove(shape_uri) 

333 return nested_fields 

334 

335 

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. 

343 

344 Argomenti: 

345 entity_type (str): L'URI del tipo di entità. 

346 

347 Restituisce: 

348 list: Una lista di URI di proprietà nell'ordine desiderato. 

349 """ 

350 if not display_rules: 

351 return [] 

352 

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 [] 

366 

367 

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à. 

374 

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. 

378 

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 

386 

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)} 

389 

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 ) 

398 

399 

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 [] 

416 

417 

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 ] 

430 

431 

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] 

452 

453 

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. 

472 

473 Argomenti: 

474 form_fields (dict): I campi del form con possibili modifiche dalle regole di 

475 visualizzazione. 

476 

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 

496 

497 

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. 

511 

512 Argomenti: 

513 form_fields (dict): I campi del form iniziali estratti dalle shape SHACL. 

514 

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") 

522 

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 

540 

541 

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. 

553 

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 ) 

589 

590 

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 [] 

599 

600 # Handle case where parent_prop is not a dictionary 

601 if not isinstance(parent_prop, dict): 

602 return nested_fields 

603 

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) 

610 

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 

626 

627 return result_fields 

628 

629 

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" 

636 

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") 

652 

653 

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. 

660 

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"] 

679 

680 

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. 

688 

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"] 

696 

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 ) 

710 

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 ) 

718 

719 connecting_property = next( 

720 (str(row.property) for row in select_results(connecting_property_results)), None 

721 ) 

722 

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"] 

741 

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 } 

749 

750 

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. 

764 

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 

773 

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 ) 

785 

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 } 

805 

806 if "intermediateRelation" in original_field: 

807 new_field["intermediateRelation"] = original_field[ 

808 "intermediateRelation" 

809 ] 

810 

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 

817 

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) 

822 

823 form_fields[entity_key][str(prop["property"])] = new_field_info_list 

824 

825 

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 

840 

841 

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 ) 

874 

875 results = execute_shacl_query( 

876 shacl, query, {"shape": URIRef(shape_uri), "predicate": URIRef(predicate_uri)} 

877 ) 

878 

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 

884 

885 

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. 

902 

903 Args: 

904 shacl: The SHACL graph 

905 display_rules: The display rules configuration 

906 app: Flask application instance 

907 

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 {} 

915 

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 ) 

928 

929 

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. 

937 

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. 

942 

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) 

949 

950 

951def extract_additional_properties(shacl: Graph, shape_uri: str) -> dict[str, str]: 

952 """ 

953 Estrae proprietà aggiuntive da una shape SHACL. 

954 

955 Argomenti: 

956 shape_uri (str): L'URI della shape SHACL. 

957 

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 ) 

975 

976 additional_properties_results = shacl.query( 

977 additional_properties_query, 

978 initBindings={"shape": URIRef(shape_uri)}, 

979 ) 

980 

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 

986 

987 return additional_properties