Coverage for changes_metadata_manager / zenodo_upload.py: 75%

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

2# 

3# SPDX-License-Identifier: ISC 

4 

5import argparse 

6import json 

7import os 

8import re 

9import signal 

10import tempfile 

11import time 

12import unicodedata 

13import zipfile 

14from collections import defaultdict 

15from collections.abc import Mapping 

16from concurrent.futures import ProcessPoolExecutor, as_completed 

17from contextlib import contextmanager 

18from datetime import date 

19from pathlib import Path 

20 

21import requests 

22import yaml 

23from rdflib import Graph, URIRef 

24from rich.progress import ( 

25 Progress, 

26 SpinnerColumn, 

27 TextColumn, 

28 BarColumn, 

29 MofNCompleteColumn, 

30 TimeElapsedColumn, 

31 TimeRemainingColumn, 

32) 

33 

34from piccione.upload.on_zenodo import ( 

35 get_headers, 

36 main as piccione_upload, 

37 publish_draft as piccione_publish_draft, 

38) 

39 

40from changes_metadata_manager.folder_metadata_builder import ( 

41 BASE_URI, 

42 FOLDER_TO_ID, 

43 KG_PATH, 

44 SKIP_FOLDERS, 

45 STAGE_STEPS, 

46 extract_id_from_folder_name, 

47 load_kg, 

48 scan_folder_structure, 

49) 

50from changes_metadata_manager.zenodo_metadata import LICENSE_URI_TO_ZENODO 

51 

52 

53class LiteralBlockDumper(yaml.SafeDumper): 

54 pass 

55 

56 

57def _literal_str_representer(dumper: yaml.SafeDumper, data): 

58 if "\n" in data: 

59 return dumper.represent_scalar("tag:yaml.org,2002:str", data, style="|") 

60 return dumper.represent_scalar("tag:yaml.org,2002:str", data) 

61 

62 

63LiteralBlockDumper.add_representer(str, _literal_str_representer) 

64 

65CREATORS_LOOKUP_PATH = Path(__file__).parent.parent / "data" / "creators_lookup.yaml" 

66 

67 

68def slugify(text: str) -> str: 

69 text = unicodedata.normalize("NFKD", text) 

70 text = text.encode("ascii", "ignore").decode("ascii") 

71 text = re.sub(r"[^\w\s-]", "", text.lower()) 

72 return re.sub(r"[-\s]+", "-", text).strip("-") 

73 

74 

75STEP_TO_STAGE = { 

76 "00": "raw", 

77 "01": "rawp", 

78 "02": "dcho", 

79 "03": "dchoo", 

80 "04": "dchoo", 

81 "05": "dchoo", 

82 "06": "dchoo", 

83} 

84 

85STAGE_DEFINING_STEP: dict[str, str] = { 

86 "raw": "00", 

87 "rawp": "01", 

88 "dcho": "02", 

89 "dchoo": "03", 

90} 

91 

92CRM = "http://www.cidoc-crm.org/cidoc-crm/" 

93CRMDIG = "http://www.cidoc-crm.org/extensions/crmdig/" 

94AAT = "http://vocab.getty.edu/aat/" 

95 

96P70I = URIRef(f"{CRM}P70i_is_documented_in") 

97P3_HAS_NOTE = URIRef(f"{CRM}P3_has_note") 

98P14_CARRIED_OUT_BY = URIRef(f"{CRM}P14_carried_out_by") 

99P1_IS_IDENTIFIED_BY = URIRef(f"{CRM}P1_is_identified_by") 

100P190_HAS_SYMBOLIC_CONTENT = URIRef(f"{CRM}P190_has_symbolic_content") 

101P74_HAS_RESIDENCE = URIRef(f"{CRM}P74_has_current_or_former_residence") 

102P32_USED_GENERAL_TECHNIQUE = URIRef(f"{CRM}P32_used_general_technique") 

103P16_USED_SPECIFIC_OBJECT = URIRef(f"{CRM}P16_used_specific_object") 

104E21_PERSON = URIRef(f"{CRM}E21_Person") 

105RDF_TYPE = URIRef("http://www.w3.org/1999/02/22-rdf-syntax-ns#type") 

