Coverage for changes_metadata_manager / patch / source_notices.py: 88%
128 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 collections import Counter
9from pathlib import Path
10from typing import cast
12import requests
13from piccione.upload.on_zenodo import text_to_html
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 load_kg
25from changes_metadata_manager.patch.prepared_updates import (
26 CACHE_FILENAME,
27 MANIFEST_FILENAME,
28 REQUEST_DELAY,
29 apply_prepared_updates,
30 extract_package_context,
31 prepare_output_directory,
32 write_payload,
33)
34from changes_metadata_manager.zenodo_api import ZenodoRecordCache, fetch_record
35from changes_metadata_manager.zenodo_metadata import (
36 ZenodoUpdatePayload,
37 build_zenodo_update_payload,
38 extract_content_license,
39 extract_entity_id,
40 extract_stage,
41 zenodo_payload_differences,
42)
43from changes_metadata_manager.zenodo_upload import (
44 EXTERNAL_SOURCE_NOTICE,
45 RESTRICTED_NOTICE,
46 _atomic_write_json,
47 select_missing_files_notice,
48)
50console = Console()
52OLD_NOTICE_HTML = text_to_html(RESTRICTED_NOTICE)
53NEW_NOTICE_HTML = text_to_html(EXTERNAL_SOURCE_NOTICE)
56def _classify_notice(record: dict) -> tuple[str, str | None]:
57 descriptions = record["metadata"]["additional_descriptions"]
58 old_notices = [
59 item
60 for item in descriptions
61 if item["type"]["id"] == "notes" and item["description"] == OLD_NOTICE_HTML
62 ]
63 new_notices = [
64 item
65 for item in descriptions
66 if item["type"]["id"] == "notes" and item["description"] == NEW_NOTICE_HTML
67 ]
68 if len(old_notices) == 1 and not new_notices:
69 return "would_patch", None
70 if len(new_notices) == 1 and not old_notices:
71 return "already_correct", None
72 if not old_notices and not new_notices:
73 return "blocked", "Expected source notice not found"
74 return "blocked", "Unexpected source notice state"
77def _payload_with_source_notice(record: dict) -> ZenodoUpdatePayload:
78 payload = build_zenodo_update_payload(record)
79 descriptions = cast("list[dict]", payload["metadata"]["additional_descriptions"])
80 for item in descriptions:
81 if item["type"]["id"] == "notes" and item["description"] == OLD_NOTICE_HTML:
82 item["description"] = NEW_NOTICE_HTML
83 return payload
84 raise ValueError("Expected source notice not found")
87def _source_notice_candidates(
88 kg: Graph, drafts: list[dict]
89) -> list[tuple[dict, list[str], str]]:
90 candidates: list[tuple[dict, list[str], str]] = []
91 for draft in drafts:
92 if draft["status"] == "failed":
93 continue
94 config_path = Path(draft["config_file"])
95 config, entity_ids, stage = extract_package_context(config_path)
96 notice = select_missing_files_notice(
97 kg, entity_ids, stage, extract_content_license(config)
98 )
99 if notice == EXTERNAL_SOURCE_NOTICE:
100 candidates.append((draft, entity_ids, stage))
101 return candidates
104def prepare_source_notice_patches(
105 drafts_path: Path, kg_path: Path, output_dir: Path
106) -> Path:
107 prepare_output_directory(output_dir)
108 console.print(f"Loading KG from {kg_path}...")
109 kg = load_kg(kg_path)
110 with open(drafts_path) as file:
111 drafts = json.load(file)
113 candidates = _source_notice_candidates(kg, drafts)
114 manifest_path = output_dir / MANIFEST_FILENAME
115 manifest: list[dict] = []
116 stats: Counter[str] = Counter()
117 _atomic_write_json(manifest_path, manifest)
118 cache_path = drafts_path.parent / CACHE_FILENAME
120 console.print(f"Checking {len(candidates)} candidate records...")
121 with (
122 ZenodoRecordCache(cache_path) as cache,
123 Progress(
124 SpinnerColumn(),
125 TextColumn("[progress.description]{task.description}"),
126 BarColumn(),
127 MofNCompleteColumn(),
128 ) as progress,
129 ):
130 task = progress.add_task("Auditing", total=len(candidates))
131 for draft, entity_ids, packaged_stage in candidates:
132 record_id = str(draft["draft_id"])
133 log_entry = {
134 "record_id": draft["draft_id"],
135 "entity_ids": entity_ids,
136 "stage": packaged_stage,
137 }
138 progress.update(task, description=f"Record {record_id}")
140 if draft["status"] != "published":
141 log_entry["status"] = "blocked"
142 log_entry["reason"] = f"Unsupported record status: {draft['status']}"
143 manifest.append(log_entry)
144 stats["blocked"] += 1
145 _atomic_write_json(manifest_path, manifest)
146 progress.advance(task)
147 continue
149 cache_hit = False
150 try:
151 zenodo_url = draft["zenodo_url"].rstrip("/")
152 cached_record = cache.get(zenodo_url, record_id)
153 if cached_record is None:
154 record, has_edit_draft = fetch_record(
155 zenodo_url,
156 record_id,
157 draft["access_token"],
158 draft["user_agent"],
159 )
160 cache.set(zenodo_url, record_id, record, has_edit_draft)
161 else:
162 record, has_edit_draft = cached_record
163 cache_hit = True
165 remote_stage = extract_stage(record)
166 if remote_stage != packaged_stage:
167 raise ValueError(
168 f"Package stage {packaged_stage} does not match remote stage "
169 f"{remote_stage}"
170 )
171 remote_entity_id = extract_entity_id(record)
172 if remote_entity_id not in entity_ids:
173 raise ValueError(
174 f"Remote entity {remote_entity_id} not found in package entities"
175 )
177 remote_status, reason = _classify_notice(record)
178 if has_edit_draft:
179 log_entry["status"] = "blocked"
180 log_entry["reason"] = "An edit draft already exists"
181 elif remote_status == "blocked":
182 log_entry["status"] = "blocked"
183 log_entry["reason"] = reason
184 elif remote_status == "already_correct":
185 log_entry["status"] = "already_correct"
186 else:
187 payload = _payload_with_source_notice(record)
188 payload_differences = zenodo_payload_differences(
189 payload, record, {"additional_descriptions"}
190 )
191 if payload_differences:
192 log_entry["status"] = "blocked"
193 log_entry["reason"] = (
194 "Remote metadata cannot be preserved in an update payload"
195 )
196 log_entry["differences"] = payload_differences
197 else:
198 payload_file = f"{record_id}.yaml"
199 write_payload(output_dir / payload_file, payload)
200 log_entry["status"] = "would_patch"
201 log_entry["payload_file"] = payload_file
202 except (requests.RequestException, ValueError) as exc:
203 log_entry["status"] = "error"
204 log_entry["error"] = str(exc)
205 console.print(f"\n[red][FAILED][/red] Record {record_id}: {exc}")
207 manifest.append(log_entry)
208 stats[log_entry["status"]] += 1
209 _atomic_write_json(manifest_path, manifest)
210 progress.advance(task)
211 if not cache_hit:
212 time.sleep(REQUEST_DELAY)
214 console.print()
215 console.print("[bold]Results:[/bold]")
216 for status in ("would_patch", "already_correct", "blocked", "error"):
217 console.print(f" {status}: {stats[status]}")
218 console.print(f" Manifest: {manifest_path}")
219 return manifest_path
222def apply_source_notice_patches(drafts_path: Path, output_dir: Path) -> Path:
223 return apply_prepared_updates(drafts_path, output_dir)
226if __name__ == "__main__": # pragma: no cover
227 parser = argparse.ArgumentParser(
228 description="Patch external source notices on affected Zenodo records"
229 )
230 subparsers = parser.add_subparsers(dest="command", required=True)
232 prepare_parser = subparsers.add_parser(
233 "prepare", help="Audit records and prepare update payloads"
234 )
235 prepare_parser.add_argument("drafts_json", type=Path, help="Path to drafts.json")
236 prepare_parser.add_argument(
237 "kg_path", type=Path, help="Path to knowledge graph (kg.ttl)"
238 )
239 prepare_parser.add_argument(
240 "output_dir", type=Path, help="Directory for prepared payloads"
241 )
243 apply_parser = subparsers.add_parser(
244 "apply", help="Apply and publish prepared update payloads"
245 )
246 apply_parser.add_argument("drafts_json", type=Path, help="Path to drafts.json")
247 apply_parser.add_argument(
248 "output_dir", type=Path, help="Directory containing prepared payloads"
249 )
251 args = parser.parse_args()
252 if args.command == "prepare":
253 prepare_source_notice_patches(args.drafts_json, args.kg_path, args.output_dir)
254 else:
255 apply_source_notice_patches(args.drafts_json, args.output_dir)