Coverage for lode / reader / provenance_graph.py: 96%

1import os 

2from pathlib import Path 

3from urllib.parse import urlparse, quote 

4from rdflib import Dataset, Graph, URIRef, OWL, RDF 

5from rdflib.namespace import SKOS 

6 

7_ONTOLOGY_TYPES = (OWL.Ontology, SKOS.ConceptScheme) 

8QUAD_FORMATS = frozenset({"trig", "nquads"}) 

9 

10class ProvenanceGraph(Dataset): 

11 """Dataset a named graph (uno per ontologia sorgente) che si presenta al pipeline come un Graph.""" 

12 

13 def __init__(self, **kw): 

14 super().__init__(default_union=True, **kw) 

15 

16 def __iter__(self): 

17 return self.triples((None, None, None)) 

18 

19 def add_document(self, g: Graph, fallback: str) -> URIRef: 

20 ctx = self.ontology_iri(g, fallback) 

21 self.graph(ctx).__iadd__(g) 

22 return ctx 

23 

24 @staticmethod 

25 def ontology_iri(g: Graph, fallback: str) -> URIRef: 

26 for t in _ONTOLOGY_TYPES: 

27 for s in g.subjects(RDF.type, t): 

28 if isinstance(s, URIRef): 

29 return s 

30 parsed = urlparse(fallback) 

31 if parsed.scheme in {"http", "https", "file"}: 

32 return URIRef(fallback) 

33 if not parsed.scheme or len(parsed.scheme) == 1: 

34 return URIRef(Path(os.path.abspath(fallback)).as_uri()) 

35 return URIRef(f"urn:lode:source:{quote(fallback, safe='')}")