Coverage for src/api/metaapi.py: 99%

1#!/usr/bin/python 

2# -*- coding: utf-8 -*- 

3# Copyright (c) 2022, Arcangelo Massari <arcangelo.massari@unibo.it> 

4# 

5# Permission to use, copy, modify, and/or distribute this software for any purpose 

6# with or without fee is hereby granted, provided that the above copyright notice 

7# and this permission notice appear in all copies. 

8# 

9# THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH 

10# REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND 

11# FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, INDIRECT, 

12# OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, 

13# DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS 

14# ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS 

15# SOFTWARE. 

16 

17__author__ = "Arcangelo Massari" 

18 

19from typing import Tuple 

20 

21URI_TYPE_DICT = { 

22 "http://purl.org/spar/doco/Abstract": "abstract", 

23 "http://purl.org/spar/fabio/ArchivalDocument": "archival document", 

24 "http://purl.org/spar/fabio/AudioDocument": "audio document", 

25 "http://purl.org/spar/fabio/Book": "book", 

26 "http://purl.org/spar/fabio/BookChapter": "book chapter", 

27 "http://purl.org/spar/fabio/ExpressionCollection": "book section", 

28 "http://purl.org/spar/fabio/BookSeries": "book series", 

29 "http://purl.org/spar/fabio/BookSet": "book set", 

30 "http://purl.org/spar/fabio/ComputerProgram": "computer program", 

31 "http://purl.org/spar/doco/Part": "book part", 

32 "http://purl.org/spar/fabio/Expression": "", 

33 "http://purl.org/spar/fabio/DataFile": "dataset", 

34 "http://purl.org/spar/fabio/DataManagementPlan": "data management plan", 

35 "http://purl.org/spar/fabio/Thesis": "dissertation", 

36 "http://purl.org/spar/fabio/Editorial": "editorial", 

37 "http://purl.org/spar/fabio/Journal": "journal", 

38 "http://purl.org/spar/fabio/JournalArticle": "journal article", 

39 "http://purl.org/spar/fabio/JournalEditorial": "journal editorial", 

40 "http://purl.org/spar/fabio/JournalIssue": "journal issue", 

41 "http://purl.org/spar/fabio/JournalVolume": "journal volume", 

42 "http://purl.org/spar/fabio/Newspaper": "newspaper", 

43 "http://purl.org/spar/fabio/NewspaperArticle": "newspaper article", 

44 "http://purl.org/spar/fabio/NewspaperIssue": "newspaper issue", 

45 "http://purl.org/spar/fr/ReviewVersion": "peer review", 

46 "http://purl.org/spar/fabio/AcademicProceedings": "proceedings", 

47 "http://purl.org/spar/fabio/Preprint": "preprint", 

48 "http://purl.org/spar/fabio/Presentation": "presentation", 

49 "http://purl.org/spar/fabio/ProceedingsPaper": "proceedings article", 

50 "http://purl.org/spar/fabio/ReferenceBook": "reference book", 

51 "http://purl.org/spar/fabio/ReferenceEntry": "reference entry", 

52 "http://purl.org/spar/fabio/ReportDocument": "report", 

53 "http://purl.org/spar/fabio/RetractionNotice": "retraction notice", 

54 "http://purl.org/spar/fabio/Series": "series", 

55 "http://purl.org/spar/fabio/SpecificationDocument": "standard", 

56 "http://purl.org/spar/fabio/WebContent": "web content", 

57} 

58 

59 

60def generate_id_search(ids: str) -> Tuple[str]: 

61 id_searches = list() 

62 omid_values = [] 

63 other_values = [] 

64 

65 for identifier in ids.split("__"): 

66 scheme_literal_value = identifier.split(":", maxsplit=1) 

67 scheme = scheme_literal_value[0].lower() 

68 literal_value = scheme_literal_value[1] 

69 literal_value = literal_value.lower() if scheme == "doi" else literal_value 

70 if scheme == "omid": 

71 omid_values.append( 

72 "{{ BIND(<https://w3id.org/oc/meta/" + literal_value + "> AS ?res) }}" 

73 ) 

74 elif scheme in { 

75 "doi", 

76 "issn", 

77 "isbn", 

78 "openalex", 

79 "pmid", 

80 "pmcid", 

81 "url", 

82 "wikidata", 

83 "wikipedia", 

84 }: 

