Coverage for lode / viewer / base_viewer.py: 90%

529 statements  

« prev     ^ index     » next       coverage.py v7.13.0, created at 2026-10-05 16:02 +0000

1# base_viewer.py 

2import hashlib 

3import re 

4from typing import Dict, List, Optional, Tuple 

5from lode.models import Literal, Model, Resource, Statement 

6import urllib.parse 

7 

8from rdflib import Graph, URIRef, BNode, Literal as RDFlibLiteral 

9from rdflib.namespace import RDF, OWL 

10 

11from lode.reader.provenance_graph import ProvenanceGraph 

12from lode.helpers.rdf import iri_local_name 

13 

14SERIALIZATION_FORMATS = [ 

15 {"label": "Turtle", "fmt": "turtle", "mime": "text/turtle", "ext": "ttl", "icon": "bi-filetype-raw"}, 

16 {"label": "RDF/XML", "fmt": "xml", "mime": "application/rdf+xml", "ext": "rdf", "icon": "bi-file-code"}, 

17 {"label": "JSON-LD", "fmt": "json-ld", "mime": "application/ld+json", "ext": "jsonld", "icon": "bi-filetype-json"}, 

18 {"label": "N-Triples", "fmt": "nt", "mime": "application/n-triples", "ext": "nt", "icon": "bi-file-text"}, 

19] 

20 

21QUAD_SERIALIZATION_FORMATS = [ 

22 {"label": "TriG", "fmt": "trig", "mime": "application/trig", "ext": "trig", "icon": "bi-filetype-raw"}, 

23 {"label": "JSON-LD", "fmt": "json-ld", "mime": "application/ld+json", "ext": "jsonld", "icon": "bi-filetype-json"}, 

24 {"label": "N-Quads", "fmt": "nquads", "mime": "application/n-quads", "ext": "nq", "icon": "bi-file-text"}, 

25] 

26 

27def formats_for(graph): 

28 return QUAD_SERIALIZATION_FORMATS if isinstance(graph, ProvenanceGraph) else SERIALIZATION_FORMATS 

29 

30class BaseViewer: 

31 """Base viewer per visualizzare istanze estratte dal Reader.""" 

32 

33 def __init__(self, reader): 

34 self.reader = reader 

35 self._cache = reader._instance_cache # it uses  

36 self._internal_iris = None 

37 self._ontology_ns = None 

38 

39 def _get_ontology_ns(self) -> str: 

40 if self._ontology_ns is None: 

41 self._ontology_ns = '' 

42 for inst in self.get_all_instances(): 

43 if isinstance(inst, Model): 

44 nid = inst.get_has_identifier() 

45 if nid: 

46 self._ontology_ns = nid if nid.endswith(('/', '#')) else nid + '/' 

47 break 

48 return self._ontology_ns 

49 

50 def _has_local_triples(self, uri) -> bool: 

51 if not uri: 

52 return False 

53 return (URIRef(str(uri)), None, None) in self.reader._graph 

54 

55 def _formats(self): 

56 return formats_for(self.reader._graph) 

57 

58 def _with_provenance(self, sub): 

59 """Con moduli attivi: ricolloca le triple di `sub` nei named graph d'origine.""" 

60 ds = self.reader._graph 

61 if not isinstance(ds, ProvenanceGraph): 

62 return sub 

63 out = ProvenanceGraph() 

64 for prefix, ns in ds.namespaces(): 

65 out.bind(prefix, ns) 

66 for s, p, o in sub: 

67 for *_, c in ds.quads((s, p, o, None)): 

68 out.graph(c).add((s, p, o)) 

69 return out 

70 

71 def _is_internal(self, uri) -> bool: 

72 if not uri: 

73 return False 

74 if self._internal_iris is None: 

75 self._internal_iris = { 

76 str(i.get_has_identifier()) 

77 for i in self.get_toc_instances() 

78 if i.get_has_identifier() 

79 } 

80 return str(uri) in self._internal_iris 

81 

82 def get_all_instances(self) -> List: 

83 """Ottiene tutte le istanze (esclusi literal).""" 

84 instances = [] 

85 

86 for uri_id, instance_set in self._cache.items(): 

87 if isinstance(uri_id, str) and uri_id.startswith("LITERAL::"): 

88 continue 

89 

90 instance_list = instance_set if isinstance(instance_set, set) else [instance_set] 

91 instances.extend(instance_list) 

92 

93 return instances 

94 

95 def get_instances_from_single_resource(self, resource_uri: str) -> Optional[set]: 

96 """Ottiene istanze per un URI specifico dalla cache.""" 

97 # Cerca l'URI nella cache 

98 for uri_id in self._cache.keys(): 

99 if str(uri_id) == resource_uri: 

100 return self._cache[uri_id] 

101 

102 return None 

103 

104 def _get_best_label(self, resource, language: Optional[str] = None) -> Optional[str]: 

105 """Gets the best label to display: language > preferred_label > label > identifier.""" 

106 

107 # 1. Force 'en' as the absolute default if no language makes it this far 

