Coverage for oc_meta / lib / cleaner.py: 96%
239 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-07-25 10:39 +0000
1# SPDX-FileCopyrightText: 2019 Silvio Peroni <silvio.peroni@unibo.it>
2# SPDX-FileCopyrightText: 2019-2020 Fabio Mariani <fabio.mariani555@gmail.com>
3# SPDX-FileCopyrightText: 2021 Simone Persiani <iosonopersia@gmail.com>
4# SPDX-FileCopyrightText: 2021-2026 Arcangelo Massari <arcangelo.massari@unibo.it>
5#
6# SPDX-License-Identifier: ISC
8import html
9import re
10from collections import OrderedDict
11from datetime import datetime
12from typing import Tuple, Union
14from dateutil.parser import parse
15from oc_ds_converter.oc_idmanager import (
16 DOIManager,
17 ISBNManager,
18 ISSNManager,
19 ORCIDManager,
20)
22from oc_meta.lib.master_of_regex import (
23 RE_COMMA_AND_SPACES,
24 RE_INVALID_VI_PATTERNS,
25 RE_ISSUES_VALID_PATTERNS,
26 RE_VOLUMES_VALID_PATTERNS,
27 split_name_and_ids,
28)
30_HYPHEN_TRANS = str.maketrans(
31 {
32 "\u00ad": "\u002d", # Soft hyphen
33 "\u06d4": "\u002d", # Arabic Full Stop
34 "\u2010": "\u002d", # Hyphen
35 "\u2011": "\u002d", # Non-breaking Hyphen
36 "\u2012": "\u002d", # Figure Dash
37 "\u2013": "\u002d", # En-Dash
38 "\u2014": "\u002d", # Em-Dash
39 "\u2043": "\u002d", # Hyphen Bullet
40 "\u2212": "\u002d", # Minus Sign
41 "\u2796": "\u002d", # Heavy Minus Sign
42 "\u2cba": "\u002d", # Coptic Capital Letter Dialect-p Ni
43 "\ufe58": "\u002d", # Small Em Dash
44 }
45)
47_SPACE_TRANS = str.maketrans(
48 {
49 "\u0009": "\u0020", # Character Tabulation
50 "\u00a0": "\u0020", # No-break space
51 "\u200b": "\u0020", # Zero width space
52 "\u202f": "\u0020", # Narrow no-break space
53 "\u2003": "\u0020", # Em Space
54 "\u2005": "\u0020", # Four-Per-Em Space
55 "\u2009": "\u0020", # Thin Space
56 }
57)
59# Translation table for control characters and extended ASCII to space
60# Covers: 0x00-0x1F (control chars), 0x7F (DEL), 0x80-0xFF (extended ASCII)
61_ASCII_CONTROL_TRANS = str.maketrans(
62 {chr(i): " " for i in range(0x00, 0x20)}
63 | {chr(0x7F): " "}
64 | {chr(i): " " for i in range(0x80, 0x100)}
65)
67_DOI_MANAGER = DOIManager(use_api_service=False, storage_manager=None)
68_ISBN_MANAGER = ISBNManager()
69_ISSN_MANAGER = ISSNManager()
70_ORCID_MANAGER = ORCIDManager(use_api_service=False, storage_manager=None)
73def normalize_hyphens(string: str) -> str:
74 """
75 It replaces any hyphen, dash and minus sign with a hyphen-minus character.
76 This is done for pages, IDs and dates.
78 .. list-table:: Comparison between the various characters similar to hyphen-minus
79 :widths: 25 25 50
80 :header-rows: 1
82 * - UTF-8
83 - SIGN
84 - NAME
85 * - U+002D
86 - -
87 - Hyphen-minus
88 * - U+00AD
89 -
90 - Soft hyphen
91 * - U+06D4
92 - ۔
93 - Arabic Full Stop
94 * - U+2010
95 - ‐
96 - Hyphen
97 * - U+2011
98 - −
99 - Non-breaking Hyphen
100 * - U+2012
101 - –
102 - Figure Dash
103 * - U+2013
104 - –
105 - En-Dash
106 * - U+2014
107 - —
108 - Em-Dash
109 * - U+2043
110 - ⁃
111 - Hyphen Bullet
112 * - U+2212
113 - −
114 - Minus Sign
115 * - U+2796
116 - ➖
117 - Heavy Minus Sign
118 * - U+2CBA
119 - Ⲻ
120 - Coptic Capital Letter Dialect-p Ni
121 * - U+FE58
122 - ﹘
123 - Small Em Dash
125 :returns: str -- the string with normalized hyphens
126 """
127 return string.translate(_HYPHEN_TRANS)
130def normalize_spaces(string: str) -> str:
131 """
132 It replaces any ambiguous spaces with a space.
134 .. list-table:: List of the various characters similar to the space
135 :widths: 25 25 50
136 :header-rows: 1
138 * - UTF-8
139 - NAME
140 * - U+0020
141 - Space
142 * - U+0009
143 - Character Tabulation
144 * - U+00A0
145 - No-break space
146 * - U+200B
147 - Zero width space
148 * - U+202F
149 - Narrow no-break space
150 * - U+2003
151 - Em Space
152 * - U+2005
153 - Four-Per-Em Space
154 * - U+2009
155 - Thin Space
157 :returns: str -- the string with normalized spaces
158 """
159 return string.translate(_SPACE_TRANS).replace(" ", "\u0020")
162def clean_title(string: str, normalize: bool = True) -> str:
163 """
164 Concerning titles of bibliographic resources ('venue' and 'title' columns),
165 every word in the title is capitalized except for those that have capitals within them
166 (probably acronyms, e.g. 'FaBiO and CiTO'). This exception, however, does not include entirely capitalized titles.
167 Finally, null characters and spaces are removed.
169 :returns: str -- The cleaned title
170 """
171 title = string
172 if normalize:
173 if title.isupper():
174 title = title.lower()
175 words = title.split()
176 for i, w in enumerate(words):
177 if not any(x.isupper() for x in w):
178 words[i] = w.title()
179 return " ".join(words)
180 return title
183def _date_parse_hack(date: str) -> str:
184 dt = parse(date, default=datetime(2001, 1, 1))
185 dt2 = parse(date, default=datetime(2002, 2, 2))
187 if dt.year == dt2.year and dt.month == dt2.month and dt.day == dt2.day:
188 clean_date = parse(date).strftime("%Y-%m-%d")
189 elif dt.year == dt2.year and dt.month == dt2.month:
190 clean_date = parse(date).strftime("%Y-%m")
191 elif dt.year == dt2.year:
192 clean_date = parse(date).strftime("%Y")
193 else:
194 clean_date = ""
195 return clean_date
198def clean_date(string: str) -> str:
199 """
200 It tries to parse a date-string into a datetime object,
201 considering both the validity of the format (YYYYY-MM-DD) and the value (e.g. 30 February is not a valid date).
202 For example, a date 2020-02-30 will become 2020-02, because the day is invalid.
203 On the other hand, 2020-27-12 will become 2020 since the day
204 and month are invalid.
205 If the year is not valid (e.g.year >9999) data would be totally discarded.
207 :returns: str -- The cleaned date or an empty string
208 """
209 date = string
210 try:
211 date = _date_parse_hack(date)
212 except ValueError:
213 try:
214 # e.g. 2021-12-17
215 if len(date) == 10:
216 try:
217 # Maybe only the day is invalid, try year-month
218 new_date = date[:-3]
219 date = _date_parse_hack(new_date)
220 except ValueError:
221 try:
222 # Maybe only the month is invalid, try year
223 new_date = date[:-6]
224 date = _date_parse_hack(new_date)
225 except ValueError:
226 date = ""
227 # e.g. 2021-12
228 elif len(date) == 7:
229 # Maybe only the month is invalid, try year
230 try:
231 new_date = date[:-3]
232 date = _date_parse_hack(new_date)
233 except ValueError:
234 date = ""
235 else:
236 date = ""
237 except ValueError:
238 date = ""
239 return date
242def clean_name(string: str) -> str:
243 """
244 The first letter of each element of the name is capitalized and superfluous spaces are removed.
246 :returns: str -- The cleaned name
247 """
248 name = string
249 if "," in name:
250 split_name = RE_COMMA_AND_SPACES.split(name)
251 first_name = split_name[1].split()
252 for i, w in enumerate(first_name):
253 first_name[i] = clean_title(w)
254 new_first_name = " ".join(first_name)
255 surname = split_name[0].split()
256 for i, w in enumerate(surname):
257 surname[i] = clean_title(w)
258 new_surname = " ".join(surname)
259 if new_surname:
260 return new_surname + ", " + new_first_name
261 return ""
262 split_name = name.split()
263 for i, w in enumerate(split_name):
264 split_name[i] = clean_title(w)
265 return " ".join(split_name)
268def clean_agent_name(string: str) -> str:
269 """
270 Clean a responsible agent name (author, editor, publisher).
272 Removes unwanted characters while preserving letters, numbers, spaces,
273 '&', apostrophes, and dots preceded by letters. Numbers and '&' are
274 kept for organization names (e.g., "3M", "Smith & Sons").
275 Normalizes hyphens, decodes HTML entities, and removes extra spaces.
277 :returns: str -- The cleaned agent name.
278 """
279 unwanted_characters = {"[", "]", ";", "?"}
280 chars = []
281 for i, c in enumerate(string):
282 if c == ".":
283 if i > 0 and string[i - 1].isalpha():
284 chars.append(c)
285 elif c not in unwanted_characters:
286 chars.append(c)
287 clean_string = " ".join("".join(chars).split())
288 clean_string = html.unescape(clean_string)
289 clean_string = clean_string.translate(_HYPHEN_TRANS)
290 return clean_string
293def _normalize_ra_name(raw_name: str) -> str:
294 """Normalize a RA name into one of: '', 'Full Name', 'Last, First', 'Last, '.
296 Returns '' when the name is absent, literally 'Not Available', or a
297 comma-separated pair whose surname is missing. Bare names are run
298 through :func:`clean_agent_name` to drop bracket / punctuation junk.
299 """
300 name = raw_name.strip()
301 if not name:
302 return ""
303 if "," in name:
304 last, _, first = name.partition(",")
305 last = last.strip()
306 first = first.strip()
307 if last.lower() == "not available":
308 last = ""
309 if first.lower() == "not available":
310 first = ""
311 if not last:
312 return ""
313 return f"{last}, {first}" if first else f"{last}, "
314 cleaned = clean_agent_name(name)
315 if cleaned.lower() == "not available":
316 return ""
317 return cleaned
320def clean_ra_list(ra_list: list) -> list:
321 """
322 Clean a list of responsible agents: normalize names, drop 'Not Available'
323 entries, remove duplicates that share a name and at least one id, and
324 strip identifiers that appear under more than one agent.
326 :returns: list -- The cleaned responsible agents' list
327 """
329 # Phase 1: parse each entry into (key, name, ids). The key groups entries
330 # that belong to the same id bucket: named entries by their normalized
331 # name, nameless (ids-only) entries by the raw input so each stays
332 # distinct.
333 parsed: list[tuple[str, str, list[str]]] = []
334 agents_ids: OrderedDict[str, OrderedDict[str, None]] = OrderedDict()
335 for ra in ra_list:
336 raw_name, ids_str = split_name_and_ids(ra)
337 name = _normalize_ra_name(raw_name)
338 ids = ids_str.split()
339 if not name and not ids:
340 continue
341 key = name or ra
342 parsed.append((key, name, ids))
343 if ids:
344 agents_ids.setdefault(key, OrderedDict()).update(OrderedDict.fromkeys(ids))
346 # Phase 2: identifiers bucketed under more than one key are shared and
347 # must be dropped — they cannot unambiguously identify a single agent.
348 id_occurrences: dict[str, int] = {}
349 for bucket in agents_ids.values():
350 for identifier in bucket:
351 id_occurrences[identifier] = id_occurrences.get(identifier, 0) + 1
352 shared_ids = {i for i, count in id_occurrences.items() if count > 1}
354 # Phase 3: emit cleaned entries in input order, dropping later duplicates
355 # that share at least one surviving id with a previous entry of the same
356 # name.
357 output: list[str] = []
358 seen_ids_by_name: OrderedDict[str, set[str]] = OrderedDict()
359 for _, name, ids in parsed:
360 kept_ids = [i for i in ids if i not in shared_ids]
361 kept_ids_str = " ".join(kept_ids)
362 if not name:
363 output.append(f"[{kept_ids_str}]")
364 continue
365 kept_set = set(kept_ids)
366 if name in seen_ids_by_name and seen_ids_by_name[name] & kept_set:
367 continue
368 seen_ids_by_name.setdefault(name, set()).update(kept_set)
369 output.append(f"{name} [{kept_ids_str}]" if kept_ids else name)
370 return output
373def normalize_id(string: str) -> Union[str, None]:
374 """
375 This function verifies and normalizes identifiers whose schema corresponds to a DOI, an ISSN, an ISBN or an ORCID.
377 :returns: Union[str, None] -- The normalized identifier if it is valid, None otherwise
378 """
379 identifier = string.split(":", 1)
380 schema = identifier[0].lower()
381 value = identifier[1]
382 if schema == "doi":
383 valid_id = (
384 _DOI_MANAGER.normalise(value, include_prefix=True)
385 if _DOI_MANAGER.syntax_ok(value)
386 else None
387 )
388 elif schema == "isbn":
389 valid_id = (
390 _ISBN_MANAGER.normalise(value, include_prefix=True)
391 if _ISBN_MANAGER.is_valid(value, get_extra_info=False)
392 else None
393 )
394 elif schema == "issn":
395 if value == "0000-0000":
396 valid_id = None
397 else:
398 try:
399 valid_id = (
400 _ISSN_MANAGER.normalise(value, include_prefix=True)
401 if _ISSN_MANAGER.is_valid(value, get_extra_info=False)
402 else None
403 )
404 except ValueError:
405 print(value)
406 raise (ValueError)
407 elif schema == "orcid":
408 valid_id = (
409 _ORCID_MANAGER.normalise(value, include_prefix=True)
410 if _ORCID_MANAGER.is_valid(value, get_extra_info=False)
411 else None
412 )
413 else:
414 valid_id = f"{schema}:{value}"
415 return valid_id
418def clean_volume_and_issue(row: dict) -> None:
419 output = {"volume": "", "issue": "", "pub_date": ""}
420 for field in {"volume", "issue"}:
421 vi = row[field]
422 vi = normalize_hyphens(vi)
423 vi = normalize_spaces(vi).strip()
424 vi = html.unescape(vi)
425 for compiled_pattern, strategy in RE_INVALID_VI_PATTERNS.items():
426 capturing_groups = compiled_pattern.search(vi)
427 if capturing_groups:
428 if strategy == "del":
429 row[field] = ""
430 elif strategy == "do_nothing":
431 row[field] = vi
432 elif strategy == "s)":
433 row[field] = f"{vi}s)"
434 else:
435 row[field] = ""
436 whatever, volume, issue, pub_date = _fix_invalid_vi(
437 capturing_groups, strategy
438 )
439 row[field] = whatever if whatever else row[field]
440 output["volume"] = volume if volume else ""
441 output["issue"] = issue if issue else ""
442 output["pub_date"] = pub_date if pub_date else ""
443 row["volume"] = output["volume"] if not row["volume"] else row["volume"]
444 row["issue"] = output["issue"] if not row["issue"] else row["issue"]
445 row["pub_date"] = output["pub_date"] if not row["pub_date"] else row["pub_date"]
446 switch_vi = {"volume": "", "issue": ""}
447 for field in {"volume", "issue"}:
448 vi = row[field]
449 for compiled_pattern in RE_VOLUMES_VALID_PATTERNS:
450 if compiled_pattern.search(vi):
451 if field == "issue":
452 switch_vi["volume"] = vi
453 for compiled_pattern in RE_ISSUES_VALID_PATTERNS:
454 if compiled_pattern.search(vi):
455 if field == "volume":
456 switch_vi["issue"] = vi
457 if switch_vi["volume"] and switch_vi["issue"]:
458 row["volume"] = switch_vi["volume"]
459 row["issue"] = switch_vi["issue"]
460 elif switch_vi["volume"] and not row["volume"]:
461 row["volume"] = switch_vi["volume"]
462 row["issue"] = ""
463 row["type"] = (
464 "journal volume" if row["type"] == "journal issue" else row["type"]
465 )
466 elif switch_vi["issue"] and not row["issue"]:
467 row["issue"] = switch_vi["issue"]
468 row["volume"] = ""
469 row["type"] = (
470 "journal issue" if row["type"] == "journal volume" else row["type"]
471 )
474def _fix_invalid_vi(
475 capturing_groups: re.Match, strategy: str
476) -> Tuple[str | None, str | None, str | None, str | None]:
477 vol_group = 1 if "vol_iss" in strategy else 2
478 iss_group = 1 if "iss_vol" in strategy else 2
479 whatever = None
480 volume = None
481 issue = None
482 pub_date = None
483 if "vol" in strategy and "iss" in strategy:
484 volume = capturing_groups.group(vol_group)
485 issue = capturing_groups.group(iss_group)
486 if "year" in strategy:
487 pub_date = capturing_groups.group(3)
488 elif strategy == "all":
489 whatever = capturing_groups.group(1)
490 elif strategy == "sep":
491 first = capturing_groups.group(1)
492 second = capturing_groups.group(2)
493 whatever = f"{first}-{second}"
494 return whatever, volume, issue, pub_date
497def remove_ascii(string: str) -> str:
498 clean_string = string.translate(_ASCII_CONTROL_TRANS)
499 return " ".join(clean_string.split())