Coverage for changes_metadata_manager / patch / license_metadata.py: 86%
136 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-07-17 16:43 +0000
1# SPDX-FileCopyrightText: 2025-2026 Arcangelo Massari <arcangelo.massari@unibo.it>
2#
3# SPDX-License-Identifier: ISC
5import argparse
6import json
7import time
8from copy import deepcopy
9from pathlib import Path
10from typing import cast
12import requests
13from piccione.upload.on_zenodo import text_to_html
14from rich.console import Console
15from rich.progress import (
16 BarColumn,
17 MofNCompleteColumn,
18 Progress,
19 SpinnerColumn,
20 TextColumn,
21)
23from changes_metadata_manager.folder_metadata_builder import load_kg
24from changes_metadata_manager.zenodo_api import (
25 create_edit_draft,
26 fetch_record,
27 publish_draft,
28 update_draft,
29)
30from changes_metadata_manager.zenodo_metadata import (
31 ZenodoUpdatePayload,
32 build_zenodo_update_payload,
33 extract_content_license,
34 extract_entity_id,
35 extract_stage,
36 zenodo_payload_differences,
37)
38from changes_metadata_manager.zenodo_upload import (
39 CC0_DISCLAIMER,
40 build_rights,
41 extract_license_for_entity_stage,
42)
44console = Console()
46REQUEST_DELAY = 2
47DEFAULT_USER_AGENT = "changes-metadata-manager/1.0.0"
50def _has_cc0_disclaimer(metadata: dict) -> bool:
51 if "additional_descriptions" not in metadata:
52 return False
53 for description in metadata["additional_descriptions"]:
54 if "D. Lgs. 42/2004" in description["description"]:
55 return True
56 return False
59def _rebuild_additional_descriptions(
60 current: list[dict], correct_license: str | None
61) -> list[dict]:
62 rebuilt = [
63 deepcopy(description)
64 for description in current
65 if "D. Lgs. 42/2004" not in description["description"]
66 ]
67 if correct_license == "cc0-1.0":
68 rebuilt.append(
69 {
70 "description": text_to_html(CC0_DISCLAIMER),
71 "type": {"id": "notes"},
72 }
73 )
74 return rebuilt
77def _payload_with_license(
78 record: dict, correct_license: str | None
79) -> ZenodoUpdatePayload:
80 payload = build_zenodo_update_payload(record)
81 metadata = payload["metadata"]
82 current_descriptions = (
83 cast("list[dict]", metadata["additional_descriptions"])
84 if "additional_descriptions" in metadata
85 else []
86 )
87 metadata["rights"] = build_rights(correct_license)
88 metadata["additional_descriptions"] = _rebuild_additional_descriptions(
89 current_descriptions, correct_license
90 )
91 return payload
94def patch_drafts(
95 drafts_path: Path,
96 kg_path: Path,
97 *,
98 dry_run: bool = False,
99) -> None:
100 console.print(f"Loading KG from {kg_path}...")
101 kg = load_kg(kg_path)
103 with open(drafts_path) as file:
104 drafts = json.load(file)
106 stats = {
107 "patched": 0,
108 "blocked": 0,
109 "skipped_correct": 0,
110 "skipped_failed": 0,
111 "skipped_no_kg_license": 0,
112 "errors": 0,
113 }
114 patch_log: list[dict] = []
115 entries_to_check = []
116 for entry in drafts:
117 if entry["status"] == "failed":
118 stats["skipped_failed"] += 1
119 else:
120 entries_to_check.append(entry)
122 console.print(f"Checking {len(entries_to_check)} records...")
124 with Progress(
125 SpinnerColumn(),
126 TextColumn("[progress.description]{task.description}"),
127 BarColumn(),
128 MofNCompleteColumn(),
129 ) as progress:
130 task = progress.add_task("Patching", total=len(entries_to_check))
132 for entry in entries_to_check:
133 record_id = entry["draft_id"]
134 record_id_text = str(record_id)
135 log_entry = {"record_id": record_id}
136 progress.update(task, description=f"Record {record_id}")
138 if entry["status"] not in ("published", "uploaded"):
139 log_entry["status"] = "blocked"
140 log_entry["reason"] = f"Unsupported record status: {entry['status']}"
141 patch_log.append(log_entry)
142 stats["blocked"] += 1
143 progress.advance(task)
144 continue
146 zenodo_url = entry["zenodo_url"].rstrip("/")
147 access_token = entry["access_token"]
148 user_agent = (
149 entry["user_agent"] if "user_agent" in entry else DEFAULT_USER_AGENT
150 )
151 is_published = entry["status"] == "published"
153 try:
154 record, has_edit_draft = fetch_record(
155 zenodo_url, record_id_text, access_token, user_agent
156 )
157 stage = extract_stage(record)
158 entity_id = extract_entity_id(record)
159 correct_license = extract_license_for_entity_stage(kg, entity_id, stage)
160 if correct_license is None:
161 stats["skipped_no_kg_license"] += 1
162 progress.advance(task)
163 continue
165 zenodo_metadata = record["metadata"]
166 current_license = extract_content_license(zenodo_metadata)
167 needs_rights_fix = correct_license != current_license
168 needs_disclaimer_fix = (
169 correct_license == "cc0-1.0"
170 ) != _has_cc0_disclaimer(zenodo_metadata)
172 if not needs_rights_fix and not needs_disclaimer_fix:
173 stats["skipped_correct"] += 1
174 progress.advance(task)
175 continue
177 log_entry.update(
178 {
179 "entity_id": entity_id,
180 "stage": stage,
181 "old_license": current_license,
182 "new_license": correct_license,
183 "rights_changed": needs_rights_fix,
184 "disclaimer_changed": needs_disclaimer_fix,
185 }
186 )
188 if is_published and has_edit_draft:
189 log_entry["status"] = "blocked"
190 log_entry["reason"] = "An edit draft already exists"
191 elif not is_published and not has_edit_draft:
192 log_entry["status"] = "blocked"
193 log_entry["reason"] = "Draft not found for unpublished record"
194 else:
195 payload = _payload_with_license(record, correct_license)
196 payload_differences = zenodo_payload_differences(
197 payload,
198 record,
199 {"rights", "additional_descriptions"},
200 )
201 if payload_differences:
202 log_entry["status"] = "blocked"
203 log_entry["reason"] = (
204 "Remote metadata cannot be preserved in an update payload"
205 )
206 log_entry["differences"] = payload_differences
207 elif dry_run:
208 console.print(
209 f" [cyan]DRY RUN[/cyan] {record_id}: "
210 f"{current_license} → {correct_license}"
211 )
212 log_entry["status"] = "dry_run"
213 else:
214 draft_record = (
215 create_edit_draft(
216 zenodo_url,
217 record_id_text,
218 access_token,
219 user_agent,
220 )
221 if is_published
222 else record
223 )
224 payload = _payload_with_license(draft_record, correct_license)
225 payload_differences = zenodo_payload_differences(
226 payload,
227 draft_record,
228 {"rights", "additional_descriptions"},
229 )
230 if payload_differences:
231 log_entry["status"] = "blocked"
232 log_entry["reason"] = (
233 "Edit draft metadata cannot be preserved in an "
234 "update payload"
235 )
236 log_entry["differences"] = payload_differences
237 else:
238 update_draft(
239 zenodo_url,
240 record_id_text,
241 access_token,
242 user_agent,
243 payload,
244 )
245 if is_published:
246 publish_draft(
247 zenodo_url,
248 record_id_text,
249 access_token,
250 user_agent,
251 )
252 log_entry["status"] = "patched"
253 except (requests.RequestException, ValueError) as exc:
254 log_entry["status"] = "error"
255 log_entry["error"] = str(exc)
256 console.print(f"\n[red][FAILED][/red] Record {record_id}: {exc}")
258 patch_log.append(log_entry)
259 if log_entry["status"] in ("patched", "dry_run"):
260 stats["patched"] += 1
261 elif log_entry["status"] == "blocked":
262 stats["blocked"] += 1
263 elif log_entry["status"] == "error":
264 stats["errors"] += 1
265 progress.advance(task)
266 time.sleep(REQUEST_DELAY)
268 log_path = drafts_path.parent / "patch_license_log.json"
269 with open(log_path, "w") as file:
270 json.dump(patch_log, file, indent=2)
272 console.print()
273 console.print("[bold]Results:[/bold]")
274 console.print(f" Patched: {stats['patched']}")
275 console.print(f" Blocked: {stats['blocked']}")
276 console.print(f" Already correct: {stats['skipped_correct']}")
277 console.print(f" Skipped (failed): {stats['skipped_failed']}")
278 console.print(f" Skipped (no KG license): {stats['skipped_no_kg_license']}")
279 console.print(f" Errors: {stats['errors']}")
280 console.print(f" Log: {log_path}")
283if __name__ == "__main__": # pragma: no cover
284 parser = argparse.ArgumentParser(
285 description="Patch license metadata on Zenodo records"
286 )
287 parser.add_argument("drafts_json", type=Path, help="Path to drafts.json")
288 parser.add_argument("kg_path", type=Path, help="Path to knowledge graph (kg.ttl)")
289 parser.add_argument(
290 "--dry-run", action="store_true", help="Show changes without applying"
291 )
292 args = parser.parse_args()
293 patch_drafts(args.drafts_json, args.kg_path, dry_run=args.dry_run)