108 target_lang = language.strip().lower() if language else "en" 

109 

110 # Check Preferred Labels for the target language 

111 labels = resource.get_has_preferred_label() 

112 for label in labels: 

113 if hasattr(label, 'get_has_language') and label.get_has_language(): 

114 if label.get_has_language().lower().startswith(target_lang): 

115 # RETURN RAW LABEL 

116 return label.get_has_value() 

117 

118 # Check Normal Labels for the target language 

119 labels = resource.get_has_label() 

120 for label in labels: 

121 if hasattr(label, 'get_has_language') and label.get_has_language(): 

122 if label.get_has_language().lower().startswith(target_lang): 

123 # RETURN RAW LABEL 

124 return label.get_has_value() 

125 

126 # --- DETERMINISTIC FALLBACKS --- 

127 # If we reach here, no English label exists. 

128 # We sort them by language tag. 

129 

130 labels = resource.get_has_preferred_label() 

131 if labels: 

132 sorted_labels = sorted(labels, key=lambda x: str(x.get_has_language() or "")) 

133 # RETURN RAW LABEL 

134 return sorted_labels[0].get_has_value() 

135 

136 labels = resource.get_has_label() 

137 if labels: 

138 sorted_labels = sorted(labels, key=lambda x: str(x.get_has_language() or "")) 

139 # RETURN RAW LABEL 

140 return sorted_labels[0].get_has_value() 

141 

142 # Final Fallback: The URI Identifier 

143 resource_id = resource.get_has_identifier() 

144 if resource_id: 

145 clean_resource_id = iri_local_name(resource_id) 

146 # ONLY clean the identifier! 

147 return self._clean_name(self, clean_resource_id) 

148 

149 return None 

150 

151 def get_view_data(self, resource_uri: Optional[str] = None, language: Optional[str] = None) -> Dict: 

152 """ 

153 Main entry point called by the API 

154 Subclasses should override this to define their specific view strategy 

155 """ 

156 # If no language is provided, the default would be English 

157 language = language.strip() if language else "en" 

158 

159 # Fallback: generic flat list 

160 all_instances = self.get_all_instances() 

161 metadata_dict = self._find_and_format_metadata(all_instances, language) 

162 

163 if resource_uri: 

164 data = self._handle_single_resource(resource_uri, language) 

165 data['metadata'] = metadata_dict 

166 data['export_formats'] = self._formats() 

167 return data 

168 

169 return { 

170 'metadata': metadata_dict, 

171 'entities': self._format_entities(all_instances, language), 

172 'export_formats': self._formats(), 

173 } 

174 

175 def _is_toc_entity(self, instance) -> bool: 

176 """Browsable (own card + clickable link) iff its type is in get_toc_config.  

177 Resources not listed in get_o_config are shown when mentioned as plain text  

178 in cards (external-ref), just never linked.""" 

179 toc_keys = {key for key, _id, _title in self.get_toc_config()} 

180 if type(instance).__name__ not in toc_keys: 

181 return False 

182 return self._has_local_triples(instance.get_has_identifier()) 

183 

184 def get_toc_instances(self) -> List: 

185 return [i for i in self.get_all_instances() if self._is_toc_entity(i)] 

186 

187 def _handle_single_resource(self, resource_uri: str, language: Optional[str] = None) -> Dict: 

188 """ 

189 Standard logic for displaying a single resource. 

190 Returns the specific dictionary structure required by viewer.html 

191 """ 

192 instance_set = self.get_instances_from_single_resource(resource_uri) 

193 

194 if not instance_set: 

195 return {'error': f'Resource {resource_uri} not found'} 

196 

197 instances = list(instance_set) if isinstance(instance_set, set) else [instance_set] 

198 instances = [i for i in instances if self._is_toc_entity(i)] 

199 if not instances: 

200 return {'error': f'{resource_uri} is not a browsable entity'} 

201 

202 return { 

203 'single_resource': True, 

204 'entities': self._format_entities(instances, language), 

205 'groupped_view': False, 

206 'sections': None 

207 } 

208 

209 def _build_grouped_view(self, group_definitions: List[Tuple[str, str, str]], language: Optional[str] = None) -> Dict: 

210 """ 

211 Constructs the 'Table of Contents' view. 

212 Adds optional 'tree' to sections whose class_key supports hierarchy. 

213 """ 

214 all_instances = self.get_all_instances() 

215 sections = [] 

216 

217 for class_key, section_id, section_title in group_definitions: 

218 instances = [ 

219 inst for inst in all_instances 

220 if type(inst).__name__ == class_key and self._is_toc_entity(inst) 

221 ] 

222 

223 if instances: 

224 section = { 

225 'id': section_id, 

226 'title': section_title, 

227 'entities': self._format_entities(instances, language), 

228 } 

229 

230 parent_getter = self._HIERARCHY_PARENT_GETTERS.get(class_key) 

231 if parent_getter: 

232 section['tree'] = self._build_hierarchy(instances, parent_getter, language) 

233 

