Coverage for oc_ocdm / reader.py: 88%
285 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-07-10 09:13 +0000
1#!/usr/bin/python
3# SPDX-FileCopyrightText: 2020-2022 Simone Persiani <iosonopersia@gmail.com>
4# SPDX-FileCopyrightText: 2022-2026 Arcangelo Massari <arcangelo.massari@unibo.it>
5#
6# SPDX-License-Identifier: ISC
8# -*- coding: utf-8 -*-
9from __future__ import annotations
11import json
12import os
13from collections.abc import Callable
14from importlib import import_module
15from typing import TYPE_CHECKING, BinaryIO, TextIO, cast
16from zipfile import ZipFile
18import orjson
19from rdflib import Dataset, Graph, URIRef
20from rdflib.term import Node
21from SPARQLWrapper import POST
22from triplelite import TripleLite, from_rdflib
24from oc_ocdm._types import ContextMap, JsonLdDocument, JsonObject, JsonValue, SparqlResultRows
25from oc_ocdm.constants import RDF_TYPE
26from oc_ocdm.graph.graph_entity import GraphEntity
27from oc_ocdm.support.reporter import Reporter
28from oc_ocdm.support.sparql import SPARQLEndpointError, sparql_query
29from oc_ocdm.support.support import build_graph_from_results, normalize_graph_literals
31if TYPE_CHECKING:
32 from typing import List, Optional
34 from oc_ocdm.graph.graph_set import GraphSet
36_validate = cast(Callable[..., tuple[object, object, object]], getattr(import_module("pyshacl"), "validate"))
39def _transform_jsonld_value(value: JsonValue, uri_fn: Callable[[str], str]) -> JsonValue:
40 if isinstance(value, dict):
41 if "@id" in value:
42 return {"@id": uri_fn(cast(str, value["@id"]))}
43 result: JsonObject = {}
44 if "@value" in value:
45 result["@value"] = value["@value"]
46 if "@type" in value:
47 result["@type"] = uri_fn(cast(str, value["@type"]))
48 if "@language" in value:
49 result["@language"] = value["@language"]
50 return result
51 return value
54def _transform_jsonld_entity(entity: JsonObject, uri_fn: Callable[[str], str]) -> JsonObject:
55 transformed: JsonObject = {}
56 for key, value in entity.items():
57 if key == "@id":
58 transformed["@id"] = uri_fn(cast(str, value))
59 elif key == "@type":
60 transformed["@type"] = (
61 [uri_fn(cast(str, t)) for t in value] if isinstance(value, list) else [uri_fn(cast(str, value))]
62 )
63 elif key.startswith("@"):
64 continue
65 else:
66 new_key = uri_fn(key)
67 if isinstance(value, list):
68 transformed[new_key] = [_transform_jsonld_value(v, uri_fn) for v in value]
69 else:
70 transformed[new_key] = _transform_jsonld_value(value, uri_fn)
71 return transformed
74def transform_jsonld_graphs(data: JsonLdDocument, uri_fn: Callable[[str], str]) -> JsonLdDocument:
75 result: JsonLdDocument = []
76 for graph_obj in data:
77 new_graph: JsonObject = {}
78 if "@id" in graph_obj:
79 new_graph["@id"] = uri_fn(cast(str, graph_obj["@id"]))
80 if "@graph" in graph_obj:
81 new_graph["@graph"] = [
82 _transform_jsonld_entity(entity, uri_fn) for entity in cast(list[JsonObject], graph_obj["@graph"])
83 ]
84 result.append(new_graph)
85 return result
88def _expand_uri(curie: str, prefix_to_ns: dict[str, str]) -> str:
89 colon = curie.find(":")
90 if colon > 0:
91 prefix = curie[:colon]
92 ns = prefix_to_ns.get(prefix)
93 if ns is not None:
94 return ns + curie[colon + 1 :]
95 return curie
98def _expand_jsonld(data: JsonLdDocument, prefix_to_ns: dict[str, str]) -> JsonLdDocument:
99 return transform_jsonld_graphs(data, lambda uri: _expand_uri(uri, prefix_to_ns))
102class Reader(object):
103 def __init__(
104 self,
105 repok: Optional[Reporter] = None,
106 reperr: Optional[Reporter] = None,
107 context_map: Optional[ContextMap] = None,
108 ) -> None:
110 if context_map is not None:
111 self.context_map: ContextMap = context_map
112 else:
113 self.context_map: ContextMap = {}
114 for context_url in self.context_map:
115 ctx_file_path = self.context_map[context_url]
116 if isinstance(ctx_file_path, str) and os.path.isfile(ctx_file_path):
117 # This expensive operation is done only when it's really needed
118 with open(ctx_file_path, "rt", encoding="utf-8") as ctx_f:
119 self.context_map[context_url] = cast(JsonObject, json.load(ctx_f))
121 if repok is None:
122 self.repok: Reporter = Reporter(prefix="[Reader: INFO] ")
123 else:
124 self.repok: Reporter = repok
126 if reperr is None:
127 self.reperr: Reporter = Reporter(prefix="[Reader: ERROR] ")
128 else:
129 self.reperr: Reporter = reperr
131 def load(self, rdf_file_path: str) -> Optional[Dataset]:
132 self.repok.new_article()
133 self.reperr.new_article()
135 loaded_graph: Optional[Dataset] = None
136 if os.path.isfile(rdf_file_path):
137 try:
138 loaded_graph = self._load_graph(rdf_file_path)
139 except Exception as e:
140 self.reperr.add_sentence(
141 "[1] "
142 "It was impossible to handle the format used for "
143 "storing the file (stored in the temporary path) "
144 f"'{rdf_file_path}'. Additional details: {e}"
145 )
146 else:
147 self.reperr.add_sentence(f"[2] The file specified ('{rdf_file_path}') doesn't exist.")
149 return loaded_graph
151 _EXT_TO_FORMATS: dict[str, list[str]] = {
152 ".json": ["json-ld"],
153 ".jsonld": ["json-ld"],
154 ".xml": ["rdfxml"],
155 ".rdf": ["rdfxml"],
156 ".ttl": ["turtle"],
157 ".trig": ["trig"],
158 ".nt": ["nt11"],
159 ".nq": ["nquads"],
160 }
161 _ALL_FORMATS: list[str] = ["json-ld", "rdfxml", "turtle", "trig", "nt11", "nquads"]
163 @staticmethod
164 def _formats_for_file(file_name: str) -> list[str]:
165 ext = os.path.splitext(file_name)[1].lower()
166 preferred = Reader._EXT_TO_FORMATS.get(ext)
167 if preferred is not None:
168 return preferred + [f for f in Reader._ALL_FORMATS if f not in preferred]
169 return Reader._ALL_FORMATS
171 def _load_graph(self, file_path: str) -> Dataset:
172 loaded_graph = Dataset()
174 if file_path.endswith(".zip"):
175 try:
176 with ZipFile(file=file_path, mode="r") as archive:
177 for zf_name in archive.namelist():
178 formats = self._formats_for_file(zf_name)
179 with archive.open(zf_name) as f:
180 if self._try_parse(loaded_graph, cast(BinaryIO, f), formats):
181 for graph in loaded_graph.graphs():
182 normalize_graph_literals(graph)
183 return loaded_graph
184 except Exception as e:
185 raise IOError(f"Error opening or reading zip file '{file_path}': {e}")
186 else:
187 formats = self._formats_for_file(file_path)
188 try:
189 with open(file_path, "rt", encoding="utf-8") as f:
190 if self._try_parse(loaded_graph, f, formats):
191 for graph in loaded_graph.graphs():
192 normalize_graph_literals(graph)
193 return loaded_graph
194 except Exception as e:
195 raise IOError(f"Error opening or reading file '{file_path}': {e}")
197 raise IOError(f"It was impossible to load the file '{file_path}' with supported formats.")
199 def _try_parse(self, graph: Dataset, file_obj: TextIO | BinaryIO, formats: List[str]) -> bool:
200 for cur_format in formats:
201 file_obj.seek(0)
202 try:
203 if cur_format == "json-ld":
204 json_ld_file = cast(JsonObject | JsonLdDocument, json.load(file_obj))
205 if isinstance(json_ld_file, dict):
206 json_ld_file = [json_ld_file]
207 for json_ld_resource in json_ld_file:
208 context_url = json_ld_resource["@context"] if "@context" in json_ld_resource else None
209 if isinstance(context_url, str) and context_url in self.context_map:
210 context_data = self.context_map[context_url]
211 if isinstance(context_data, dict) and "@context" in context_data:
212 json_ld_resource["@context"] = context_data["@context"]
213 graph.parse(data=json.dumps(json_ld_file, ensure_ascii=False), format=cur_format)
214 else:
215 graph.parse(file=file_obj, format=cur_format)
216 return True
217 except Exception:
218 continue
219 return False
221 def load_jsonld_dict(self, rdf_file_path: str) -> JsonLdDocument:
222 if rdf_file_path.endswith(".zip"):
223 with ZipFile(file=rdf_file_path, mode="r") as archive:
224 for zf_name in archive.namelist():
225 ext = os.path.splitext(zf_name)[1].lower()
226 if ext in (".json", ".jsonld"):
227 with archive.open(zf_name) as f:
228 data = cast(JsonObject | JsonLdDocument, orjson.loads(f.read()))
229 break
230 else:
231 raise IOError(f"No JSON/JSON-LD file found inside ZIP archive '{rdf_file_path}'.")
232 else:
233 with open(rdf_file_path, "rb") as f:
234 data = cast(JsonObject | JsonLdDocument, orjson.loads(f.read()))
235 if isinstance(data, dict):
236 data = [data]
237 prefix_to_ns: dict[str, str] | None = None
238 for graph_obj in data:
239 ctx_url = graph_obj["@context"] if "@context" in graph_obj else None
240 if isinstance(ctx_url, str) and ctx_url in self.context_map:
241 ctx = self.context_map[ctx_url]
242 if isinstance(ctx, dict) and "@context" in ctx:
243 context_value = ctx["@context"]
244 if isinstance(context_value, dict):
245 ctx = context_value
246 if isinstance(ctx, dict):
247 prefix_to_ns = {k: v for k, v in ctx.items() if isinstance(v, str) and not k.startswith("@")}
248 break
249 if prefix_to_ns is not None:
250 data = _expand_jsonld(data, prefix_to_ns)
251 return data
253 def graph_validation(self, graph: Graph, closed: bool = False) -> Graph:
254 valid_graph: Graph = Graph(identifier=graph.identifier)
255 sg = Graph()
256 if closed:
257 sg.parse(os.path.join("oc_ocdm", "resources", "shacle_closed.ttl"))
258 else:
259 sg.parse(os.path.join("oc_ocdm", "resources", "shacle.ttl"))
260 _, report_result, _ = _validate(
261 graph,
262 shacl_graph=sg,
263 ont_graph=None,
264 inference=None,
265 abort_on_first=False,
266 allow_infos=False,
267 allow_warnings=False,
268 meta_shacl=False,
269 advanced=False,
270 js=False,
271 debug=False,
272 )
273 if not isinstance(report_result, Graph):
274 raise TypeError(f"Expected Graph from SHACL validation, got {type(report_result)}")
275 invalid_nodes: set[Node] = set()
276 for triple in report_result.triples((None, URIRef("http://www.w3.org/ns/shacl#focusNode"), None)):
277 invalid_nodes.add(triple[2])
278 for s in graph.subjects(unique=True):
279 if isinstance(s, URIRef) and s not in invalid_nodes:
280 for valid_subject_triple in graph.triples((s, None, None)):
281 valid_graph.add(valid_subject_triple)
282 return valid_graph
284 @staticmethod
285 def import_entities_from_graph(
286 g_set: GraphSet,
287 results: SparqlResultRows | TripleLite | Graph | Dataset,
288 resp_agent: str,
289 enable_validation: bool = False,
290 closed: bool = False,
291 ) -> List[GraphEntity]:
292 if isinstance(results, list):
293 graph: TripleLite | Graph = build_graph_from_results(results)
294 elif isinstance(results, Dataset):
295 merged = TripleLite()
296 for tl in from_rdflib(results):
297 for triple in tl.triples((None, None, None)):
298 merged.add(triple)
299 graph = merged
300 elif isinstance(results, Graph):
301 graph = results
302 else:
303 graph = results
304 if enable_validation:
305 reader = Reader()
306 if not isinstance(graph, Graph):
307 graph = graph.to_rdflib()
308 graph = reader.graph_validation(graph, closed)
309 if isinstance(graph, Graph):
310 graph = from_rdflib(graph)[0]
311 imported_entities: List[GraphEntity] = []
312 for subject in graph.subjects():
313 types: List[str] = [o.value for o in graph.objects(subject, RDF_TYPE)]
314 preexisting = graph.subgraph(subject)
315 if GraphEntity.iri_note in types:
316 imported_entities.append(
317 g_set.add_an(resp_agent=resp_agent, res=subject, preexisting_graph=preexisting)
318 )
319 elif GraphEntity.iri_role_in_time in types:
320 imported_entities.append(
321 g_set.add_ar(resp_agent=resp_agent, res=subject, preexisting_graph=preexisting)
322 )
323 elif GraphEntity.iri_bibliographic_reference in types:
324 imported_entities.append(
325 g_set.add_be(resp_agent=resp_agent, res=subject, preexisting_graph=preexisting)
326 )
327 elif GraphEntity.iri_expression in types:
328 imported_entities.append(
329 g_set.add_br(resp_agent=resp_agent, res=subject, preexisting_graph=preexisting)
330 )
331 elif GraphEntity.iri_citation in types:
332 imported_entities.append(
333 g_set.add_ci(resp_agent=resp_agent, res=subject, preexisting_graph=preexisting)
334 )
335 elif GraphEntity.iri_discourse_element in types:
336 imported_entities.append(
337 g_set.add_de(resp_agent=resp_agent, res=subject, preexisting_graph=preexisting)
338 )
339 elif GraphEntity.iri_identifier in types:
340 imported_entities.append(
341 g_set.add_id(resp_agent=resp_agent, res=subject, preexisting_graph=preexisting)
342 )
343 elif GraphEntity.iri_singleloc_pointer_list in types:
344 imported_entities.append(
345 g_set.add_pl(resp_agent=resp_agent, res=subject, preexisting_graph=preexisting)
346 )
347 elif GraphEntity.iri_agent in types:
348 imported_entities.append(
349 g_set.add_ra(resp_agent=resp_agent, res=subject, preexisting_graph=preexisting)
350 )
351 elif GraphEntity.iri_manifestation in types:
352 imported_entities.append(
353 g_set.add_re(resp_agent=resp_agent, res=subject, preexisting_graph=preexisting)
354 )
355 elif GraphEntity.iri_intextref_pointer in types:
356 imported_entities.append(
357 g_set.add_rp(resp_agent=resp_agent, res=subject, preexisting_graph=preexisting)
358 )
359 return imported_entities
361 @staticmethod
362 def import_entity_from_triplestore(
363 g_set: GraphSet, ts_url: str, res: str, resp_agent: str, enable_validation: bool = False
364 ) -> GraphEntity:
365 query: str = f"SELECT ?s ?p ?o WHERE {{BIND (<{res}> AS ?s). ?s ?p ?o.}}"
366 try:
367 result = sparql_query(ts_url, query, max_retries=3, backoff_factor=2.5)["results"]["bindings"]
369 if not result:
370 raise ValueError(f"The requested entity {res} was not found in the triplestore.")
372 imported_entities: List[GraphEntity] = Reader.import_entities_from_graph(
373 g_set, result, resp_agent, enable_validation
374 )
375 if len(imported_entities) <= 0:
376 raise ValueError("The requested entity was not recognized as a proper OCDM entity.")
377 return imported_entities[0]
379 except ValueError:
380 raise
381 except SPARQLEndpointError as e:
382 print(f"[3] Could not import entity due to communication problems: {e}")
383 raise
385 @staticmethod
386 def import_entities_from_triplestore(
387 g_set: GraphSet,
388 ts_url: str,
389 entities: List[str],
390 resp_agent: str,
391 enable_validation: bool = False,
392 batch_size: int = 1000,
393 ) -> List[GraphEntity]:
394 if not entities:
395 raise ValueError("No entities provided for import")
396 if batch_size <= 0:
397 raise ValueError("Batch size must be greater than zero")
399 imported_entities: List[GraphEntity] = []
401 try:
402 for i in range(0, len(entities), batch_size):
403 batch = entities[i : i + batch_size]
404 requested_entities = set(batch)
405 values = " ".join(f"<{entity}>" for entity in batch)
406 query = f"SELECT ?s ?p ?o WHERE {{ VALUES ?s {{ {values} }} ?s ?p ?o . }}"
407 results = sparql_query(
408 ts_url,
409 query,
410 method=POST,
411 max_retries=3,
412 backoff_factor=2.5,
413 )["results"]["bindings"]
415 returned_entities = {result["s"]["value"] for result in results}
416 not_found_entities = requested_entities - returned_entities
417 if not_found_entities:
418 entities_str = ", ".join(sorted(not_found_entities))
419 raise ValueError(f"The requested entities were not found in the triplestore: {entities_str}")
421 batch_entities = Reader.import_entities_from_graph(
422 g_set=g_set, results=results, resp_agent=resp_agent, enable_validation=enable_validation
423 )
424 imported_entity_uris = {str(entity.res) for entity in batch_entities}
425 not_imported_entities = requested_entities - imported_entity_uris
426 if not_imported_entities:
427 entities_str = ", ".join(sorted(not_imported_entities))
428 raise ValueError(
429 f"The following entities could not be imported as GraphEntity instances: {entities_str}"
430 )
431 imported_entities.extend(batch_entities)
433 except SPARQLEndpointError as e:
434 print(f"[3] Could not import batch due to communication problems: {e}")
435 raise
437 return imported_entities