Coverage for changes_metadata_manager / patch / prepared_updates.py: 86%
207 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 json
6import os
7import tempfile
8import time
9import zipfile
10from collections import Counter
11from pathlib import Path
12from typing import cast
14import requests
15import yaml
16from rich.console import Console
17from rich.progress import (
18 BarColumn,
19 MofNCompleteColumn,
20 Progress,
21 SpinnerColumn,
22 TextColumn,
23)
25from changes_metadata_manager.folder_metadata_builder import extract_id_from_folder_name
26from changes_metadata_manager.zenodo_api import (
27 ZenodoRecordCache,
28 create_edit_draft,
29 publish_draft,
30 update_draft,
31)
32from changes_metadata_manager.zenodo_metadata import (
33 ZenodoUpdatePayload,
34 extract_stage_from_filenames,
35)
36from changes_metadata_manager.zenodo_upload import _atomic_write_json
38console = Console()
40CACHE_FILENAME = "creator_names_cache.sqlite3"
41MANIFEST_FILENAME = "manifest.json"
42APPLY_LOG_FILENAME = "apply_log.json"
43PREPARE_STATUSES = {"would_patch", "already_correct", "blocked", "error"}
44REQUEST_DELAY = 2
47def write_payload(path: Path, payload: ZenodoUpdatePayload) -> None:
48 fd, tmp_path = tempfile.mkstemp(dir=path.parent, suffix=".tmp")
49 with os.fdopen(fd, "w") as file:
50 yaml.safe_dump(payload, file, sort_keys=False, allow_unicode=True)
51 os.replace(tmp_path, path)
54def load_manifest(path: Path) -> list[dict]:
55 with open(path) as file:
56 manifest = json.load(file)
57 if not isinstance(manifest, list):
58 raise ValueError(f"Invalid manifest: {path}")
60 entries: list[dict] = []
61 record_ids: set[str] = set()
62 for item in manifest:
63 if not isinstance(item, dict):
64 raise ValueError(f"Invalid manifest entry: {item}")
65 if "record_id" not in item or "status" not in item:
66 raise ValueError(f"Invalid manifest entry: {item}")
67 record_id = str(item["record_id"])
68 if record_id in record_ids:
69 raise ValueError(f"Duplicate record in manifest: {record_id}")
70 record_ids.add(record_id)
71 if item["status"] not in PREPARE_STATUSES:
72 raise ValueError(f"Invalid manifest status for record {record_id}")
73 if item["status"] == "would_patch":
74 expected_payload_file = f"{record_id}.yaml"
75 if (
76 "payload_file" not in item
77 or item["payload_file"] != expected_payload_file
78 ):
79 raise ValueError(f"Invalid payload file for record {record_id}")
80 elif "payload_file" in item:
81 raise ValueError(f"Unexpected payload file for record {record_id}")
82 entries.append(item)
83 return entries
86def prepare_output_directory(output_dir: Path) -> None:
87 output_dir.mkdir(parents=True, exist_ok=True)
88 existing_names = {path.name for path in output_dir.iterdir()}
89 if not existing_names:
90 return
91 if MANIFEST_FILENAME not in existing_names:
92 raise ValueError(f"Output directory contains unmanaged files: {output_dir}")
94 manifest = load_manifest(output_dir / MANIFEST_FILENAME)
95 managed_names = {MANIFEST_FILENAME, APPLY_LOG_FILENAME}
96 managed_names.update(
97 entry["payload_file"] for entry in manifest if "payload_file" in entry
98 )
99 unmanaged_names = existing_names - managed_names
100 if unmanaged_names:
101 names = ", ".join(sorted(unmanaged_names))
102 raise ValueError(f"Output directory contains unmanaged files: {names}")
104 for name in existing_names:
105 path = output_dir / name
106 if not path.is_file():
107 raise ValueError(f"Managed output is not a file: {path}")
108 for name in existing_names:
109 (output_dir / name).unlink()
112def load_package_config(config_path: Path) -> dict:
113 with open(config_path) as file:
114 config = yaml.safe_load(file)
115 if not isinstance(config, dict):
116 raise ValueError(f"Invalid package config: {config_path}")
117 return config
120def _package_archive_path(config: dict, config_path: Path) -> Path:
121 archive_paths = config["files"]
122 if len(archive_paths) != 1:
123 raise ValueError(f"Expected one package archive in config: {config_path}")
124 return Path(archive_paths[0])
127def _extract_archive_entity_ids(archive_path: Path) -> list[str]:
128 with zipfile.ZipFile(archive_path) as archive:
129 folder_names = {name.split("/", 1)[0] for name in archive.namelist()}
130 return sorted(
131 {extract_id_from_folder_name(folder_name) for folder_name in folder_names}
132 )
135def extract_package_entity_ids(config_path: Path) -> list[str]:
136 config = load_package_config(config_path)
137 return _extract_archive_entity_ids(_package_archive_path(config, config_path))
140def extract_package_context(config_path: Path) -> tuple[dict, list[str], str]:
141 config = load_package_config(config_path)
142 archive_path = _package_archive_path(config, config_path)
143 stage = extract_stage_from_filenames([archive_path.name])
144 return config, _extract_archive_entity_ids(archive_path), stage
147def _load_payloads(
148 output_dir: Path, manifest: list[dict]
149) -> dict[str, ZenodoUpdatePayload]:
150 payloads: dict[str, ZenodoUpdatePayload] = {}
151 for entry in manifest:
152 if entry["status"] != "would_patch":
153 continue
154 record_id = str(entry["record_id"])
155 payload_path = output_dir / entry["payload_file"]
156 with open(payload_path) as file:
157 payload = yaml.safe_load(file)
158 if not isinstance(payload, dict):
159 raise ValueError(f"Invalid payload for record {record_id}")
160 required_fields = {"access", "files", "metadata"}
161 allowed_fields = required_fields | {"custom_fields"}
162 if set(payload) - allowed_fields or not required_fields <= set(payload):
163 raise ValueError(f"Invalid payload fields for record {record_id}")
164 if any(not isinstance(payload[field], dict) for field in required_fields):
165 raise ValueError(f"Invalid payload structure for record {record_id}")
166 if "custom_fields" in payload and not isinstance(
167 payload["custom_fields"], dict
168 ):
169 raise ValueError(f"Invalid payload structure for record {record_id}")
170 payloads[record_id] = cast(ZenodoUpdatePayload, payload)
171 return payloads
174def _load_drafts_for_records(
175 drafts_path: Path, record_ids: set[str]
176) -> dict[str, dict]:
177 with open(drafts_path) as file:
178 drafts = json.load(file)
179 if not isinstance(drafts, list):
180 raise ValueError(f"Invalid drafts file: {drafts_path}")
182 drafts_by_id: dict[str, dict] = {}
183 for draft in drafts:
184 if not isinstance(draft, dict) or "draft_id" not in draft:
185 raise ValueError(f"Invalid draft entry: {draft}")
186 record_id = str(draft["draft_id"])
187 if record_id not in record_ids:
188 continue
189 if record_id in drafts_by_id:
190 raise ValueError(f"Duplicate draft for record {record_id}")
191 required_fields = {"zenodo_url", "access_token", "user_agent", "status"}
192 if not required_fields <= set(draft):
193 raise ValueError(f"Incomplete draft for record {record_id}")
194 if draft["status"] != "published":
195 raise ValueError(f"Record {record_id} is not published")
196 drafts_by_id[record_id] = draft
198 missing_ids = record_ids - drafts_by_id.keys()
199 if missing_ids:
200 ids = ", ".join(sorted(missing_ids))
201 raise ValueError(f"Records missing from drafts file: {ids}")
202 return drafts_by_id
205def _load_apply_log(path: Path, record_ids: set[str]) -> list[dict]:
206 if not path.exists():
207 return []
208 with open(path) as file:
209 apply_log = json.load(file)
210 if not isinstance(apply_log, list):
211 raise ValueError(f"Invalid apply log: {path}")
213 entries: list[dict] = []
214 attempted_ids: set[str] = set()
215 for item in apply_log:
216 if not isinstance(item, dict):
217 raise ValueError(f"Invalid apply log entry: {item}")
218 if "record_id" not in item or "status" not in item:
219 raise ValueError(f"Invalid apply log entry: {item}")
220 record_id = str(item["record_id"])
221 if record_id not in record_ids or record_id in attempted_ids:
222 raise ValueError(f"Invalid record in apply log: {record_id}")
223 if item["status"] not in {"patched", "error"}:
224 raise ValueError(f"Invalid apply status for record {record_id}")
225 attempted_ids.add(record_id)
226 entries.append(item)
227 return entries
230def apply_prepared_updates(drafts_path: Path, output_dir: Path) -> Path:
231 manifest = load_manifest(output_dir / MANIFEST_FILENAME)
232 payloads = _load_payloads(output_dir, manifest)
233 record_ids = set(payloads)
234 drafts_by_id = _load_drafts_for_records(drafts_path, record_ids)
235 log_path = output_dir / APPLY_LOG_FILENAME
236 apply_log = _load_apply_log(log_path, record_ids)
237 attempted_ids = {str(entry["record_id"]) for entry in apply_log}
238 _atomic_write_json(log_path, apply_log)
239 stats: Counter[str] = Counter()
240 stats["skipped"] = len(attempted_ids)
241 entries_to_apply = [
242 entry
243 for entry in manifest
244 if entry["status"] == "would_patch"
245 and str(entry["record_id"]) not in attempted_ids
246 ]
247 cache_path = drafts_path.parent / CACHE_FILENAME
249 console.print(f"Applying {len(entries_to_apply)} prepared records...")
250 with (
251 ZenodoRecordCache(cache_path) as cache,
252 Progress(
253 SpinnerColumn(),
254 TextColumn("[progress.description]{task.description}"),
255 BarColumn(),
256 MofNCompleteColumn(),
257 ) as progress,
258 ):
259 task = progress.add_task("Applying", total=len(entries_to_apply))
260 for entry in entries_to_apply:
261 record_id = str(entry["record_id"])
262 draft = drafts_by_id[record_id]
263 zenodo_url = draft["zenodo_url"].rstrip("/")
264 log_entry = {"record_id": entry["record_id"]}
265 progress.update(task, description=f"Record {record_id}")
266 cache.invalidate(zenodo_url, record_id)
268 try:
269 create_edit_draft(
270 zenodo_url,
271 record_id,
272 draft["access_token"],
273 draft["user_agent"],
274 )
275 update_draft(
276 zenodo_url,
277 record_id,
278 draft["access_token"],
279 draft["user_agent"],
280 payloads[record_id],
281 )
282 publish_draft(
283 zenodo_url,
284 record_id,
285 draft["access_token"],
286 draft["user_agent"],
287 )
288 log_entry["status"] = "patched"
289 except requests.RequestException as exc:
290 log_entry["status"] = "error"
291 log_entry["error"] = str(exc)
292 console.print(f"\n[red][FAILED][/red] Record {record_id}: {exc}")
294 apply_log.append(log_entry)
295 stats[log_entry["status"]] += 1
296 _atomic_write_json(log_path, apply_log)
297 progress.advance(task)
298 time.sleep(REQUEST_DELAY)
300 console.print()
301 console.print("[bold]Results:[/bold]")
302 for status in ("patched", "error", "skipped"):
303 console.print(f" {status}: {stats[status]}")
304 console.print(f" Log: {log_path}")
305 return log_path