234 sections.append(section) 

235 

236 return { 

237 'grouped_view': True, 

238 'sections': sections, 

239 'export_formats': self._formats() 

240 } 

241 

242 def _format_entities(self, instances: List, language: Optional[str] = None) -> List[Dict]: 

243 """ 

244 Converts Python Models -> HTML Template Dictionary. 

245 Ensures consistent keys ('type', 'uri', 'label') across all viewers. 

246 """ 

247 all_instances = self.get_all_instances() 

248 

249 entities = [] 

250 for instance in instances: 

251 uri = instance.has_identifier if hasattr(instance, 'has_identifier') else None 

252 

253 if not uri: 

254 continue 

255 

256 #Create a safe HTML ID to facilitate on-page navigation 

257 safe_id = hashlib.md5(str(uri).encode('utf-8')).hexdigest() 

258 

259 # Extract internal attributes (SuperClasses, etc.) 

260 relations = {} 

261 if hasattr(instance, '__dict__'): 

262 for attr, value in instance.__dict__.items(): 

263 if not attr.startswith('_') and value: 

264 # Skip attributes that are handled elsewhere or are empty 

265 # Clean up name: 

266 clean_name = self._clean_name(self,attr) 

267 

268 # Process value (could be a list of objects)clean 

269 # We use the helper to get clean text for each item 

270 formatted_values = [] 

271 if isinstance(value, list): 

272 for v in value: 

273 # Nested list (es. has_property_chain: lista di catene) 

274 if isinstance(v, list): 

275 chain_dict = self._resolve_chain_value(v, language) 

276 if chain_dict and chain_dict['text']: 

277 formatted_values.append(chain_dict) 

278 continue 

279 # Normal list 

280 val_dict = self._resolve_resource_value(v, language) 

281 if val_dict['text']: formatted_values.append(val_dict) 

282 else: 

283 val_dict = self._resolve_resource_value(value, language) 

284 if val_dict['text']: formatted_values.append(val_dict) 

285 

286 if formatted_values: 

287 if attr in self._FULL_IRI_ATTRS: 

288 for fv in formatted_values: 

289 if fv.get('link'): 

290 fv['text'] = str(fv['link']) 

291 if clean_name not in relations: 

292 relations[clean_name] = [] 

293 for v in formatted_values: 

294 if v not in relations[clean_name]: # Valentina FIX: this do not add duplicates in metadata values  

295 relations[clean_name].append(v) 

296 # Sort relations 

297 ordered_relations = {} 

298 

299 # Define the exact order you 

300 priority_order = [ 

301 "is sub property of", 

302 "domain", 

303 "range", 

304 "is inverse Of", 

305 "property chain", 

306 "is disjoint with" 

307 ] 

308 

309 # Extract and insert the priority keys first 

310 for key in priority_order: 

311 if key in relations: 

312 ordered_relations[key] = relations[key] 

313 

314 # Insert all remaining keys alphabetically 

315 for key in sorted(relations.keys()): 

316 if key not in ordered_relations: 

317 ordered_relations[key] = relations[key] 

318 

319 characteristics = {} 

320 char_attributes = [ 

321 'is_functional', 'is_inverse_functional', 'is_transitive', 

322 'is_symmetric', 'is_asymmetric', 'is_reflexive', 'is_irreflexive' 

323 ] 

324 

325 for attr in char_attributes: 

326 getter = f"get_{attr}" 

327 # Safely get the boolean value 

328 is_true = getattr(instance, getter)() if hasattr(instance, getter) else getattr(instance, attr, False) 

329 

330 # Convert 'is_asymmetric' to 'is asymmetric' for the HTML label 

331 if is_true: 

332 display_name = attr.replace('_', ' ') 

333 characteristics[display_name] = True 

334 

335 # Extract Statement Entities 

336 statements = self._format_statement(all_instances, instance, language) 

337 type_inst = type(instance).__name__.replace(" ", "_") 

338 

339 is_dep = getattr(instance, 'get_is_deprecated')() if hasattr(instance, 'get_is_deprecated') else getattr( 

340 instance, 'is_deprecated', False) 

341 

342 entities.append({ 

343 'type': type_inst, 

344 'uri': uri, 

345 'label': self._get_best_label(instance, language), 

346 'anchor_id': f"id_{safe_id}_{type_inst}", 

347 'relations': ordered_relations, 

348 'statements': statements, 

349 'characteristics': characteristics, 

350 'is_deprecated': bool(is_dep), 

351 'provenance': self._build_provenance_subgraph(instance), 

352 'is_external': not self._is_internal(uri), 

353 }) 

354 

355 entities.sort(key=lambda x: (x['label'] or x['uri']).lower()) 

356 return entities 

357 

358 def _find_and_format_metadata(self, all_instances: List[Resource], language=None) -> Dict: 

359 """ 

360 Searches for the Model and its Statements and formatting them 

361 for the template. 

362 """ 

363 ontology_model = None 

364 

365 # 1. Find the Model (skip imported/closure ontologies) 