106L23_USED_SOFTWARE = URIRef(f"{CRMDIG}L23_used_software_or_firmware") 

107 

108AAT_TECHNIQUE_LABELS: dict[str, str] = { 

109 f"{AAT}300266792": "digital photography", 

110 f"{AAT}300429747": "optical scanning", 

111} 

112 

113SLUG_LABELS: dict[str, str] = { 

114 "3df_zephyr": "3DF Zephyr", 

115 "adobe_photoshop_2023": "Adobe Photoshop 2023", 

116 "agisoft_metashape": "Agisoft Metashape", 

117 "artec_eva": "Artec Eva", 

118 "artec_leo": "Artec Leo", 

119 "artec_spider": "Artec Spider", 

120 "artec_studio_14": "Artec Studio 14", 

121 "artec_studio_15": "Artec Studio 15", 

122 "artec_studio_16": "Artec Studio 16", 

123 "artec_studio_19": "Artec Studio 19", 

124 "aton": "ATON", 

125 "blender": "Blender", 

126 "canon_eos_6d": "Canon EOS 6D", 

127 "chad-ap": "CHAD-AP", 

128 "cloudcompare": "CloudCompare", 

129 "gestaltor": "Gestaltor", 

130 "gimp": "GIMP", 

131 "heritrace": "HERITRACE", 

132 "instalod": "InstaLOD", 

133 "instant_meshes": "Instant Meshes", 

134 "lente_24-70_f2_8_l": "Canon EF 24-70mm f/2.8L", 

135 "meshlab": "MeshLab", 

136 "metashape": "Agisoft Metashape", 

137 "microscopio_digitale_bresciani": "Microscopio Digitale Bresciani", 

138 "modo": "Modo", 

139 "morph-kgc": "Morph-KGC", 

140 "nextcloud": "Nextcloud", 

141 "nikkor_35mm": "Nikkor 35mm", 

142 "nikkor_50mm": "Nikkor 50mm", 

143 "nikon_d3300": "Nikon D3300", 

144 "nikon_d5200": "Nikon D5200", 

145 "nikon_d5600": "Nikon D5600", 

146 "nikon_d7200": "Nikon D7200", 

147 "nikon_d750": "Nikon D750", 

148 "panasonic_dmc-lx100": "Panasonic DMC-LX100", 

149 "sony_alpha_6100": "Sony Alpha 6100", 

150 "sony_alpha_7_i": "Sony Alpha 7 I", 

151 "substance_3d_painter": "Substance 3D Painter", 

152 "substance_painter": "Substance Painter", 

153 "zbrush": "ZBrush", 

154} 

155 

156 

157def _format_slug(slug: str) -> str: 

158 return SLUG_LABELS[slug] 

159 

160 

161def load_creators_lookup(path: Path) -> dict[str, dict]: 

162 with open(path) as f: 

163 data = yaml.safe_load(f) 

164 return { 

165 creator["name_in_rdf"]: { 

166 "family_name": creator["family_name"], 

167 "given_name": creator["given_name"], 

168 "affiliation": creator["affiliation"], 

169 "orcid": creator["orcid"], 

170 } 

171 for creator in data["creators"] 

172 } 

173 

174 

175def _format_creator(creator_data: dict, role: str) -> dict: 

176 return { 

177 "person_or_org": { 

178 "type": "personal", 

179 "family_name": creator_data["family_name"], 

180 "given_name": creator_data["given_name"], 

181 "identifiers": [{"scheme": "orcid", "identifier": creator_data["orcid"]}], 

182 }, 

183 "role": {"id": role}, 

184 "affiliations": [{"name": creator_data["affiliation"]}], 

185 } 

186 

187 

188METADATA_STEP = "05" 

189 

190 

191def _extract_actor_names(graph: Graph, act_uri: URIRef) -> set[str]: 

192 names = set() 

193 for _, _, actor_uri in graph.triples((act_uri, P14_CARRIED_OUT_BY, None)): 

194 assert (actor_uri, RDF_TYPE, E21_PERSON) in graph 

195 for _, _, apl_uri in graph.triples((actor_uri, P1_IS_IDENTIFIED_BY, None)): 

196 for _, _, name in graph.triples((apl_uri, P190_HAS_SYMBOLIC_CONTENT, None)): 

