Coverage for oc_meta / run / patches / fix_dangling_ars.py: 83%
551 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-07-25 10:39 +0000
1# SPDX-FileCopyrightText: 2026 Arcangelo Massari <arcangelo.massari@unibo.it>
2#
3# SPDX-License-Identifier: ISC
5from __future__ import annotations
7import argparse
8import csv
9import hashlib
10import multiprocessing
11import os
12import shutil
13import signal
14import tempfile
15from collections import Counter, defaultdict
16from collections.abc import Mapping
17from concurrent.futures import ProcessPoolExecutor
18from dataclasses import dataclass
19from datetime import datetime, timezone
20from typing import Protocol, cast
22import orjson
23import requests
24from oc_ocdm.graph import GraphSet
25from oc_ocdm.graph.entities.bibliographic.bibliographic_resource import (
26 BibliographicResource,
27)
28from oc_ocdm.graph.graph_entity import GraphEntity
29from rich_argparse import RichHelpFormatter
30from triplelite import RDFTerm, TripleLite, from_rdflib
32from oc_meta.core.editor import MetaEditor
33from oc_meta.lib.agent_metadata import (
34 AgentIdentifier,
35 AgentMetadata,
36 AgentMetadataClient,
37 ApiCache,
38 WorkMetadata,
39)
40from oc_meta.lib.console import console, create_progress
41from oc_meta.lib.rdf_patch import (
42 HAS_IDENTIFIER,
43 HAS_LITERAL_VALUE,
44 HAS_NEXT,
45 IS_DOCUMENT_CONTEXT_FOR,
46 PROV_SPECIALIZATION_OF,
47 USES_IDENTIFIER_SCHEME,
48 AuditConfig,
49 EntityFileLocator,
50 batches as _batches,
51 data_files as _data_files,
52 ensure_parent as _ensure_parent,
53 first as _first,
54 ids as _ids,
55 literals as _literals,
56 load_audit_config,
57 load_available_entities,
58 load_entities as _load_entities,
59 provenance_path as _provenance_path,
60 read_json_object as _read_json_object,
61 sha256 as _sha256,
62 snapshot_number as _snapshot_number,
63 write_json as _write_json,
64)
65from oc_meta.run.merge.entities import REINDEX_SENTINEL_FILENAME
66from oc_meta.run.meta.generate_csv import FIELDNAMES, URI_TYPE_DICT
68PLAN_VERSION = 2
69PROVIDERS = ("crossref", "datacite")
70PROV_INVALIDATED_AT_TIME = "http://www.w3.org/ns/prov#invalidatedAtTime"
71DCTERMS_TITLE = "http://purl.org/dc/terms/title"
73csv.field_size_limit(2**31 - 1)
75_stop_requested = False
76_existing_roles: frozenset[str] = frozenset()
77_target_roles: frozenset[str] = frozenset()
78_fork_context = multiprocessing.get_context("fork")
79_forkserver_context = multiprocessing.get_context("forkserver")
82@dataclass(frozen=True, slots=True)
83class WorkRecord:
84 uri: str
85 role_uris: tuple[str, ...]
86 identifier_uris: tuple[str, ...]
89@dataclass(frozen=True, slots=True)
90class IdentifierRecord:
91 uri: str
92 scheme: str
93 value: str
96class WorkProviderClient(Protocol):
97 def crossref(self, doi: str) -> WorkMetadata | None: ...
99 def datacite(self, doi: str) -> WorkMetadata | None: ...
102def _handle_signal(signum: int, frame: object) -> None:
103 del signum, frame
104 global _stop_requested
105 _stop_requested = True
108def _object_sha256(value: object) -> str:
109 return hashlib.sha256(orjson.dumps(value, option=orjson.OPT_SORT_KEYS)).hexdigest()
112def _process_pool(workers: int) -> ProcessPoolExecutor:
113 return ProcessPoolExecutor(max_workers=workers, mp_context=_fork_context)
116def _scan_entity_uri_batch(paths: list[str]) -> set[str]:
117 return {uri for path in paths for uri in _load_entities(path)}
120def scan_entity_uris(files: list[str], workers: int) -> set[str]:
121 result = set()
122 batches = _batches(files, 24)
123 with _process_pool(workers) as executor:
124 partial_results = executor.map(_scan_entity_uri_batch, batches)
125 with create_progress() as progress:
126 task = progress.add_task("Indexing agent roles", total=len(files))
127 for batch, partial in zip(batches, partial_results):
128 result.update(partial)
129 progress.advance(task, len(batch))
130 return result
133def _init_dangling_scan(existing_roles: frozenset[str]) -> None:
134 global _existing_roles
135 _existing_roles = existing_roles
138def _scan_dangling_work_batch(
139 paths: list[str],
140) -> tuple[dict[str, WorkRecord], dict[str, tuple[str, ...]]]:
141 works = {}
142 missing_by_work = {}
143 for path in paths:
144 for uri, entity in _load_entities(path).items():
145 role_uris = tuple(_ids(entity, IS_DOCUMENT_CONTEXT_FOR))
146 missing = tuple(sorted(set(role_uris) - _existing_roles))
147 if not missing:
148 continue
149 works[uri] = WorkRecord(uri, role_uris, tuple(_ids(entity, HAS_IDENTIFIER)))
150 missing_by_work[uri] = missing
151 return works, missing_by_work
154def _init_role_scan(target_roles: frozenset[str]) -> None:
155 global _target_roles
156 _target_roles = target_roles
159def _scan_role_context_batch(paths: list[str]) -> dict[str, list[str]]:
160 contexts: dict[str, list[str]] = defaultdict(list)
161 for path in paths:
162 for uri, entity in _load_entities(path).items():
163 for role_uri in _ids(entity, IS_DOCUMENT_CONTEXT_FOR):
164 if role_uri in _target_roles:
165 contexts[role_uri].append(uri)
166 return contexts
169def _scan_role_link_batch(paths: list[str]) -> set[tuple[str, str]]:
170 links = set()
171 for path in paths:
172 for uri, entity in _load_entities(path).items():
173 for next_uri in _ids(entity, HAS_NEXT):
174 if uri in _target_roles or next_uri in _target_roles:
175 links.add((uri, next_uri))
176 return links
179def _scan_contexts(
180 br_files: list[str], target_roles: frozenset[str], workers: int
181) -> dict[str, tuple[str, ...]]:
182 contexts: dict[str, list[str]] = defaultdict(list)
183 batches = _batches(br_files, 24)
184 with ProcessPoolExecutor(
185 max_workers=workers,
186 mp_context=_fork_context,
187 initializer=_init_role_scan,
188 initargs=(target_roles,),
189 ) as executor:
190 partial_results = executor.map(_scan_role_context_batch, batches)
191 with create_progress() as progress:
192 task = progress.add_task("Checking role contexts", total=len(br_files))
193 for batch, partial in zip(batches, partial_results):
194 for role_uri, work_uris in partial.items():
195 contexts[role_uri].extend(work_uris)
196 progress.advance(task, len(batch))
197 return {role_uri: tuple(sorted(contexts[role_uri])) for role_uri in target_roles}
200def find_dangling_works(
201 config: AuditConfig, workers: int
202) -> tuple[
203 dict[str, WorkRecord],
204 dict[str, dict[str, object]],
205 dict[str, tuple[str, ...]],
206 dict[str, tuple[str, ...]],
207]:
208 role_files = _data_files(os.path.join(config.rdf_dir, "ar"), config.zip_output)
209 existing_roles = frozenset(scan_entity_uris(role_files, workers))
210 br_files = _data_files(os.path.join(config.rdf_dir, "br"), config.zip_output)
211 works = {}
212 missing_by_work = {}
213 batches = _batches(br_files, 24)
214 with ProcessPoolExecutor(
215 max_workers=workers,
216 mp_context=_fork_context,
217 initializer=_init_dangling_scan,
218 initargs=(existing_roles,),
219 ) as executor:
220 partial_results = executor.map(_scan_dangling_work_batch, batches)
221 with create_progress() as progress:
222 task = progress.add_task(
223 "Finding dangling role references", total=len(br_files)
224 )
225 for batch, (partial_works, partial_missing) in zip(
226 batches, partial_results
227 ):
228 works.update(partial_works)
229 missing_by_work.update(partial_missing)
230 progress.advance(task, len(batch))
231 target_roles = frozenset(
232 role_uri for work in works.values() for role_uri in work.role_uris
233 )
234 locator = EntityFileLocator(
235 config.rdf_dir, config.dir_split, config.items_per_file, config.zip_output
236 )
237 role_entities = load_available_entities(
238 set(target_roles.intersection(existing_roles)), locator, workers
239 )
240 return (
241 works,
242 role_entities,
243 missing_by_work,
244 _scan_contexts(br_files, target_roles, workers),
245 )
248def scan_role_links(
249 config: AuditConfig, role_uris: set[str], workers: int
250) -> set[tuple[str, str]]:
251 if not role_uris:
252 return set()
253 role_files = _data_files(os.path.join(config.rdf_dir, "ar"), config.zip_output)
254 batches = _batches(role_files, 24)
255 links = set()
256 with ProcessPoolExecutor(
257 max_workers=workers,
258 mp_context=_fork_context,
259 initializer=_init_role_scan,
260 initargs=(frozenset(role_uris),),
261 ) as executor:
262 partial_results = executor.map(_scan_role_link_batch, batches)
263 with create_progress() as progress:
264 task = progress.add_task("Checking role links", total=len(role_files))
265 for batch, partial in zip(batches, partial_results):
266 links.update(partial)
267 progress.advance(task, len(batch))
268 return links
271def _identifier_record(uri: str, entity: dict[str, object]) -> IdentifierRecord | None:
272 scheme_uri = _first(_ids(entity, USES_IDENTIFIER_SCHEME))
273 value = _first(_literals(entity, HAS_LITERAL_VALUE))
274 if not scheme_uri or not value:
275 return None
276 return IdentifierRecord(uri, scheme_uri.rsplit("/", 1)[-1], value)
279def _provenance_status_batch(
280 paths: list[tuple[str, frozenset[str]]],
281) -> dict[str, str]:
282 snapshots_by_entity: dict[str, list[dict[str, object]]] = defaultdict(list)
283 for path, targets in paths:
284 if not os.path.exists(path):
285 continue
286 for snapshot in _load_entities(path).values():
287 specialization = _first(_ids(snapshot, PROV_SPECIALIZATION_OF))
288 if specialization in targets:
289 snapshots_by_entity[specialization].append(snapshot)
290 statuses = {}
291 for uri, snapshots in snapshots_by_entity.items():
292 latest = max(
293 snapshots,
294 key=lambda snapshot: _snapshot_number(cast(str, snapshot["@id"])),
295 )
296 statuses[uri] = (
297 "latest_snapshot_invalidated"
298 if _literals(latest, PROV_INVALIDATED_AT_TIME)
299 else "latest_snapshot_active"
300 )
301 return statuses
304def load_provenance_statuses(
305 uris: set[str], locator: EntityFileLocator, workers: int
306) -> dict[str, str]:
307 targets_by_path: dict[str, set[str]] = defaultdict(set)
308 for uri in uris:
309 path = _provenance_path(locator.path(uri), locator.zip_output)
310 targets_by_path[path].add(uri)
311 tasks = [(path, frozenset(targets)) for path, targets in targets_by_path.items()]
312 statuses = {}
313 with ProcessPoolExecutor(
314 max_workers=workers, mp_context=_forkserver_context
315 ) as executor:
316 for partial in executor.map(_provenance_status_batch, _batches(tasks, 24)):
317 statuses.update(partial)
318 return {uri: statuses[uri] if uri in statuses else "no_snapshot" for uri in uris}
321def _load_local_context(
322 works: dict[str, WorkRecord],
323 role_entities: dict[str, dict[str, object]],
324 missing_by_work: dict[str, tuple[str, ...]],
325 locator: EntityFileLocator,
326 workers: int,
327) -> tuple[dict[str, IdentifierRecord], dict[str, dict[str, object]], dict[str, str]]:
328 work_entities = load_available_entities(set(works), locator, workers)
329 identifier_uris = {
330 identifier_uri
331 for work in works.values()
332 for identifier_uri in work.identifier_uris
333 }
334 identifier_entities = load_available_entities(identifier_uris, locator, workers)
335 identifiers = {
336 uri: record
337 for uri, entity in identifier_entities.items()
338 if (record := _identifier_record(uri, entity)) is not None
339 }
340 raw_entities = dict(work_entities)
341 raw_entities.update(role_entities)
342 raw_entities.update(identifier_entities)
343 missing_uris = {uri for uris in missing_by_work.values() for uri in uris}
344 provenance = load_provenance_statuses(missing_uris, locator, workers)
345 return identifiers, raw_entities, provenance
348def _work_identifiers(
349 work: WorkRecord, identifiers: Mapping[str, IdentifierRecord]
350) -> dict[str, str]:
351 result = {}
352 for uri in work.identifier_uris:
353 if uri in identifiers and identifiers[uri].scheme not in result:
354 record = identifiers[uri]
355 result[record.scheme] = record.value
356 return result
359def _types(entity: dict[str, object]) -> list[str]:
360 value = entity["@type"] if "@type" in entity else []
361 if isinstance(value, str):
362 return [value]
363 if not isinstance(value, list):
364 return []
365 return [item for item in value if isinstance(item, str)]
368def _local_type(entity: dict[str, object]) -> str:
369 for entity_type in _types(entity):
370 mapped = URI_TYPE_DICT[entity_type] if entity_type in URI_TYPE_DICT else ""
371 if mapped:
372 return mapped
373 return ""
376def _omid(br_uri: str) -> str:
377 if "/br/" not in br_uri:
378 raise ValueError(f"Cannot derive a BR OMID from URI: {br_uri}")
379 return f"omid:br/{br_uri.rsplit('/br/', 1)[1]}"
382def _serialize_name(
383 family: str,
384 given: str,
385 name: str,
386 identifiers: tuple[AgentIdentifier, ...],
387) -> str:
388 if family or given:
389 rendered_name = f"{family}, {given}" if given else f"{family},"
390 else:
391 rendered_name = name
392 rendered_identifiers = " ".join(
393 f"{identifier['scheme']}:{identifier['value']}" for identifier in identifiers
394 )
395 if rendered_identifiers:
396 return (
397 f"{rendered_name} [{rendered_identifiers}]"
398 if rendered_name
399 else f"[{rendered_identifiers}]"
400 )
401 return rendered_name
404def serialize_agents(agents: list[AgentMetadata]) -> str:
405 return "; ".join(
406 _serialize_name(
407 agent["family"],
408 agent["given"],
409 agent["name"],
410 agent["identifiers"],
411 )
412 for agent in agents
413 )
416def work_csv_row(
417 work: WorkRecord,
418 work_entity: dict[str, object],
419 identifiers: Mapping[str, str],
420 metadata: WorkMetadata,
421) -> dict[str, str]:
422 doi = identifiers["doi"]
423 publisher = _serialize_name(
424 "",
425 "",
426 metadata["publisher"],
427 metadata["publisher_identifiers"],
428 )
429 return {
430 "id": f"{_omid(work.uri)} doi:{doi}",
431 "title": _first(_literals(work_entity, DCTERMS_TITLE)),
432 "author": serialize_agents(metadata["author"]),
433 "issue": "",
434 "volume": "",
435 "venue": "",
436 "page": "",
437 "pub_date": "",
438 "type": _local_type(work_entity),
439 "publisher": publisher,
440 "editor": serialize_agents(metadata["editor"]),
441 }
444def _select_provider(
445 provider: WorkProviderClient, doi: str
446) -> tuple[
447 str | None,
448 WorkMetadata | None,
449 list[str],
450 list[dict[str, str]],
451]:
452 attempted = []
453 errors = []
454 if not doi:
455 return None, None, attempted, errors
456 attempted.append("crossref")
457 try:
458 metadata = provider.crossref(doi)
459 except requests.RequestException as error:
460 errors.append(
461 {
462 "provider": "crossref",
463 "error": type(error).__name__,
464 "message": str(error),
465 }
466 )
467 metadata = None
468 if metadata is not None:
469 return "crossref", metadata, attempted, errors
470 attempted.append("datacite")
471 try:
472 metadata = provider.datacite(doi)
473 except requests.RequestException as error:
474 errors.append(
475 {
476 "provider": "datacite",
477 "error": type(error).__name__,
478 "message": str(error),
479 }
480 )
481 metadata = None
482 if metadata is not None:
483 return "datacite", metadata, attempted, errors
484 return None, None, attempted, errors
487def _operation_payload(operation: dict[str, object]) -> dict[str, object]:
488 payload = dict(operation)
489 payload.pop("operation_id")
490 return payload
493def _finalize_operation(operation: dict[str, object]) -> dict[str, object]:
494 operation["operation_id"] = _object_sha256(operation)[:20]
495 return operation
498def _verify_operation_id(operation: dict[str, object]) -> None:
499 operation_id = operation["operation_id"]
500 if not isinstance(operation_id, str):
501 raise ValueError("Operation ID must be a string")
502 if _object_sha256(_operation_payload(operation))[:20] != operation_id:
503 raise ValueError(f"Correction plan has modified operation {operation_id}")
506def _operation_preconditions(
507 work: WorkRecord,
508 role_entities: Mapping[str, dict[str, object]],
509 raw_entities: Mapping[str, dict[str, object]],
510 missing: tuple[str, ...],
511 contexts: Mapping[str, tuple[str, ...]],
512 links: set[tuple[str, str]],
513 owned_missing: set[str],
514 provenance: Mapping[str, str],
515) -> dict[str, object]:
516 return {
517 "br_entity": raw_entities[work.uri],
518 "identifier_entities": {
519 uri: raw_entities[uri]
520 for uri in sorted(work.identifier_uris)
521 if uri in raw_entities
522 },
523 "role_entities": {
524 uri: role_entities[uri]
525 for uri in sorted(work.role_uris)
526 if uri in role_entities
527 },
528 "role_references": list(work.role_uris),
529 "dangling_ar_references": list(missing),
530 "owned_missing_provenance": {
531 uri: provenance[uri] for uri in sorted(owned_missing)
532 },
533 "contexts": {uri: list(contexts[uri]) for uri in sorted(set(work.role_uris))},
534 "has_next_edges": [
535 [source, target]
536 for source, target in sorted(links)
537 if source in work.role_uris or target in work.role_uris
538 ],
539 }
542def build_repair_plan(
543 works: dict[str, WorkRecord],
544 role_entities: dict[str, dict[str, object]],
545 identifiers: dict[str, IdentifierRecord],
546 raw_entities: dict[str, dict[str, object]],
547 missing_by_work: dict[str, tuple[str, ...]],
548 contexts: dict[str, tuple[str, ...]],
549 provenance: dict[str, str],
550 links: set[tuple[str, str]],
551 provider: WorkProviderClient,
552) -> tuple[list[dict[str, object]], dict[str, list[dict[str, str]]]]:
553 affected_brs = set(works)
554 target_roles = {role_uri for work in works.values() for role_uri in work.role_uris}
555 owners = {
556 role_uri: min(
557 br_uri for br_uri, work in works.items() if role_uri in work.role_uris
558 )
559 for role_uri in target_roles
560 }
561 boundary_links = {
562 edge for edge in links if (edge[0] in target_roles) != (edge[1] in target_roles)
563 }
564 operations = []
565 rows_by_provider: dict[str, list[dict[str, str]]] = {
566 provider_name: [] for provider_name in PROVIDERS
567 }
568 with create_progress() as progress:
569 task = progress.add_task("Querying work metadata", total=len(works))
570 for br_uri in sorted(works):
571 if _stop_requested:
572 break
573 work = works[br_uri]
574 work_identifiers = _work_identifiers(work, identifiers)
575 blockers = []
576 for role_uri in sorted(set(work.role_uris)):
577 external_contexts = sorted(set(contexts[role_uri]) - affected_brs)
578 if external_contexts:
579 blockers.append(
580 {
581 "type": "shared_agent_role",
582 "ar": role_uri,
583 "external_contexts": external_contexts,
584 }
585 )
586 for source, target in sorted(boundary_links):
587 if source in work.role_uris or target in work.role_uris:
588 blockers.append(
589 {
590 "type": "external_has_next",
591 "source": source,
592 "target": target,
593 }
594 )
595 selected_provider = None
596 metadata = None
597 attempted_providers: list[str] = []
598 provider_errors: list[dict[str, str]] = []
599 if not blockers:
600 (
601 selected_provider,
602 metadata,
603 attempted_providers,
604 provider_errors,
605 ) = _select_provider(
606 provider,
607 work_identifiers["doi"] if "doi" in work_identifiers else "",
608 )
609 if metadata is None:
610 blockers.append(
611 {
612 "type": "provider_unavailable",
613 "doi": work_identifiers["doi"]
614 if "doi" in work_identifiers
615 else "",
616 "attempted_providers": attempted_providers,
617 "errors": provider_errors,
618 }
619 )
620 existing_roles = sorted(
621 role_uri
622 for role_uri in set(work.role_uris)
623 if role_uri in role_entities and owners[role_uri] == br_uri
624 )
625 owned_missing = {
626 role_uri
627 for role_uri in missing_by_work[br_uri]
628 if owners[role_uri] == br_uri
629 }
630 invalidate_missing = sorted(
631 role_uri
632 for role_uri in owned_missing
633 if provenance[role_uri] == "latest_snapshot_active"
634 )
635 work_entity = raw_entities[br_uri]
636 operation = _finalize_operation(
637 {
638 "work": {
639 "br": br_uri,
640 "title": _first(_literals(work_entity, DCTERMS_TITLE)),
641 "type": _local_type(work_entity),
642 "identifiers": work_identifiers,
643 },
644 "provider": {
645 "selected": selected_provider,
646 "attempted": attempted_providers,
647 "errors": provider_errors,
648 "author_count": len(metadata["author"])
649 if metadata is not None
650 else 0,
651 "editor_count": len(metadata["editor"])
652 if metadata is not None
653 else 0,
654 "publisher_present": bool(metadata["publisher"])
655 if metadata is not None
656 else False,
657 },
658 "blockers": blockers,
659 "actions": {
660 "remove_role_references": list(work.role_uris),
661 "delete_existing_ars": existing_roles,
662 "invalidate_missing_ars": invalidate_missing,
663 },
664 "preconditions": _operation_preconditions(
665 work,
666 role_entities,
667 raw_entities,
668 missing_by_work[br_uri],
669 contexts,
670 links,
671 owned_missing,
672 provenance,
673 ),
674 }
675 )
676 operations.append(operation)
677 if metadata is not None and selected_provider is not None and not blockers:
678 rows_by_provider[selected_provider].append(
679 work_csv_row(work, work_entity, work_identifiers, metadata)
680 )
681 progress.advance(task)
682 return operations, rows_by_provider
685def _write_csv(path: str, rows: list[dict[str, str]]) -> None:
686 _ensure_parent(path)
687 with open(path, "w", newline="", encoding="utf-8") as stream:
688 writer = csv.DictWriter(stream, fieldnames=FIELDNAMES)
689 writer.writeheader()
690 writer.writerows(rows)
693def write_provider_csvs(
694 output_dir: str, rows_by_provider: Mapping[str, list[dict[str, str]]]
695) -> dict[str, dict[str, object]]:
696 output_dir = os.path.abspath(output_dir)
697 parent = os.path.dirname(output_dir)
698 os.makedirs(parent, exist_ok=True)
699 staging_dir = tempfile.mkdtemp(
700 prefix=f".{os.path.basename(output_dir)}.", dir=parent
701 )
702 backup_dir = f"{staging_dir}.previous"
703 try:
704 for provider_name in PROVIDERS:
705 _write_csv(
706 os.path.join(staging_dir, provider_name, "input.csv"),
707 rows_by_provider[provider_name],
708 )
709 if os.path.exists(output_dir):
710 os.replace(output_dir, backup_dir)
711 try:
712 os.replace(staging_dir, output_dir)
713 except OSError:
714 if os.path.exists(backup_dir):
715 os.replace(backup_dir, output_dir)
716 raise
717 if os.path.exists(backup_dir):
718 shutil.rmtree(backup_dir)
719 finally:
720 if os.path.exists(staging_dir):
721 shutil.rmtree(staging_dir)
722 return {
723 provider_name: {
724 "path": os.path.join(output_dir, provider_name, "input.csv"),
725 "sha256": _sha256(os.path.join(output_dir, provider_name, "input.csv")),
726 "rows": len(rows_by_provider[provider_name]),
727 }
728 for provider_name in PROVIDERS
729 }
732def analyze_dangling_ars(
733 config_path: str,
734 report_path: str,
735 csv_output_dir: str,
736 cache_path: str,
737 mailto: str,
738 workers: int,
739 refresh_cache: bool,
740) -> dict[str, object]:
741 global _stop_requested
742 _stop_requested = False
743 config_path = os.path.abspath(config_path)
744 report_path = os.path.abspath(report_path)
745 csv_output_dir = os.path.abspath(csv_output_dir)
746 cache_path = os.path.abspath(cache_path)
747 config = load_audit_config(config_path)
748 works, role_entities, missing_by_work, contexts = find_dangling_works(
749 config, workers
750 )
751 target_roles = {role_uri for work in works.values() for role_uri in work.role_uris}
752 links = scan_role_links(config, target_roles, workers)
753 locator = EntityFileLocator(
754 config.rdf_dir, config.dir_split, config.items_per_file, config.zip_output
755 )
756 identifiers, raw_entities, provenance = _load_local_context(
757 works, role_entities, missing_by_work, locator, workers
758 )
759 _ensure_parent(cache_path)
760 api_cache = ApiCache(cache_path)
761 provider = AgentMetadataClient(
762 mailto=mailto, cache=api_cache, refresh_cache=refresh_cache
763 )
764 try:
765 operations, rows_by_provider = build_repair_plan(
766 works,
767 role_entities,
768 identifiers,
769 raw_entities,
770 missing_by_work,
771 contexts,
772 provenance,
773 links,
774 provider,
775 )
776 finally:
777 provider.close()
778 api_cache.close()
779 blockers = [
780 blocker
781 for operation in operations
782 for blocker in cast(list[dict[str, object]], operation["blockers"])
783 ]
784 complete = not _stop_requested and len(operations) == len(works)
785 executable = complete and not blockers
786 csv_files = (
787 write_provider_csvs(csv_output_dir, rows_by_provider) if executable else {}
788 )
789 report: dict[str, object] = {
790 "version": PLAN_VERSION,
791 "complete": complete,
792 "executable": executable,
793 "generated_at": datetime.now(timezone.utc).isoformat(),
794 "config": config_path,
795 "config_sha256": _sha256(config_path),
796 "rdf_dir": config.rdf_dir,
797 "api_cache": cache_path,
798 "csv_output_dir": csv_output_dir,
799 "csv_files": csv_files,
800 "operations_sha256": _object_sha256(operations),
801 "summary": {
802 "affected_brs": len(works),
803 "planned_operations": len(operations),
804 "dangling_ar_references": len(
805 {uri for missing in missing_by_work.values() for uri in missing}
806 ),
807 "existing_ars_to_delete": len(role_entities),
808 "blocker_counts": dict(
809 sorted(
810 Counter(cast(str, blocker["type"]) for blocker in blockers).items()
811 )
812 ),
813 "provider_counts": dict(
814 sorted(
815 Counter(
816 cast(
817 str,
818 cast(dict[str, object], operation["provider"])["selected"],
819 )
820 for operation in operations
821 if cast(dict[str, object], operation["provider"])["selected"]
822 is not None
823 ).items()
824 )
825 ),
826 },
827 "operations": operations,
828 }
829 _write_json(report_path, report)
830 return report
833def _plan_operations(plan: dict[str, object]) -> list[dict[str, object]]:
834 if "version" not in plan or plan["version"] != PLAN_VERSION:
835 version = plan["version"] if "version" in plan else "missing"
836 raise ValueError(
837 f"Unsupported correction plan version: {version}; expected {PLAN_VERSION}"
838 )
839 raw_operations = plan["operations"]
840 if not isinstance(raw_operations, list) or not all(
841 isinstance(operation, dict) for operation in raw_operations
842 ):
843 raise ValueError("Correction plan operations must be a list of objects")
844 operations = cast(list[dict[str, object]], raw_operations)
845 for operation in operations:
846 _verify_operation_id(operation)
847 if plan["operations_sha256"] != _object_sha256(operations):
848 raise ValueError("Correction plan operations were modified")
849 return operations
852def _verify_csv_files(plan: dict[str, object]) -> None:
853 csv_output_dir = plan["csv_output_dir"]
854 csv_files = plan["csv_files"]
855 if not isinstance(csv_output_dir, str) or not isinstance(csv_files, dict):
856 raise ValueError("Correction plan CSV metadata is invalid")
857 if set(csv_files) != set(PROVIDERS):
858 raise ValueError("Correction plan must contain Crossref and DataCite CSVs")
859 for provider_name in PROVIDERS:
860 entry = csv_files[provider_name]
861 if not isinstance(entry, dict):
862 raise ValueError(f"Invalid CSV metadata for {provider_name}")
863 expected_path = os.path.join(csv_output_dir, provider_name, "input.csv")
864 if entry["path"] != expected_path:
865 raise ValueError(f"Unexpected CSV path for {provider_name}")
866 if not os.path.isfile(expected_path):
867 raise ValueError(f"Planned CSV is missing: {expected_path}")
868 if entry["sha256"] != _sha256(expected_path):
869 raise ValueError(f"Planned CSV changed: {expected_path}")
872def _load_progress(path: str, plan_sha256: str, operations_sha256: str) -> set[str]:
873 if not os.path.exists(path):
874 return set()
875 progress = _read_json_object(path)
876 if progress["plan_sha256"] != plan_sha256:
877 raise ValueError("Progress file belongs to a different correction plan")
878 if progress["operations_sha256"] != operations_sha256:
879 raise ValueError("Progress file contains a different operation set")
880 completed = progress["completed_operations"]
881 if not isinstance(completed, list) or not all(
882 isinstance(operation_id, str) for operation_id in completed
883 ):
884 raise ValueError("Invalid completed_operations in progress file")
885 return set(cast(list[str], completed))
888def _save_progress(
889 path: str,
890 plan_sha256: str,
891 operations_sha256: str,
892 completed: set[str],
893) -> None:
894 _write_json(
895 path,
896 {
897 "plan_sha256": plan_sha256,
898 "operations_sha256": operations_sha256,
899 "completed_operations": sorted(completed),
900 },
901 )
904def _current_contexts(
905 config: AuditConfig, role_uris: set[str], workers: int
906) -> dict[str, tuple[str, ...]]:
907 if not role_uris:
908 return {}
909 br_files = _data_files(os.path.join(config.rdf_dir, "br"), config.zip_output)
910 return _scan_contexts(br_files, frozenset(role_uris), workers)
913def _capture_preconditions(
914 operation: dict[str, object],
915 config: AuditConfig,
916 contexts: Mapping[str, tuple[str, ...]],
917 links: set[tuple[str, str]],
918 completed_deleted_roles: set[str],
919 workers: int,
920) -> dict[str, object]:
921 work_summary = cast(dict[str, object], operation["work"])
922 br_uri = cast(str, work_summary["br"])
923 locator = EntityFileLocator(
924 config.rdf_dir, config.dir_split, config.items_per_file, config.zip_output
925 )
926 work_entities = load_available_entities({br_uri}, locator, workers)
927 if br_uri not in work_entities:
928 raise RuntimeError(f"Stale plan: bibliographic resource is missing: {br_uri}")
929 work_entity = work_entities[br_uri]
930 work = WorkRecord(
931 br_uri,
932 tuple(_ids(work_entity, IS_DOCUMENT_CONTEXT_FOR)),
933 tuple(_ids(work_entity, HAS_IDENTIFIER)),
934 )
935 role_entities = load_available_entities(set(work.role_uris), locator, workers)
936 missing = tuple(sorted(set(work.role_uris) - role_entities.keys()))
937 identifier_entities = load_available_entities(
938 set(work.identifier_uris), locator, workers
939 )
940 raw_entities = {br_uri: work_entity, **role_entities, **identifier_entities}
941 planned_preconditions = cast(dict[str, object], operation["preconditions"])
942 owned_provenance = cast(
943 dict[str, str], planned_preconditions["owned_missing_provenance"]
944 )
945 provenance = load_provenance_statuses(set(owned_provenance), locator, workers)
946 current = _operation_preconditions(
947 work,
948 role_entities,
949 raw_entities,
950 missing,
951 contexts,
952 links,
953 set(owned_provenance),
954 provenance,
955 )
956 current_roles = cast(dict[str, object], current["role_entities"])
957 planned_roles = cast(dict[str, object], planned_preconditions["role_entities"])
958 if completed_deleted_roles.intersection(current_roles):
959 raise RuntimeError(f"Stale plan: a deleted agent role reappeared for {br_uri}")
960 current["role_entities"] = {
961 uri: entity
962 for uri, entity in current_roles.items()
963 if uri not in completed_deleted_roles
964 }
965 expected = dict(planned_preconditions)
966 expected["role_entities"] = {
967 uri: entity
968 for uri, entity in planned_roles.items()
969 if uri not in completed_deleted_roles
970 }
971 expected["dangling_ar_references"] = sorted(
972 set(cast(list[str], expected["dangling_ar_references"]))
973 | completed_deleted_roles.intersection(work.role_uris)
974 )
975 current_edges = cast(list[list[str]], current["has_next_edges"])
976 planned_edges = cast(list[list[str]], expected["has_next_edges"])
977 new_edges_to_deleted_roles = {
978 tuple(edge)
979 for edge in current_edges
980 if completed_deleted_roles.intersection(edge)
981 } - {tuple(edge) for edge in planned_edges}
982 if new_edges_to_deleted_roles:
983 raise RuntimeError(
984 f"Stale plan: role links changed for {br_uri}: "
985 f"{sorted(new_edges_to_deleted_roles)}"
986 )
987 current["has_next_edges"] = [
988 edge for edge in current_edges if not completed_deleted_roles.intersection(edge)
989 ]
990 expected["has_next_edges"] = [
991 edge for edge in planned_edges if not completed_deleted_roles.intersection(edge)
992 ]
993 if current != expected:
994 raise RuntimeError(f"Stale plan: local RDF state changed for {br_uri}")
995 return current
998def _import_exact_entities(
999 editor: MetaEditor,
1000 g_set: GraphSet,
1001 uris: set[str],
1002 locator: EntityFileLocator,
1003) -> None:
1004 uris_by_path: dict[str, set[str]] = defaultdict(set)
1005 for uri in uris:
1006 uris_by_path[locator.path(uri)].add(uri)
1007 for path, path_uris in sorted(uris_by_path.items()):
1008 if not os.path.exists(path):
1009 continue
1010 graph = editor.reader.load(path)
1011 if graph is None:
1012 continue
1013 merged = TripleLite()
1014 for context in from_rdflib(graph):
1015 for triple in context.triples((None, None, None)):
1016 merged.add(triple)
1017 for uri in sorted(path_uris):
1018 preexisting = merged.subgraph(uri)
1019 if "/br/" in uri:
1020 g_set.add_br(editor.resp_agent, res=uri, preexisting_graph=preexisting)
1021 elif "/ar/" in uri:
1022 g_set.add_ar(editor.resp_agent, res=uri, preexisting_graph=preexisting)
1023 else:
1024 raise ValueError(f"Unsupported entity in correction operation: {uri}")
1027def _bibliographic_resource(g_set: GraphSet, uri: str) -> BibliographicResource:
1028 entity = g_set.get_entity(uri)
1029 if not isinstance(entity, BibliographicResource):
1030 raise ValueError(f"Bibliographic resource not imported: {uri}")
1031 return entity
1034def _apply_operation(
1035 editor: MetaEditor,
1036 operation: dict[str, object],
1037 locator: EntityFileLocator,
1038) -> None:
1039 work = cast(dict[str, object], operation["work"])
1040 br_uri = cast(str, work["br"])
1041 actions = cast(dict[str, object], operation["actions"])
1042 delete_existing = set(cast(list[str], actions["delete_existing_ars"]))
1043 g_set = GraphSet(
1044 editor.base_iri,
1045 supplier_prefix=editor.supplier_prefix,
1046 custom_counter_handler=editor.counter_handler,
1047 wanted_label=False,
1048 )
1049 _import_exact_entities(editor, g_set, {br_uri, *delete_existing}, locator)
1050 br = _bibliographic_resource(g_set, br_uri)
1051 for role_uri in cast(list[str], actions["remove_role_references"]):
1052 br.g.remove(
1053 (
1054 br.res,
1055 GraphEntity.iri_is_document_context_for,
1056 RDFTerm("uri", role_uri),
1057 )
1058 )
1059 for role_uri in sorted(delete_existing):
1060 role = g_set.get_entity(role_uri)
1061 if role is None:
1062 raise ValueError(f"Agent role not imported: {role_uri}")
1063 role.mark_as_to_be_deleted()
1064 for role_uri in cast(list[str], actions["invalidate_missing_ars"]):
1065 g_set.add_ar(editor.resp_agent, res=role_uri).mark_as_to_be_deleted()
1066 editor.save(g_set, editor.supplier_prefix)
1069def _write_reindex_sentinel(path: str, plan_path: str) -> None:
1070 with open(path, "w", encoding="utf-8") as stream:
1071 stream.write(
1072 f"{plan_path} changed local RDF files.\nRe-index the data and provenance "
1073 "triplestores from the RDF files, then delete this file. Run Meta for "
1074 "the Crossref and DataCite CSV directories only after this plan is "
1075 "complete.\n"
1076 )
1079def execute_plan(
1080 config_path: str,
1081 plan_path: str,
1082 resp_agent: str,
1083 progress_path: str,
1084 execution_report_path: str,
1085 workers: int,
1086) -> dict[str, object]:
1087 global _stop_requested
1088 _stop_requested = False
1089 config_path = os.path.abspath(config_path)
1090 plan_path = os.path.abspath(plan_path)
1091 plan = _read_json_object(plan_path)
1092 operations = _plan_operations(plan)
1093 if plan["complete"] is not True or plan["executable"] is not True:
1094 raise ValueError("The correction plan is not executable")
1095 if plan["config_sha256"] != _sha256(config_path):
1096 raise ValueError("The Meta configuration changed after plan generation")
1097 if any(cast(list[object], operation["blockers"]) for operation in operations):
1098 raise ValueError("An executable correction plan cannot contain blockers")
1099 _verify_csv_files(plan)
1100 plan_sha256 = _sha256(plan_path)
1101 operations_sha256 = cast(str, plan["operations_sha256"])
1102 completed = _load_progress(progress_path, plan_sha256, operations_sha256)
1103 operation_ids = {cast(str, operation["operation_id"]) for operation in operations}
1104 unknown_completed = completed - operation_ids
1105 if unknown_completed:
1106 raise ValueError(
1107 f"Progress file contains unknown operations: {sorted(unknown_completed)}"
1108 )
1109 sentinel_path = os.path.join(os.path.dirname(plan_path), REINDEX_SENTINEL_FILENAME)
1110 if os.path.exists(sentinel_path) and not os.path.exists(progress_path):
1111 raise RuntimeError(
1112 f"{sentinel_path} exists. Re-index the triplestores and remove the "
1113 "sentinel before starting another correction plan."
1114 )
1115 pending = [
1116 operation
1117 for operation in operations
1118 if cast(str, operation["operation_id"]) not in completed
1119 ]
1120 completed_deleted_roles = {
1121 role_uri
1122 for operation in operations
1123 if cast(str, operation["operation_id"]) in completed
1124 for role_uri in cast(
1125 list[str],
1126 cast(dict[str, object], operation["actions"])["delete_existing_ars"],
1127 )
1128 }
1129 pending_role_uris = {
1130 role_uri
1131 for operation in pending
1132 for role_uri in cast(
1133 list[str],
1134 cast(dict[str, object], operation["preconditions"])["role_references"],
1135 )
1136 }
1137 config = load_audit_config(config_path)
1138 contexts = _current_contexts(config, pending_role_uris, workers)
1139 links = scan_role_links(config, pending_role_uris, workers)
1140 for operation in pending:
1141 _capture_preconditions(
1142 operation,
1143 config,
1144 contexts,
1145 links,
1146 completed_deleted_roles,
1147 workers,
1148 )
1149 locator = EntityFileLocator(
1150 config.rdf_dir, config.dir_split, config.items_per_file, config.zip_output
1151 )
1152 attempted = 0
1153 if pending:
1154 _save_progress(progress_path, plan_sha256, operations_sha256, completed)
1155 editor = MetaEditor(config_path, resp_agent, save_queries=True)
1156 editor.rdf_files_only = True
1157 try:
1158 with create_progress() as progress:
1159 task = progress.add_task(
1160 "Applying correction plan", total=len(operations)
1161 )
1162 progress.advance(task, len(completed))
1163 for operation in pending:
1164 if _stop_requested:
1165 break
1166 attempted += 1
1167 _apply_operation(editor, operation, locator)
1168 completed.add(cast(str, operation["operation_id"]))
1169 _save_progress(
1170 progress_path, plan_sha256, operations_sha256, completed
1171 )
1172 progress.advance(task)
1173 finally:
1174 if attempted:
1175 _write_reindex_sentinel(sentinel_path, plan_path)
1176 complete = len(completed) == len(operations) and not _stop_requested
1177 execution_report: dict[str, object] = {
1178 "plan": plan_path,
1179 "plan_sha256": plan_sha256,
1180 "operations_sha256": operations_sha256,
1181 "generated_at": datetime.now(timezone.utc).isoformat(),
1182 "complete": complete,
1183 "planned_operations": len(operations),
1184 "completed_operations": sorted(completed),
1185 "reindex_sentinel": sentinel_path if os.path.exists(sentinel_path) else None,
1186 }
1187 _write_json(execution_report_path, execution_report)
1188 if complete and os.path.exists(progress_path):
1189 os.remove(progress_path)
1190 return execution_report
1193def _validate_uri(uri: str) -> None:
1194 if not uri.startswith(("http://", "https://")) or any(
1195 character in uri for character in "<> \t\r\n"
1196 ):
1197 raise ValueError(f"Invalid URI: {uri}")
1200def main() -> None: # pragma: no cover
1201 parser = argparse.ArgumentParser(
1202 description=(
1203 "Find bibliographic resources that reference missing agent roles, "
1204 "remove all their contributor roles, and prepare Crossref or DataCite "
1205 "CSV input for Meta."
1206 ),
1207 formatter_class=RichHelpFormatter,
1208 )
1209 parser.add_argument("-c", "--config", required=True, help="Meta YAML config")
1210 mode = parser.add_mutually_exclusive_group(required=True)
1211 mode.add_argument(
1212 "--dry-run", action="store_true", help="Generate a plan without changing RDF"
1213 )
1214 mode.add_argument("--execute", metavar="PLAN", help="Execute a complete plan")
1215 parser.add_argument("--report-file", help="Dry-run JSON plan path")
1216 parser.add_argument(
1217 "--csv-output-dir", help="Directory for Crossref and DataCite Meta CSVs"
1218 )
1219 parser.add_argument("--cache-file", help="SQLite API cache path")
1220 parser.add_argument("--mailto", help="Contact email sent to metadata APIs")
1221 parser.add_argument(
1222 "--refresh-cache", action="store_true", help="Refresh cached API responses"
1223 )
1224 parser.add_argument(
1225 "--workers",
1226 type=int,
1227 default=min(os.cpu_count() or 1, 16),
1228 help="Processes used for local RDF scanning",
1229 )
1230 parser.add_argument("-r", "--resp-agent", help="Provenance responsible-agent URI")
1231 parser.add_argument("--progress-file", help="Execution progress JSON path")
1232 parser.add_argument("--execution-report", help="Execution result JSON path")
1233 args = parser.parse_args()
1234 if args.workers < 1:
1235 parser.error("--workers must be positive")
1236 signal.signal(signal.SIGINT, _handle_signal)
1237 signal.signal(signal.SIGTERM, _handle_signal)
1239 if args.dry_run:
1240 if not args.report_file or not args.csv_output_dir or not args.mailto:
1241 parser.error(
1242 "--report-file, --csv-output-dir, and --mailto are required with --dry-run"
1243 )
1244 cache_path = args.cache_file or f"{args.report_file}.cache.sqlite"
1245 report = analyze_dangling_ars(
1246 config_path=args.config,
1247 report_path=args.report_file,
1248 csv_output_dir=args.csv_output_dir,
1249 cache_path=cache_path,
1250 mailto=args.mailto,
1251 workers=args.workers,
1252 refresh_cache=args.refresh_cache,
1253 )
1254 summary = cast(dict[str, object], report["summary"])
1255 console.print(
1256 f"Plan written to [cyan]{os.path.abspath(args.report_file)}[/cyan]. "
1257 f"Affected BRs: [cyan]{summary['affected_brs']}[/cyan]; dangling AR "
1258 f"references: [cyan]{summary['dangling_ar_references']}[/cyan]."
1259 )
1260 if report["executable"] is not True:
1261 raise SystemExit(1)
1262 return
1264 if not args.resp_agent:
1265 parser.error("--resp-agent is required with --execute")
1266 _validate_uri(args.resp_agent)
1267 plan_path = os.path.abspath(cast(str, args.execute))
1268 progress_path = os.path.abspath(args.progress_file or f"{plan_path}.progress.json")
1269 execution_report_path = os.path.abspath(
1270 args.execution_report or f"{plan_path}.execution.json"
1271 )
1272 result = execute_plan(
1273 config_path=args.config,
1274 plan_path=plan_path,
1275 resp_agent=args.resp_agent,
1276 progress_path=progress_path,
1277 execution_report_path=execution_report_path,
1278 workers=args.workers,
1279 )
1280 console.print(
1281 f"Execution report written to [cyan]{execution_report_path}[/cyan]. "
1282 f"Completed operations: "
1283 f"[cyan]{len(cast(list[str], result['completed_operations']))}[/cyan]."
1284 )
1285 if result["complete"] is not True:
1286 raise SystemExit(1)
1289if __name__ == "__main__":
1290 main()