366 imported = getattr(self.reader, "_imported_uris", set()) or set() 

367 models = [i for i in all_instances if isinstance(i, Model)] 

368 ontology_model = next( 

369 (m for m in models if str(m.get_has_identifier()) not in imported), 

370 models[0] if models else None, 

371 ) 

372 

373 if not ontology_model: 

374 return {} 

375 

376 # 2. Prepare Output 

377 data = { 

378 'uri': [self._resolve_resource_value(ontology_model, language)], 

379 'label': [self._resolve_resource_value(self._get_best_label(ontology_model, language))] 

380 } 

381 

382 # 3: Dynamic Extraction of Structural Data --- 

383 # Methods we want to skip because they belong in the Header or Annotations 

384 ignore_methods = [ 

385 'get_has_identifier', 

386 'get_has_label', 

387 'get_has_subject', 

388 'get_has_predicate', 

389 'get_has_object', 

390 'get_has_namespaces', 

391 ] 

392 entry = {'text': None, 'link': None} 

393 # 1. Loop through ALL attributes and methods of the Model object 

394 for attr_name in dir(ontology_model): 

395 # 2. Look specifically for getter methods 

396 if attr_name.startswith('get_') and attr_name not in ignore_methods: 

397 

398 method = getattr(ontology_model, attr_name) 

399 

400 if callable(method): 

401 try: 

402 values = method() 

403 except AttributeError: 

404 # skip attributes that are not initialized 

405 continue 

406 

407 if values: 

408 # 4. Auto-format the key name 

409 clean_key = self._clean_name(self, attr_name) 

410 

411 # 5. Ensure values are in a list 

412 if not isinstance(values, list): 

413 values = [values] 

414 

415 # 6. Extract the actual text values 

416 extracted_values = [] 

417 for val in values: 

418 entry = self._resolve_resource_value(val, language) 

419 if entry['text']: 

420 # 7. Check if the link if an visualizable ontology or not 

421 if entry.get('link'): 

422 entry['is_visualizable'] = self._is_likely_ontology(entry['link']) 

423 else: 

424 entry['is_visualizable'] = False 

425 

426 extracted_values.append(entry) 

427 

428 # 8. Add to structural data ONLY if we found valid text 

429 if extracted_values: 

430 data[clean_key] = extracted_values 

431 

432 # Namespaces (dict prefix -> URI), handled explicitly: not a Resource 

433 ns = ontology_model.get_has_namespaces() 

434 if ns: 

435 data['_namespaces'] = [ 

436 {'prefix': prefix or 'default namespace', 'uri': uri} 

437 for prefix, uri in sorted(ns.items(), key=lambda kv: (kv[0] or '')) 

438 ] 

439 

440 # 9. Statements 

441 data.update(self._format_statement(all_instances, ontology_model, language)) 

442 

443 # Ordine alfabetico su tutti i metadati (uri/label in testa) 

444 pinned = ['uri', 'label'] 

445 rest = sorted(k for k in data if k not in pinned) 

446 data = {k: data[k] for k in pinned if k in data} | {k: data[k] for k in rest} 

447 

448 return data 

449 

450 def _resolve_chain_value(self, chain, language=None) -> dict: 

451 """Renderizza una property chain ordinata (lista di Relation) come un 

452 unico valore con parti cliccabili unite dall'operatore di composizione.""" 

453 members = [] 

454 for m in chain: 

455 md = self._resolve_resource_value(m, language) 

456 if md.get('text'): 

457 members.append(md) 

458 if not members: 

459 return None 

460 parts = [] 

461 for i, md in enumerate(members): 

462 if i > 0: 

463 parts.append({'text': ' o ', 'link': None}) # ring operator 

464 parts.append(md) 

465 return { 

466 'text': ' o '.join(md['text'] for md in members), 

467 'link': None, 

468 'lan': None, 

469 'parts': parts, 

470 'type': 'PropertyChain', 

471 } 

472 

473 def _resolve_resource_value(self, obj, language=None) -> dict: 

474 """Helper: Extracts text and link from any object.""" 

475 handler_dic = { 

476 'text': None, 

477 'link': None, 

478 'lan': None, 

479 'parts': None, # This key is for restrictions 

480 'type': None, 

481 'is_deprecated': False, 

482 'is_imported': False, 

483 'is_external': False, 

484 } 

485 

486 if not obj: return handler_dic 

487 

488 # --- 1. INTERCEPT RESTRICTIONS --- 

489 restriction_types = ["Restriction", "PropertyConceptRestriction", "Quantifier", "Cardinality", "TruthFunction", 

490 "OneOf", "Value", "DatatypeRestriction"] 

491 obj_type = type(obj).__name__ 

492 

493 if obj_type in restriction_types: 

494 is_dep = getattr(obj, 'get_is_deprecated')() if hasattr(obj, 'get_is_deprecated') else getattr(obj, 

495 'is_deprecated', 

496 False) 

497 handler_dic['is_deprecated'] = bool(is_dep) 

498 

499 # Recursively parse the restriction into clickable parts 