197 names.add(str(name)) 

198 return names 

199 

200 

201def extract_authors_for_entity_stage( 

202 graph: Graph, entity_ids: list[str], stage: str 

203) -> set[str]: 

204 steps = [s for s in STAGE_STEPS[stage] if s != METADATA_STEP] 

205 authors: set[str] = set() 

206 for entity_id in entity_ids: 

207 for step in steps: 

208 authors |= _extract_actor_names( 

209 graph, URIRef(f"{BASE_URI}/act/{entity_id}/{step}/1") 

210 ) 

211 return authors 

212 

213 

214def extract_metadata_authors(graph: Graph, entity_ids: list[str]) -> set[str]: 

215 authors: set[str] = set() 

216 for entity_id in entity_ids: 

217 authors |= _extract_actor_names( 

218 graph, URIRef(f"{BASE_URI}/act/{entity_id}/05/1") 

219 ) 

220 return authors 

221 

222 

223def _build_creators( 

224 author_names: set[str], role: str, creators_lookup: dict[str, dict] 

225) -> list[dict]: 

226 missing_names = sorted(author_names.difference(creators_lookup)) 

227 if missing_names: 

228 raise ValueError(f"Creators missing from lookup: {', '.join(missing_names)}") 

229 return [ 

230 _format_creator(creators_lookup[name], role) for name in sorted(author_names) 

231 ] 

232 

233 

234def build_creators_for_entity_stage( 

235 graph: Graph, entity_ids: list[str], stage: str, creators_lookup: dict[str, dict] 

236) -> list[dict]: 

237 author_names = extract_authors_for_entity_stage(graph, entity_ids, stage) 

238 return _build_creators(author_names, "researcher", creators_lookup) 

239 

240 

241def build_metadata_creators( 

242 graph: Graph, entity_ids: list[str], creators_lookup: dict[str, dict] 

243) -> list[dict]: 

244 author_names = extract_metadata_authors(graph, entity_ids) 

245 return _build_creators(author_names, "datacurator", creators_lookup) 

246 

247 

248def merge_creators( 

249 digitization_creators: list[dict], metadata_creators: list[dict] 

250) -> list[dict]: 

251 seen_orcids: set[str] = set() 

252 merged: list[dict] = [] 

253 for creator in digitization_creators: 

254 orcid = creator["person_or_org"]["identifiers"][0]["identifier"] 

255 seen_orcids.add(orcid) 

256 merged.append(creator) 

257 for creator in metadata_creators: 

258 orcid = creator["person_or_org"]["identifiers"][0]["identifier"] 

259 if orcid not in seen_orcids: 

260 seen_orcids.add(orcid) 

261 merged.append(creator) 

262 return merged 

263 

264 

265def extract_licensed_entity_stages(graph: Graph) -> set[tuple[str, str]]: 

266 pattern = re.compile(rf"^{re.escape(BASE_URI)}/lic/([^/]+)/(\d{{2}})/1$") 

267 licensed = set() 

268 for s, p, o in graph.triples((None, P70I, None)): 

269 match = pattern.match(str(s)) 

270 if match: 

271 entity_id, step = match.groups() 

272 stage = STEP_TO_STAGE.get(step) 

273 if stage: 

274 licensed.add((entity_id, stage)) 

275 return licensed 

276 

277 

278def group_folders_by_entity(structure: dict) -> dict[str, list[tuple[str, str, dict]]]: 

279 groups = defaultdict(list) 

280 for sala_name, sala_items in structure["structure"].items(): 

281 for folder_name, subfolders in sala_items.items(): 

282 if folder_name in SKIP_FOLDERS: 

283 continue 

284 entity_id = extract_id_from_folder_name(folder_name) 

285 if entity_id in FOLDER_TO_ID.values(): 

286 group_id = entity_id 

287 else: 

288 group_id = entity_id.rstrip("abcdefghijklmnopqrstuvwxyz") 

289 groups[group_id].append((sala_name, folder_name, subfolders)) 

290 return dict(groups) 

291 

292 

293STAGES = ("raw", "rawp", "dcho", "dchoo") 

294 

295 

296def _extract_license_from_meta(stage_dir: Path, stage: str) -> str | None: 

