Coverage for src/oc_graphenricher/APIs/__init__.py: 92%
339 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#
5# SPDX-License-Identifier: ISC
7from __future__ import annotations
9import json
10import logging
11import unicodedata
12from abc import ABC, abstractmethod
13from http import HTTPStatus
14from pathlib import Path
15from time import sleep
16from typing import cast
17from urllib.parse import quote
19import Levenshtein
20import requests
21import requests_cache
22from oc_ocdm.graph.graph_entity import GraphEntity
23from requests.exceptions import RequestException
25LOGGER = logging.getLogger(__name__)
27USER_AGENT = "GraphEnricher (via OpenCitations - http://opencitations.net; mailto:contact@opencitations.net)"
28CROSSREF_ROWS = 4
29TITLE_KEYWORD_LIMIT = 4
30CROSSREF_TITLE_THRESHOLD = 0.8
31YEAR_LENGTH = 4
32DEFAULT_RETRY_ATTEMPTS = 3
33DEFAULT_BACKOFF_FACTOR = 1.0
34RETRY_STATUS_CODES = {
35 HTTPStatus.TOO_MANY_REQUESTS,
36 HTTPStatus.INTERNAL_SERVER_ERROR,
37 HTTPStatus.BAD_GATEWAY,
38 HTTPStatus.SERVICE_UNAVAILABLE,
39 HTTPStatus.GATEWAY_TIMEOUT,
40}
42JsonDict = dict[str, object]
43AuthorTuple = tuple[str | None, str | None, str | None, object]
44IdentifierTuple = tuple[str | None, str | None]
47def _default_headers(content_type: str | None = None) -> dict[str, str]:
48 headers = {"User-Agent": USER_AGENT}
49 if content_type is not None:
50 headers["Content-Type"] = content_type
51 return headers
54def _normalize_ascii(value: str) -> str:
55 return unicodedata.normalize("NFKD", value).encode("ASCII", "ignore").decode("utf-8")
58def _get_with_retries(
59 url: str,
60 headers: dict[str, str],
61 timeout: float,
62 label: str,
63 params: dict[str, str] | None = None,
64 max_attempts: int = DEFAULT_RETRY_ATTEMPTS,
65 backoff_factor: float = DEFAULT_BACKOFF_FACTOR,
66) -> requests.Response | None:
67 last_error: object = "no attempts"
68 for attempt in range(max_attempts):
69 try:
70 if params is None:
71 response = requests.get(url, headers=headers, timeout=timeout)
72 else:
73 response = requests.get(url, headers=headers, timeout=timeout, params=params)
74 except RequestException as error:
75 last_error = error
76 else:
77 if response.status_code == HTTPStatus.OK:
78 return response
79 if response.status_code not in RETRY_STATUS_CODES:
80 return None
81 last_error = HTTPStatus(response.status_code)
83 if attempt + 1 < max_attempts:
84 sleep(backoff_factor * 2**attempt)
86 LOGGER.warning("%s retry attempts exhausted for %s: %r", label, url, last_error)
87 return None
90class QueryInterface(ABC):
91 def __init__(self) -> None:
92 """
93 Initialize the query interface.
95 The interface installs the shared requests cache used by concrete query backends.
96 """
97 requests_cache.install_cache("GraphEnricher_cache")
99 @abstractmethod
100 def query(self, *args: object, **kwargs: object) -> object:
101 raise NotImplementedError
104class VIAF(QueryInterface):
105 def __init__(self) -> None:
106 """
107 Initialize the VIAF query backend.
109 The backend extracts VIAF identifiers for authors by querying viaf.org.
110 """
111 super().__init__()
112 self.headers = {
113 "User-Agent": USER_AGENT,
114 "Accept": "application/json",
115 }
116 self.api_url = (
117 'http://www.viaf.org/viaf/search?local.title+all+"{}"&query=local.names+all+"{}"'
118 "&sortKeys=holdingscount&recordSchema=BriefVIAF"
119 )
121 def query(self, given_name: str, family_name: str, title: str) -> str | None:
122 """
123 Having specified the author's names and the title of a paper, extract a VIAF.
125 :param given_name: author's given name
126 :param family_name: author's family name
127 :param title: paper's title
128 :return: VIAF, if exists, otherwise None
129 """
130 name = f"{given_name} {family_name}".strip()
131 query = self.api_url.format(quote(title), quote(name))
132 response = _get_with_retries(query, self.headers, 60, "[GraphEnricher-VIAF]")
133 if response is None:
134 return None
136 data = response.json()
137 response_data = data["searchRetrieveResponse"]
138 number_of_records = response_data["numberOfRecords"]["content"]
139 if int(number_of_records) != 1:
140 return None
142 record_data = response_data["records"]["record"]["recordData"]
143 return str(record_data["v:VIAFCluster"]["v:viafID"]["content"])
146class Wikidata(QueryInterface):
147 def __init__(self) -> None:
148 """
149 Initialize the Wikidata query backend.
151 The backend queries Wikidata by means of another identifier and checks whether a related entity exists.
152 """
153 super().__init__()
154 self.headers = {
155 "User-Agent": USER_AGENT,
156 "Accept": "application/json",
157 }
158 self.api_url = "https://query.wikidata.org/sparql"
159 self.base_query = """
160 SELECT ?item WHERE {{
161 ?item p:{property} ?x.
162 ?x ps:{property} "{literal}".
163 }} LIMIT 1
164 """
165 self.schema_properties = {
166 "doi": "P356",
167 "issn": "P236",
168 "orcid": "P496",
169 "viaf": "P214",
170 "pmid": "P698",
171 "pmcid": "P932",
172 }
174 def query(self, entity: str, schema: str) -> str | None:
175 """
176 Query Wikidata, given the literal of an identifier and its schema.
178 :param entity: the literal of the given identifier
179 :param schema: the schema of the given identifier
180 :return: Wikidata ID if found, otherwise None
181 """
182 if schema not in self.schema_properties:
183 return None
185 literal = entity.upper() if schema == "doi" else entity
186 query = self.base_query.format(property=self.schema_properties[schema], literal=literal)
187 response = _get_with_retries(
188 self.api_url,
189 self.headers,
190 60,
191 "[GraphEnricher-Wikidata]",
192 params={"format": "json", "query": query},
193 )
194 if response is None:
195 return None
197 try:
198 data = response.json()
199 return data["results"]["bindings"][0]["item"]["value"].split("/")[-1]
200 except IndexError:
201 return None
204class Crossref(QueryInterface):
205 def __init__(
206 self,
207 crossref_min_similarity_score: float = 0.95,
208 max_iteration: int = DEFAULT_RETRY_ATTEMPTS,
209 sec_to_wait: float = DEFAULT_BACKOFF_FACTOR,
210 headers: dict[str, str] | None = None,
211 timeout: int = 30,
212 *,
213 is_json: bool = True,
214 ) -> None:
215 """
216 Initialize the Crossref query backend.
218 The backend extracts DOIs, ISSNs, and publisher IDs.
219 """
220 super().__init__()
222 self.max_iteration = max_iteration
223 self.sec_to_wait = sec_to_wait
224 self.headers = headers if headers is not None else _default_headers()
225 self.timeout = timeout
226 self.is_json = is_json
227 self.crossref_min_similarity_score = crossref_min_similarity_score
228 self.__crossref_doi_url = "https://api.crossref.org/works/"
229 self.__crossref_journal_url = "https://api.crossref.org/journals/"
230 stopwords_path = Path(__file__).with_name("stopwords-it.txt")
231 with stopwords_path.open(encoding="utf-8") as stopwords_file:
232 self.stoplist = {line.strip() for line in stopwords_file}
234 def _cleaning_title(self, title: str) -> str:
235 """
236 Clean a given title, filtering the words according to a stoplist.
238 :param title: the title string
239 :return: the cleaned title
240 """
241 keywords = [word for word in title.split(" ") if word not in self.stoplist]
242 return " ".join(keywords[:TITLE_KEYWORD_LIMIT])
244 def query_journal(self, issn: str) -> list[str] | None:
245 """
246 Query Crossref to get a list of other ISSNs for an ISSN.
248 :param issn: the ISSN of the bibliographic entity
249 :return: a list that contains any other ISSN found, otherwise an empty list
250 """
251 query = self.__crossref_journal_url + issn
252 response = _get_with_retries(
253 query,
254 self.headers,
255 self.timeout,
256 "[GraphEnricher-Crossref]",
257 max_attempts=self.max_iteration,
258 backoff_factor=self.sec_to_wait,
259 )
260 if response is None:
261 return None
263 data = response.json()
264 new_issn = cast("list[str]", data["message"]["ISSN"])
265 if issn in new_issn:
266 new_issn.remove(issn)
267 return new_issn
269 def query_publisher(self, doi: str) -> str | None:
270 """
271 Extract the identifier of a publisher starting from a given DOI.
273 :param doi: the DOI of the paper
274 :return: a string representing the ID of the publisher, otherwise None
275 """
276 url_cr = self.__crossref_doi_url + doi
277 response = _get_with_retries(
278 url_cr,
279 self.headers,
280 self.timeout,
281 "[GraphEnricher-Crossref-publisher]",
282 max_attempts=self.max_iteration,
283 backoff_factor=self.sec_to_wait,
284 )
285 if response is None:
286 return None
288 data = response.json()
289 return str(data["message"]["member"])
291 def query(self, fullnames: list[tuple[str | None, str | None]], title: str, year: str | None) -> str | None:
292 """
293 Extract the DOI, given authors, title and year of publication.
295 :param fullnames: a list composed of a tuple of <name, family_name> (e.g.: [ ("Gabriele", "Pisciotta") ]
296 :param title: the title of the paper
297 :param year: a string that represent the year of publication
298 :return: the DOI found, otherwise None
299 """
300 author_query, exist_author, name, surname = self.__author_query(fullnames)
301 query = f"query.bibliographic={self._cleaning_title(title)}{author_query}"
302 query += f"&rows={CROSSREF_ROWS}&select=DOI,title,author,issued"
303 url_cr = f"https://api.crossref.org/works?{query}"
304 response = _get_with_retries(
305 url_cr,
306 self.headers,
307 self.timeout,
308 "[GraphEnricher-Crossref]",
309 max_attempts=self.max_iteration,
310 backoff_factor=self.sec_to_wait,
311 )
312 if response is None:
313 return None
315 data = response.json()
316 return self.__best_doi(data, title, year, exist_author=exist_author, name=name, surname=surname)
318 def __author_query(self, fullnames: list[tuple[str | None, str | None]]) -> tuple[str, bool, str, str]:
319 query = ""
320 exist_author = False
321 name = ""
322 surname = ""
323 for fullname in fullnames:
324 name, surname = self.__normalized_fullname(fullname)
325 if name or surname:
326 exist_author = True
327 query += f"&query.author={name} {surname}".rstrip()
328 return query, exist_author, name, surname
330 def __normalized_fullname(self, fullname: tuple[str | None, str | None]) -> tuple[str, str]:
331 name = fullname[0].lower() if fullname[0] is not None else ""
332 surname = fullname[1].lower() if fullname[1] is not None else ""
333 return name, surname
335 def __best_doi(
336 self,
337 data: JsonDict,
338 title: str,
339 year: str | None,
340 *,
341 exist_author: bool,
342 name: str,
343 surname: str,
344 ) -> str | None:
345 doi = None
346 message = cast("JsonDict", data["message"])
347 items = cast("list[JsonDict]", message["items"])
348 if items:
349 possible = [
350 (
351 *self.__score_crossref_item(
352 item,
353 title,
354 year,
355 exist_author=exist_author,
356 name=name,
357 surname=surname,
358 ),
359 item,
360 )
361 for item in items
362 ]
363 best_title, best_authors, _best_year, best_item = sorted(possible, key=lambda item: item[:3])[-1]
364 if best_title > CROSSREF_TITLE_THRESHOLD and (not exist_author or best_authors >= 1) and "DOI" in best_item:
365 doi = str(best_item["DOI"])
366 return doi
368 def __score_crossref_item(
369 self,
370 item: JsonDict,
371 title: str,
372 year: str | None,
373 *,
374 exist_author: bool,
375 name: str,
376 surname: str,
377 ) -> tuple[float, int, int]:
378 point_year = self.__year_score(item, year)
379 point_authors = self.__author_score(item, exist_author=exist_author, name=name, surname=surname)
380 point_title = self.__title_score(item, title)
381 return point_title, point_authors, point_year
383 def __year_score(self, item: JsonDict, year: str | None) -> int:
384 year_int = self.__year_int(year)
385 if year_int is None or "issued" not in item:
386 return 0
387 issued = cast("JsonDict", item["issued"])
388 if "date-parts" not in issued:
389 return 0
390 date_parts = cast("list[list[int | None]]", issued["date-parts"])
391 if not date_parts or not date_parts[0] or date_parts[0][0] is None:
392 return 0
393 return 3 if int(date_parts[0][0]) == year_int else 0
395 def __year_int(self, year: str | None) -> int | None:
396 if year is None:
397 return None
398 year_string = str(year)
399 if "-" in year_string:
400 for element_of_year in year_string.split("-"):
401 if len(element_of_year) == YEAR_LENGTH:
402 year_string = element_of_year
403 break
404 return int(year_string)
406 def __author_score(self, item: JsonDict, *, exist_author: bool, name: str, surname: str) -> int:
407 if not exist_author or "author" not in item:
408 return 0
409 point_authors = 0
410 authors = cast("list[JsonDict]", item["author"])
411 for author in authors:
412 if "family" not in author:
413 continue
414 family = cast("str", author["family"]).lower()
415 if "given" not in author:
416 if family == surname:
417 point_authors += 1
418 continue
419 given = cast("str", author["given"]).lower()
420 if family == surname and given == name:
421 point_authors += 2
422 elif family == surname and name and given and given[0] == name[0]:
423 point_authors += 1
424 return point_authors
426 def __title_score(self, item: JsonDict, title: str) -> float:
427 if "title" not in item:
428 return 0
429 title_values = cast("list[str]", item["title"])
430 if not title_values:
431 return 0
432 return Levenshtein.ratio(title, title_values[0].lower())
435class ORCID(QueryInterface):
436 def __init__(
437 self,
438 max_iteration: int = DEFAULT_RETRY_ATTEMPTS,
439 sec_to_wait: float = DEFAULT_BACKOFF_FACTOR,
440 headers: dict[str, str] | None = None,
441 timeout: int = 30,
442 repok: object = None,
443 reperr: object = None,
444 *,
445 is_json: bool = True,
446 ) -> None:
447 """
448 Initialize the ORCID query backend.
450 The backend extracts ORCID identifiers.
451 """
452 del repok, reperr
453 super().__init__()
455 self.max_iteration = max_iteration
456 self.sec_to_wait = sec_to_wait
457 self.headers = headers if headers is not None else _default_headers("application/json")
458 self.timeout = timeout
459 self.is_json = is_json
460 self.__orcid_api_url = "https://pub.orcid.org/v2.1/search?q="
461 self.__personal_url = "https://pub.orcid.org/v2.1/%s/personal-details"
463 def query(self, authors: list[AuthorTuple], identifiers: list[IdentifierTuple]) -> list[AuthorTuple] | None:
464 """
465 Given a list of authors and a list of identifiers, returns the ORCIDs in the list of authors.
467 :param authors: a list of tuples in the following form [ (name, family_name, ORCID, ar_object) ]
468 :param identifiers: a list of identifiers of the bibliographic resource
469 :return: the authors list enriched with the ORCID identifier
470 """
471 if len(identifiers) == 0:
472 return None
474 records = self._get_orcid_records(identifiers, authors)
475 to_return: dict[tuple[str | None, str | None], str] = {}
476 if records is not None:
477 records_data = cast("JsonDict", records)
478 results = cast("list[JsonDict]", records_data["result"])
479 for record in results:
480 orcid_identifier = cast("JsonDict", record["orcid-identifier"])
481 orcid_id = str(orcid_identifier["path"])
482 self.__collect_matching_orcid(authors, to_return, orcid_id)
484 return [(author[0], author[1], to_return.get((author[0], author[1])), author[3]) for author in authors]
486 def _get_orcid_records(
487 self,
488 identifiers: list[IdentifierTuple],
489 family_names: list[AuthorTuple],
490 ) -> JsonDict | str | None:
491 identifier_query = self.__identifier_query(identifiers)
492 family_query = self.__family_query(family_names)
493 if identifier_query and family_query:
494 cur_query = f"{identifier_query} AND ({family_query})"
495 elif identifier_query:
496 cur_query = identifier_query
497 else:
498 cur_query = family_query
500 if cur_query == "":
501 return None
503 query_url = self.__orcid_api_url + quote(cur_query)
504 return self.__get_data(query_url)
506 def __collect_matching_orcid(
507 self,
508 authors: list[AuthorTuple],
509 to_return: dict[tuple[str | None, str | None], str],
510 orcid_id: str,
511 ) -> None:
512 personal_details = self.__get_data(self.__personal_url % orcid_id.upper())
513 if personal_details is None:
514 return
516 personal_data = cast("JsonDict", personal_details)
517 name = cast("JsonDict", personal_data["name"])
518 given_names = cast("JsonDict", name["given-names"])
519 family_name_data = cast("JsonDict", name["family-name"])
520 given_name = str(given_names["value"]).lower()
521 family_name = str(family_name_data["value"]).lower()
523 for author in authors:
524 if self.__matches_author(author, given_name, family_name) and to_return.get((author[0], author[1])) is None:
525 to_return[(author[0], author[1])] = orcid_id.upper()
527 def __matches_author(self, author: AuthorTuple, given_name: str, family_name: str) -> bool:
528 if author[2] is not None or author[1] is None:
529 return False
530 if author[1].lower() not in family_name:
531 return False
532 return author[0] is None or author[0].lower() in given_name
534 def __identifier_query(self, identifiers: list[IdentifierTuple]) -> str:
535 terms = []
536 for scheme, value in identifiers:
537 if value is None:
538 continue
539 if scheme == GraphEntity.iri_doi:
540 terms.extend(self.__doi_terms(value))
541 elif scheme == GraphEntity.iri_isbn:
542 terms.append(f'isbn:"{value}"')
543 elif scheme == GraphEntity.iri_pmid:
544 terms.append(f'pmid-self:"{value}"')
545 if not terms:
546 return ""
547 return f"({' OR '.join(terms)})"
549 def __doi_terms(self, doi_string: str) -> list[str]:
550 terms = [f'doi-self:"{doi_string}"']
551 doi_string_l = doi_string.lower()
552 doi_string_u = doi_string.upper()
553 if doi_string_l != doi_string:
554 terms.append(f'doi-self:"{doi_string_l}"')
555 if doi_string_u != doi_string:
556 terms.append(f'doi-self:"{doi_string_u}"')
557 return terms
559 def __family_query(self, family_names: list[AuthorTuple]) -> str:
560 terms = []
561 for given_names, family_name, _orcid, _entity in family_names:
562 if family_name is None:
563 continue
564 term = f'family-name:"{_normalize_ascii(family_name)}"'
565 if given_names:
566 term += f' AND given-names:"{_normalize_ascii(given_names)}"'
567 terms.append(term)
568 return " OR ".join(terms)
570 def __get_data(self, get_url: str) -> JsonDict | str | None:
571 """
572 Send requests.
574 :param get_url: the URL to query
575 :return: results if found, otherwise None
576 """
577 response = _get_with_retries(
578 get_url,
579 self.headers,
580 self.timeout,
581 "[GraphEnricher-ORCID]",
582 max_attempts=self.max_iteration,
583 backoff_factor=self.sec_to_wait,
584 )
585 if response is None:
586 return None
587 if self.is_json:
588 return cast("JsonDict", json.loads(response.text))
589 return response.text
592class OpenAlex(QueryInterface):
593 def __init__(self) -> None:
594 super().__init__()
595 self.headers = {"User-Agent": USER_AGENT}
596 self.api_url_works = "https://api.openalex.org/works"
597 self.api_url_sources = "https://api.openalex.org/sources"
599 def query(self, entity: str, schema: str) -> list[str] | None:
600 schema = schema.lower()
601 query = self.__query_url(entity, schema)
602 result = None
603 if query is None:
604 LOGGER.warning("[GraphEnricher-OpenAlex]:The specified schema '%s' is not supported", schema)
605 else:
606 response = _get_with_retries(query, self.headers, 60, "[GraphEnricher-OpenAlex]")
607 if response is not None:
608 result = self.__response_results(response)
609 return result
611 def __response_results(self, response: requests.Response) -> list[str] | None:
612 data = response.json()
613 results = cast("list[JsonDict]", data["results"])
614 if not results:
615 return None
616 return [str(result["id"]).replace("https://openalex.org/", "") for result in results]
618 def __query_url(self, entity: str, schema: str) -> str | None:
619 if schema in {"doi", "pmid", "pmcid"}:
620 return f"{self.api_url_works}?filter={schema}:{entity}&select=id"
621 if schema == "issn":
622 return f"{self.api_url_sources}?filter={schema}:{entity}&select=id"
623 return None