Coverage for changes_metadata_manager / zenodo_metadata.py: 86%

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

2# 

3# SPDX-License-Identifier: ISC 

4 

5import re 

6from collections.abc import Iterable 

7from copy import deepcopy 

8from typing import TypedDict, cast 

9 

10from typing_extensions import NotRequired 

11 

12ENTITY_URI_PATTERN = re.compile(r"/itm/([^/]+)/ob\d+/\d+$") 

13STAGE_ARCHIVE_PATTERN = re.compile(r"-(raw|rawp|dcho|dchoo)\.zip$") 

14LICENSE_URI_TO_ZENODO = { 

15 "https://creativecommons.org/publicdomain/zero/1.0/": "cc0-1.0", 

16 "https://creativecommons.org/licenses/by/4.0/": "cc-by-4.0", 

17 "https://creativecommons.org/licenses/by-nc/4.0/": "cc-by-nc-4.0", 

18 "https://creativecommons.org/licenses/by-sa/4.0/": "cc-by-sa-4.0", 

19 "https://creativecommons.org/licenses/by-nc-sa/4.0/": "cc-by-nc-sa-4.0", 

20} 

21 

22 

23class ZenodoUpdatePayload(TypedDict): 

24 access: dict[str, object] 

25 files: dict[str, object] 

26 metadata: dict[str, object] 

27 custom_fields: NotRequired[dict[str, object]] 

28 

29 

30def _strip_vocabulary_titles(value: object) -> object: 

31 if isinstance(value, dict): 

32 if set(value) == {"id", "title"}: 

33 return {"id": value["id"]} 

34 return {str(key): _strip_vocabulary_titles(item) for key, item in value.items()} 

35 if isinstance(value, list): 

36 return [_strip_vocabulary_titles(item) for item in value] 

37 return value 

38 

39 

40def build_zenodo_update_payload(record: dict) -> ZenodoUpdatePayload: 

41 remote_access = record["access"] 

42 access: dict[str, object] = { 

43 "record": remote_access["record"], 

44 "files": remote_access["files"], 

45 } 

46 if "embargo" in remote_access and remote_access["embargo"]["active"]: 

47 access["embargo"] = deepcopy(remote_access["embargo"]) 

48 

49 remote_files = record["files"] 

50 files: dict[str, object] = {"enabled": remote_files["enabled"]} 

51 for field in ("default_preview", "order"): 

52 if field in remote_files and remote_files[field] is not None: 

53 files[field] = deepcopy(remote_files[field]) 

54 

55 metadata = cast("dict[str, object]", _strip_vocabulary_titles(record["metadata"])) 

56 payload = ZenodoUpdatePayload( 

57 access=access, 

58 files=files, 

59 metadata=metadata, 

60 ) 

61 if "custom_fields" in record: 

62 payload["custom_fields"] = deepcopy(record["custom_fields"]) 

63 return payload 

64 

65 

66def extract_entity_id(record: dict) -> str: 

67 for identifier in record["metadata"]["identifiers"]: 

68 match = ENTITY_URI_PATTERN.search(identifier["identifier"]) 

69 if match: 

70 return match.group(1) 

71 raise ValueError( 

72 f"No CHANGES entity URI found in identifiers: " 

73 f"{record['metadata']['identifiers']}" 

74 ) 

75 

76 

77def extract_stage_from_filenames(filenames: Iterable[str]) -> str: 

78 stages = { 

79 match.group(1) 

80 for filename in filenames 

81 if (match := STAGE_ARCHIVE_PATTERN.search(filename)) 

82 } 

83 if len(stages) != 1: 

84 raise ValueError(f"Expected one stage archive, found: {sorted(stages)}") 

85 return stages.pop() 

86 

87 

88def extract_stage(record: dict) -> str: 

89 return extract_stage_from_filenames(record["files"]["entries"]) 

90 

91 

92def extract_content_license(metadata: dict) -> str | None: 

93 if "rights" not in metadata: 

94 return None 

95 for right in metadata["rights"]: 

96 if "title" not in right or "en" not in right["title"]: 

97 continue 

98 title = right["title"]["en"] 

99 if "(Content license)" in title: 

100 if "link" not in right: 

101 continue 

102 link = right["link"] 

103 if "zero" in link: 

104 return "cc0-1.0" 

105 if "by-nc-sa" in link: 

106 return "cc-by-nc-sa-4.0" 

107 if "by-nc" in link: 

108 return "cc-by-nc-4.0" 

109 if "by-sa" in link: 

110 return "cc-by-sa-4.0" 

111 if "by" in link: 

112 return "cc-by-4.0" 

113 return None 

114 

115 

116def _value_differences(expected: object, actual: object, path: str) -> list[str]: 

117 if isinstance(expected, dict): 

118 if not isinstance(actual, dict): 

119 return [path] 

120 differences: list[str] = [] 

121 for key, value in expected.items(): 

122 child_path = f"{path}/{key}" 

123 if key not in actual: 

124 differences.append(child_path) 

125 else: 

126 differences.extend(_value_differences(value, actual[key], child_path)) 

127 return differences 

128 

129 if isinstance(expected, list): 

130 if not isinstance(actual, list) or len(expected) != len(actual): 

131 return [path] 

132 differences = [] 

133 for index, (expected_item, actual_item) in enumerate(zip(expected, actual)): 

134 differences.extend( 

135 _value_differences(expected_item, actual_item, f"{path}/{index}") 

136 ) 

137 return differences 

138 

139 return [] if expected == actual else [path] 

140 

141 

142def zenodo_payload_differences( 

143 payload: ZenodoUpdatePayload, 

144 record: dict, 

145 ignored_metadata_fields: set[str], 

146) -> list[str]: 

147 expected_metadata = { 

148 key: value 

149 for key, value in payload["metadata"].items() 

150 if key not in ignored_metadata_fields 

151 } 

152 remote_metadata = { 

153 key: value 

154 for key, value in record["metadata"].items() 

155 if key not in ignored_metadata_fields 

156 } 

157 differences = [ 

158 f"/metadata/{key}" for key in remote_metadata.keys() - expected_metadata.keys() 

159 ] 

160 differences.extend( 

161 _value_differences(expected_metadata, remote_metadata, "/metadata") 

162 ) 

163 differences.extend( 

164 _value_differences(payload["access"], record["access"], "/access") 

165 ) 

166 differences.extend(_value_differences(payload["files"], record["files"], "/files")) 

167 expected_custom_fields = ( 

168 payload["custom_fields"] if "custom_fields" in payload else {} 

169 ) 

170 remote_custom_fields = record["custom_fields"] if "custom_fields" in record else {} 

171 differences.extend( 

172 _value_differences( 

173 expected_custom_fields, remote_custom_fields, "/custom_fields" 

174 ) 

175 ) 

176 return sorted(differences)