297 g = Graph() 

298 g.parse(stage_dir / "meta.ttl", format="turtle") 

299 expected_step = STAGE_DEFINING_STEP[stage] 

300 for s, _, o in g.triples((None, P70I, None)): 

301 s_str = str(s) 

302 if "/lic/" in s_str: 

303 step_match = re.search(r"/(\d{2})/\d+$", s_str) 

304 if step_match and step_match.group(1) == expected_step: 

305 return LICENSE_URI_TO_ZENODO.get(str(o)) 

306 return None 

307 

308 

309def create_stage_zip( 

310 entity_id: str, 

311 stage: str, 

312 folders: list[tuple[str, str, dict]], 

313 root: Path, 

314 output_dir: Path, 

315 title: str, 

316) -> tuple[Path, str | None] | None: 

317 stage_dirs: list[tuple[str, str, Path]] = [] 

318 license_id: str | None = None 

319 for sala_name, folder_name, stages_dict in folders: 

320 stage_name_in_folder = None 

321 for name in stages_dict: 

322 if name.lower() == stage: 

323 stage_name_in_folder = name 

324 break 

325 if stage_name_in_folder is None: 

326 continue 

327 stage_dir = root / sala_name / folder_name / stage_name_in_folder 

328 stage_dirs.append((folder_name, stage_name_in_folder, stage_dir)) 

329 folder_license = _extract_license_from_meta(stage_dir, stage) 

330 if folder_license: 

331 license_id = folder_license 

332 if not stage_dirs: 

333 return None 

334 sala_slug = slugify(folders[0][0]) 

335 title_slug = slugify(title) 

336 zip_path = output_dir / f"{sala_slug}-{title_slug}-{entity_id}-{stage}.zip" 

337 with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf: 

338 for folder_name, stage_name_in_folder, stage_dir in stage_dirs: 

339 for file_path in stage_dir.rglob("*"): 

340 if not file_path.is_file(): 

341 continue 

342 if license_id or file_path.name in ("meta.ttl", "prov.trig"): 

343 rel = file_path.relative_to(stage_dir) 

344 arc_name = f"{folder_name}/{stage_name_in_folder}/{rel}" 

345 zf.write(file_path, arc_name) 

346 return zip_path, license_id 

347 

348 

349def _get_label(graph: Graph, uri: URIRef) -> str | None: 

350 for _, _, apl_uri in graph.triples((uri, P1_IS_IDENTIFIED_BY, None)): 

351 for _, _, name in graph.triples((apl_uri, P190_HAS_SYMBOLIC_CONTENT, None)): 

352 return str(name) 

353 return None 

354 

355 

356def extract_keeper_info( 

357 graph: Graph, entity_ids: list[str] 

358) -> tuple[str | None, str | None]: 

359 for entity_id in entity_ids: 

360 custody_uri = URIRef(f"{BASE_URI}/act/{entity_id}/ob08/1") 

361 for _, _, keeper_uri in graph.triples((custody_uri, P14_CARRIED_OUT_BY, None)): 

362 assert isinstance(keeper_uri, URIRef) 

363 keeper_name = _get_label(graph, keeper_uri) 

364 location_name = None 

365 for _, _, place_uri in graph.triples((keeper_uri, P74_HAS_RESIDENCE, None)): 

366 assert isinstance(place_uri, URIRef) 

367 location_name = _get_label(graph, place_uri) 

368 return keeper_name, location_name 

369 return None, None 

370 

371 

372def extract_entity_title(graph: Graph, entity_ids: list[str]) -> str: 

373 for entity_id in entity_ids: 

374 item_uri = URIRef(f"{BASE_URI}/itm/{entity_id}/ob00/1") 

375 for _, _, note_value in graph.triples((item_uri, P3_HAS_NOTE, None)): 

376 note = str(note_value) 

377 return re.split(r"\n|\\n", note)[0].strip() 

378 return f"Entity {entity_ids[0]}" 

379 

380 

381def extract_acquisition_technique(graph: Graph, entity_ids: list[str]) -> str | None: 

382 for entity_id in entity_ids: 

383 act_uri = URIRef(f"{BASE_URI}/act/{entity_id}/00/1") 