500 parts = self._parse_restriction(obj, language) 

501 

502 handler_dic['parts'] = parts 

503 handler_dic['text'] = "".join([p['text'] for p in parts if p.get('text')]) 

504 handler_dic['link'] = None # Forces Jinja to ignore the blank node URI 

505 handler_dic['is_external'] = False 

506 handler_dic['is_imported'] = False 

507 handler_dic['type'] = obj_type 

508 return handler_dic 

509 

510 # --- 2. String Handling --- 

511 if isinstance(obj, str): 

512 handler_dic['text'] = obj 

513 return handler_dic 

514 

515 # --- 3. Literal Handling --- 

516 if type(obj).__name__ == 'Literal': 

517 if hasattr(obj, 'get_has_value') and obj.get_has_value(): 

518 

519 is_dep = getattr(obj, 'get_is_deprecated')() if hasattr(obj, 'get_is_deprecated') else getattr(obj, 

520 'is_deprecated', 

521 False) 

522 handler_dic['is_deprecated'] = bool(is_dep) 

523 

524 lit_lang = obj.get_has_language() 

525 

526 # If a language is requested (e.g., 'pt') 

527 if language: 

528 target_lang = language.strip().lower() 

529 

530 # If the literal has a tag, and it DOES NOT match 'pt', destroy it. 

531 if lit_lang and not lit_lang.lower().startswith(target_lang): 

532 return handler_dic 

533 

534 # if not lit_lang: 

535 # return handler_dic 

536 

537 handler_dic['text'] = obj.get_has_value() 

538 handler_dic['lan'] = lit_lang 

539 handler_dic['type'] = obj_type 

540 return handler_dic 

541 

542 raw_str = str(obj) 

543 if "object at" not in raw_str: 

544 handler_dic['text'] = raw_str 

545 handler_dic['type'] = obj_type 

546 return handler_dic 

547 

548 # --- 4. Statement (nested annotation) --- 

549 if obj_type == 'Statement': 

550 handler_dic['type'] = 'Statement' 

551 handler_dic['link'] = None # don't link to the BNode id 

552 inner_pred = obj.get_has_predicate() 

553 inner_obj = obj.get_has_object() 

554 

555 # Resolve the inner (predicate, value) pair recursively 

556 inner = { 

557 'predicate': self._get_best_label(inner_pred, language) if inner_pred else None, 

558 'value': self._resolve_resource_value(inner_obj, language) if inner_obj else None, 

559 } 

560 # Walk Statement's own annotations (anything else in __dict__ besides 

561 # the reification slots), so dc:date, rdf:value siblings show up too. 

562 skip = {'has_subject', 'has_predicate', 'has_object', 

563 'has_identifier', 'is_positive_statement'} 

564 siblings = [] 

565 for attr, value in obj.__dict__.items(): 

566 if attr.startswith('_') or attr in skip or not value: 

567 continue 

568 values = value if isinstance(value, list) else [value] 

569 for v in values: 

570 resolved = self._resolve_resource_value(v, language) 

571 if resolved.get('text') or resolved.get('nested'): 

572 siblings.append({ 

573 'predicate': self._clean_name(self, attr), 

574 'value': resolved, 

575 }) 

576 handler_dic['nested'] = {'main': inner, 'siblings': siblings} 

577 # Pick a primary text for collapsed display: the inner value's text 

578 handler_dic['text'] = (inner.get('value') or {}).get('text') or '[Statement]' 

579 return handler_dic 

580 

581 # --- 5. Normal Resource Handling (Concepts, Properties, Individuals) --- 

582 if hasattr(obj, 'get_has_identifier'): 

583 handler_dic['link'] = obj.get_has_identifier() 

584 link = handler_dic['link'] 

585 ns = self._get_ontology_ns() 

586 internal = self._is_internal(link) 

587 native = bool(link) and bool(ns) and str(link).startswith(ns) 

588 handler_dic['is_external'] = not internal 

589 handler_dic['is_imported'] = internal and not native 

590 is_dep = getattr(obj, 'get_is_deprecated')() if hasattr(obj, 'get_is_deprecated') else getattr(obj, 'is_deprecated', False) 

591 handler_dic['is_deprecated'] = bool(is_dep) 

592 

593 try: 

594 handler_dic['text'] = self._get_best_label(obj, language) 

595 handler_dic['type'] = obj_type 

596 except AttributeError: 

597 handler_dic['text'] = handler_dic['link'] 

598 handler_dic['type'] = obj_type 

599 

600 # Fallbacks 

601 if not handler_dic['text'] and handler_dic['link']: 

602 handler_dic['text'] = handler_dic['link'] 

603 handler_dic['type'] = obj_type 

604 

605 return handler_dic 

606 

607 def _format_statement(self, instances, target_instance, language=None) -> Dict: 

608 """ 

609 Extracts all statements where the subject matches the given target_instance. 

610 """ 

611 statements = {} 

612 

613 # 1. Validate the target instance 

614 if target_instance is None: 

615 return statements 

