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

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("{{ BIND(<https://w3id.org/oc/meta/"+literal_value+"> AS ?res) }}") 

72 elif scheme in {'doi', 'issn', 'isbn', 'openalex', 'pmid', 'pmcid', 'url', 'wikidata', 'wikipedia'}: 

73 other_values.append(''' 

74 {{ 

75 { 

76 ?identifier literal:hasLiteralValue "'''+literal_value+'''" 

77 } 

78 UNION 

79 { 

80 ?identifier literal:hasLiteralValue "'''+literal_value+'''"^^<http://www.w3.org/2001/XMLSchema#string> 

81 } 

82 ?identifier datacite:usesIdentifierScheme datacite:'''+scheme+'''; 

83 ^datacite:hasIdentifier ?res. 

84 ?res a fabio:Expression. 

85 }} 

86 ''') 

87 

88 if omid_values: 

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

90 

91 if other_values: 

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

93 

94 ids_search = " UNION ".join(id_searches) 

95 return ids_search, 

96 

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

98 scheme_literal_value = identifier.split(':') 

99 if len(scheme_literal_value) == 2: 

100 scheme = scheme_literal_value[0] 

101 literal_value = scheme_literal_value[1] 

102 else: 

103 scheme = 'orcid' 

104 literal_value = scheme_literal_value[0] 

105 if scheme == 'omid': 

106 return '<https://w3id.org/oc/meta/{0}> ^pro:isHeldBy ?knownRole.'.format(literal_value), 

107 else: 

108 return ''' 

109 { 

110 ?knownPersonIdentifier literal:hasLiteralValue "'''+literal_value+'''" 

111 } 

112 UNION 

113 { 

114 ?knownPersonIdentifier literal:hasLiteralValue "'''+literal_value+'''"^^<http://www.w3.org/2001/XMLSchema#string> 

115 } 

116 ?knownPersonIdentifier datacite:usesIdentifierScheme datacite:'''+scheme+'''; 

117 ^datacite:hasIdentifier ?knownPerson. 

118 ?knownPerson ^pro:isHeldBy ?knownRole. 

119 ''', 

120 

121def create_metadata_output(results): 

122 header = results[0] 

123 output_results = [header] 

124 for result in results[1:]: 

125 output_result = list() 

126 for i, data in enumerate(result): 

127 if i == header.index('type'): 

128 beautiful_type = __postprocess_type(data[1]) 

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

130 elif i == header.index('author') or i == header.index('editor') or i == header.index('publisher'): 

131 ordered_list = process_ordered_list(data[1]) 

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

133 else: 

134 output_result.append(data) 

135 output_results.append(output_result) 

136 return output_results, True 

137 

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

139 if type_uri: 

140 type_string = URI_TYPE_DICT[type_uri] 

141 else: 

142 type_string = '' 

143 return type_string 

144 

145def process_ordered_list(items): 

146 if not items: 

147 return items 

148 items_dict = {} 

149 role_to_name = {} 

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

151 if len(l_author) == 0: 

152 return "" 

153 for item in l_author: 

154 parts = item.split(':') 

155 name = ':'.join(parts[:-2]) 

156 current_role = parts[-2] 

157 next_role = parts[-1] if parts[-1] != '' else None 

158 items_dict[current_role] = next_role 

159 role_to_name[current_role] = name 

160 

161 ordered_items = [] 

162 start_role = next(iter(role for role, next_role in items_dict.items() if not role in items_dict.values())) 

163 

164 current_role = start_role 

165 while current_role: 

166 ordered_items.append(role_to_name[current_role]) 

167 current_role = items_dict.get(current_role, '') 

168 

169 return "; ".join(ordered_items)