384 for _, _, technique_uri in graph.triples( 

385 (act_uri, P32_USED_GENERAL_TECHNIQUE, None) 

386 ): 

387 return AAT_TECHNIQUE_LABELS[str(technique_uri)] 

388 return None 

389 

390 

391def extract_devices(graph: Graph, entity_ids: list[str]) -> list[str]: 

392 devices: set[str] = set() 

393 for entity_id in entity_ids: 

394 act_uri = URIRef(f"{BASE_URI}/act/{entity_id}/00/1") 

395 for _, _, obj_uri in graph.triples((act_uri, P16_USED_SPECIFIC_OBJECT, None)): 

396 uri_str = str(obj_uri) 

397 if "/dev/" in uri_str: 

398 slug = uri_str.split("/dev/")[1].split("/")[0] 

399 devices.add(_format_slug(slug)) 

400 return sorted(devices) 

401 

402 

403def extract_software_for_stage( 

404 graph: Graph, entity_ids: list[str], stage: str 

405) -> list[str]: 

406 steps = [s for s in STAGE_STEPS[stage] if s != METADATA_STEP] 

407 software: set[str] = set() 

408 for entity_id in entity_ids: 

409 for step in steps: 

410 act_uri = URIRef(f"{BASE_URI}/act/{entity_id}/{step}/1") 

411 for _, _, sfw_uri in graph.triples((act_uri, L23_USED_SOFTWARE, None)): 

412 uri_str = str(sfw_uri) 

413 slug = uri_str.split("/sfw/")[1].split("/")[0] 

414 software.add(_format_slug(slug)) 

415 return sorted(software) 

416 

417 

418STAGE_TITLE_NAMES = { 

419 "raw": "Raw", 

420 "rawp": "Processed raw model", 

421 "dcho": "Digital Cultural Heritage Object", 

422 "dchoo": "Optimized Digital Cultural Heritage Object", 

423} 

424 

425STAGE_DESCRIPTION_NAMES = { 

426 "raw": "Raw acquisition data", 

427 "rawp": "Processed raw model", 

428 "dcho": "Digital Cultural Heritage Object", 

429 "dchoo": "Optimized Digital Cultural Heritage Object", 

430} 

431 

432STAGE_DESCRIPTIONS = { 

433 "raw": "This dataset contains the raw material generated during the acquisition phase.", 

434 "rawp": "This dataset contains the preliminary output from the photogrammetry or scanner software after initial data processing but without any interpolation or geometry corrections.", 

435 "dcho": "This dataset contains the version that includes interpolation, gap filling, and resolution of geometric issues, resulting in a refined and improved model.", 

436 "dchoo": "This dataset contains the version optimised for real-time online interaction.", 

437} 

438 

439PROPAGATED_FIELDS = ( 

440 "zenodo_url", 

441 "access_token", 

442 "user_agent", 

443 "subjects", 

444 "publication_date", 

445 "version", 

446 "community", 

447 "contributors", 

448 "funding", 

449 "references", 

450 "dates", 

451) 

452 

453 

454def extract_license_for_entity_stage( 

455 graph: Graph, entity_id: str, stage: str 

456) -> str | None: 

457 step = STAGE_DEFINING_STEP[stage] 

458 lic_uri = URIRef(f"{BASE_URI}/lic/{entity_id}/{step}/1") 

459 for _, _, license_url in graph.triples((lic_uri, P70I, None)): 

460 return LICENSE_URI_TO_ZENODO.get(str(license_url)) 

461 return None 

462 

463 

464CC0_DISCLAIMER = ( 

465 "No copyright or related rights are claimed in these digital reproductions. " 

466 "The files are released under CC0 1.0 Universal (Public Domain Dedication).\n" 

467 "\n" 

468 "Please note that the original works may qualify as cultural heritage assets " 

469 "under Italian law (D. Lgs. 42/2004). Consequently, although the digital " 

470 "reproductions are released under CC0, certain uses — and in particular " 

471 "commercial uses — may be subject to specific authorisations, restrictions, " 

472 "or fees pursuant to the applicable provisions governing the reproduction " 

473 "and publication of cultural heritage assets. Users are therefore responsible " 

474 "for ensuring compliance with Italian cultural heritage regulations before " 

475 "undertaking any commercial exploitation of the images." 

476) 