616 

617 for instance in instances: 

618 if isinstance(instance, Statement): 

619 subj = instance.get_has_subject() 

620 

621 # 2. Identity comparison (handles punning: same URI, different Python instances) 

622 if subj is not target_instance: 

623 continue 

624 

625 predicate = instance.get_has_predicate() 

626 obj = instance.get_has_object() 

627 

628 if obj: 

629 obj_data = self._resolve_resource_value(obj, language) 

630 

631 # 3. Predicate Resolution 

632 pred_label = self._get_best_label(predicate, language) if predicate else "Annotation" 

633 

634 if pred_label not in statements: 

635 statements[pred_label] = [] 

636 

637 # 4. Resolve Object and Prevent Duplicates 

638 if obj: 

639 obj_data = self._resolve_resource_value(obj, language) 

640 

641 

642 if obj_data not in statements[pred_label]: 

643 statements[pred_label].append(obj_data) 

644 

645 return statements 

646 

647 @staticmethod 

648 def _clean_name(self, name: str) -> str: 

649 if not name: return "" 

650 

651 name = re.sub(r'^(get_has_|get_|has_)', '', name) 

652 # Put space between camelCase letters 

653 name = re.sub(r'([a-z0-9])([A-Z])', r'\1 \2', name) 

654 name = name.replace('_', ' ') 

655 

656 # Strip extra spaces 

657 name = ' '.join(name.split()) 

658 

659 # lowercase everything 

660 name = name.lower() 

661 

662 return name 

663 

664 def _parse_restriction(self, obj, language=None, _seen=None) -> list: 

665 """ 

666 Recursively unpacks nested restrictions into a list of display parts 

667 (each part being a dict with 'text' and 'link'). 

668 Guards against cyclic class expressions (possible once owl:imports 

669 completes cross-ontology definitions) via a visited set. 

670 """ 

671 if not obj: return [] 

672 

673 if _seen is None: 

674 _seen = set() 

675 

676 # 1. Handle lists of restrictions/concepts (e.g., in TruthFunctions or OneOf) 

677 if isinstance(obj, list) or isinstance(obj, set): 

678 parts = [] 

679 for i, item in enumerate(obj): 

680 if i > 0: 

681 parts.append({'text': ', ', 'link': None}) 

682 parts.extend(self._parse_restriction(item, language, _seen)) 

683 return parts 

684 

685 obj_type = type(obj).__name__ 

686 restriction_types = ["Restriction", "PropertyConceptRestriction", "Quantifier", "Cardinality", "TruthFunction", 

687 "OneOf", "Value", "DatatypeRestriction"] 

688 

689 # 2. If it is a Restriction, recursively unpack its specific components 

690 if obj_type in restriction_types: 

691 # cycle guard: a class expression already being expanded upstream 

692 if id(obj) in _seen: 

693 return [{'text': '…', 'link': None}] 

694 _seen.add(id(obj)) 

695 

696 parts = [] 

697 

698 # Helper to safely call getter methods (e.g., get_applies_on_property) 

699 def _get(instance, prop_name, default=None): 

700 getter = f"get_{prop_name}" 

701 if hasattr(instance, getter): 

702 res = getattr(instance, getter)() 

703 return res if res is not None else default 

704 return getattr(instance, prop_name, default) 

705 

706 if obj_type == "Quantifier": 

707 prop = _get(obj, 'applies_on_property') 

708 quant = _get(obj, 'has_quantifier_type', "some") 

709 concept = _get(obj, 'applies_on_concept') 

710 

711 # Safely check if the is_inverse flag is True 

712 is_inv = obj.get_is_inverse() if hasattr(obj, 'get_is_inverse') else getattr(obj, 'is_inverse', False) 

713 if is_inv: 

714 parts.append({'text': 'inverse of ', 'link': None, 'type': 'Text'}) 

715 

716 parts.extend(self._parse_restriction(prop, language, _seen)) 

717 parts.append({'text': f' {quant} ', 'link': None}) 

718 parts.extend(self._parse_restriction(concept, language, _seen)) 

719 

720 elif obj_type == "Cardinality": 

721 prop = _get(obj, 'applies_on_property') 

722 card = _get(obj, 'has_cardinality_type', "exactly") 

723 card_num = _get(obj, 'has_cardinality') 

724 concept = _get(obj, 'applies_on_concept') 

725 

726 # Safely check if the is_inverse flag is True 

727 is_inv = obj.get_is_inverse() if hasattr(obj, 'get_is_inverse') else getattr(obj, 'is_inverse', False) 

728 if is_inv: 

729 parts.append({'text': 'inverse of ', 'link': None, 'type': 'Text'}) 

730 

731 parts.extend(self._parse_restriction(prop, language, _seen)) 

732 parts.append({'text': f' {card} {card_num}', 'link': None}) 

733 parts.extend(self._parse_restriction(concept, language, _seen)) 

734 

735 elif obj_type == "TruthFunction": 

736 operator = _get(obj, 'has_logical_operator', "and") 

