Coverage for oc_meta / lib / bibliographic_matching.py: 94%
137 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 time
8from urllib.parse import quote
10from oc_ds_converter.oc_idmanager.support import call_api
11from oc_ocdm.graph.graph_entity import GraphEntity
12from rapidfuzz.distance import Levenshtein
14from oc_meta.lib.sparql import execute_sparql
16DATACITE_DOI = GraphEntity.iri_doi
17DATACITE_ISSN = GraphEntity.iri_issn
18DATACITE_HAS_ID = GraphEntity.iri_has_identifier
19DATACITE_USES_SCHEME = GraphEntity.iri_uses_identifier_scheme
20LITERAL_HAS_VALUE = GraphEntity.iri_has_literal_value
21DCTERMS_TITLE = GraphEntity.iri_title
22PRISM_PUB_DATE = GraphEntity.iri_has_publication_date
23PRISM_START_PAGE = GraphEntity.iri_starting_page
24PRISM_END_PAGE = GraphEntity.iri_ending_page
25FRBR_PART_OF = GraphEntity.iri_part_of
26FRBR_EMBODIMENT = GraphEntity.iri_embodiment
27FABIO_HAS_SEQ_ID = GraphEntity.iri_has_sequence_identifier
28FABIO_JOURNAL_VOLUME = GraphEntity.iri_journal_volume
29FABIO_JOURNAL_ISSUE = GraphEntity.iri_journal_issue
30PRO_IS_DOC_CONTEXT = GraphEntity.iri_is_document_context_for
31PRO_WITH_ROLE = GraphEntity.iri_with_role
32PRO_IS_HELD_BY = GraphEntity.iri_is_held_by
33PRO_AUTHOR = GraphEntity.iri_author
34FOAF_FAMILY_NAME = GraphEntity.iri_family_name
35FOAF_GIVEN_NAME = GraphEntity.iri_given_name
36OCO_HAS_NEXT = GraphEntity.iri_has_next
38CROSSREF_API = "https://api.crossref.org/works/"
39CROSSREF_RATE_LIMIT = 50
40MATCHING_THRESHOLD = 25.0
43def crossref_headers(mailto: str) -> dict[str, str]:
44 return {
45 "Accept": "application/json",
46 "User-Agent": f"oc_meta/mailto:{mailto}",
47 }
50def fetch_crossref_metadata(
51 doi: str, cache: dict[str, dict | None], mailto: str
52) -> dict | None:
53 if doi in cache:
54 return cache[doi]
55 time.sleep(1.0 / CROSSREF_RATE_LIMIT)
56 url = CROSSREF_API + quote(doi, safe="")
57 response = call_api(url=url, headers=crossref_headers(mailto))
58 if response is None or not isinstance(response, dict):
59 cache[doi] = None
60 return None
61 msg: dict = response["message"] # type: ignore[assignment]
62 titles: list = msg.get("title", [])
63 authors: list = msg.get("author", [])
64 first_author: dict = authors[0] if authors else {}
65 date_parts: list | None = msg.get("issued", {}).get("date-parts")
66 year = str(date_parts[0][0]) if date_parts and date_parts[0] else ""
67 page: str = msg.get("page", "")
68 pages = page.split("-", 1) if page else []
69 meta = {
70 "title": titles[0].lower().strip() if titles else "",
71 "first_author_family": first_author.get("family", "").lower().strip(),
72 "first_author_given": first_author.get("given", "").strip(),
73 "year": year,
74 "venue": (msg.get("container-title") or [""])[0].lower().strip(),
75 "issn": (msg.get("ISSN") or [""])[0],
76 "volume": msg.get("volume", ""),
77 "issue": msg.get("issue", ""),
78 "start_page": pages[0].strip() if pages else "",
79 "end_page": pages[1].strip() if len(pages) > 1 else "",
80 }
81 cache[doi] = meta
82 return meta
85def fetch_triplestore_metadata(endpoint: str, br_uri: str) -> dict:
86 query = f"""
87 SELECT ?title ?date ?venue_title ?venue_issn ?volume ?issue
88 ?start_page ?end_page
89 ?ar ?author_family ?author_given ?ar_next WHERE {{
90 OPTIONAL {{ <{br_uri}> <{DCTERMS_TITLE}> ?title }}
91 OPTIONAL {{ <{br_uri}> <{PRISM_PUB_DATE}> ?date }}
92 OPTIONAL {{
93 <{br_uri}> <{FRBR_PART_OF}>+ ?venue .
94 ?venue <{DCTERMS_TITLE}> ?venue_title .
95 OPTIONAL {{
96 ?venue <{DATACITE_HAS_ID}> ?venue_id_ent .
97 ?venue_id_ent <{DATACITE_USES_SCHEME}> <{DATACITE_ISSN}> .
98 ?venue_id_ent <{LITERAL_HAS_VALUE}> ?venue_issn .
99 }}
100 }}
101 OPTIONAL {{
102 <{br_uri}> <{FRBR_PART_OF}>* ?vol_parent .
103 ?vol_parent <{FRBR_PART_OF}> ?vol_container .
104 ?vol_container a <{FABIO_JOURNAL_VOLUME}> .
105 ?vol_container <{FABIO_HAS_SEQ_ID}> ?volume .
106 }}
107 OPTIONAL {{
108 <{br_uri}> <{FRBR_PART_OF}>* ?iss_parent .
109 ?iss_parent <{FRBR_PART_OF}> ?iss_container .
110 ?iss_container a <{FABIO_JOURNAL_ISSUE}> .
111 ?iss_container <{FABIO_HAS_SEQ_ID}> ?issue .
112 }}
113 OPTIONAL {{
114 <{br_uri}> <{FRBR_EMBODIMENT}> ?re .
115 ?re <{PRISM_START_PAGE}> ?start_page .
116 OPTIONAL {{ ?re <{PRISM_END_PAGE}> ?end_page }}
117 }}
118 OPTIONAL {{
119 <{br_uri}> <{PRO_IS_DOC_CONTEXT}> ?ar .
120 ?ar <{PRO_WITH_ROLE}> <{PRO_AUTHOR}> .
121 ?ar <{PRO_IS_HELD_BY}> ?ra .
122 ?ra <{FOAF_FAMILY_NAME}> ?author_family .
123 OPTIONAL {{ ?ra <{FOAF_GIVEN_NAME}> ?author_given }}
124 OPTIONAL {{ ?ar <{OCO_HAS_NEXT}> ?ar_next }}
125 }}
126 }}
127 """
128 result = execute_sparql(endpoint, query)
129 bindings = result["results"]["bindings"]
130 if not bindings:
131 return {}
133 first = bindings[0]
134 _val = _binding_value
136 ar_to_family: dict[str, str] = {}
137 ar_to_given: dict[str, str] = {}
138 ar_to_next: dict[str, str] = {}
139 for row in bindings:
140 ar = _val(row, "ar")
141 if ar:
142 family = _val(row, "author_family")
143 if family:
144 ar_to_family[ar] = family
145 given = _val(row, "author_given")
146 if given:
147 ar_to_given[ar] = given
148 nxt = _val(row, "ar_next")
149 if nxt:
150 ar_to_next[ar] = nxt
152 first_author_family = ""
153 first_author_given = ""
154 if ar_to_family:
155 pointed_to = set(ar_to_next.values())
156 first_ar_candidates = [ar for ar in ar_to_family if ar not in pointed_to]
157 first_ar = (
158 first_ar_candidates[0] if first_ar_candidates else next(iter(ar_to_family))
159 )
160 first_author_family = ar_to_family[first_ar].lower().strip()
161 first_author_given = ar_to_given.get(first_ar, "").strip()
163 year = _val(first, "date") or ""
164 if year and len(year) >= 4:
165 year = year[:4]
167 return {
168 "title": (_val(first, "title") or "").lower().strip(),
169 "first_author_family": first_author_family,
170 "first_author_given": first_author_given,
171 "year": year,
172 "venue": (_val(first, "venue_title") or "").lower().strip(),
173 "issn": _val(first, "venue_issn") or "",
174 "volume": _val(first, "volume") or "",
175 "issue": _val(first, "issue") or "",
176 "start_page": _val(first, "start_page") or "",
177 "end_page": _val(first, "end_page") or "",
178 }
181def _binding_value(row: dict, key: str) -> str:
182 return row[key]["value"] if key in row else ""
185def compute_matching_score(meta_a: dict, meta_b: dict) -> float:
186 m_first_author = _score_first_author(
187 meta_a["first_author_family"],
188 meta_b["first_author_family"],
189 meta_a["first_author_given"],
190 meta_b["first_author_given"],
191 )
192 m_title = _score_title(meta_a["title"], meta_b["title"])
193 m_source = _score_source(
194 meta_a["venue"],
195 meta_b["venue"],
196 meta_a["issn"],
197 meta_b["issn"],
198 )
199 m_other = _score_other(meta_a, meta_b)
200 return 7 * m_first_author + 14 * m_title + 5 * m_source + 14 * m_other
203def _score_first_author(
204 family_a: str, family_b: str, given_a: str, given_b: str
205) -> float:
206 if not family_a or not family_b:
207 return 0.0
208 max_len = max(len(family_a), len(family_b))
209 dist = Levenshtein.distance(family_a, family_b)
210 family_sim = 0.8 * (1.0 - dist / max_len)
211 initial_a = given_a[0].lower() if given_a else ""
212 initial_b = given_b[0].lower() if given_b else ""
213 initial_match = 0.2 if (initial_a and initial_b and initial_a == initial_b) else 0.0
214 return family_sim + initial_match
217def _score_title(title_a: str, title_b: str) -> float:
218 if not title_a or not title_b:
219 return 0.0
220 max_len = max(len(title_a), len(title_b))
221 dist = Levenshtein.distance(title_a, title_b)
222 return 1.0 - dist / max_len
225def _score_source(venue_a: str, venue_b: str, issn_a: str, issn_b: str) -> float:
226 if issn_a and issn_b and issn_a == issn_b:
227 return 1.0
228 if not venue_a or not venue_b:
229 return 0.0
230 min_len = min(len(venue_a), len(venue_b))
231 if min_len == 0:
232 return 0.0
233 dist = Levenshtein.distance(venue_a, venue_b)
234 len_diff = abs(len(venue_a) - len(venue_b))
235 score = 1.0 - (dist - len_diff) / min_len
236 return max(score, 0.0)
239def _score_other(meta_a: dict, meta_b: dict) -> float:
240 score = 0.0
241 if meta_a["year"] and meta_b["year"] and meta_a["year"] == meta_b["year"]:
242 score += 0.1
243 if meta_a["volume"] and meta_b["volume"] and meta_a["volume"] == meta_b["volume"]:
244 score += 0.2
245 if meta_a["issue"] and meta_b["issue"] and meta_a["issue"] == meta_b["issue"]:
246 score += 0.1
247 if (
248 meta_a["start_page"]
249 and meta_b["start_page"]
250 and meta_a["start_page"] == meta_b["start_page"]
251 ):
252 score += 0.3
253 if (
254 meta_a["end_page"]
255 and meta_b["end_page"]
256 and meta_a["end_page"] == meta_b["end_page"]
257 ):
258 score += 0.3
259 return score