477 

478 

479CHAD_AP_URL = "https://w3id.org/dharc/ontology/chad-ap" 

480 

481RESTRICTED_NOTICE = ( 

482 "The digital object files are not included in this dataset " 

483 "because the holding institution did not grant permission for their publication. " 

484 "Only metadata and provenance files are provided." 

485) 

486 

487EXTERNAL_SOURCE_NOTICE = ( 

488 "The digital object files are not included in this dataset because they have " 

489 "been either gathered from an existing platform or provided directly by " 

490 "colleagues, without any formal permission to republish them in their original " 

491 "form. Only metadata and provenance files are provided." 

492) 

493 

494 

495def has_defining_activity(graph: Graph, entity_ids: list[str], stage: str) -> bool: 

496 step = STAGE_DEFINING_STEP[stage] 

497 return any( 

498 any(graph.triples((URIRef(f"{BASE_URI}/act/{entity_id}/{step}/1"), None, None))) 

499 for entity_id in entity_ids 

500 ) 

501 

502 

503def select_missing_files_notice( 

504 graph: Graph, 

505 entity_ids: list[str], 

506 stage: str, 

507 content_license: str | None, 

508) -> str | None: 

509 if content_license is not None: 

510 return None 

511 if has_defining_activity(graph, entity_ids, stage): 

512 return RESTRICTED_NOTICE 

513 return EXTERNAL_SOURCE_NOTICE 

514 

515 

516def build_enhanced_description( 

517 stage: str, 

518 title: str, 

519 keeper_name: str | None = None, 

520 keeper_location: str | None = None, 

521) -> str: 

522 parts = [ 

523 f'{STAGE_DESCRIPTION_NAMES[stage]} of "{title}" from the Aldrovandi Digital Twin.', 

524 ] 

525 if keeper_name: 

526 keeper_line = f"The original object is held by {keeper_name}" 

527 if keeper_location: 

528 keeper_line += f" ({keeper_location})" 

529 keeper_line += "." 

530 parts.append(keeper_line) 

531 parts.append(STAGE_DESCRIPTIONS[stage]) 

532 parts.append( 

533 f'Includes metadata (meta.ttl) and provenance (prov.trig) files following the <a href="{CHAD_AP_URL}">CHAD-AP</a> ontology.', 

534 ) 

535 return " ".join(parts) + "\n" 

536 

537 

538WORKFLOW_DOI_URL = "https://doi.org/10.46298/transformations.14773" 

539 

540 

541def build_methods_description( 

542 graph: Graph, 

543 entity_ids: list[str], 

544 stage: str, 

545) -> str: 

546 parts = [ 

547 f"Acquisition and digitization followed the reproducible workflow documented in " 

548 f'<a href="{WORKFLOW_DOI_URL}">doi:10.46298/transformations.14773</a>.', 

549 ] 

550 technique = extract_acquisition_technique(graph, entity_ids) 

551 devices = extract_devices(graph, entity_ids) 

552 if technique: 

553 line = f"Data was acquired using {technique}" 

554 if devices: 

555 line += f" ({', '.join(devices)})" 

556 line += "." 

557 parts.append(line) 

558 software = extract_software_for_stage(graph, entity_ids, stage) 

559 if software: 

560 parts.append(f"Processing software: {', '.join(software)}.") 

561 parts.append( 

562 f'Metadata follows the <a href="{CHAD_AP_URL}">Cultural Heritage Acquisition and ' 

563 f"Digitisation Application Profile (CHAD-AP)</a> based on CIDOC-CRM.", 

564 ) 

565 return "\n\n".join(parts) + "\n" 

566 

567 

568def build_entity_uri(entity_ids: list[str]) -> str: 

569 return f"{BASE_URI}/itm/{entity_ids[0]}/ob00/1" 

570 

571 