737 concepts = _get(obj, 'applies_on_concept', []) 

738 if not isinstance(concepts, list): concepts = [concepts] 

739 

740 # Safely check if the is_inverse flag is True 

741 is_inv = obj.get_is_inverse() if hasattr(obj, 'get_is_inverse') else getattr(obj, 'is_inverse', False) 

742 if is_inv: 

743 parts.append({'text': 'inverse of ', 'link': None, 'type': 'Text'}) 

744 

745 parts.append({'text': '(', 'link': None}) 

746 for i, c in enumerate(concepts): 

747 if i > 0: 

748 parts.append({'text': f' {operator} ', 'link': None}) 

749 parts.extend(self._parse_restriction(c, language, _seen)) 

750 parts.append({'text': ')', 'link': None}) 

751 

752 elif obj_type == "OneOf": 

753 resources = _get(obj, 'applies_on_resource', []) 

754 if not isinstance(resources, list): resources = [resources] 

755 

756 # Safely check if the is_inverse flag is True 

757 is_inv = obj.get_is_inverse() if hasattr(obj, 'get_is_inverse') else getattr(obj, 'is_inverse', False) 

758 if is_inv: 

759 parts.append({'text': 'inverse of ', 'link': None, 'type': 'Text'}) 

760 

761 parts.append({'text': 'one of { ', 'link': None}) 

762 for i, r in enumerate(resources): 

763 if i > 0: 

764 parts.append({'text': ', ', 'link': None}) 

765 parts.extend(self._parse_restriction(r, language, _seen)) 

766 parts.append({'text': ' }', 'link': None}) 

767 

768 elif obj_type == "Value": 

769 prop = _get(obj, 'applies_on_property') 

770 resource = _get(obj, 'applies_on_resource') 

771 

772 # Safely check if the is_inverse flag is True 

773 is_inv = obj.get_is_inverse() if hasattr(obj, 'get_is_inverse') else getattr(obj, 'is_inverse', False) 

774 if is_inv: 

775 parts.append({'text': 'inverse of ', 'link': None, 'type': 'Text'}) 

776 

777 parts.extend(self._parse_restriction(prop, language, _seen)) 

778 parts.append({'text': ' value ', 'link': None}) 

779 parts.extend(self._parse_restriction(resource, language, _seen)) 

780 

781 elif obj_type == "DatatypeRestriction": 

782 concept = _get(obj, 'applies_on_concept') # Datatype(xsd:string) 

783 constraint = _get(obj, 'has_constraint') # "pattern" 

784 value = _get(obj, 'has_restriction_value') # Literal("[0-9]+") 

785 

786 # Safely check if the is_inverse flag is True 

787 is_inv = obj.get_is_inverse() if hasattr(obj, 'get_is_inverse') else getattr(obj, 'is_inverse', False) 

788 if is_inv: 

789 parts.append({'text': 'inverse of ', 'link': None, 'type': 'Text'}) 

790 

791 parts.extend(self._parse_restriction(concept, language, _seen)) 

792 if constraint: 

793 parts.append({'text': f' with {constraint} ', 'link': None, 'type': 'Text'}) 

794 else: 

795 parts.append({'text': ' restricted by ', 'link': None, 'type': 'Text'}) 

796 

797 parts.extend(self._parse_restriction(value, language, _seen)) 

798 

799 return parts 

800 

801 # 3. Base Case: We hit a non-blank node (Concept, Relation, String) 

802 # Send it to the main resolver to extract its URI link and clean text. 

803 resolved = self._resolve_resource_value(obj, language) 

804 

805 if resolved.get('text'): 

806 return [{'text': resolved['text'], 'link': resolved.get('link'), 'type': resolved.get('type'), 'is_external': resolved.get('is_external', False)}] 

807 

808 return [] 

809 

810 def _is_likely_ontology(self, url: str) -> bool: 

811 """Determines if a URL likely points to an ontology based on naming patterns.""" 

812 if not url or not isinstance(url, str) or not url.startswith('http'): 

813 return False 

814 

815 # 1. Clean the string 

816 clean_url = url.strip().lower() 

817 

818 # 2. Safely parse the URL to ignore fragments (#) and queries (?) for extension checking 

819 parsed = urllib.parse.urlparse(clean_url) 

820 path = parsed.path.rstrip('/') 

821 

822 # 3. Standard Ontology File Extensions 

823 ontology_exts = ['.owl', '.rdf', '.ttl', '.nt', '.n3', '.jsonld'] 

824 if any(path.endswith(ext) for ext in ontology_exts): 

825 return True 

826 

827 # Match registries against the parsed hostname, not the raw string: 

828 # a substring check would also match e.g. https://evil.com/w3id.org/x 

829 host = parsed.hostname or '' 

830 

831 def _host_is(domain: str) -> bool: 

832 return host == domain or host.endswith('.' + domain) 

833 

834 # 4. Common Ontology Registries & Permanent URLs 

835 ontology_domains = [ 

836 'w3id.org', 

837 'purl.org', 

838 'xmlns.com' # Often used for FOAF 

839 ] 

