Coverage for oc_ocdm / graph / graph_set.py: 87%
199 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-07-06 20:05 +0000
1# SPDX-FileCopyrightText: 2020-2022 Simone Persiani <iosonopersia@gmail.com>
2# SPDX-FileCopyrightText: 2023-2026 Arcangelo Massari <arcangelo.massari@unibo.it>
3#
4# SPDX-License-Identifier: ISC
6from __future__ import annotations
8from io import BytesIO
9from typing import TYPE_CHECKING, cast
11from rdflib import Graph
12from triplelite import RDFTerm, SubgraphView, TripleLite, from_rdflib
14from oc_ocdm.abstract_set import AbstractSet
15from oc_ocdm.counter_handler.counter_handler import CounterHandler
16from oc_ocdm.counter_handler.filesystem_counter_handler import FilesystemCounterHandler
17from oc_ocdm.counter_handler.in_memory_counter_handler import InMemoryCounterHandler
18from oc_ocdm.graph.entities.bibliographic.agent_role import AgentRole
19from oc_ocdm.graph.entities.bibliographic.bibliographic_reference import BibliographicReference
20from oc_ocdm.graph.entities.bibliographic.bibliographic_resource import BibliographicResource
21from oc_ocdm.graph.entities.bibliographic.citation import Citation
22from oc_ocdm.graph.entities.bibliographic.discourse_element import DiscourseElement
23from oc_ocdm.graph.entities.bibliographic.pointer_list import PointerList
24from oc_ocdm.graph.entities.bibliographic.reference_annotation import ReferenceAnnotation
25from oc_ocdm.graph.entities.bibliographic.reference_pointer import ReferencePointer
26from oc_ocdm.graph.entities.bibliographic.resource_embodiment import ResourceEmbodiment
27from oc_ocdm.graph.entities.bibliographic.responsible_agent import ResponsibleAgent
28from oc_ocdm.graph.entities.identifier import Identifier
29from oc_ocdm.graph.graph_entity import GraphEntity
30from oc_ocdm.support.sparql import sparql_construct
31from oc_ocdm.support.support import get_count, get_prefix, get_short_name
33if TYPE_CHECKING:
34 from typing import ClassVar, Dict, List, Optional, Set
37class GraphSet(AbstractSet[GraphEntity]):
38 # Labels
39 labels: ClassVar[Dict[str, str]] = {
40 "an": "annotation",
41 "ar": "agent role",
42 "be": "bibliographic entry",
43 "br": "bibliographic resource",
44 "ci": "citation",
45 "de": "discourse element",
46 "id": "identifier",
47 "pl": "single location pointer list",
48 "ra": "responsible agent",
49 "re": "resource embodiment",
50 "rp": "in-text reference pointer",
51 }
53 def __init__(
54 self,
55 base_iri: str,
56 info_dir: str | None = "",
57 supplier_prefix: str = "",
58 wanted_label: bool = True,
59 custom_counter_handler: CounterHandler | None = None,
60 ) -> None:
61 super(GraphSet, self).__init__()
62 # The following variable maps a URIRef with the related graph entity
63 self.res_to_entity: Dict[str, GraphEntity] = {}
64 self.base_iri: str = base_iri
65 self.info_dir: str | None = info_dir
66 self.supplier_prefix: str = supplier_prefix
67 self.wanted_label: bool = wanted_label
68 # Graphs
69 # The following structure of URL is quite important for the other classes
70 # developed and should not be changed. The only part that can change is the
71 # value of the base_iri
72 self.g_an: str = base_iri + "an/"
73 self.g_ar: str = base_iri + "ar/"
74 self.g_be: str = base_iri + "be/"
75 self.g_br: str = base_iri + "br/"
76 self.g_ci: str = base_iri + "ci/"
77 self.g_de: str = base_iri + "de/"
78 self.g_id: str = base_iri + "id/"
79 self.g_pl: str = base_iri + "pl/"
80 self.g_ra: str = base_iri + "ra/"
81 self.g_re: str = base_iri + "re/"
82 self.g_rp: str = base_iri + "rp/"
84 if custom_counter_handler:
85 self.counter_handler = custom_counter_handler
86 elif info_dir is not None and info_dir != "":
87 self.counter_handler = FilesystemCounterHandler(info_dir, supplier_prefix)
88 else:
89 self.counter_handler = InMemoryCounterHandler()
91 def get_entity(self, res: str) -> Optional[GraphEntity]:
92 if res in self.res_to_entity:
93 return self.res_to_entity[res]
95 # Add resources related to bibliographic entities
96 def add_an(
97 self,
98 resp_agent: str | None,
99 source: str | None = None,
100 res: str | None = None,
101 preexisting_graph: SubgraphView | None = None,
102 ) -> ReferenceAnnotation:
103 if res is not None and get_short_name(res) != "an":
104 raise ValueError(f"Given res: <{res}> is inappropriate for a ReferenceAnnotation entity.")
105 if res is not None and res in self.res_to_entity:
106 return cast(ReferenceAnnotation, self.res_to_entity[res])
107 cur_g, count, label = self._add(self.g_an, "an", res)
108 return ReferenceAnnotation(
109 cur_g, self, GraphEntity.iri_note, res, resp_agent, source, count, label, "an", preexisting_graph
110 )
112 def add_ar(
113 self,
114 resp_agent: str | None,
115 source: str | None = None,
116 res: str | None = None,
117 preexisting_graph: SubgraphView | None = None,
118 ) -> AgentRole:
119 if res is not None and get_short_name(res) != "ar":
120 raise ValueError(f"Given res: <{res}> is inappropriate for an AgentRole entity.")
121 if res is not None and res in self.res_to_entity:
122 return cast(AgentRole, self.res_to_entity[res])
123 cur_g, count, label = self._add(self.g_ar, "ar", res)
124 return AgentRole(
125 cur_g, self, GraphEntity.iri_role_in_time, res, resp_agent, source, count, label, "ar", preexisting_graph
126 )
128 def add_be(
129 self,
130 resp_agent: str | None,
131 source: str | None = None,
132 res: str | None = None,
133 preexisting_graph: SubgraphView | None = None,
134 ) -> BibliographicReference:
135 if res is not None and get_short_name(res) != "be":
136 raise ValueError(f"Given res: <{res}> is inappropriate for a BibliographicReference entity.")
137 if res is not None and res in self.res_to_entity:
138 return cast(BibliographicReference, self.res_to_entity[res])
139 cur_g, count, label = self._add(self.g_be, "be", res)
140 return BibliographicReference(
141 cur_g,
142 self,
143 GraphEntity.iri_bibliographic_reference,
144 res,
145 resp_agent,
146 source,
147 count,
148 label,
149 "be",
150 preexisting_graph,
151 )
153 def add_br(
154 self,
155 resp_agent: str | None,
156 source: str | None = None,
157 res: str | None = None,
158 preexisting_graph: SubgraphView | None = None,
159 ) -> BibliographicResource:
160 if res is not None and get_short_name(res) != "br":
161 raise ValueError(f"Given res: <{res}> is inappropriate for a BibliographicResource entity.")
162 if res is not None and res in self.res_to_entity:
163 return cast(BibliographicResource, self.res_to_entity[res])
164 cur_g, count, label = self._add(self.g_br, "br", res)
165 return BibliographicResource(
166 cur_g, self, GraphEntity.iri_expression, res, resp_agent, source, count, label, "br", preexisting_graph
167 )
169 def add_ci(
170 self,
171 resp_agent: str | None,
172 source: str | None = None,
173 res: str | None = None,
174 preexisting_graph: SubgraphView | None = None,
175 ) -> Citation:
176 if res is not None and get_short_name(res) != "ci":
177 raise ValueError(f"Given res: <{res}> is inappropriate for a Citation entity.")
178 if res is not None and res in self.res_to_entity:
179 return cast(Citation, self.res_to_entity[res])
180 cur_g, count, label = self._add(self.g_ci, "ci", res)
181 return Citation(
182 cur_g, self, GraphEntity.iri_citation, res, resp_agent, source, count, label, "ci", preexisting_graph
183 )
185 def add_de(
186 self,
187 resp_agent: str | None,
188 source: str | None = None,
189 res: str | None = None,
190 preexisting_graph: SubgraphView | None = None,
191 ) -> DiscourseElement:
192 if res is not None and get_short_name(res) != "de":
193 raise ValueError(f"Given res: <{res}> is inappropriate for a DiscourseElement entity.")
194 if res is not None and res in self.res_to_entity:
195 return cast(DiscourseElement, self.res_to_entity[res])
196 cur_g, count, label = self._add(self.g_de, "de", res)
197 return DiscourseElement(
198 cur_g,
199 self,
200 GraphEntity.iri_discourse_element,
201 res,
202 resp_agent,
203 source,
204 count,
205 label,
206 "de",
207 preexisting_graph,
208 )
210 def add_id(
211 self,
212 resp_agent: str | None,
213 source: str | None = None,
214 res: str | None = None,
215 preexisting_graph: SubgraphView | None = None,
216 ) -> Identifier:
217 if res is not None and get_short_name(res) != "id":
218 raise ValueError(f"Given res: <{res}> is inappropriate for an Identifier entity.")
219 if res is not None and res in self.res_to_entity:
220 return cast(Identifier, self.res_to_entity[res])
221 cur_g, count, label = self._add(self.g_id, "id", res)
222 return Identifier(
223 cur_g, self, GraphEntity.iri_identifier, res, resp_agent, source, count, label, "id", preexisting_graph
224 )
226 def add_pl(
227 self,
228 resp_agent: str | None,
229 source: str | None = None,
230 res: str | None = None,
231 preexisting_graph: SubgraphView | None = None,
232 ) -> PointerList:
233 if res is not None and get_short_name(res) != "pl":
234 raise ValueError(f"Given res: <{res}> is inappropriate for a PointerList entity.")
235 if res is not None and res in self.res_to_entity:
236 return cast(PointerList, self.res_to_entity[res])
237 cur_g, count, label = self._add(self.g_pl, "pl", res)
238 return PointerList(
239 cur_g,
240 self,
241 GraphEntity.iri_singleloc_pointer_list,
242 res,
243 resp_agent,
244 source,
245 count,
246 label,
247 "pl",
248 preexisting_graph,
249 )
251 def add_rp(
252 self,
253 resp_agent: str | None,
254 source: str | None = None,
255 res: str | None = None,
256 preexisting_graph: SubgraphView | None = None,
257 ) -> ReferencePointer:
258 if res is not None and get_short_name(res) != "rp":
259 raise ValueError(f"Given res: <{res}> is inappropriate for a ReferencePointer entity.")
260 if res is not None and res in self.res_to_entity:
261 return cast(ReferencePointer, self.res_to_entity[res])
262 cur_g, count, label = self._add(self.g_rp, "rp", res)
263 return ReferencePointer(
264 cur_g,
265 self,
266 GraphEntity.iri_intextref_pointer,
267 res,
268 resp_agent,
269 source,
270 count,
271 label,
272 "rp",
273 preexisting_graph,
274 )
276 def add_ra(
277 self,
278 resp_agent: str | None,
279 source: str | None = None,
280 res: str | None = None,
281 preexisting_graph: SubgraphView | None = None,
282 ) -> ResponsibleAgent:
283 if res is not None and get_short_name(res) != "ra":
284 raise ValueError(f"Given res: <{res}> is inappropriate for a ResponsibleAgent entity.")
285 if res is not None and res in self.res_to_entity:
286 return cast(ResponsibleAgent, self.res_to_entity[res])
287 cur_g, count, label = self._add(self.g_ra, "ra", res)
288 return ResponsibleAgent(
289 cur_g, self, GraphEntity.iri_agent, res, resp_agent, source, count, label, "ra", preexisting_graph
290 )
292 def add_re(
293 self,
294 resp_agent: str | None,
295 source: str | None = None,
296 res: str | None = None,
297 preexisting_graph: SubgraphView | None = None,
298 ) -> ResourceEmbodiment:
299 if res is not None and get_short_name(res) != "re":
300 raise ValueError(f"Given res: <{res}> is inappropriate for a ResourceEmbodiment entity.")
301 if res is not None and res in self.res_to_entity:
302 return cast(ResourceEmbodiment, self.res_to_entity[res])
303 cur_g, count, label = self._add(self.g_re, "re", res)
304 return ResourceEmbodiment(
305 cur_g, self, GraphEntity.iri_manifestation, res, resp_agent, source, count, label, "re", preexisting_graph
306 )
308 def _add(
309 self, graph_url: str, short_name: str, res: str | None = None
310 ) -> tuple[TripleLite, str | None, str | None]:
311 cur_g = TripleLite(identifier=graph_url)
313 count: Optional[str] = None
314 label: Optional[str] = None
315 supplier_prefix = get_prefix(res) if res is not None else self.supplier_prefix
316 if res is not None:
317 try:
318 res_count: int = int(get_count(res))
319 except ValueError:
320 res_count: int = -1
321 if res_count > self.counter_handler.read_counter(short_name, supplier_prefix=supplier_prefix):
322 self.counter_handler.set_counter(res_count, short_name, supplier_prefix=supplier_prefix)
323 return cur_g, count, label
325 count = supplier_prefix + str(
326 self.counter_handler.increment_counter(short_name, supplier_prefix=supplier_prefix)
327 )
329 if self.wanted_label:
330 label = "%s %s [%s/%s]" % (self.labels[short_name], count, short_name, count)
332 return cur_g, count, label
334 def get_orphans(self) -> List[GraphEntity]:
335 full_set_of_entities: Set[str] = set(self.res_to_entity.keys())
336 referenced_entities: Set[str] = set()
337 for res, entity in self.res_to_entity.items():
338 for obj in entity.g.objects(subject=res, predicate=None):
339 if obj.type == "uri":
340 referenced_entities.add(obj.value)
341 set_of_orphan_res: Set[str] = full_set_of_entities - referenced_entities
343 result_list: List[GraphEntity] = []
344 for orphan_res in set_of_orphan_res:
345 entity: Optional[GraphEntity] = self.get_entity(orphan_res)
346 if entity is not None:
347 result_list.append(entity)
349 return result_list
351 def remove_orphans_from_triplestore(self, ts_url: str, resp_agent: str) -> None:
352 for entity_res, entity in self.res_to_entity.items():
353 if entity.to_be_deleted:
354 query: str = f"CONSTRUCT {{?s ?p ?o}} WHERE {{?s ?p ?o ; ?p_1 <{entity_res}>}}"
355 nt_bytes = sparql_construct(ts_url, query)
356 if nt_bytes:
357 from oc_ocdm.reader import Reader
359 rdflib_g = Graph()
360 rdflib_g.parse(BytesIO(nt_bytes), format="nt")
361 graphs = from_rdflib(rdflib_g)
362 imported_entities: List[GraphEntity] = Reader.import_entities_from_graph(
363 self, graphs[0], resp_agent
364 )
365 for imported_entity in imported_entities:
366 imported_entity.g.remove((imported_entity.res, None, RDFTerm("uri", str(entity_res))))
368 def commit_changes(self):
369 for res, entity in self.res_to_entity.items():
370 entity.commit_changes()
371 if entity.to_be_deleted:
372 del self.res_to_entity[res]
374 def get_an(self) -> tuple[ReferenceAnnotation, ...]:
375 return tuple(entity for entity in self.res_to_entity.values() if isinstance(entity, ReferenceAnnotation))
377 def get_ar(self) -> tuple[AgentRole, ...]:
378 return tuple(entity for entity in self.res_to_entity.values() if isinstance(entity, AgentRole))
380 def get_be(self) -> tuple[BibliographicReference, ...]:
381 return tuple(entity for entity in self.res_to_entity.values() if isinstance(entity, BibliographicReference))
383 def get_br(self) -> tuple[BibliographicResource, ...]:
384 return tuple(entity for entity in self.res_to_entity.values() if isinstance(entity, BibliographicResource))
386 def get_ci(self) -> tuple[Citation, ...]:
387 return tuple(entity for entity in self.res_to_entity.values() if isinstance(entity, Citation))
389 def get_de(self) -> tuple[DiscourseElement, ...]:
390 return tuple(entity for entity in self.res_to_entity.values() if isinstance(entity, DiscourseElement))
392 def get_id(self) -> tuple[Identifier, ...]:
393 return tuple(entity for entity in self.res_to_entity.values() if isinstance(entity, Identifier))
395 def get_pl(self) -> tuple[PointerList, ...]:
396 return tuple(entity for entity in self.res_to_entity.values() if isinstance(entity, PointerList))
398 def get_rp(self) -> tuple[ReferencePointer, ...]:
399 return tuple(entity for entity in self.res_to_entity.values() if isinstance(entity, ReferencePointer))
401 def get_ra(self) -> tuple[ResponsibleAgent, ...]:
402 return tuple(entity for entity in self.res_to_entity.values() if isinstance(entity, ResponsibleAgent))
404 def get_re(self) -> tuple[ResourceEmbodiment, ...]:
405 return tuple(entity for entity in self.res_to_entity.values() if isinstance(entity, ResourceEmbodiment))