572LICENSE_INFO = { 

573 "cc0-1.0": { 

574 "title": "Creative Commons Zero v1.0 Universal", 

575 "link": "https://creativecommons.org/publicdomain/zero/1.0/", 

576 }, 

577 "cc-by-4.0": { 

578 "title": "Creative Commons Attribution 4.0 International", 

579 "link": "https://creativecommons.org/licenses/by/4.0/", 

580 }, 

581 "cc-by-nc-4.0": { 

582 "title": "Creative Commons Attribution Non Commercial 4.0 International", 

583 "link": "https://creativecommons.org/licenses/by-nc/4.0/", 

584 }, 

585 "cc-by-sa-4.0": { 

586 "title": "Creative Commons Attribution Share Alike 4.0 International", 

587 "link": "https://creativecommons.org/licenses/by-sa/4.0/", 

588 }, 

589 "cc-by-nc-sa-4.0": { 

590 "title": "Creative Commons Attribution Non Commercial Share Alike 4.0 International", 

591 "link": "https://creativecommons.org/licenses/by-nc-sa/4.0/", 

592 }, 

593} 

594 

595 

596def build_rights(content_license: str | None) -> list[dict]: 

597 metadata_info = LICENSE_INFO["cc0-1.0"] 

598 rights = [ 

599 { 

600 "title": {"en": f"{metadata_info['title']} (Metadata license)"}, 

601 "description": {"en": "Applies to metadata files: meta.ttl, prov.trig"}, 

602 "link": metadata_info["link"], 

603 } 

604 ] 

605 if content_license and content_license in LICENSE_INFO: 

606 content_info = LICENSE_INFO[content_license] 

607 rights.append( 

608 { 

609 "title": {"en": f"{content_info['title']} (Content license)"}, 

610 "description": { 

611 "en": "Applies to all data files except meta.ttl and prov.trig" 

612 }, 

613 "link": content_info["link"], 

614 } 

615 ) 

616 return rights 

617 

618 

619def generate_zenodo_config( 

620 stage: str, 

621 zip_path: Path, 

622 title: str, 

623 base_config: dict, 

624 creators: list[dict], 

625 methods_description: str, 

626 license: str | None = None, 

627 entity_uri: str | None = None, 

628 keeper_name: str | None = None, 

629 keeper_location: str | None = None, 

630 missing_files_notice: str | None = None, 

631) -> dict: 

632 description = build_enhanced_description(stage, title, keeper_name, keeper_location) 

633 

634 config: dict = { 

635 "title": f"{title} - {STAGE_TITLE_NAMES[stage]} - Aldrovandi Digital Twin", 

636 "description": description, 

637 "resource_type": {"id": "dataset"}, 

638 "publisher": "Zenodo", 

639 "access": {"record": "public", "files": "public"}, 

640 "files": [str(zip_path.absolute())], 

641 "creators": creators, 

642 "publication_date": date.today().isoformat(), 

643 "rights": build_rights(license), 

644 } 

645 

646 additional_descriptions: list[dict] = [ 

647 { 

648 "description": methods_description, 

649 "type": {"id": "methods"}, 

650 }, 

651 { 

652 "description": base_config["notes"], 

653 "type": {"id": "notes"}, 

654 }, 

655 ] 

656 if missing_files_notice is not None: 

657 additional_descriptions.append( 

658 { 

659 "description": missing_files_notice, 

660 "type": {"id": "notes"}, 

661 } 

662 ) 

663 if license == "cc0-1.0": 

664 additional_descriptions.append( 

665 { 

666 "description": CC0_DISCLAIMER, 

667 "type": {"id": "notes"}, 

668 } 

669 ) 

670 config["additional_descriptions"] = additional_descriptions 

671 

672 config["locations"] = { 

673 "features": [ 

674 { 

675 "geometry": { 

676 "type": "Point", 

677 "coordinates": [loc["lon"], loc["lat"]], 

678 }, 

679 "place": loc["place"], 

680 "description": loc["description"], 

681 } 

682 for loc in base_config["locations"] 

683 ] 

684 } 

685 

686 for field in PROPAGATED_FIELDS: 

687 if field in base_config and field not in config: 

688 config[field] = base_config[field] 

689 

690 if "related_identifiers" in base_config: 

691 converted = [] 

692 for ri in base_config["related_identifiers"]: 

693 entry: dict = { 

694 "identifier": ri["identifier"], 

695 "relation_type": {"id": ri["relation"]}, 

696 } 

697 if "resource_type" in ri: 

698 entry["resource_type"] = {"id": ri["resource_type"]} 

