Coverage for changes_metadata_manager / export_zenodo_csv.py: 100%

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

2# 

3# SPDX-License-Identifier: ISC 

4 

5import argparse 

6import csv 

7import json 

8import os 

9import tempfile 

10import time 

11from pathlib import Path 

12 

13from rich.progress import ( 

14 BarColumn, 

15 MofNCompleteColumn, 

16 Progress, 

17 SpinnerColumn, 

18 TextColumn, 

19) 

20 

21from changes_metadata_manager.zenodo_api import fetch_latest_published_record 

22from changes_metadata_manager.zenodo_metadata import LICENSE_URI_TO_ZENODO 

23 

24CSV_FIELDNAMES = [ 

25 "Numero su DMP", 

26 "Caso di studio", 

27 "Autore/i", 

28 "Tipo", 

29 "Titolo", 

30 "Data pubblicazione", 

31 "DOI", 

32 "URL", 

33 "Repository", 

34 "Licenza", 

35 "Note", 

36] 

37REQUEST_DELAY = 0.5 

38 

39 

40def _format_creators(metadata: dict) -> str: 

41 creators: list[str] = [] 

42 for creator in metadata["creators"]: 

43 person = creator["person_or_org"] 

44 orcid = next( 

45 identifier["identifier"] 

46 for identifier in person["identifiers"] 

47 if identifier["scheme"] == "orcid" 

48 ) 

49 creators.append( 

50 f"{person['family_name']}, {person['given_name']} [orcid:{orcid}]" 

51 ) 

52 return "; ".join(creators) 

53 

54 

55def _format_licenses(metadata: dict) -> str: 

56 licenses: list[str] = [] 

57 for right in metadata["rights"]: 

58 short_name = LICENSE_URI_TO_ZENODO[right["link"]] 

59 context = right["title"]["en"].rsplit(" (", maxsplit=1)[1].removesuffix(")") 

60 licenses.append(f"{short_name} ({context})") 

61 return "; ".join(licenses) 

62 

63 

64def _build_row(record: dict) -> dict[str, str]: 

65 metadata = record["metadata"] 

66 return { 

67 "Numero su DMP": "", 

68 "Caso di studio": "Aldrovandi", 

69 "Autore/i": _format_creators(metadata), 

70 "Tipo": metadata["resource_type"]["title"]["en"], 

71 "Titolo": metadata["title"], 

72 "Data pubblicazione": metadata["publication_date"], 

73 "DOI": record["pids"]["doi"]["identifier"], 

74 "URL": record["links"]["self_html"], 

75 "Repository": metadata["publisher"], 

76 "Licenza": _format_licenses(metadata), 

77 "Note": "", 

78 } 

79 

80 

81def _write_csv(path: Path, rows: list[dict[str, str]]) -> None: 

82 descriptor, temporary_path = tempfile.mkstemp(dir=path.parent, suffix=".tmp") 

83 with os.fdopen(descriptor, "w", newline="") as file: 

84 writer = csv.DictWriter(file, fieldnames=CSV_FIELDNAMES) 

85 writer.writeheader() 

86 writer.writerows(rows) 

87 os.replace(temporary_path, path) 

88 

89 

90def export_zenodo_csv(drafts_path: Path, output_path: Path | None = None) -> Path: 

91 with open(drafts_path) as file: 

92 drafts = json.load(file) 

93 

94 records = [draft for draft in drafts if draft["draft_id"]] 

95 rows: list[dict[str, str]] = [] 

96 with Progress( 

97 SpinnerColumn(), 

98 TextColumn("[progress.description]{task.description}"), 

99 BarColumn(), 

100 MofNCompleteColumn(), 

101 ) as progress: 

102 task = progress.add_task("Fetching published records", total=len(records)) 

103 for draft in records: 

104 record_id = str(draft["draft_id"]) 

105 progress.update(task, description=f"Fetching record {record_id}") 

106 record = fetch_latest_published_record( 

107 draft["zenodo_url"].rstrip("/"), 

108 record_id, 

109 draft["access_token"], 

110 draft["user_agent"], 

111 ) 

112 rows.append(_build_row(record)) 

113 progress.advance(task) 

114 time.sleep(REQUEST_DELAY) 

115 

116 if output_path is None: 

117 destination = drafts_path.parent / "doi_table.csv" 

118 else: 

119 destination = output_path 

120 _write_csv(destination, rows) 

121 print(f"DOI table written to {destination}") 

122 return destination 

123 

124 

125def main() -> None: # pragma: no cover 

126 parser = argparse.ArgumentParser( 

127 description="Export published Zenodo records to the DMP CSV format" 

128 ) 

129 parser.add_argument("drafts_json", type=Path, help="Path to drafts.json") 

130 parser.add_argument("--output", type=Path, help="Output CSV path") 

131 args = parser.parse_args() 

132 export_zenodo_csv(args.drafts_json, args.output) 

133 

134 

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

136 main()