Coverage for changes_metadata_manager / patch / creator_names.py: 90%
138 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 re
8import time
9from collections import Counter
10from copy import deepcopy
11from pathlib import Path
13import requests
14from rdflib import Graph
15from rich.console import Console
16from rich.progress import (
17 BarColumn,
18 MofNCompleteColumn,
19 Progress,
20 SpinnerColumn,
21 TextColumn,
22)
24from changes_metadata_manager.folder_metadata_builder import (
25 BASE_URI,
26 load_kg,
27)
28from changes_metadata_manager.zenodo_api import (
29 ZenodoRecordCache,
30 fetch_record,
31)
32from changes_metadata_manager.zenodo_metadata import (
33 ZenodoUpdatePayload,
34 build_zenodo_update_payload,
35 extract_stage,
36 zenodo_payload_differences,
37)
38from changes_metadata_manager.zenodo_upload import (
39 CREATORS_LOOKUP_PATH,
40 _atomic_write_json,
41 build_creators_for_entity_stage,
42 build_metadata_creators,
43 load_creators_lookup,
44 merge_creators,
45)
46from changes_metadata_manager.patch.prepared_updates import (
47 CACHE_FILENAME,
48 MANIFEST_FILENAME,
49 REQUEST_DELAY,
50 apply_prepared_updates,
51 extract_package_entity_ids,
52 prepare_output_directory,
53 write_payload,
54)
56console = Console()
57ACTIVITY_ENTITY_PATTERN = re.compile(rf"^{re.escape(BASE_URI)}/act/([^/]+)/")
60def _creator_orcid(creator: dict) -> str:
61 identifiers = creator["person_or_org"]["identifiers"]
62 orcids = [
63 identifier["identifier"]
64 for identifier in identifiers
65 if identifier["scheme"] == "orcid"
66 ]
67 if len(orcids) != 1:
68 raise ValueError(f"Expected one ORCID for creator: {creator}")
69 return orcids[0]
72def _creator_signature(creator: dict) -> tuple[str, str, str, str, tuple[str, ...]]:
73 person = creator["person_or_org"]
74 affiliations = tuple(sorted(item["name"] for item in creator["affiliations"]))
75 return (
76 person["type"],
77 person["family_name"],
78 person["given_name"],
79 creator["role"]["id"],
80 affiliations,
81 )
84def _creators_by_orcid(
85 creators: list[dict],
86) -> dict[str, tuple[str, str, str, str, tuple[str, ...]]]:
87 creators_by_orcid = {
88 _creator_orcid(creator): _creator_signature(creator) for creator in creators
89 }
90 if len(creators_by_orcid) != len(creators):
91 raise ValueError("Duplicate creator ORCID")
92 return creators_by_orcid
95def _classify_creators(
96 current_creators: list[dict],
97 expected_creators: list[dict],
98) -> tuple[str, set[str]]:
99 current = _creators_by_orcid(current_creators)
100 expected = _creators_by_orcid(expected_creators)
101 missing_orcids = expected.keys() - current.keys()
103 if current == expected:
104 return "already_correct", set()
106 expected_without_missing = {
107 orcid: signature
108 for orcid, signature in expected.items()
109 if orcid not in missing_orcids
110 }
111 if missing_orcids and current == expected_without_missing:
112 return "would_patch", missing_orcids
114 return "blocked", missing_orcids
117def _entity_ids(kg: Graph) -> set[str]:
118 entity_ids: set[str] = set()
119 for subject in kg.subjects():
120 match = ACTIVITY_ENTITY_PATTERN.match(str(subject))
121 if match:
122 entity_ids.add(match.group(1))
123 return entity_ids
126def _payload_with_creators(
127 record: dict, expected_creators: list[dict]
128) -> ZenodoUpdatePayload:
129 payload = build_zenodo_update_payload(record)
130 payload["metadata"]["creators"] = deepcopy(expected_creators)
131 return payload
134def prepare_creator_name_patches(
135 drafts_path: Path,
136 kg_path: Path,
137 output_dir: Path,
138 *,
139 creators_lookup_path: Path = CREATORS_LOOKUP_PATH,
140) -> Path:
141 prepare_output_directory(output_dir)
142 console.print(f"Loading KG from {kg_path}...")
143 kg = load_kg(kg_path)
144 kg_entity_ids = _entity_ids(kg)
145 creators_lookup = load_creators_lookup(creators_lookup_path)
146 with open(drafts_path) as file:
147 drafts = json.load(file)
149 records = [draft for draft in drafts if draft["status"] != "failed"]
150 manifest_path = output_dir / MANIFEST_FILENAME
151 manifest: list[dict] = []
152 stats: Counter[str] = Counter()
153 _atomic_write_json(manifest_path, manifest)
154 cache_path = drafts_path.parent / CACHE_FILENAME
156 console.print(f"Checking {len(records)} records...")
157 with (
158 ZenodoRecordCache(cache_path) as cache,
159 Progress(
160 SpinnerColumn(),
161 TextColumn("[progress.description]{task.description}"),
162 BarColumn(),
163 MofNCompleteColumn(),
164 ) as progress,
165 ):
166 task = progress.add_task("Auditing", total=len(records))
167 for draft in records:
168 record_id = str(draft["draft_id"])
169 log_entry = {"record_id": draft["draft_id"]}
170 progress.update(task, description=f"Record {record_id}")
172 if draft["status"] != "published":
173 log_entry["status"] = "blocked"
174 log_entry["reason"] = f"Unsupported record status: {draft['status']}"
175 manifest.append(log_entry)
176 stats["blocked"] += 1
177 _atomic_write_json(manifest_path, manifest)
178 progress.advance(task)
179 continue
181 cache_hit = False
182 try:
183 zenodo_url = draft["zenodo_url"].rstrip("/")
184 cached_record = cache.get(zenodo_url, record_id)
185 if cached_record is None:
186 record, has_edit_draft = fetch_record(
187 zenodo_url,
188 record_id,
189 draft["access_token"],
190 draft["user_agent"],
191 )
192 cache.set(zenodo_url, record_id, record, has_edit_draft)
193 else:
194 record, has_edit_draft = cached_record
195 cache_hit = True
196 stage = extract_stage(record)
197 entity_ids = extract_package_entity_ids(Path(draft["config_file"]))
198 missing_entity_ids = set(entity_ids).difference(kg_entity_ids)
199 if missing_entity_ids:
200 missing_ids = ", ".join(sorted(missing_entity_ids))
201 raise ValueError(
202 f"No KG activities found for entities: {missing_ids}"
203 )
204 expected_creators = merge_creators(
205 build_creators_for_entity_stage(
206 kg, entity_ids, stage, creators_lookup
207 ),
208 build_metadata_creators(kg, entity_ids, creators_lookup),
209 )
210 remote_status, missing_orcids = _classify_creators(
211 record["metadata"]["creators"], expected_creators
212 )
213 missing_creators = [
214 {
215 "name": (
216 f"{creator['person_or_org']['given_name']} "
217 f"{creator['person_or_org']['family_name']}"
218 ),
219 "orcid": _creator_orcid(creator),
220 }
221 for creator in expected_creators
222 if _creator_orcid(creator) in missing_orcids
223 ]
224 log_entry.update(
225 {
226 "entity_ids": entity_ids,
227 "stage": stage,
228 "missing_creators": missing_creators,
229 }
230 )
232 if has_edit_draft:
233 log_entry["status"] = "blocked"
234 log_entry["reason"] = "An edit draft already exists"
235 elif remote_status == "blocked":
236 log_entry["status"] = "blocked"
237 log_entry["reason"] = (
238 "Creator differences are not limited to missing creators"
239 )
240 elif remote_status == "already_correct":
241 log_entry["status"] = "already_correct"
242 else:
243 payload = _payload_with_creators(record, expected_creators)
244 payload_differences = zenodo_payload_differences(
245 payload, record, {"creators"}
246 )
247 if payload_differences:
248 log_entry["status"] = "blocked"
249 log_entry["reason"] = (
250 "Remote metadata cannot be preserved in an update payload"
251 )
252 log_entry["differences"] = payload_differences
253 else:
254 payload_file = f"{record_id}.yaml"
255 write_payload(output_dir / payload_file, payload)
256 log_entry["status"] = "would_patch"
257 log_entry["payload_file"] = payload_file
258 except (requests.RequestException, ValueError) as exc:
259 log_entry["status"] = "error"
260 log_entry["error"] = str(exc)
261 console.print(f"\n[red][FAILED][/red] Record {record_id}: {exc}")
263 manifest.append(log_entry)
264 stats[log_entry["status"]] += 1
265 _atomic_write_json(manifest_path, manifest)
266 progress.advance(task)
267 if not cache_hit:
268 time.sleep(REQUEST_DELAY)
270 console.print()
271 console.print("[bold]Results:[/bold]")
272 for status in ("would_patch", "already_correct", "blocked", "error"):
273 console.print(f" {status}: {stats[status]}")
274 console.print(f" Manifest: {manifest_path}")
275 return manifest_path
278def apply_creator_name_patches(drafts_path: Path, output_dir: Path) -> Path:
279 return apply_prepared_updates(drafts_path, output_dir)
282if __name__ == "__main__": # pragma: no cover
283 parser = argparse.ArgumentParser(
284 description="Patch missing creators on affected Zenodo records"
285 )
286 subparsers = parser.add_subparsers(dest="command", required=True)
288 prepare_parser = subparsers.add_parser(
289 "prepare", help="Audit records and prepare update payloads"
290 )
291 prepare_parser.add_argument("drafts_json", type=Path, help="Path to drafts.json")
292 prepare_parser.add_argument(
293 "kg_path", type=Path, help="Path to knowledge graph (kg.ttl)"
294 )
295 prepare_parser.add_argument(
296 "output_dir", type=Path, help="Directory for prepared payloads"
297 )
299 apply_parser = subparsers.add_parser(
300 "apply", help="Apply and publish prepared update payloads"
301 )
302 apply_parser.add_argument("drafts_json", type=Path, help="Path to drafts.json")
303 apply_parser.add_argument(
304 "output_dir", type=Path, help="Directory containing prepared payloads"
305 )
307 args = parser.parse_args()
308 if args.command == "prepare":
309 prepare_creator_name_patches(args.drafts_json, args.kg_path, args.output_dir)
310 else:
311 apply_creator_name_patches(args.drafts_json, args.output_dir)