699 if "scheme" in ri: 

700 entry["scheme"] = ri["scheme"] 

701 converted.append(entry) 

702 config["related_identifiers"] = converted 

703 

704 if entity_uri: 

705 config["identifiers"] = [{"identifier": entity_uri, "scheme": "url"}] 

706 

707 return config 

708 

709 

710def _get_sub_entity_ids(folders: list[tuple[str, str, dict]]) -> list[str]: 

711 entity_ids: list[str] = [] 

712 for _, folder_name, _ in folders: 

713 entity_id = extract_id_from_folder_name(folder_name) 

714 if entity_id not in entity_ids: 

715 entity_ids.append(entity_id) 

716 return entity_ids 

717 

718 

719_worker_kg: Graph 

720_worker_base_config: dict 

721_worker_creators_lookup: dict 

722 

723 

724def _init_worker(kg_path: Path, base_config: dict, creators_lookup: dict) -> None: 

725 global _worker_kg, _worker_base_config, _worker_creators_lookup 

726 _worker_kg = load_kg(kg_path) 

727 _worker_base_config = base_config 

728 _worker_creators_lookup = creators_lookup 

729 

730 

731def _process_entity( 

732 entity_id: str, 

733 folders: list[tuple[str, str, dict]], 

734 root: Path, 

735 zips_dir: Path, 

736 configs_dir: Path, 

737) -> None: 

738 kg = _worker_kg 

739 base_config = _worker_base_config 

740 creators_lookup = _worker_creators_lookup 

741 entity_ids = _get_sub_entity_ids(folders) 

742 title = extract_entity_title(kg, entity_ids) 

743 keeper_name, keeper_location = extract_keeper_info(kg, entity_ids) 

744 sala_slug = slugify(folders[0][0]) 

745 title_slug = slugify(title) 

746 metadata_creators = build_metadata_creators(kg, entity_ids, creators_lookup) 

747 for stage in STAGES: 

748 result = create_stage_zip(entity_id, stage, folders, root, zips_dir, title) 

749 if result is None: 

750 continue 

751 zip_path, license = result 

752 digitization_creators = build_creators_for_entity_stage( 

753 kg, entity_ids, stage, creators_lookup 

754 ) 

755 creators = merge_creators(digitization_creators, metadata_creators) 

756 entity_uri = build_entity_uri(entity_ids) 

757 methods_description = build_methods_description(kg, entity_ids, stage) 

758 missing_files_notice = select_missing_files_notice( 

759 kg, entity_ids, stage, license 

760 ) 

761 config = generate_zenodo_config( 

762 stage, 

763 zip_path, 

764 title, 

765 base_config, 

766 creators, 

767 methods_description, 

768 license, 

769 entity_uri, 

770 keeper_name, 

771 keeper_location, 

772 missing_files_notice, 

773 ) 

774 config_path = configs_dir / f"{sala_slug}-{title_slug}-{entity_id}-{stage}.yaml" 

775 with open(config_path, "w") as f: 

776 yaml.dump( 

777 config, 

778 f, 

779 Dumper=LiteralBlockDumper, 

780 default_flow_style=False, 

781 allow_unicode=True, 

782 sort_keys=False, 

783 ) 

784 

785 

786def prepare_all( 

787 root: Path, 

788 zenodo_base_config_path: Path, 

789 output_dir: Path, 

790 kg_path: Path = KG_PATH, 

791) -> None: 

792 structure = scan_folder_structure(root) 

793 entity_groups = group_folders_by_entity(structure) 

794 

795 with open(zenodo_base_config_path) as f: 

796 base_config = yaml.safe_load(f) 

797 

798 creators_lookup = load_creators_lookup(CREATORS_LOOKUP_PATH) 

799 

800 zips_dir = output_dir / "zips" 

801 configs_dir = output_dir / "configs" 

802 zips_dir.mkdir(parents=True, exist_ok=True) 

803 configs_dir.mkdir(parents=True, exist_ok=True) 

804 

805 with ProcessPoolExecutor( 

806 initializer=_init_worker, 

807 initargs=(kg_path, base_config, creators_lookup), 

808 ) as executor: 

809 futures = { 

810 executor.submit(