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
« 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
8from rdflib import Graph, URIRef, BNode, Literal as RDFlibLiteral
9from rdflib.namespace import RDF, OWL
11from lode.reader.provenance_graph import ProvenanceGraph
12from lode.helpers.rdf import iri_local_name
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]
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]
27def formats_for(graph):
28 return QUAD_SERIALIZATION_FORMATS if isinstance(graph, ProvenanceGraph) else SERIALIZATION_FORMATS
30class BaseViewer:
31 """Base viewer per visualizzare istanze estratte dal Reader."""
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
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
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
55 def _formats(self):
56 return formats_for(self.reader._graph)
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
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
82 def get_all_instances(self) -> List:
83 """Ottiene tutte le istanze (esclusi literal)."""
84 instances = []
86 for uri_id, instance_set in self._cache.items():
87 if isinstance(uri_id, str) and uri_id.startswith("LITERAL::"):
88 continue
90 instance_list = instance_set if isinstance(instance_set, set) else [instance_set]
91 instances.extend(instance_list)
93 return instances
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]
102 return None
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."""
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"
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()
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()
126 # --- DETERMINISTIC FALLBACKS ---
127 # If we reach here, no English label exists.
128 # We sort them by language tag.
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()
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()
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)
149 return None
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"
159 # Fallback: generic flat list
160 all_instances = self.get_all_instances()
161 metadata_dict = self._find_and_format_metadata(all_instances, language)
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
169 return {
170 'metadata': metadata_dict,
171 'entities': self._format_entities(all_instances, language),
172 'export_formats': self._formats(),
173 }
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())
184 def get_toc_instances(self) -> List:
185 return [i for i in self.get_all_instances() if self._is_toc_entity(i)]
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)
194 if not instance_set:
195 return {'error': f'Resource {resource_uri} not found'}
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'}
202 return {
203 'single_resource': True,
204 'entities': self._format_entities(instances, language),
205 'groupped_view': False,
206 'sections': None
207 }
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 = []
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 ]
223 if instances:
224 section = {
225 'id': section_id,
226 'title': section_title,
227 'entities': self._format_entities(instances, language),
228 }
230 parent_getter = self._HIERARCHY_PARENT_GETTERS.get(class_key)
231 if parent_getter:
232 section['tree'] = self._build_hierarchy(instances, parent_getter, language)
234 sections.append(section)
236 return {
237 'grouped_view': True,
238 'sections': sections,
239 'export_formats': self._formats()
240 }
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()
249 entities = []
250 for instance in instances:
251 uri = instance.has_identifier if hasattr(instance, 'has_identifier') else None
253 if not uri:
254 continue
256 #Create a safe HTML ID to facilitate on-page navigation
257 safe_id = hashlib.md5(str(uri).encode('utf-8')).hexdigest()
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)
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)
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 = {}
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 ]
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]
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]
319 characteristics = {}
320 char_attributes = [
321 'is_functional', 'is_inverse_functional', 'is_transitive',
322 'is_symmetric', 'is_asymmetric', 'is_reflexive', 'is_irreflexive'
323 ]
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)
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
335 # Extract Statement Entities
336 statements = self._format_statement(all_instances, instance, language)
337 type_inst = type(instance).__name__.replace(" ", "_")
339 is_dep = getattr(instance, 'get_is_deprecated')() if hasattr(instance, 'get_is_deprecated') else getattr(
340 instance, 'is_deprecated', False)
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 })
355 entities.sort(key=lambda x: (x['label'] or x['uri']).lower())
356 return entities
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
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 )
373 if not ontology_model:
374 return {}
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 }
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:
398 method = getattr(ontology_model, attr_name)
400 if callable(method):
401 try:
402 values = method()
403 except AttributeError:
404 # skip attributes that are not initialized
405 continue
407 if values:
408 # 4. Auto-format the key name
409 clean_key = self._clean_name(self, attr_name)
411 # 5. Ensure values are in a list
412 if not isinstance(values, list):
413 values = [values]
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
426 extracted_values.append(entry)
428 # 8. Add to structural data ONLY if we found valid text
429 if extracted_values:
430 data[clean_key] = extracted_values
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 ]
440 # 9. Statements
441 data.update(self._format_statement(all_instances, ontology_model, language))
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}
448 return data
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 }
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 }
486 if not obj: return handler_dic
488 # --- 1. INTERCEPT RESTRICTIONS ---
489 restriction_types = ["Restriction", "PropertyConceptRestriction", "Quantifier", "Cardinality", "TruthFunction",
490 "OneOf", "Value", "DatatypeRestriction"]
491 obj_type = type(obj).__name__
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)
499 # Recursively parse the restriction into clickable parts
500 parts = self._parse_restriction(obj, language)
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
510 # --- 2. String Handling ---
511 if isinstance(obj, str):
512 handler_dic['text'] = obj
513 return handler_dic
515 # --- 3. Literal Handling ---
516 if type(obj).__name__ == 'Literal':
517 if hasattr(obj, 'get_has_value') and obj.get_has_value():
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)
524 lit_lang = obj.get_has_language()
526 # If a language is requested (e.g., 'pt')
527 if language:
528 target_lang = language.strip().lower()
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
534 # if not lit_lang:
535 # return handler_dic
537 handler_dic['text'] = obj.get_has_value()
538 handler_dic['lan'] = lit_lang
539 handler_dic['type'] = obj_type
540 return handler_dic
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
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()
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
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)
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
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
605 return handler_dic
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 = {}
613 # 1. Validate the target instance
614 if target_instance is None:
615 return statements
617 for instance in instances:
618 if isinstance(instance, Statement):
619 subj = instance.get_has_subject()
621 # 2. Identity comparison (handles punning: same URI, different Python instances)
622 if subj is not target_instance:
623 continue
625 predicate = instance.get_has_predicate()
626 obj = instance.get_has_object()
628 if obj:
629 obj_data = self._resolve_resource_value(obj, language)
631 # 3. Predicate Resolution
632 pred_label = self._get_best_label(predicate, language) if predicate else "Annotation"
634 if pred_label not in statements:
635 statements[pred_label] = []
637 # 4. Resolve Object and Prevent Duplicates
638 if obj:
639 obj_data = self._resolve_resource_value(obj, language)
642 if obj_data not in statements[pred_label]:
643 statements[pred_label].append(obj_data)
645 return statements
647 @staticmethod
648 def _clean_name(self, name: str) -> str:
649 if not name: return ""
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('_', ' ')
656 # Strip extra spaces
657 name = ' '.join(name.split())
659 # lowercase everything
660 name = name.lower()
662 return name
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 []
673 if _seen is None:
674 _seen = set()
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
685 obj_type = type(obj).__name__
686 restriction_types = ["Restriction", "PropertyConceptRestriction", "Quantifier", "Cardinality", "TruthFunction",
687 "OneOf", "Value", "DatatypeRestriction"]
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))
696 parts = []
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)
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')
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'})
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))
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')
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'})
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))
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]
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'})
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})
752 elif obj_type == "OneOf":
753 resources = _get(obj, 'applies_on_resource', [])
754 if not isinstance(resources, list): resources = [resources]
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'})
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})
768 elif obj_type == "Value":
769 prop = _get(obj, 'applies_on_property')
770 resource = _get(obj, 'applies_on_resource')
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'})
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))
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]+")
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'})
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'})
797 parts.extend(self._parse_restriction(value, language, _seen))
799 return parts
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)
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)}]
808 return []
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
815 # 1. Clean the string
816 clean_url = url.strip().lower()
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('/')
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
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 ''
831 def _host_is(domain: str) -> bool:
832 return host == domain or host.endswith('.' + domain)
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
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
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
857 return False
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 }
868 _FULL_IRI_ATTRS = {'is_defined_by'}
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
877 by_uri = {}
878 for inst in instances:
879 uri = getattr(inst, 'has_identifier', None)
880 if uri:
881 by_uri[str(uri)] = inst
883 children_map = {uri: [] for uri in by_uri}
884 has_internal_parent = set()
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)
897 roots = [uri for uri in by_uri if uri not in has_internal_parent]
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}
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
925 # ========== PROVENANCE: subgraph serialisation for each card ==========
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 ]
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}"
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)