Coverage for oc_meta / lib / agent_metadata.py: 77%
259 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-07-25 10:39 +0000
1# SPDX-FileCopyrightText: 2026 Arcangelo Massari <arcangelo.massari@unibo.it>
2#
3# SPDX-License-Identifier: ISC
5from __future__ import annotations
7import json
8import sqlite3
9import time
10from collections.abc import Mapping
11from dataclasses import dataclass, field
12from datetime import datetime, timezone
13from typing import TypeAlias, TypedDict, cast
14from urllib.parse import quote
16import requests
17from oc_ds_converter.oc_idmanager import ORCIDManager
19JsonValue: TypeAlias = (
20 None | bool | int | float | str | list["JsonValue"] | dict[str, "JsonValue"]
21)
22JsonObject: TypeAlias = dict[str, JsonValue]
24CROSSREF_BASE = "https://api.crossref.org/works/"
25DATACITE_BASE = "https://api.datacite.org/dois/"
26OPENALEX_WORKS = "https://api.openalex.org/works"
27ORCID_BASE = "https://pub.orcid.org/v3.0/"
29_ORCID_MANAGER = ORCIDManager(use_api_service=False)
32class AgentMetadata(TypedDict):
33 family: str
34 given: str
35 name: str
36 orcid: str | None
37 identifiers: tuple[AgentIdentifier, ...]
38 position: int
39 role: str
42class AgentIdentifier(TypedDict):
43 scheme: str
44 value: str
47class WorkMetadata(TypedDict):
48 identifier: str
49 source: str
50 author: list[AgentMetadata]
51 editor: list[AgentMetadata]
52 publisher: str
53 publisher_identifiers: tuple[AgentIdentifier, ...]
56class OrcidProfile(TypedDict):
57 orcid: str
58 given: str
59 family: str
60 name: str
63def agents_for_role(work: WorkMetadata, role: str) -> list[AgentMetadata]:
64 if role == "author":
65 return work["author"]
66 if role == "editor":
67 return work["editor"]
68 if role == "publisher" and work["publisher"]:
69 return [
70 AgentMetadata(
71 family="",
72 given="",
73 name=work["publisher"],
74 orcid=None,
75 identifiers=work["publisher_identifiers"],
76 position=0,
77 role="publisher",
78 )
79 ]
80 return []
83def normalize_orcid(orcid: str) -> str:
84 normalized = orcid.strip()
85 for prefix in ("https://orcid.org/", "http://orcid.org/", "orcid:"):
86 if normalized.casefold().startswith(prefix):
87 normalized = normalized[len(prefix) :]
88 break
89 normalized = normalized.upper()
90 compact = normalized.replace("-", "")
91 if (
92 len(compact) == 16
93 and compact[:15].isdigit()
94 and (compact[-1].isdigit() or compact[-1] == "X")
95 ):
96 return "-".join(compact[index : index + 4] for index in range(0, 16, 4))
97 return normalized
100def is_valid_orcid(orcid: str) -> bool:
101 normalized = normalize_orcid(orcid)
102 return _ORCID_MANAGER.syntax_ok(normalized) and _ORCID_MANAGER.check_digit(
103 normalized
104 )
107def _orcid_or_none(value: str) -> str | None:
108 normalized = normalize_orcid(value)
109 return normalized if is_valid_orcid(normalized) else None
112def _object(value: JsonValue) -> JsonObject:
113 return cast(JsonObject, value) if isinstance(value, dict) else {}
116def _objects(value: JsonValue) -> list[JsonObject]:
117 if not isinstance(value, list):
118 return []
119 return [cast(JsonObject, item) for item in value if isinstance(item, dict)]
122def _string(value: JsonValue) -> str:
123 return value if isinstance(value, str) else ""
126def _name_identifiers(value: JsonValue) -> tuple[AgentIdentifier, ...]:
127 identifiers = []
128 for identifier in _objects(value):
129 scheme = _string(identifier.get("nameIdentifierScheme")).upper()
130 if scheme == "ORCID":
131 orcid = _orcid_or_none(_string(identifier.get("nameIdentifier")))
132 if orcid is not None:
133 identifiers.append(AgentIdentifier(scheme="orcid", value=orcid))
134 elif scheme == "ROR":
135 value = _string(identifier.get("nameIdentifier")).rstrip("/")
136 if value:
137 identifiers.append(
138 AgentIdentifier(scheme="ror", value=value.rsplit("/", 1)[-1])
139 )
140 return tuple(identifiers)
143def _orcid_identifier(value: str) -> tuple[AgentIdentifier, ...]:
144 orcid = _orcid_or_none(value)
145 if orcid is None:
146 return ()
147 return (AgentIdentifier(scheme="orcid", value=orcid),)
150def _member_identifier(value: JsonValue) -> tuple[AgentIdentifier, ...]:
151 if not isinstance(value, (str, int)) or isinstance(value, bool):
152 return ()
153 member = str(value).rstrip("/").rsplit("/", 1)[-1]
154 if not member:
155 return ()
156 return (AgentIdentifier(scheme="crossref", value=member),)
159def _publisher_identifiers(value: JsonValue) -> tuple[AgentIdentifier, ...]:
160 publisher = _object(value)
161 scheme = _string(publisher.get("publisherIdentifierScheme")).upper()
162 identifier = _string(publisher.get("publisherIdentifier")).rstrip("/")
163 if scheme != "ROR" or not identifier:
164 return ()
165 return (AgentIdentifier(scheme="ror", value=identifier.rsplit("/", 1)[-1]),)
168def parse_crossref_work(data: JsonObject, identifier: str = "") -> WorkMetadata:
169 message = _object(data.get("message"))
170 authors = []
171 for position, author in enumerate(_objects(message.get("author"))):
172 identifiers = _orcid_identifier(_string(author.get("ORCID")))
173 authors.append(
174 AgentMetadata(
175 family=_string(author.get("family")),
176 given=_string(author.get("given")),
177 name=_string(author.get("name")),
178 orcid=identifiers[0]["value"] if identifiers else None,
179 identifiers=identifiers,
180 position=position,
181 role="author",
182 )
183 )
184 editors = []
185 for position, editor in enumerate(_objects(message.get("editor"))):
186 identifiers = _orcid_identifier(_string(editor.get("ORCID")))
187 editors.append(
188 AgentMetadata(
189 family=_string(editor.get("family")),
190 given=_string(editor.get("given")),
191 name=_string(editor.get("name")),
192 orcid=identifiers[0]["value"] if identifiers else None,
193 identifiers=identifiers,
194 position=position,
195 role="editor",
196 )
197 )
198 return WorkMetadata(
199 identifier=identifier or _string(message.get("DOI")),
200 source="crossref",
201 author=authors,
202 editor=editors,
203 publisher=_string(message.get("publisher")),
204 publisher_identifiers=_member_identifier(message.get("member")),
205 )
208def parse_datacite_work(data: JsonObject, identifier: str = "") -> WorkMetadata:
209 attributes = _object(_object(data.get("data")).get("attributes"))
210 authors = []
211 for position, creator in enumerate(_objects(attributes.get("creators"))):
212 identifiers = _name_identifiers(creator.get("nameIdentifiers"))
213 orcid = next(
214 (
215 identifier["value"]
216 for identifier in identifiers
217 if identifier["scheme"] == "orcid"
218 ),
219 None,
220 )
221 authors.append(
222 AgentMetadata(
223 family=_string(creator.get("familyName")),
224 given=_string(creator.get("givenName")),
225 name=_string(creator.get("name")),
226 orcid=orcid,
227 identifiers=identifiers,
228 position=position,
229 role="author",
230 )
231 )
232 editors = []
233 for contributor in _objects(attributes.get("contributors")):
234 if _string(contributor.get("contributorType")) != "Editor":
235 continue
236 identifiers = _name_identifiers(contributor.get("nameIdentifiers"))
237 orcid = next(
238 (
239 identifier["value"]
240 for identifier in identifiers
241 if identifier["scheme"] == "orcid"
242 ),
243 None,
244 )
245 editors.append(
246 AgentMetadata(
247 family=_string(contributor.get("familyName")),
248 given=_string(contributor.get("givenName")),
249 name=_string(contributor.get("name")),
250 orcid=orcid,
251 identifiers=identifiers,
252 position=len(editors),
253 role="editor",
254 )
255 )
256 publisher_value = attributes.get("publisher")
257 publisher = (
258 _string(_object(publisher_value).get("name"))
259 if isinstance(publisher_value, dict)
260 else _string(publisher_value)
261 )
262 return WorkMetadata(
263 identifier=identifier or _string(_object(data.get("data")).get("id")),
264 source="datacite",
265 author=authors,
266 editor=editors,
267 publisher=publisher,
268 publisher_identifiers=_publisher_identifiers(publisher_value),
269 )
272def parse_openalex_work(data: JsonObject, identifier: str = "") -> WorkMetadata:
273 authors = []
274 for position, authorship in enumerate(_objects(data.get("authorships"))):
275 author = _object(authorship.get("author"))
276 display_name = _string(author.get("display_name"))
277 raw_name = _string(authorship.get("raw_author_name")) or display_name
278 identifiers = _orcid_identifier(_string(author.get("orcid")))
279 authors.append(
280 AgentMetadata(
281 family="",
282 given="",
283 name=raw_name,
284 orcid=identifiers[0]["value"] if identifiers else None,
285 identifiers=identifiers,
286 position=position,
287 role="author",
288 )
289 )
290 return WorkMetadata(
291 identifier=identifier or _string(data.get("id")),
292 source="openalex",
293 author=authors,
294 editor=[],
295 publisher="",
296 publisher_identifiers=(),
297 )
300def parse_orcid_profile(data: JsonObject, orcid: str) -> OrcidProfile:
301 person_value = data.get("person")
302 person = _object(person_value) if isinstance(person_value, dict) else data
303 name = _object(person.get("name"))
304 given = _string(_object(name.get("given-names")).get("value"))
305 family = _string(_object(name.get("family-name")).get("value"))
306 return OrcidProfile(
307 orcid=normalize_orcid(orcid),
308 given=given,
309 family=family,
310 name=" ".join(part for part in (given, family) if part),
311 )
314class ApiCache:
315 def __init__(self, path: str) -> None:
316 self.connection = sqlite3.connect(path)
317 self.connection.execute(
318 """
319 CREATE TABLE IF NOT EXISTS api_response (
320 source TEXT NOT NULL,
321 cache_key TEXT NOT NULL,
322 status INTEGER NOT NULL,
323 body TEXT,
324 fetched_at TEXT NOT NULL,
325 PRIMARY KEY (source, cache_key)
326 )
327 """
328 )
329 self.connection.commit()
331 def get(self, source: str, cache_key: str) -> tuple[int, JsonObject | None] | None:
332 row = self.connection.execute(
333 "SELECT status, body FROM api_response WHERE source = ? AND cache_key = ?",
334 (source, cache_key),
335 ).fetchone()
336 if row is None:
337 return None
338 status = cast(int, row[0])
339 body = cast(str | None, row[1])
340 return status, cast(JsonObject, json.loads(body)) if body else None
342 def set(
343 self, source: str, cache_key: str, status: int, body: JsonObject | None
344 ) -> None:
345 self.connection.execute(
346 """
347 INSERT INTO api_response (source, cache_key, status, body, fetched_at)
348 VALUES (?, ?, ?, ?, ?)
349 ON CONFLICT(source, cache_key) DO UPDATE SET
350 status = excluded.status,
351 body = excluded.body,
352 fetched_at = excluded.fetched_at
353 """,
354 (
355 source,
356 cache_key,
357 status,
358 json.dumps(body, ensure_ascii=False) if body is not None else None,
359 datetime.now(timezone.utc).isoformat(),
360 ),
361 )
362 self.connection.commit()
364 def close(self) -> None:
365 self.connection.close()
368@dataclass(slots=True)
369class AgentMetadataClient:
370 mailto: str
371 cache: ApiCache
372 refresh_cache: bool = False
373 openalex_api_key: str = ""
374 timeout: int = 30
375 max_attempts: int = 4
376 session: requests.Session = field(init=False)
378 def __post_init__(self) -> None:
379 self.session = requests.Session()
380 self.session.headers.update(
381 {
382 "Accept": "application/json",
383 "User-Agent": f"oc_meta-agent-audit/1.0 (mailto:{self.mailto})",
384 }
385 )
387 def _request(
388 self,
389 source: str,
390 cache_key: str,
391 url: str,
392 params: Mapping[str, str | int] | None = None,
393 headers: Mapping[str, str] | None = None,
394 ) -> JsonObject | None:
395 if not self.refresh_cache:
396 cached = self.cache.get(source, cache_key)
397 if cached is not None:
398 return cached[1] if cached[0] == 200 else None
399 for attempt in range(self.max_attempts):
400 try:
401 response = self.session.get(
402 url, params=params, headers=headers, timeout=self.timeout
403 )
404 except requests.RequestException:
405 if attempt + 1 == self.max_attempts:
406 raise
407 time.sleep(2**attempt)
408 continue
409 if response.status_code == 200:
410 body = cast(JsonObject, response.json())
411 self.cache.set(source, cache_key, response.status_code, body)
412 return body
413 if response.status_code == 404:
414 self.cache.set(source, cache_key, response.status_code, None)
415 return None
416 if response.status_code == 429 or response.status_code >= 500:
417 if attempt + 1 == self.max_attempts:
418 response.raise_for_status()
419 retry_after = response.headers.get("Retry-After")
420 time.sleep(float(retry_after) if retry_after else 2**attempt)
421 continue
422 response.raise_for_status()
423 return None
425 def crossref(self, doi: str) -> WorkMetadata | None:
426 data = self._request(
427 "crossref",
428 doi.lower(),
429 CROSSREF_BASE + quote(doi, safe=""),
430 {"mailto": self.mailto},
431 )
432 return parse_crossref_work(data, doi) if data is not None else None
434 def datacite(self, doi: str) -> WorkMetadata | None:
435 data = self._request(
436 "datacite", doi.lower(), DATACITE_BASE + quote(doi, safe="")
437 )
438 return parse_datacite_work(data, doi) if data is not None else None
440 def openalex_work(
441 self, doi: str = "", openalex_id: str = ""
442 ) -> WorkMetadata | None:
443 if openalex_id:
444 normalized = openalex_id.rsplit("/", 1)[-1].upper()
445 params = (
446 {"api_key": self.openalex_api_key} if self.openalex_api_key else None
447 )
448 data = self._request(
449 "openalex", normalized, f"{OPENALEX_WORKS}/{normalized}", params
450 )
451 return parse_openalex_work(data, normalized) if data is not None else None
452 if not doi:
453 return None
454 params = {"filter": f"doi:{doi}", "per-page": 1}
455 if self.openalex_api_key:
456 params["api_key"] = self.openalex_api_key
457 data = self._request(
458 "openalex",
459 f"doi:{doi.lower()}",
460 OPENALEX_WORKS,
461 params,
462 )
463 if data is None:
464 return None
465 results = _objects(data.get("results"))
466 return parse_openalex_work(results[0], doi) if results else None
468 def orcid(self, orcid: str) -> OrcidProfile | None:
469 normalized = normalize_orcid(orcid)
470 data = self._request(
471 "orcid",
472 normalized,
473 ORCID_BASE + quote(normalized, safe="") + "/person",
474 headers={"Accept": "application/vnd.orcid+json"},
475 )
476 return parse_orcid_profile(data, normalized) if data is not None else None
478 def work_sources(self, doi: str, openalex_id: str = "") -> list[WorkMetadata]:
479 works = []
480 primary = self.crossref(doi) if doi else None
481 if primary is None and doi:
482 primary = self.datacite(doi)
483 if primary is not None:
484 works.append(primary)
485 openalex = self.openalex_work(doi, openalex_id)
486 if openalex is not None:
487 works.append(openalex)
488 return works
490 def all_work_sources(self, doi: str, openalex_id: str = "") -> list[WorkMetadata]:
491 works = []
492 if doi:
493 crossref = self.crossref(doi)
494 if crossref is not None:
495 works.append(crossref)
496 datacite = self.datacite(doi)
497 if datacite is not None:
498 works.append(datacite)
499 openalex = self.openalex_work(doi, openalex_id)
500 if openalex is not None:
501 works.append(openalex)
502 return works
504 def close(self) -> None:
505 self.session.close()