Coverage for oc_meta / run / orcid_process.py: 87%

1# SPDX-FileCopyrightText: 2019-2020 Fabio Mariani <fabio.mariani555@gmail.com> 

2# SPDX-FileCopyrightText: 2021-2026 Arcangelo Massari <arcangelo.massari@unibo.it> 

3# 

4# SPDX-License-Identifier: ISC 

5 

6import os 

7import re 

8from argparse import ArgumentParser 

9 

10from bs4 import BeautifulSoup 

11from oc_ds_converter.oc_idmanager import DOIManager 

12from rich.console import Console 

13 

14from oc_meta.lib.console import create_progress 

15from oc_meta.lib.csvmanager import CSVManager 

16from oc_meta.lib.master_of_regex import orcid_pattern 

17 

18 

19class IndexOrcidDoi: 

20 def __init__(self, output_path: str, threshold: int = 10000): 

21 self.file_counter = 0 

22 self.threshold = threshold 

23 self.console = Console() 

24 self.console.print("[cyan][INFO][/cyan] Loading existing CSV files") 

25 self.orcid_re = re.compile(orcid_pattern) 

26 self.doimanager = DOIManager(use_api_service=False) 

27 self.csvstorage = CSVManager(output_path=output_path) 

28 self.cache = self._build_cache() 

29 

30 def _build_cache(self) -> set[str]: 

31 cache = set() 

32 for values in self.csvstorage.data.values(): 

33 for value in values: 

34 orcid = self._extract_orcid(value) 

35 if orcid: 

36 cache.add(orcid) 

37 return cache 

38 

39 def _extract_orcid(self, text: str) -> str | None: 

40 match = self.orcid_re.search(text) 

41 return match.group(0) if match else None 

42 

43 def explorer(self, summaries_path: str) -> None: 

44 self.console.print("[cyan][INFO][/cyan] Counting files to process") 

45 files_to_process = [ 

46 os.path.join(fold, filename) 

47 for fold, _, files in os.walk(summaries_path) 

48 for filename in files 

49 if filename.endswith(".xml") 

50 and self._extract_orcid(filename) not in self.cache 

51 ] 

52 processed_files = len(self.cache) 

53 del self.cache 

54 progress = create_progress() 

55 with progress: 

56 task = progress.add_task("Processing files", total=len(files_to_process)) 

57 for file in files_to_process: 

58 self._process_file(file) 

59 self.file_counter += 1 

60 if self.file_counter % self.threshold == 0: 

61 start = processed_files + self.file_counter - self.threshold + 1 

62 end = processed_files + self.file_counter 

63 self.csvstorage.dump_data(f"{start}-{end}.csv") 

64 progress.advance(task) 

65 if self.csvstorage.data_to_store: 

66 start = ( 

67 processed_files 

68 + self.file_counter 

69 - (self.file_counter % self.threshold) 

70 + 1 

71 ) 

72 end = processed_files + self.file_counter 

73 self.csvstorage.dump_data(f"{start}-{end}.csv") 

74 

75 def _process_file(self, file_path: str) -> None: 

76 orcid = self._extract_orcid(file_path) 

77 if not orcid: 

78 return 

79 with open(file_path, "r", encoding="utf-8") as xml_file: 

80 xml_soup = BeautifulSoup(xml_file, "xml") 

81 name = self._extract_name(xml_soup) 

82 author = f"{name} [{orcid}]" if name else f"[{orcid}]" 

83 valid_doi = False 

84 for el in xml_soup.find_all("common:external-id"): 

85 id_type = el.find("common:external-id-type") 

86 rel = el.find("common:external-id-relationship") 

87 if not (id_type and rel): 

88 continue 

89 if id_type.get_text().lower() != "doi" or rel.get_text().lower() != "self": 

90 continue 

91 doi_el = el.find("common:external-id-value") 

92 if not doi_el: 

93 continue 

94 doi = self.doimanager.normalise(doi_el.get_text()) 

95 if not doi: 

96 continue 

97 valid_doi = True 

98 self.csvstorage.add_value(doi, author) 

99 if not valid_doi: 

100 self.csvstorage.add_value("None", f"[{orcid}]") 

101 

102 def _extract_name(self, xml_soup: BeautifulSoup) -> str | None: 

103 family_name_el = xml_soup.find("personal-details:family-name") 

104 given_name_el = xml_soup.find("personal-details:given-names") 

105 if family_name_el and given_name_el: 

106 return f"{family_name_el.get_text()}, {given_name_el.get_text()}" 

107 if family_name_el: 

108 return family_name_el.get_text() 

109 if given_name_el: 

110 return given_name_el.get_text() 

111 return None 

112 

113 

114if __name__ == "__main__": # pragma: no cover 

115 arg_parser = ArgumentParser( 

116 "orcid_process.py", 

117 description="Build a CSV index of DOIs associated with ORCIDs from XML summary files", 

118 ) 

119 arg_parser.add_argument( 

120 "-out", 

121 "--output", 

122 dest="output_path", 

123 required=True, 

124 help="Output directory for CSV files", 

125 ) 

126 arg_parser.add_argument( 

127 "-s", 

128 "--summaries", 

129 dest="summaries_path", 

130 required=True, 

131 help="Directory containing ORCID XML summaries (scanned recursively)", 

132 ) 

133 arg_parser.add_argument( 

134 "-t", 

135 "--threshold", 

136 dest="threshold", 

137 type=int, 

138 default=10000, 

139 help="Number of files to process before saving a CSV chunk (default: 10000)", 

140 ) 

141 args = arg_parser.parse_args() 

142 iod = IndexOrcidDoi(output_path=args.output_path, threshold=args.threshold) 

143 iod.explorer(summaries_path=args.summaries_path)