840 if any(_host_is(domain) for domain in ontology_domains): 

841 return True 

842 

843 # 5. Core W3C Namespaces 

844 w3c_core = ['/1999/02/22-rdf-syntax-ns', '/2000/01/rdf-schema', '/2002/07/owl'] 

845 if _host_is('w3.org') and any(ns in path for ns in w3c_core): 

846 return True 

847 

848 # 6. Broad Keywords & Famous Standard Acronyms 

849 # Looking for these anywhere in the URL string 

850 path_keywords = [ 

851 'ontology', 'vocab', 'schema', # General terms 

852 'skos', 'foaf', 'crm', 'cidoc' # Famous standards 

853 ] 

854 if any(keyword in clean_url for keyword in path_keywords): 

855 return True 

856 

857 return False 

858 

859 # Mappa class_key -> nome del getter del parent (None = no hierarchy) 

860 _HIERARCHY_PARENT_GETTERS = { 

861 'Concept': 'get_is_sub_concept_of', 

862 'Relation': 'get_is_sub_property_of', 

863 'Attribute': 'get_is_sub_property_of', 

864 'Property': 'get_is_sub_property_of', 

865 # Annotation, Individual: nessuna gerarchia -> non in mappa 

866 } 

867 

868 _FULL_IRI_ATTRS = {'is_defined_by'} 

869 

870 def _build_hierarchy(self, instances, parent_getter_name, language=None): 

871 """ 

872 Costruisce un albero (forest) di dict. 

873 anchor_id usa lo stesso formato di _format_entities: id_{md5}_{ClassName} 

874 """ 

875 import hashlib 

876 

877 by_uri = {} 

878 for inst in instances: 

879 uri = getattr(inst, 'has_identifier', None) 

880 if uri: 

881 by_uri[str(uri)] = inst 

882 

883 children_map = {uri: [] for uri in by_uri} 

884 has_internal_parent = set() 

885 

886 for uri, inst in by_uri.items(): 

887 getter = getattr(inst, parent_getter_name, None) 

888 if not callable(getter): 

889 continue 

890 parents = getter() or [] 

891 for p in parents: 

892 p_uri = getattr(p, 'has_identifier', None) 

893 if p_uri and str(p_uri) in by_uri and str(p_uri) != uri: 

894 children_map[str(p_uri)].append(uri) 

895 has_internal_parent.add(uri) 

896 

897 roots = [uri for uri in by_uri if uri not in has_internal_parent] 

898 

899 def node_dict(uri, visited): 

900 if uri in visited: 

901 return None 

902 visited = visited | {uri} 

903 inst = by_uri[uri] 

904 label = self._get_best_label(inst, language) or uri 

905 type_inst = type(inst).__name__.replace(" ", "_") 

906 safe_id = hashlib.md5(str(uri).encode('utf-8')).hexdigest() 

907 anchor_id = f"id_{safe_id}_{type_inst}" 

908 kids = [] 

909 for child_uri in sorted( 

910 children_map.get(uri, []), 

911 key=lambda u: (self._get_best_label(by_uri[u], language) or u).lower() 

912 ): 

913 child_node = node_dict(child_uri, visited) 

914 if child_node: 

915 kids.append(child_node) 

916 return {'label': label, 'anchor_id': anchor_id, 'uri': uri, 'children': kids} 

917 

918 tree = [] 

919 for uri in sorted(roots, key=lambda u: (self._get_best_label(by_uri[u], language) or u).lower()): 

920 node = node_dict(uri, set()) 

921 if node: 

922 tree.append(node) 

923 return tree 

924 

925 # ========== PROVENANCE: subgraph serialisation for each card ========== 

926 

927 def _build_provenance_subgraph(self, instance) -> List[Dict[str, str]]: 

928 sub = self.reader.get_provenance_subgraph(instance) 

929 if not any(True for _ in sub): 

930 return [] 

931 sub = self._with_provenance(sub) 

932 return [ 

933 {"label": f["label"], "code": self._safe_serialize(sub, f["fmt"])} 

934 for f in self._formats() 

935 ] 

936 

937 @staticmethod 

938 def _safe_serialize(g, fmt: str) -> str: 

939 try: 

940 return g.serialize(format=fmt) 

941 except Exception as e: 

942 return f"# serialization error ({fmt}): {e}" 

943 

944 def export_resource(self, resource_uri: str, fmt: str = "turtle") -> str: 

945 """Serializza il subgraph della singola risorsa (unione su punning).""" 

946 g = Graph() 

947 for prefix, ns in self.reader._graph.namespaces(): 

948 g.bind(prefix, ns) 

949 instances = self.get_instances_from_single_resource(resource_uri) 

950 if instances: 

951 inst_iter = instances if isinstance(instances, set) else [instances] 

952 for inst in inst_iter: 

953 for t in self.reader.get_provenance_subgraph(inst): # già esiste 

954 g.add(t) 

955 return self._safe_serialize(self._with_provenance(g), fmt)