85 other_values.append( 

86 ''' 

87 {{ 

88 ?identifier literal:hasLiteralValue "''' 

89 + literal_value 

90 + """"; 

91 datacite:usesIdentifierScheme datacite:""" 

92 + scheme 

93 + """; 

94 ^datacite:hasIdentifier ?res. 

95 ?res a fabio:Expression. 

96 }} 

97 """ 

98 ) 

99 

100 if omid_values: 

101 id_searches.append("?res a fabio:Expression." + " UNION ".join(omid_values)) 

102 

103 if other_values: 

104 id_searches.append(" UNION ".join(other_values)) 

105 

106 ids_search = " UNION ".join(id_searches) 

107 return (ids_search,) 

108 

109 

110def generate_ra_search(identifier: str) -> Tuple[str]: 

111 scheme_literal_value = identifier.split(":") 

112 if len(scheme_literal_value) == 2: 

113 scheme = scheme_literal_value[0] 

114 literal_value = scheme_literal_value[1] 

115 else: 

116 scheme = "orcid" 

117 literal_value = scheme_literal_value[0] 

118 if scheme == "omid": 

119 return ( 

120 "<https://w3id.org/oc/meta/{0}> ^pro:isHeldBy ?knownRole.".format( 

121 literal_value 

122 ), 

123 ) 

124 else: 

125 return ( 

126 ''' 

127 ?knownPersonIdentifier literal:hasLiteralValue "''' 

128 + literal_value 

129 + """"; 

130 datacite:usesIdentifierScheme datacite:""" 

131 + scheme 

132 + """; 

133 ^datacite:hasIdentifier ?knownPerson. 

134 ?knownPerson ^pro:isHeldBy ?knownRole. 

135 """, 

136 ) 

137 

138 

139def create_metadata_output(results): 

140 header = results[0] 

141 output_results = [header] 

142 for result in results[1:]: 

143 output_result = list() 

144 for i, data in enumerate(result): 

145 if i == header.index("id"): 

146 sorted_ids = __sort_ids(data[1]) 

147 output_result.append((sorted_ids, sorted_ids)) 

148 elif i == header.index("venue"): 

149 sorted_venue = __sort_venue_ids(data[1]) 

150 output_result.append((sorted_venue, sorted_venue)) 

151 elif i == header.index("type"): 

152 beautiful_type = __postprocess_type(data[1]) 

153 output_result.append((data[0], beautiful_type)) 

154 elif ( 

155 i == header.index("author") 

156 or i == header.index("editor") 

157 or i == header.index("publisher") 

158 ): 

159 ordered_list = process_ordered_list(data[1]) 

160 output_result.append((data[0], ordered_list)) 

161 else: 

162 output_result.append(data) 

163 output_results.append(output_result) 

164 return output_results, True 

165 

166 

167def __sort_ids(ids: str) -> str: 

168 *external_ids, omid = ids.split(" ") 

169 return " ".join(sorted(external_ids) + [omid]) 

170 

171 

172def __sort_venue_ids(venue: str) -> str: 

173 if not venue: 

174 return venue 

175 name, ids = venue.rsplit(" [", 1) 

176 return f"{name} [{__sort_ids(ids.removesuffix(']'))}]" 

177 

178 

179def __postprocess_type(type_uri: str) -> str: 

180 if type_uri: 

181 type_string = URI_TYPE_DICT[type_uri] 

182 else: 

183 type_string = "" 

184 return type_string 

185 

186 

187def process_ordered_list(items): 

188 if not items: 

189 return items 

190 items_dict = {} 

191 role_to_name = {} 

192 l_author = [item for item in items.split("|") if item is not None and item != ""] 

193 if len(l_author) == 0: 

194 return "" 

195 for item in l_author: 

196 parts = item.split(":") 

197 name = ":".join(parts[:-2]) 

198 current_role = parts[-2] 

199 next_role = parts[-1] if parts[-1] != "" else None 

200 items_dict[current_role] = next_role 

201 role_to_name[current_role] = name 

202 

203 ordered_items = [] 

204 start_role = next( 

205 iter( 

206 role 

207 for role, next_role in items_dict.items() 

208 if role not in items_dict.values() 

209 ) 

210 ) 

211 

212 current_role = start_role 

213 while current_role: 

214 ordered_items.append(role_to_name[current_role]) 

215 current_role = items_dict.get(current_role, "") 

216 

217 return "; ".join(ordered_items)