Coverage for src/oc_graphenricher/enricher/__init__.py: 96%
222 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-06 11:55 +0000
1# SPDX-FileCopyrightText: 2021 Gabriele Pisciotta <ga.pisciotta@gmail.com>
2# SPDX-FileCopyrightText: 2021 Simone Persiani
3# SPDX-FileCopyrightText: 2026 Arcangelo Massari <arcangelo.massari@unibo.it>
4# SPDX-FileCopyrightText: 2026 Ilaria De Dominicis <ilaria.dedominicis2@studio.unibo.it>
5#
6# SPDX-License-Identifier: ISC
8from __future__ import annotations
10import contextlib
11import logging
12import sys
13from datetime import datetime, timezone
14from typing import TYPE_CHECKING
16import requests_cache
17from oc_ocdm.graph.entities.bibliographic.responsible_agent import ResponsibleAgent
18from oc_ocdm.graph.graph_entity import GraphEntity
19from oc_ocdm.prov.prov_set import ProvSet
20from tqdm import tqdm
21from tqdm.contrib import DummyTqdmFile
23from oc_graphenricher._identifiers import supported_br_identifiers
24from oc_graphenricher._storage import store_graph_set, store_provenance
25from oc_graphenricher.APIs import ORCID, VIAF, Crossref, OpenAlex, Wikidata
27if TYPE_CHECKING:
28 from collections.abc import Iterable, Iterator
29 from typing import Protocol, TextIO
31 from oc_ocdm.graph.entities.bibliographic.bibliographic_resource import BibliographicResource
32 from oc_ocdm.graph.graph_set import GraphSet
34 from oc_graphenricher.storage import Storage
36 class IdentifierLike(Protocol):
37 def get_literal_value(self) -> str | None: ...
40LOGGER = logging.getLogger(__name__)
43class GraphEnricher:
44 def __init__(
45 self,
46 graph_set: GraphSet,
47 storage: Storage,
48 *,
49 debug: bool = False,
50 checkpoint_interval: int | None = None,
51 use_wikidata: bool = True,
52 use_viaf: bool = True,
53 use_orcid: bool = True,
54 ) -> None:
55 """
56 Initialize the enricher.
58 The enricher adds missing identifiers to entities in an OCDM graph set.
60 :param graph_set: graph set to be enriched.
61 :param storage: output storage configuration
62 :param debug: a bool flag to enable richer output
63 :param checkpoint_interval: save the graph every N processed Bibliographic Resources (BRs)
64 :param use_wikidata: a bool flag to enable or disable Wikidata queries (default: True)
65 :param use_viaf: a bool flag to enable or disable VIAF queries (default: True)
66 :param use_orcid: a bool flag to enable or disable ORCID queries (default: True)
67 """
68 if checkpoint_interval is not None and checkpoint_interval <= 0:
69 message = "checkpoint_interval must be greater than 0."
70 raise ValueError(message)
72 requests_cache.install_cache("GraphEnricher_cache")
74 self.resp_agent = "https://w3id.org/oc/meta/prov/pa/2"
75 self.crossref_api = Crossref()
76 self.orcid_api = ORCID()
77 self.viaf_api = VIAF()
78 self.wikidata_api = Wikidata()
79 self.openalex_api = OpenAlex()
80 self.graph_set = graph_set
81 self.debug = debug
82 self.added_identifier_count = 0
83 self.storage = storage
84 self.checkpoint_interval = checkpoint_interval
85 self.use_wikidata = use_wikidata
86 self.use_viaf = use_viaf
87 self.use_orcid = use_orcid
89 def enrich(self) -> None:
90 """
91 Iterate over each BR contained in the graph set.
93 For each BR, excluding journal issues and journal volumes, it reads the identifiers already contained in the
94 graph set and checks whether the BR already has a DOI, ISSN, Wikidata ID, and OpenAlex ID:
96 - If an ISSN is available, it queries Crossref to extract related ISSNs.
97 - If no DOI is available and the title is usable, it queries Crossref to find one from the BR metadata.
98 - If no Wikidata ID is available, it queries Wikidata by using the BR identifiers.
99 - If no OpenAlex ID is available, it queries OpenAlex by using the BR identifiers.
101 Any new identifier found is added to the BR.
103 Then, for each AR related to the BR, it reads the identifiers already contained in the graph set and:
105 - If an author does not have an ORCID, it queries ORCID to find one.
106 - If an author does not have a VIAF, it queries VIAF to find one from the author name and BR title.
107 - If an author does not have a Wikidata ID, it queries Wikidata by using the author identifiers found so far.
108 - If a publisher does not have a Crossref ID, it queries Crossref to find one from the BR DOI.
110 Any new identifier found is added to the AR.
112 In the end it will store a new graph set and its provenance.
114 NB: Even if it's not possible to have an identifier duplicated for the same entity, it's possible that in
115 the whole graph set you could find different identifiers that share the same schema and literal. For this
116 purpose, you should use the `deduplication` module after you've enriched the graph set.
117 """
118 with self.__std_out_err_redirect_tqdm() as orig_stdout:
119 progress_bar = tqdm(self.graph_set.get_br(), file=orig_stdout, dynamic_ncols=True)
120 for br_counter, br in enumerate(progress_bar, start=1):
121 progress_bar.set_description(desc=f"Added identifiers: {self.added_identifier_count}")
122 self.__serialize_intermediate(br_counter)
123 if self.__is_journal_issue_or_volume(br):
124 continue
125 has_doi = self.__enrich_bibliographic_resource(br)
126 self.__enrich_contributors(br, has_doi)
128 self.__serialize_graphs()
130 def __serialize_intermediate(self, br_counter: int) -> None:
131 if self.checkpoint_interval is None or br_counter % self.checkpoint_interval != 0:
132 return
133 store_graph_set(self.graph_set, self.storage)
135 def __is_journal_issue_or_volume(self, br: BibliographicResource) -> bool:
136 return GraphEntity.iri_journal_issue in br.get_types() or GraphEntity.iri_journal_volume in br.get_types()
138 def __enrich_bibliographic_resource(self, br: BibliographicResource) -> str | None:
139 has_doi, has_issn, has_wikidata, has_openalex = self.__br_identifiers(br)
140 self.__add_missing_issns(br, has_issn)
141 has_doi = self.__add_missing_doi(br, has_doi)
142 if self.use_wikidata and len(has_wikidata) == 0:
143 self.__add_wikidata_to_br(br)
144 if has_openalex is None:
145 self.__add_openalex_to_br(br)
146 return has_doi
148 def __br_identifiers(self, br: BibliographicResource) -> tuple[str | None, list[str], list[str], str | None]:
149 has_doi = None
150 has_issn = []
151 has_wikidata = []
152 has_openalex = None
153 for identifier in br.get_identifiers():
154 literal = identifier.get_literal_value()
155 if literal is None:
156 continue
157 if identifier.get_scheme() == br.iri_doi:
158 has_doi = literal
159 elif identifier.get_scheme() == br.iri_issn:
160 has_issn.append(literal)
161 elif identifier.get_scheme() == br.iri_wikidata:
162 has_wikidata.append(literal)
163 elif identifier.get_scheme() == br.iri_openalex:
164 has_openalex = literal
165 return has_doi, has_issn, has_wikidata, has_openalex
167 def __add_missing_issns(self, br: BibliographicResource, has_issn: list[str]) -> None:
168 for issn in has_issn:
169 result = self.crossref_api.query_journal(issn)
170 if not result:
171 continue
172 for literal in result:
173 if literal not in has_issn:
174 self._add_id(br, literal, "issn", f"its ISSN {issn}")
175 break
177 def __add_missing_doi(self, br: BibliographicResource, has_doi: str | None) -> str | None:
178 title = br.get_title()
179 if has_doi is not None or not title or title.strip().lower() == "unknown":
180 return has_doi
181 result = self.crossref_api.query([], title, br.get_pub_date())
182 if result:
183 self._add_id(br, result, "doi", "Crossref query")
184 return result
185 return has_doi
187 def __add_wikidata_to_br(self, br: BibliographicResource) -> None:
188 for schema, literal in supported_br_identifiers(br):
189 result = self.wikidata_api.query(literal, schema)
190 if result:
191 self._add_id(br, result, "wikidata", f"its {schema.upper()} {literal}")
192 break
194 def __add_openalex_to_br(self, br: BibliographicResource) -> None:
195 for schema, literal in supported_br_identifiers(br):
196 result = self.openalex_api.query(literal, schema)
197 if result:
198 for openalex_id in result:
199 self._add_id(br, openalex_id, "openalex", f"its {schema.upper()} {literal}")
201 def __enrich_contributors(self, br: BibliographicResource, has_doi: str | None) -> None:
202 publisher_has_crossrefid = False
203 for ar in br.get_contributors():
204 role = ar.get_role_type()
205 ra = ar.get_is_held_by()
206 if ra is None:
207 continue
208 if role == GraphEntity.iri_author:
209 self.__enrich_author(br, ra)
210 if role == GraphEntity.iri_publisher:
211 publisher_has_crossrefid = self.__enrich_publisher(
212 ra,
213 has_doi,
214 publisher_has_crossrefid=publisher_has_crossrefid,
215 )
217 def __enrich_author(self, br: BibliographicResource, ra: ResponsibleAgent) -> None:
218 has_orcid, has_viaf, has_wikidata, author_id_found = self.__author_identifiers(br, ra)
219 if self.use_orcid and has_orcid is None:
220 self.__add_orcid_to_author(br, ra, author_id_found)
221 if self.use_viaf and not has_viaf:
222 self.__add_viaf_to_author(br, ra, author_id_found)
223 if self.use_wikidata and not has_wikidata:
224 self.__add_wikidata_to_author(ra, author_id_found)
226 def __author_identifiers(
227 self,
228 br: BibliographicResource,
229 ra: ResponsibleAgent,
230 ) -> tuple[str | None, str | None, str | None, list[tuple[str | None, str]]]:
231 has_orcid = None
232 has_viaf = None
233 has_wikidata = None
234 author_id_found = []
235 for author_identifier in ra.get_identifiers():
236 scheme = author_identifier.get_scheme()
237 literal = author_identifier.get_literal_value()
238 if scheme is None:
239 continue
240 if ra.iri_orcid in scheme:
241 has_orcid = literal
242 author_id_found.append((literal, "orcid"))
243 if br.iri_viaf in scheme:
244 has_viaf = literal
245 author_id_found.append((literal, "viaf"))
246 if br.iri_wikidata in scheme:
247 has_wikidata = literal
248 if has_viaf is not None and has_orcid is not None and has_wikidata is not None:
249 break
250 return has_orcid, has_viaf, has_wikidata, author_id_found
252 def __add_orcid_to_author(
253 self,
254 br: BibliographicResource,
255 ra: ResponsibleAgent,
256 author_id_found: list[tuple[str | None, str]],
257 ) -> None:
258 result = self.orcid_api.query(
259 [(ra.get_given_name(), ra.get_family_name(), None, ra)],
260 [(identifier.get_scheme(), identifier.get_literal_value()) for identifier in br.get_identifiers()],
261 )
262 if not result:
263 return
264 for _given_name, _family_name, orcid, responsible_agent in result:
265 if orcid is not None and isinstance(responsible_agent, ResponsibleAgent):
266 self._add_id(responsible_agent, orcid, "orcid")
267 author_id_found.append((orcid, "orcid"))
269 def __add_viaf_to_author(
270 self,
271 br: BibliographicResource,
272 ra: ResponsibleAgent,
273 author_id_found: list[tuple[str | None, str]],
274 ) -> None:
275 given = ra.get_given_name()
276 family = ra.get_family_name()
277 if not given and not family:
278 return
279 viaf = self.viaf_api.query(given or "", family or "", br.get_title() or "")
280 if viaf is not None:
281 self._add_id(ra, viaf, "viaf")
282 author_id_found.append((viaf, "viaf"))
284 def __add_wikidata_to_author(
285 self,
286 ra: ResponsibleAgent,
287 author_id_found: list[tuple[str | None, str]],
288 ) -> None:
289 for literal, schema in author_id_found:
290 if literal is None:
291 continue
292 result = self.wikidata_api.query(literal, schema)
293 if result:
294 self._add_id(ra, result, "wikidata", f"its {schema.upper()} {literal}")
295 break
297 def __enrich_publisher(
298 self,
299 ra: ResponsibleAgent,
300 has_doi: str | None,
301 *,
302 publisher_has_crossrefid: bool,
303 ) -> bool:
304 if publisher_has_crossrefid or has_doi is None:
305 return publisher_has_crossrefid
306 for publisher_id in ra.get_identifiers():
307 scheme = publisher_id.get_scheme()
308 if scheme is not None and GraphEntity.iri_crossref in scheme:
309 return True
310 crossref_id = self.crossref_api.query_publisher(has_doi)
311 if crossref_id:
312 self._add_id(ra, crossref_id, "crossref")
313 return False
315 def __serialize_graphs(self) -> None:
316 store_graph_set(self.graph_set, self.storage)
317 prov = self.__provenance()
318 prov.generate_provenance()
319 store_provenance(prov, self.storage)
321 def __provenance(self) -> ProvSet:
322 return ProvSet(
323 self.graph_set,
324 self.graph_set.base_iri,
325 info_dir=self.storage.info_dir,
326 wanted_label=self.storage.wanted_label,
327 custom_counter_handler=self.storage.counter_handler,
328 supplier_prefix=self.storage.supplier_prefix,
329 )
331 def _add_id(
332 self,
333 entity: BibliographicResource | ResponsibleAgent,
334 literal: str,
335 schema: str,
336 by_means_of: str | None = None,
337 ) -> None:
338 """
339 Add a new identifier to an entity.
341 :param entity: a bibliographic resource or an agent role
342 :param literal: the literal value of the identifier
343 :param schema: the schema of the identifier
344 :param by_means_of: an optional string that let you specify the API used
345 """
346 old_identifiers = entity.get_identifiers()
347 if self.__has_identifier(entity, literal):
348 if self.debug:
349 LOGGER.debug("Identifier %s already present", literal)
350 return
352 self.added_identifier_count += 1
354 new_id = self.graph_set.add_id(self.resp_agent)
355 create_identifier = {
356 "issn": new_id.create_issn,
357 "doi": new_id.create_doi,
358 "orcid": new_id.create_orcid,
359 "viaf": new_id.create_viaf,
360 "crossref": new_id.create_crossref,
361 "wikidata": new_id.create_wikidata,
362 "openalex": new_id.create_openalex,
363 }
364 create_identifier[schema](literal)
365 entity.has_identifier(new_id)
367 if self.debug:
368 self.__log_new_identifier(schema, literal, old_identifiers, entity, by_means_of)
370 def __has_identifier(self, entity: BibliographicResource | ResponsibleAgent, literal: str) -> bool:
371 return any(identifier.get_literal_value() == literal for identifier in entity.get_identifiers())
373 def __log_new_identifier(
374 self,
375 schema: str,
376 literal: str,
377 old_identifiers: Iterable[IdentifierLike],
378 entity: BibliographicResource | ResponsibleAgent,
379 by_means_of: str | None,
380 ) -> None:
381 message = f"[{datetime.now(timezone.utc).strftime('%Y/%m/%d %H:%M:%S')}] FOUND {schema}: {literal}"
382 if by_means_of is not None:
383 message += f", by means of {by_means_of}"
385 LOGGER.debug(message)
386 LOGGER.debug("OLD: %s", [identifier.get_literal_value() for identifier in old_identifiers])
387 LOGGER.debug("NEW: %s", [identifier.get_literal_value() for identifier in entity.get_identifiers()])
389 @contextlib.contextmanager
390 def __std_out_err_redirect_tqdm(self) -> Iterator[TextIO]:
391 """Redirect stdout and stderr through the TQDM progress bar."""
392 orig_out_err = sys.stdout, sys.stderr
393 try:
394 sys.stdout, sys.stderr = map(DummyTqdmFile, orig_out_err)
395 yield orig_out_err[0]
396 finally:
397 sys.stdout, sys.stderr = orig_out_err