Coverage for oc_meta / run / patches / fix_duplicate_ras.py: 85%
935 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 os
11import signal
12from collections import Counter, defaultdict
13from collections.abc import Iterator
14from concurrent.futures import ThreadPoolExecutor
15from dataclasses import asdict, dataclass
16from datetime import datetime, timezone
17from itertools import combinations
18from typing import Protocol, cast
20import orjson
21from oc_ocdm.graph import GraphSet
22from oc_ocdm.graph.entities.identifier import Identifier
23from rich_argparse import RichHelpFormatter
25from oc_meta.core.editor import MetaEditor
26from oc_meta.lib.agent_matching import (
27 AlignmentResult,
28 PersonName,
29 align_names,
30 name_score,
31 normalize_name,
32 script_family,
33)
34from oc_meta.lib.agent_metadata import (
35 AgentMetadata,
36 AgentMetadataClient,
37 ApiCache,
38 OrcidProfile,
39 WorkMetadata,
40 agents_for_role,
41 is_valid_orcid,
42 normalize_orcid,
43)
44from oc_meta.lib.console import console, create_progress
45from oc_meta.lib.rdf_patch import (
46 DATACITE_PREFIX,
47 FAMILY_NAME,
48 FOAF_NAME,
49 GIVEN_NAME,
50 HAS_IDENTIFIER,
51 HAS_LITERAL_VALUE,
52 HAS_NEXT,
53 IS_DOCUMENT_CONTEXT_FOR,
54 IS_HELD_BY,
55 PROV_SPECIALIZATION_OF,
56 ROLE_MAP,
57 USES_IDENTIFIER_SCHEME,
58 WITH_ROLE,
59 EntityFileLocator,
60 agent_role as _agent_role,
61 batches as _batches,
62 data_files as _data_files,
63 ensure_parent as _ensure_parent,
64 first as _first,
65 identifier as _identifier,
66 ids as _ids,
67 literals as _literals,
68 load_audit_config,
69 load_available_entities,
70 load_entities as _load_entities,
71 load_progress as _load_progress,
72 provenance_path as _provenance_path,
73 read_json_object as _read_json_object,
74 responsible_agent as _responsible_agent,
75 save_progress as _save_progress,
76 sha256 as _sha256,
77 snapshot_number as _snapshot_number,
78 write_json as _write_json,
79)
80from oc_meta.lib.sparql import execute_sparql
81from oc_meta.run.merge.entities import REINDEX_SENTINEL_FILENAME
83PROV_GENERATED_AT_TIME = "http://www.w3.org/ns/prov#generatedAtTime"
84PROV_WAS_ATTRIBUTED_TO = "http://www.w3.org/ns/prov#wasAttributedTo"
85PROV_HAD_PRIMARY_SOURCE = "http://www.w3.org/ns/prov#hadPrimarySource"
86DCTERMS_DESCRIPTION = "http://purl.org/dc/terms/description"
87HAS_UPDATE_QUERY = "https://w3id.org/oc/ontology/hasUpdateQuery"
88CONFIRMED_NAME_SCORE = 0.9
89AMBIGUOUS_NAME_SCORE = 0.75
90PLAN_SCHEMA_VERSION = 1
91CLUSTER_BATCH_SIZE = 5000
92REVIEW_FIELDS: tuple[str, ...] = (
93 "operation_id",
94 "csv_row",
95 "br",
96 "ar",
97 "ra",
98 "action",
99 "identifier_uri",
100 "old_value",
101 "new_value",
102 "confidence",
103 "reason",
104 "decision",
105)
107_stop_requested = False
110@dataclass(frozen=True, slots=True)
111class Cluster:
112 csv_row: int
113 survivor: str
114 members: tuple[str, ...]
117@dataclass(frozen=True, slots=True)
118class IdentifierInfo:
119 uri: str
120 scheme: str
121 value: str
124@dataclass(frozen=True, slots=True)
125class AgentInfo:
126 uri: str
127 name: PersonName
128 identifiers: tuple[IdentifierInfo, ...]
130 @property
131 def orcids(self) -> tuple[IdentifierInfo, ...]:
132 return tuple(
133 identifier
134 for identifier in self.identifiers
135 if identifier.scheme == "orcid"
136 )
139@dataclass(frozen=True, slots=True)
140class RoleInfo:
141 uri: str
142 ra: str
143 role: str
144 next_uris: tuple[str, ...]
145 holder_uris: tuple[str, ...] = ()
148@dataclass(frozen=True, slots=True)
149class WorkInfo:
150 uri: str
151 identifiers: tuple[IdentifierInfo, ...]
152 role_uris: tuple[str, ...]
154 def identifier(self, scheme: str) -> str:
155 identifier = self.identifier_info(scheme)
156 return identifier.value if identifier is not None else ""
158 def identifier_info(self, scheme: str) -> IdentifierInfo | None:
159 for identifier in self.identifiers:
160 if identifier.scheme == scheme:
161 return identifier
162 return None
165@dataclass(frozen=True, slots=True)
166class OrderedChain:
167 status: str
168 roles: tuple[RoleInfo, ...]
171@dataclass(frozen=True, slots=True)
172class WorkEvidence:
173 br: str
174 ar: str
175 next_uri: str
176 work_identifier_uri: str
177 work_identifier_scheme: str
178 work_identifier_value: str
179 role: str
180 source: str
181 matched: bool
182 name_score: float
183 api_orcid: str | None
184 api_name: str
185 contested_elsewhere: bool
188class WorkEvidenceClient(Protocol):
189 def work_sources(self, doi: str, openalex_id: str = "") -> list[WorkMetadata]: ...
192class OrcidClient(Protocol):
193 def orcid(self, orcid: str) -> OrcidProfile | None: ...
196def _handle_signal(signum: int, frame: object) -> None:
197 del signum, frame
198 global _stop_requested
199 _stop_requested = True
202def _entity_name(entity: dict[str, object]) -> PersonName:
203 return PersonName(
204 name=_first(_literals(entity, FOAF_NAME)),
205 given=_first(_literals(entity, GIVEN_NAME)),
206 family=_first(_literals(entity, FAMILY_NAME)),
207 )
210def iter_cluster_batches(
211 path: str, batch_size: int = CLUSTER_BATCH_SIZE
212) -> Iterator[list[Cluster]]:
213 csv.field_size_limit(1024 * 1024 * 1024)
214 with open(path, newline="", encoding="utf-8") as stream:
215 reader = csv.DictReader(stream)
216 if reader.fieldnames != ["surviving_entity", "merged_entities"]:
217 raise ValueError(f"Unexpected duplicate CSV header: {reader.fieldnames}")
218 batch = []
219 for csv_row, row in enumerate(reader, 2):
220 merged = tuple(
221 item.strip()
222 for item in row["merged_entities"].split(";")
223 if item.strip()
224 )
225 members = (row["surviving_entity"].strip(), *merged)
226 if not members[0] or len(members) < 2:
227 raise ValueError(f"Invalid duplicate cluster at CSV row {csv_row}")
228 if len(set(members)) != len(members):
229 raise ValueError(f"Repeated entity at CSV row {csv_row}")
230 batch.append(Cluster(csv_row, members[0], members))
231 if len(batch) == batch_size:
232 yield batch
233 batch = []
234 if batch:
235 yield batch
238def _count_duplicate_clusters(path: str) -> int:
239 with open(path, "rb") as stream:
240 next(stream, None)
241 return sum(1 for _ in stream)
244def load_target_entities(
245 uris: set[str], cache: EntityFileLocator, workers: int
246) -> dict[str, dict[str, object]]:
247 result = load_available_entities(uris, cache, workers)
248 missing = uris - result.keys()
249 if missing:
250 examples = sorted(missing)[:10]
251 raise ValueError(f"RDF entities not found: {examples} ({len(missing)} total)")
252 return result
255def _identifier_info(
256 uri: str, entities: dict[str, dict[str, object]]
257) -> IdentifierInfo | None:
258 entity = entities.get(uri)
259 if entity is None:
260 return None
261 scheme_uri = _first(_ids(entity, USES_IDENTIFIER_SCHEME))
262 value = _first(_literals(entity, HAS_LITERAL_VALUE))
263 if not scheme_uri or not value:
264 return None
265 scheme = (
266 scheme_uri[len(DATACITE_PREFIX) :]
267 if scheme_uri.startswith(DATACITE_PREFIX)
268 else scheme_uri
269 )
270 return IdentifierInfo(uri, scheme, value)
273def load_agents(
274 uris: set[str], cache: EntityFileLocator, workers: int
275) -> dict[str, AgentInfo]:
276 agent_entities = load_target_entities(uris, cache, workers)
277 identifier_uris = {
278 identifier
279 for entity in agent_entities.values()
280 for identifier in _ids(entity, HAS_IDENTIFIER)
281 }
282 identifier_entities = load_target_entities(identifier_uris, cache, workers)
283 agents = {}
284 for uri, entity in agent_entities.items():
285 identifiers = tuple(
286 identifier
287 for identifier_uri in _ids(entity, HAS_IDENTIFIER)
288 if (identifier := _identifier_info(identifier_uri, identifier_entities))
289 is not None
290 )
291 agents[uri] = AgentInfo(uri, _entity_name(entity), identifiers)
292 return agents
295def scan_candidate_clusters(
296 duplicate_path: str,
297 cache: EntityFileLocator,
298 workers: int,
299 all_api: bool,
300) -> tuple[list[Cluster], dict[str, AgentInfo], dict[int, list[str]], int, int]:
301 candidates = []
302 candidate_agents = {}
303 risks_by_row = {}
304 cluster_count = 0
305 agent_count = 0
306 total_clusters = _count_duplicate_clusters(duplicate_path)
307 with create_progress() as progress:
308 task = progress.add_task("Checking duplicate clusters", total=total_clusters)
309 for cluster_batch in iter_cluster_batches(duplicate_path):
310 if _stop_requested:
311 break
312 uris = {member for cluster in cluster_batch for member in cluster.members}
313 agents = load_agents(uris, cache, workers)
314 for cluster in cluster_batch:
315 risks = cluster_risks(cluster, agents)
316 cluster_count += 1
317 agent_count += len(cluster.members)
318 if not risks and not all_api:
319 continue
320 candidates.append(cluster)
321 risks_by_row[cluster.csv_row] = risks
322 candidate_agents.update(
323 (member, agents[member]) for member in cluster.members
324 )
325 progress.advance(task, len(cluster_batch))
326 return (
327 candidates,
328 candidate_agents,
329 risks_by_row,
330 cluster_count,
331 agent_count,
332 )
335def _load_provenance_batch(
336 paths: list[tuple[str, frozenset[str]]],
337) -> dict[str, dict[str, object]]:
338 snapshots_by_entity: dict[str, list[dict[str, object]]] = defaultdict(list)
339 for path, targets in paths:
340 if not os.path.exists(path):
341 continue
342 for snapshot in _load_entities(path).values():
343 specializations = _ids(snapshot, PROV_SPECIALIZATION_OF)
344 if not specializations or specializations[0] not in targets:
345 continue
346 snapshots_by_entity[specializations[0]].append(snapshot)
348 result = {}
349 for uri, snapshots in snapshots_by_entity.items():
350 snapshots.sort(
351 key=lambda snapshot: _snapshot_number(cast(str, snapshot["@id"]))
352 )
353 first = snapshots[0]
354 latest = snapshots[-1]
355 result[uri] = {
356 "snapshot_count": len(snapshots),
357 "created_at": _first(_literals(first, PROV_GENERATED_AT_TIME)),
358 "latest_at": _first(_literals(latest, PROV_GENERATED_AT_TIME)),
359 "latest_snapshot": cast(str, latest["@id"]),
360 "attributed_to": _ids(latest, PROV_WAS_ATTRIBUTED_TO),
361 "primary_sources": _ids(latest, PROV_HAD_PRIMARY_SOURCE),
362 "description": _first(_literals(latest, DCTERMS_DESCRIPTION)),
363 "update_query": _first(_literals(latest, HAS_UPDATE_QUERY)),
364 }
365 return result
368def load_provenance(
369 uris: set[str], cache: EntityFileLocator, workers: int
370) -> dict[str, dict[str, object]]:
371 targets_by_path: dict[str, set[str]] = defaultdict(set)
372 for uri in uris:
373 path = _provenance_path(cache.path(uri), cache.zip_output)
374 targets_by_path[path].add(uri)
375 tasks = [(path, frozenset(targets)) for path, targets in targets_by_path.items()]
376 result = {}
377 with ThreadPoolExecutor(max_workers=workers) as executor:
378 for partial in executor.map(_load_provenance_batch, _batches(tasks, 24)):
379 result.update(partial)
380 return result
383def _has_conflicting_names(names: list[PersonName]) -> bool:
384 return any(
385 name_score(left, right) < CONFIRMED_NAME_SCORE
386 for left, right in combinations(names, 2)
387 )
390def cluster_risks(cluster: Cluster, agents: dict[str, AgentInfo]) -> list[str]:
391 names = [agents[uri].name for uri in cluster.members]
392 orcids = {
393 normalize_orcid(identifier.value)
394 for uri in cluster.members
395 for identifier in agents[uri].orcids
396 }
397 risks = []
398 if _has_conflicting_names(names):
399 risks.append("conflicting_names")
400 if len(orcids) > 1:
401 risks.append("multiple_orcids")
402 if any(len(agents[uri].orcids) > 1 for uri in cluster.members):
403 risks.append("bridge_agent")
404 if len(cluster.members) >= 50:
405 risks.append("large_cluster")
406 if any(not name.display for name in names):
407 risks.append("missing_name")
408 return risks
411def _scan_roles_batch(
412 paths: list[str], target_ras: frozenset[str] | None
413) -> dict[str, RoleInfo]:
414 result = {}
415 for path in paths:
416 for uri, entity in _load_entities(path).items():
417 ras = _ids(entity, IS_HELD_BY)
418 if not ras or target_ras is not None and target_ras.isdisjoint(ras):
419 continue
420 roles = _ids(entity, WITH_ROLE)
421 result[uri] = RoleInfo(
422 uri=uri,
423 ra=ras[0],
424 role=ROLE_MAP.get(_first(roles), "unknown"),
425 next_uris=tuple(_ids(entity, HAS_NEXT)),
426 holder_uris=tuple(ras),
427 )
428 return result
431def scan_roles(
432 files: list[str], target_ras: set[str] | None, workers: int
433) -> dict[str, RoleInfo]:
434 targets = frozenset(target_ras) if target_ras is not None else None
435 result = {}
436 with ThreadPoolExecutor(max_workers=workers) as executor:
437 for partial in executor.map(
438 lambda paths: _scan_roles_batch(paths, targets), _batches(files, 24)
439 ):
440 result.update(partial)
441 if _stop_requested:
442 break
443 return result
446def _scan_works_batch(
447 paths: list[str], target_roles: frozenset[str]
448) -> dict[str, tuple[tuple[str, ...], tuple[str, ...]]]:
449 result = {}
450 for path in paths:
451 for uri, entity in _load_entities(path).items():
452 roles = tuple(_ids(entity, IS_DOCUMENT_CONTEXT_FOR))
453 if target_roles.isdisjoint(roles):
454 continue
455 result[uri] = (roles, tuple(_ids(entity, HAS_IDENTIFIER)))
456 return result
459def scan_works(
460 files: list[str], target_roles: set[str], workers: int
461) -> dict[str, tuple[tuple[str, ...], tuple[str, ...]]]:
462 targets = frozenset(target_roles)
463 result = {}
464 with ThreadPoolExecutor(max_workers=workers) as executor:
465 for partial in executor.map(
466 lambda paths: _scan_works_batch(paths, targets), _batches(files, 24)
467 ):
468 result.update(partial)
469 if _stop_requested:
470 break
471 return result
474def ordered_chain(roles: list[RoleInfo]) -> OrderedChain:
475 if not roles:
476 return OrderedChain("empty", ())
477 by_uri = {role.uri: role for role in roles}
478 if any(
479 len(role.holder_uris or ((role.ra,) if role.ra else ())) != 1 for role in roles
480 ):
481 return OrderedChain("multiple_or_missing_holders", tuple(roles))
482 if any(len(role.next_uris) > 1 for role in roles):
483 return OrderedChain("fork", tuple(roles))
484 if any(next_uri not in by_uri for role in roles for next_uri in role.next_uris):
485 return OrderedChain("dangling_or_cross_role", tuple(roles))
486 targets = {
487 next_uri for role in roles for next_uri in role.next_uris if next_uri in by_uri
488 }
489 starts = [role for role in roles if role.uri not in targets]
490 if len(starts) != 1:
491 return OrderedChain("cycle_or_multiple_heads", tuple(roles))
492 ordered = []
493 seen = set()
494 current = starts[0]
495 while current.uri not in seen:
496 seen.add(current.uri)
497 ordered.append(current)
498 next_uris = [uri for uri in current.next_uris if uri in by_uri]
499 if not next_uris:
500 break
501 current = by_uri[next_uris[0]]
502 if len(ordered) != len(roles):
503 return OrderedChain("disconnected_or_cycle", tuple(roles))
504 return OrderedChain("valid", tuple(ordered))
507def _agent_metadata_name(agent: AgentMetadata) -> PersonName:
508 return PersonName(name=agent["name"], given=agent["given"], family=agent["family"])
511def _alignment_dict(alignment: AlignmentResult) -> dict[int, tuple[int, float]]:
512 return {
513 pair.local_index: (pair.external_index, pair.score) for pair in alignment.pairs
514 }
517def _operation_id(action: str, *parts: str) -> str:
518 content = "|".join((action, *parts)).encode()
519 return hashlib.sha256(content).hexdigest()[:20]
522def _operation(
523 action: str,
524 csv_row: int,
525 reason: str,
526 confidence: float,
527 *,
528 br: str = "",
529 ar: str = "",
530 ra: str = "",
531 identifier_uri: str = "",
532 old_value: str = "",
533 new_value: str = "",
534 links: list[dict[str, str]] | None = None,
535 evidence: list[dict[str, str]] | None = None,
536) -> dict[str, object]:
537 link_value = (
538 orjson.dumps(links, option=orjson.OPT_SORT_KEYS).decode()
539 if links is not None
540 else ""
541 )
542 evidence_value = (
543 orjson.dumps(evidence, option=orjson.OPT_SORT_KEYS).decode()
544 if evidence is not None
545 else ""
546 )
547 operation_id = _operation_id(
548 action,
549 br,
550 ar,
551 ra,
552 identifier_uri,
553 old_value,
554 new_value,
555 link_value,
556 evidence_value,
557 )
558 result: dict[str, object] = {
559 "operation_id": operation_id,
560 "csv_row": csv_row,
561 "action": action,
562 "br": br,
563 "ar": ar,
564 "ra": ra,
565 "identifier_uri": identifier_uri,
566 "old_value": old_value,
567 "new_value": new_value,
568 "confidence": round(confidence, 3),
569 "reason": reason,
570 "approved": False,
571 }
572 if links is not None:
573 result["links"] = links
574 if evidence is not None:
575 result["evidence"] = evidence
576 return result
579def _computed_operation_id(operation: dict[str, object]) -> str:
580 links = operation["links"] if "links" in operation else None
581 evidence = operation["evidence"] if "evidence" in operation else None
582 link_value = (
583 orjson.dumps(links, option=orjson.OPT_SORT_KEYS).decode()
584 if links is not None
585 else ""
586 )
587 evidence_value = (
588 orjson.dumps(evidence, option=orjson.OPT_SORT_KEYS).decode()
589 if evidence is not None
590 else ""
591 )
592 return _operation_id(
593 *(cast(str, operation[field]) for field in ("action", "br", "ar", "ra")),
594 cast(str, operation["identifier_uri"]),
595 cast(str, operation["old_value"]),
596 cast(str, operation["new_value"]),
597 link_value,
598 evidence_value,
599 )
602def build_context(
603 candidate_ras: set[str],
604 rdf_dir: str,
605 zip_output: bool,
606 cache: EntityFileLocator,
607 workers: int,
608 max_evidence_works: int,
609) -> tuple[dict[str, WorkInfo], dict[str, RoleInfo], dict[str, AgentInfo]]:
610 ar_files = _data_files(os.path.join(rdf_dir, "ar"), zip_output)
611 candidate_roles = scan_roles(ar_files, candidate_ras, workers)
612 br_files = _data_files(os.path.join(rdf_dir, "br"), zip_output)
613 raw_works = scan_works(br_files, set(candidate_roles), workers)
615 contexts: dict[str, list[str]] = defaultdict(list)
616 for work_uri, (role_refs, _) in raw_works.items():
617 for role_uri in role_refs:
618 role = candidate_roles.get(role_uri)
619 if role is not None:
620 for holder_uri in role.holder_uris or (role.ra,):
621 if holder_uri in candidate_ras:
622 contexts[holder_uri].append(work_uri)
623 selected_work_uris = {
624 work_uri
625 for ra_uri in candidate_ras
626 for work_uri in sorted(set(contexts[ra_uri]))[:max_evidence_works]
627 }
628 raw_works = {
629 work_uri: data
630 for work_uri, data in raw_works.items()
631 if work_uri in selected_work_uris
632 }
634 role_uris = {
635 role_uri for role_refs, _ in raw_works.values() for role_uri in role_refs
636 }
637 role_entities = load_target_entities(role_uris, cache, workers)
638 roles = {}
639 for uri, entity in role_entities.items():
640 ras = _ids(entity, IS_HELD_BY)
641 role_types = _ids(entity, WITH_ROLE)
642 roles[uri] = RoleInfo(
643 uri=uri,
644 ra=_first(ras),
645 role=ROLE_MAP.get(_first(role_types), "unknown"),
646 next_uris=tuple(_ids(entity, HAS_NEXT)),
647 holder_uris=tuple(ras),
648 )
650 identifier_uris = {
651 identifier_uri
652 for _, identifiers in raw_works.values()
653 for identifier_uri in identifiers
654 }
655 identifier_entities = load_target_entities(identifier_uris, cache, workers)
656 works = {}
657 for uri, (role_refs, identifier_refs) in raw_works.items():
658 identifiers = tuple(
659 identifier
660 for identifier_uri in identifier_refs
661 if (identifier := _identifier_info(identifier_uri, identifier_entities))
662 is not None
663 )
664 works[uri] = WorkInfo(uri, identifiers, role_refs)
666 ra_uris = {role.ra for role in roles.values() if role.ra}
667 agents = load_agents(ra_uris, cache, workers)
668 return works, roles, agents
671def _role_chains(work: WorkInfo, roles: dict[str, RoleInfo]) -> dict[str, OrderedChain]:
672 grouped: dict[str, list[RoleInfo]] = defaultdict(list)
673 for role_uri in work.role_uris:
674 role = roles.get(role_uri)
675 if role is not None:
676 grouped[role.role].append(role)
677 return {role: ordered_chain(members) for role, members in grouped.items()}
680def _selected_work_uris(
681 candidate_ras: set[str],
682 works: dict[str, WorkInfo],
683 roles: dict[str, RoleInfo],
684 max_evidence_works: int,
685) -> set[str]:
686 contexts: dict[str, list[str]] = defaultdict(list)
687 for work in works.values():
688 for role_uri in work.role_uris:
689 role = roles.get(role_uri)
690 if role is not None:
691 for holder_uri in role.holder_uris or (role.ra,):
692 if holder_uri in candidate_ras:
693 contexts[holder_uri].append(work.uri)
694 selected = set()
695 for ra in candidate_ras:
696 selected.update(sorted(set(contexts[ra]))[:max_evidence_works])
697 return selected
700def _alignment_report(
701 chain: OrderedChain,
702 external: list[AgentMetadata],
703 agents: dict[str, AgentInfo],
704) -> tuple[AlignmentResult | None, dict[str, object]]:
705 if chain.status != "valid":
706 return None, {
707 "chain_status": chain.status,
708 "ambiguous": True,
709 "pairs": [],
710 "unmatched_local": [role.uri for role in chain.roles],
711 "unmatched_external": list(range(len(external))),
712 }
713 local_names = [agents[role.ra].name for role in chain.roles]
714 alignment = align_names(
715 local_names, [_agent_metadata_name(agent) for agent in external]
716 )
717 return alignment, {
718 "chain_status": chain.status,
719 "ambiguous": alignment.ambiguous,
720 "pairs": [
721 {
722 "ar": chain.roles[pair.local_index].uri,
723 "ra": chain.roles[pair.local_index].ra,
724 "external_position": pair.external_index,
725 "score": round(pair.score, 3),
726 }
727 for pair in alignment.pairs
728 ],
729 "unmatched_local": [
730 chain.roles[index].uri for index in alignment.unmatched_local
731 ],
732 "unmatched_external": list(alignment.unmatched_external),
733 }
736def _desired_role_order(
737 chain: OrderedChain,
738 external: list[AgentMetadata],
739 agents: dict[str, AgentInfo],
740) -> list[RoleInfo] | None:
741 if chain.status != "valid" or len(chain.roles) != len(external):
742 return None
743 available = set(range(len(chain.roles)))
744 desired = []
745 for external_agent in external:
746 scored = sorted(
747 (
748 (
749 name_score(
750 agents[chain.roles[index].ra].name,
751 _agent_metadata_name(external_agent),
752 ),
753 index,
754 )
755 for index in available
756 ),
757 reverse=True,
758 )
759 if not scored or scored[0][0] < CONFIRMED_NAME_SCORE:
760 return None
761 if len(scored) > 1 and abs(scored[0][0] - scored[1][0]) < 1e-9:
762 return None
763 _, index = scored[0]
764 desired.append(chain.roles[index])
765 available.remove(index)
766 return desired
769def _chain_links(roles: list[RoleInfo] | tuple[RoleInfo, ...]) -> list[dict[str, str]]:
770 return [
771 {
772 "ar": role.uri,
773 "old_next": _first(list(role.next_uris)),
774 "new_next": roles[index + 1].uri if index + 1 < len(roles) else "",
775 }
776 for index, role in enumerate(roles)
777 ]
780def _role_operations(
781 cluster_by_ra: dict[str, Cluster],
782 work: WorkInfo,
783 chain: OrderedChain,
784 external: list[AgentMetadata],
785 agents: dict[str, AgentInfo],
786 source: str,
787) -> list[dict[str, object]]:
788 operations = []
789 desired = _desired_role_order(chain, external, agents)
790 if desired is not None and [role.uri for role in desired] != [
791 role.uri for role in chain.roles
792 ]:
793 csv_rows = {
794 cluster_by_ra[role.ra].csv_row
795 for role in chain.roles
796 if role.ra in cluster_by_ra
797 }
798 operations.append(
799 _operation(
800 "reorder_chain",
801 min(csv_rows) if csv_rows else 0,
802 f"{source} confirms the same contributors in a different order",
803 CONFIRMED_NAME_SCORE,
804 br=work.uri,
805 links=_chain_links(desired),
806 )
807 )
808 return operations
810 if chain.status != "valid":
811 return operations
812 for position, role in enumerate(chain.roles[: len(external)]):
813 current_score = name_score(
814 agents[role.ra].name, _agent_metadata_name(external[position])
815 )
816 if current_score >= AMBIGUOUS_NAME_SCORE or role.ra not in cluster_by_ra:
817 continue
818 cluster = cluster_by_ra[role.ra]
819 candidates = sorted(
820 (
821 (
822 name_score(
823 agents[member].name, _agent_metadata_name(external[position])
824 ),
825 member,
826 )
827 for member in cluster.members
828 if member in agents and member != role.ra
829 ),
830 reverse=True,
831 )
832 if not candidates or candidates[0][0] < CONFIRMED_NAME_SCORE:
833 continue
834 if len(candidates) > 1 and abs(candidates[0][0] - candidates[1][0]) < 1e-9:
835 continue
836 score, candidate = candidates[0]
837 operations.append(
838 _operation(
839 "reassign_role",
840 cluster.csv_row,
841 f"{source} contributor at position {position} matches {candidate}",
842 score,
843 br=work.uri,
844 ar=role.uri,
845 ra=role.ra,
846 old_value=role.ra,
847 new_value=candidate,
848 )
849 )
850 return operations
853def collect_external_evidence(
854 selected_works: set[str],
855 works: dict[str, WorkInfo],
856 roles: dict[str, RoleInfo],
857 agents: dict[str, AgentInfo],
858 cluster_by_ra: dict[str, Cluster],
859 client: WorkEvidenceClient,
860) -> tuple[
861 list[dict[str, object]],
862 dict[tuple[str, str], list[WorkEvidence]],
863 dict[str, list[PersonName]],
864 list[dict[str, object]],
865]:
866 role_assessments = []
867 edge_evidence: dict[tuple[str, str], list[WorkEvidence]] = defaultdict(list)
868 names_by_orcid: dict[str, list[PersonName]] = defaultdict(list)
869 operations_by_id: dict[str, dict[str, object]] = {}
871 with create_progress() as progress:
872 task = progress.add_task("Querying work metadata", total=len(selected_works))
873 for br_uri in sorted(selected_works):
874 if _stop_requested:
875 break
876 work = works[br_uri]
877 doi = work.identifier("doi")
878 openalex_id = work.identifier("openalex")
879 sources = client.work_sources(doi, openalex_id)
880 chains = _role_chains(work, roles)
881 if not sources:
882 role_assessments.extend(
883 {
884 "br": br_uri,
885 "role": role_name,
886 "source": "",
887 "chain_status": chain.status,
888 "ambiguous": True,
889 "reason": "no_external_work_metadata",
890 "pairs": [],
891 "unmatched_local": [role.uri for role in chain.roles],
892 "unmatched_external": [],
893 }
894 for role_name, chain in chains.items()
895 )
896 for source_work in sources:
897 work_identifier_scheme = (
898 "openalex"
899 if source_work["source"] == "openalex" and openalex_id
900 else "doi"
901 )
902 work_identifier = work.identifier_info(work_identifier_scheme)
903 if work_identifier is None:
904 raise ValueError(
905 f"{source_work['source']} evidence for {br_uri} has no local "
906 f"{work_identifier_scheme} identifier"
907 )
908 for role_name, chain in chains.items():
909 external = agents_for_role(source_work, role_name)
910 if not external:
911 role_assessments.append(
912 {
913 "br": br_uri,
914 "role": role_name,
915 "source": source_work["source"],
916 "chain_status": chain.status,
917 "ambiguous": True,
918 "reason": "external_role_missing",
919 "pairs": [],
920 "unmatched_local": [role.uri for role in chain.roles],
921 "unmatched_external": [],
922 }
923 )
924 continue
925 alignment, report = _alignment_report(chain, external, agents)
926 report.update(
927 {
928 "br": br_uri,
929 "role": role_name,
930 "source": source_work["source"],
931 }
932 )
933 role_assessments.append(report)
934 for operation in _role_operations(
935 cluster_by_ra,
936 work,
937 chain,
938 external,
939 agents,
940 source_work["source"],
941 ):
942 operations_by_id[cast(str, operation["operation_id"])] = (
943 operation
944 )
945 if alignment is None:
946 continue
947 matched = _alignment_dict(alignment)
948 for local_index, role in enumerate(chain.roles):
949 if role.ra not in cluster_by_ra:
950 continue
951 for identifier in agents[role.ra].orcids:
952 normalized = normalize_orcid(identifier.value)
953 match = matched.get(local_index)
954 if match is None:
955 evidence = WorkEvidence(
956 br_uri,
957 role.uri,
958 _first(list(role.next_uris)),
959 work_identifier.uri,
960 work_identifier.scheme,
961 work_identifier.value,
962 role_name,
963 source_work["source"],
964 False,
965 0.0,
966 None,
967 "",
968 False,
969 )
970 else:
971 external_index, score = match
972 api_agent = external[external_index]
973 contested_elsewhere = any(
974 normalize_orcid(agent["orcid"] or "") == normalized
975 for index, agent in enumerate(external)
976 if index != external_index
977 )
978 evidence = WorkEvidence(
979 br_uri,
980 role.uri,
981 _first(list(role.next_uris)),
982 work_identifier.uri,
983 work_identifier.scheme,
984 work_identifier.value,
985 role_name,
986 source_work["source"],
987 score >= CONFIRMED_NAME_SCORE
988 and not alignment.ambiguous,
989 score,
990 normalize_orcid(api_agent["orcid"] or "") or None,
991 _agent_metadata_name(api_agent).display,
992 contested_elsewhere,
993 )
994 edge_evidence[(role.ra, normalized)].append(evidence)
995 for api_agent in external:
996 api_orcid = normalize_orcid(api_agent["orcid"] or "")
997 if api_orcid:
998 names_by_orcid[api_orcid].append(
999 _agent_metadata_name(api_agent)
1000 )
1001 progress.advance(task)
1002 return (
1003 role_assessments,
1004 edge_evidence,
1005 names_by_orcid,
1006 list(operations_by_id.values()),
1007 )
1010def _polluted_identifier(names: list[PersonName]) -> bool:
1011 for left, right in combinations(names, 2):
1012 if name_score(left, right) < AMBIGUOUS_NAME_SCORE:
1013 return True
1014 return False
1017def classify_identifiers(
1018 clusters: list[Cluster],
1019 risks_by_row: dict[int, list[str]],
1020 agents: dict[str, AgentInfo],
1021 edge_evidence: dict[tuple[str, str], list[WorkEvidence]],
1022 names_by_orcid: dict[str, list[PersonName]],
1023 provenance: dict[str, dict[str, object]],
1024 client: OrcidClient,
1025) -> tuple[list[dict[str, object]], list[dict[str, object]]]:
1026 assessments = []
1027 operations = []
1028 profiles: dict[str, OrcidProfile | None] = {}
1029 candidate_orcids = {
1030 normalize_orcid(identifier.value)
1031 for cluster in clusters
1032 if cluster.csv_row in risks_by_row
1033 for member in cluster.members
1034 for identifier in agents[member].orcids
1035 if is_valid_orcid(identifier.value)
1036 }
1037 with create_progress() as progress:
1038 task = progress.add_task("Querying ORCID profiles", total=len(candidate_orcids))
1039 for orcid in sorted(candidate_orcids):
1040 if _stop_requested:
1041 break
1042 profiles[orcid] = client.orcid(orcid)
1043 progress.advance(task)
1045 for cluster in clusters:
1046 if cluster.csv_row not in risks_by_row:
1047 continue
1048 for member in cluster.members:
1049 agent = agents[member]
1050 for identifier in agent.orcids:
1051 orcid = normalize_orcid(identifier.value)
1052 evidence_key = (member, orcid)
1053 evidence = (
1054 edge_evidence[evidence_key] if evidence_key in edge_evidence else []
1055 )
1056 confirmed = [item for item in evidence if item.matched]
1057 positive = [item for item in confirmed if item.api_orcid == orcid]
1058 different = [
1059 item
1060 for item in confirmed
1061 if item.api_orcid and item.api_orcid != orcid
1062 ]
1063 elsewhere = [item for item in confirmed if item.contested_elsewhere]
1064 profile = profiles.get(orcid)
1065 profile_name = (
1066 PersonName(
1067 name=profile["name"],
1068 given=profile["given"],
1069 family=profile["family"],
1070 )
1071 if profile is not None
1072 else PersonName()
1073 )
1074 profile_score = name_score(agent.name, profile_name)
1075 other_scores = sorted(
1076 (
1077 name_score(agents[other].name, profile_name),
1078 other,
1079 )
1080 for other in cluster.members
1081 if other != member and profile_name.display
1082 )
1083 best_other_score, best_other = (
1084 other_scores[-1] if other_scores else (0.0, "")
1085 )
1086 cross_script = bool(profile_name.display) and script_family(
1087 agent.name.display
1088 ) != script_family(profile_name.display)
1089 polluted = _polluted_identifier(
1090 names_by_orcid[orcid] if orcid in names_by_orcid else []
1091 )
1092 replacement_counts = Counter(
1093 item.api_orcid for item in different if item.api_orcid
1094 )
1095 replacement = ""
1096 for candidate, votes in replacement_counts.most_common():
1097 if candidate not in profiles:
1098 profiles[candidate] = client.orcid(candidate)
1099 candidate_profile = profiles[candidate]
1100 if candidate_profile is None:
1101 continue
1102 candidate_name = PersonName(
1103 name=candidate_profile["name"],
1104 given=candidate_profile["given"],
1105 family=candidate_profile["family"],
1106 )
1107 if (
1108 votes >= 2
1109 and name_score(agent.name, candidate_name)
1110 >= CONFIRMED_NAME_SCORE
1111 ):
1112 replacement = candidate
1113 break
1115 status = "manual_review"
1116 reason = "Insufficient or conflicting work evidence"
1117 if not is_valid_orcid(orcid) and confirmed:
1118 status = "verified_wrong"
1119 reason = "ORCID has an invalid format or checksum"
1120 elif not is_valid_orcid(orcid):
1121 reason = (
1122 "ORCID has an invalid format or checksum, but the local "
1123 "work responsibility is not externally confirmed"
1124 )
1125 elif (
1126 confirmed
1127 and profile_name.display
1128 and profile_score < AMBIGUOUS_NAME_SCORE
1129 and best_other_score >= CONFIRMED_NAME_SCORE
1130 and not cross_script
1131 and (positive or elsewhere or different or polluted)
1132 ):
1133 status = "verified_wrong"
1134 reason = f"ORCID profile matches {best_other}, not {member}"
1135 elif positive and (
1136 not profile_name.display
1137 or profile_score >= AMBIGUOUS_NAME_SCORE
1138 or cross_script
1139 ):
1140 status = "verified_correct"
1141 reason = "Work contributor and local agent use the same ORCID"
1143 assessment = {
1144 "csv_row": cluster.csv_row,
1145 "ra": member,
1146 "identifier_uri": identifier.uri,
1147 "orcid": orcid,
1148 "status": status,
1149 "reason": reason,
1150 "profile": profile,
1151 "profile_score": round(profile_score, 3),
1152 "best_other_ra": best_other,
1153 "best_other_score": round(best_other_score, 3),
1154 "replacement_orcid": replacement or None,
1155 "work_evidence": [asdict(item) for item in evidence],
1156 "agent_provenance": provenance.get(member),
1157 "identifier_provenance": provenance.get(identifier.uri),
1158 }
1159 assessments.append(assessment)
1160 if status != "verified_wrong":
1161 continue
1162 action = "replace_identifier" if replacement else "detach_identifier"
1163 evidence_links = [
1164 {
1165 "br": br,
1166 "ar": ar,
1167 "ra": member,
1168 "next": next_uri,
1169 "work_identifier_uri": work_identifier_uri,
1170 "work_identifier_scheme": work_identifier_scheme,
1171 "work_identifier_value": work_identifier_value,
1172 }
1173 for (
1174 br,
1175 ar,
1176 next_uri,
1177 work_identifier_uri,
1178 work_identifier_scheme,
1179 work_identifier_value,
1180 ) in sorted(
1181 {
1182 (
1183 item.br,
1184 item.ar,
1185 item.next_uri,
1186 item.work_identifier_uri,
1187 item.work_identifier_scheme,
1188 item.work_identifier_value,
1189 )
1190 for item in confirmed
1191 }
1192 )
1193 ]
1194 operations.append(
1195 _operation(
1196 action,
1197 cluster.csv_row,
1198 reason,
1199 max(best_other_score, CONFIRMED_NAME_SCORE),
1200 ra=member,
1201 identifier_uri=identifier.uri,
1202 old_value=orcid,
1203 new_value=replacement,
1204 evidence=evidence_links,
1205 )
1206 )
1207 return assessments, operations
1210def _review_chain_values(operation: dict[str, object]) -> tuple[str, str]:
1211 links = operation.get("links")
1212 if not isinstance(links, list):
1213 return cast(str, operation["old_value"]), cast(str, operation["new_value"])
1214 old_values = []
1215 new_values = []
1216 for raw_link in links:
1217 link = cast(dict[str, str], raw_link)
1218 old_values.append(f"{link['ar']} -> {link['old_next'] or '[end]'}")
1219 new_values.append(f"{link['ar']} -> {link['new_next'] or '[end]'}")
1220 return "; ".join(old_values), "; ".join(new_values)
1223def write_review_file(path: str, operations: list[dict[str, object]]) -> None:
1224 _ensure_parent(path)
1225 with open(path, "w", newline="", encoding="utf-8") as stream:
1226 writer = csv.DictWriter(stream, fieldnames=REVIEW_FIELDS)
1227 writer.writeheader()
1228 for operation in operations:
1229 old_value, new_value = _review_chain_values(operation)
1230 row = {
1231 field: operation[field]
1232 for field in REVIEW_FIELDS
1233 if field not in {"old_value", "new_value", "decision"}
1234 }
1235 row["old_value"] = old_value
1236 row["new_value"] = new_value
1237 row["decision"] = ""
1238 writer.writerow(row)
1241def _agent_report(
1242 agent: AgentInfo, provenance: dict[str, dict[str, object]]
1243) -> dict[str, object]:
1244 return {
1245 "ra": agent.uri,
1246 "name": asdict(agent.name),
1247 "normalized_name": normalize_name(agent.name.display),
1248 "provenance": provenance.get(agent.uri),
1249 "identifiers": [
1250 {
1251 "uri": identifier.uri,
1252 "scheme": identifier.scheme,
1253 "value": identifier.value,
1254 "provenance": provenance.get(identifier.uri),
1255 }
1256 for identifier in agent.identifiers
1257 ],
1258 }
1261def analyze_duplicate_ras(
1262 config_path: str,
1263 duplicate_path: str,
1264 report_path: str,
1265 review_path: str,
1266 cache_path: str,
1267 mailto: str,
1268 workers: int,
1269 max_evidence_works: int,
1270 all_api: bool,
1271 refresh_cache: bool,
1272 openalex_api_key: str,
1273) -> dict[str, object]:
1274 global _stop_requested
1275 _stop_requested = False
1276 config = load_audit_config(config_path)
1277 duplicate_path = os.path.abspath(duplicate_path)
1278 report_path = os.path.abspath(report_path)
1279 review_path = os.path.abspath(review_path)
1280 cache_path = os.path.abspath(cache_path)
1281 cache = EntityFileLocator(
1282 config.rdf_dir,
1283 config.dir_split,
1284 config.items_per_file,
1285 config.zip_output,
1286 )
1287 (
1288 clusters,
1289 agents,
1290 risks_by_row,
1291 cluster_count,
1292 agent_count,
1293 ) = scan_candidate_clusters(duplicate_path, cache, workers, all_api)
1295 candidate_ras = set(agents)
1296 if candidate_ras:
1297 works, roles, contextual_agents = build_context(
1298 candidate_ras,
1299 config.rdf_dir,
1300 config.zip_output,
1301 cache,
1302 workers,
1303 max_evidence_works,
1304 )
1305 agents.update(contextual_agents)
1306 else:
1307 works, roles = {}, {}
1308 cluster_by_ra = {
1309 member: cluster for cluster in clusters for member in cluster.members
1310 }
1311 provenance_uris = set(candidate_ras)
1312 provenance_uris.update(
1313 identifier.uri for ra in candidate_ras for identifier in agents[ra].identifiers
1314 )
1315 provenance = load_provenance(provenance_uris, cache, workers)
1316 selected_works = _selected_work_uris(
1317 candidate_ras, works, roles, max_evidence_works
1318 )
1320 _ensure_parent(cache_path)
1321 api_cache = ApiCache(cache_path)
1322 client = AgentMetadataClient(
1323 mailto=mailto,
1324 cache=api_cache,
1325 refresh_cache=refresh_cache,
1326 openalex_api_key=openalex_api_key,
1327 )
1328 try:
1329 (
1330 role_assessments,
1331 edge_evidence,
1332 names_by_orcid,
1333 role_operations,
1334 ) = collect_external_evidence(
1335 selected_works,
1336 works,
1337 roles,
1338 agents,
1339 cluster_by_ra,
1340 client,
1341 )
1342 identifier_assessments, identifier_operations = classify_identifiers(
1343 clusters,
1344 risks_by_row,
1345 agents,
1346 edge_evidence,
1347 names_by_orcid,
1348 provenance,
1349 client,
1350 )
1351 finally:
1352 client.close()
1353 api_cache.close()
1355 operations_by_id = {
1356 cast(str, operation["operation_id"]): operation
1357 for operation in (*role_operations, *identifier_operations)
1358 }
1359 operations = sorted(
1360 operations_by_id.values(),
1361 key=lambda operation: cast(str, operation["operation_id"]),
1362 )
1363 risk_counts = Counter(risk for risks in risks_by_row.values() for risk in risks)
1364 identifier_status_counts = Counter(
1365 cast(str, assessment["status"]) for assessment in identifier_assessments
1366 )
1367 operation_counts = Counter(
1368 cast(str, operation["action"]) for operation in operations
1369 )
1370 report: dict[str, object] = {
1371 "schema_version": PLAN_SCHEMA_VERSION,
1372 "complete": not _stop_requested,
1373 "generated_at": datetime.now(timezone.utc).isoformat(),
1374 "config": os.path.abspath(config_path),
1375 "config_sha256": _sha256(config_path),
1376 "duplicates": duplicate_path,
1377 "duplicates_sha256": _sha256(duplicate_path),
1378 "rdf_dir": config.rdf_dir,
1379 "api_cache": cache_path,
1380 "audit_options": {
1381 "all_api": all_api,
1382 "max_evidence_works": max_evidence_works,
1383 "refresh_cache": refresh_cache,
1384 },
1385 "review_file": review_path,
1386 "summary": {
1387 "total_clusters": cluster_count,
1388 "total_cluster_members": agent_count,
1389 "candidate_clusters": len(clusters),
1390 "candidate_agents": len(candidate_ras),
1391 "locally_consistent_clusters": cluster_count
1392 - sum(bool(risks) for risks in risks_by_row.values()),
1393 "selected_works": len(selected_works),
1394 "risk_counts": dict(sorted(risk_counts.items())),
1395 "identifier_status_counts": dict(sorted(identifier_status_counts.items())),
1396 "operation_counts": dict(sorted(operation_counts.items())),
1397 },
1398 "clusters": [
1399 {
1400 "csv_row": cluster.csv_row,
1401 "survivor": cluster.survivor,
1402 "merge_status": "blocked_pending_review",
1403 "risks": risks_by_row[cluster.csv_row],
1404 "members": [
1405 _agent_report(agents[member], provenance)
1406 for member in cluster.members
1407 ],
1408 }
1409 for cluster in clusters
1410 ],
1411 "role_assessments": role_assessments,
1412 "identifier_assessments": identifier_assessments,
1413 "operations": operations,
1414 }
1415 _write_json(report_path, report)
1416 write_review_file(review_path, operations)
1417 return report
1420def read_review_decisions(
1421 path: str, operations: list[dict[str, object]]
1422) -> list[dict[str, object]]:
1423 operations_by_id = {
1424 cast(str, operation["operation_id"]): operation for operation in operations
1425 }
1426 if len(operations_by_id) != len(operations):
1427 raise ValueError("Correction plan contains repeated operation IDs")
1428 changed_ids = sorted(
1429 operation_id
1430 for operation_id, operation in operations_by_id.items()
1431 if _computed_operation_id(operation) != operation_id
1432 )
1433 if changed_ids:
1434 raise ValueError(f"Correction plan has modified operations: {changed_ids}")
1435 decisions = {}
1436 with open(path, newline="", encoding="utf-8") as stream:
1437 reader = csv.DictReader(stream)
1438 if reader.fieldnames != list(REVIEW_FIELDS):
1439 raise ValueError(f"Unexpected review CSV header: {reader.fieldnames}")
1440 for row in reader:
1441 operation_id = row["operation_id"]
1442 if operation_id in decisions:
1443 raise ValueError(f"Repeated review operation: {operation_id}")
1444 if operation_id not in operations_by_id:
1445 raise ValueError(f"Unknown review operation: {operation_id}")
1446 operation = operations_by_id[operation_id]
1447 old_value, new_value = _review_chain_values(operation)
1448 expected = {
1449 field: str(operation[field])
1450 for field in REVIEW_FIELDS
1451 if field not in {"old_value", "new_value", "decision"}
1452 }
1453 expected["old_value"] = old_value
1454 expected["new_value"] = new_value
1455 changed = [
1456 field
1457 for field in REVIEW_FIELDS
1458 if field != "decision" and row[field] != expected[field]
1459 ]
1460 if changed:
1461 raise ValueError(
1462 f"Review row {operation_id} differs from the plan in: {changed}"
1463 )
1464 decision = row["decision"].strip().lower()
1465 if decision not in {"", "approve", "reject"}:
1466 raise ValueError(
1467 f"Invalid decision for {operation_id}: {row['decision']}"
1468 )
1469 decisions[operation_id] = decision
1470 missing = operations_by_id.keys() - decisions.keys()
1471 if missing:
1472 raise ValueError(f"Review CSV is missing operations: {sorted(missing)}")
1473 return [
1474 operation
1475 for operation_id, operation in operations_by_id.items()
1476 if decisions[operation_id] == "approve"
1477 ]
1480def _validate_uri(uri: str) -> None:
1481 if not uri.startswith(("http://", "https://")) or any(
1482 character in uri for character in "<> \t\r\n"
1483 ):
1484 raise ValueError(f"Invalid URI in correction plan: {uri}")
1487def _sparql_bindings(endpoint: str, query: str) -> list[dict[str, dict[str, str]]]:
1488 result = execute_sparql(endpoint, query, max_retries=3, backoff_factor=1)
1489 result_section = cast(dict[str, object], result["results"])
1490 return cast(list[dict[str, dict[str, str]]], result_section["bindings"])
1493def _current_objects(endpoint: str, subject: str, predicate: str) -> list[str]:
1494 _validate_uri(subject)
1495 _validate_uri(predicate)
1496 query = f"SELECT ?value WHERE {{ <{subject}> <{predicate}> ?value . }}"
1497 return sorted(
1498 binding["value"]["value"] for binding in _sparql_bindings(endpoint, query)
1499 )
1502def _find_orcid_identifier(endpoint: str, orcid: str) -> str:
1503 if not is_valid_orcid(orcid):
1504 raise ValueError(f"Invalid replacement ORCID: {orcid}")
1505 literal = orjson.dumps(normalize_orcid(orcid)).decode()
1506 query = f"""
1507 SELECT DISTINCT ?id WHERE {{
1508 ?id <{USES_IDENTIFIER_SCHEME}> <{DATACITE_PREFIX}orcid> ;
1509 <{HAS_LITERAL_VALUE}> {literal} .
1510 }}
1511 """
1512 identifiers = sorted(
1513 binding["id"]["value"] for binding in _sparql_bindings(endpoint, query)
1514 )
1515 if len(identifiers) > 1:
1516 raise ValueError(
1517 f"Replacement ORCID {orcid} has multiple identifier entities: {identifiers}"
1518 )
1519 return identifiers[0] if identifiers else ""
1522def _operation_string(operation: dict[str, object], field: str) -> str:
1523 value = operation[field]
1524 if not isinstance(value, str):
1525 raise ValueError(f"Operation field {field} must be a string")
1526 return value
1529def _operation_links(operation: dict[str, object]) -> list[dict[str, str]]:
1530 value = operation["links"]
1531 if not isinstance(value, list):
1532 raise ValueError("Reorder operation links must be a list")
1533 links = []
1534 for item in value:
1535 if not isinstance(item, dict):
1536 raise ValueError("Reorder operation link must be an object")
1537 link = cast(dict[str, object], item)
1538 parsed = {}
1539 for field in ("ar", "old_next", "new_next"):
1540 field_value = link[field]
1541 if not isinstance(field_value, str):
1542 raise ValueError(f"Reorder link field {field} must be a string")
1543 parsed[field] = field_value
1544 links.append(parsed)
1545 return links
1548def _operation_evidence(operation: dict[str, object]) -> list[dict[str, str]]:
1549 value = operation["evidence"] if "evidence" in operation else []
1550 if not isinstance(value, list):
1551 raise ValueError("Identifier operation evidence must be a list")
1552 evidence = []
1553 for item in value:
1554 if not isinstance(item, dict):
1555 raise ValueError("Identifier evidence must be an object")
1556 raw_evidence = cast(dict[str, object], item)
1557 parsed = {}
1558 for field in (
1559 "br",
1560 "ar",
1561 "ra",
1562 "next",
1563 "work_identifier_uri",
1564 "work_identifier_scheme",
1565 "work_identifier_value",
1566 ):
1567 field_value = raw_evidence[field]
1568 if not isinstance(field_value, str):
1569 raise ValueError(f"Identifier evidence field {field} must be a string")
1570 parsed[field] = field_value
1571 evidence.append(parsed)
1572 return evidence
1575def _import_entities(editor: MetaEditor, g_set: GraphSet, uris: set[str]) -> None:
1576 editor.reader.import_entities_from_triplestore(
1577 g_set=g_set,
1578 ts_url=editor.endpoint,
1579 entities=sorted(uris),
1580 resp_agent=editor.resp_agent,
1581 enable_validation=False,
1582 )
1585def _preflight_operations(
1586 editor: MetaEditor, operations: list[dict[str, object]]
1587) -> tuple[set[str], dict[str, str]]:
1588 uris = set()
1589 replacements = {}
1590 for operation in operations:
1591 action = _operation_string(operation, "action")
1592 if action in {"detach_identifier", "replace_identifier"}:
1593 ra_uri = _operation_string(operation, "ra")
1594 identifier_uri = _operation_string(operation, "identifier_uri")
1595 old_value = normalize_orcid(_operation_string(operation, "old_value"))
1596 _validate_uri(ra_uri)
1597 _validate_uri(identifier_uri)
1598 if identifier_uri not in _current_objects(
1599 editor.endpoint, ra_uri, HAS_IDENTIFIER
1600 ):
1601 raise RuntimeError(
1602 f"Stale plan: {ra_uri} no longer has identifier {identifier_uri}"
1603 )
1604 schemes = _current_objects(
1605 editor.endpoint, identifier_uri, USES_IDENTIFIER_SCHEME
1606 )
1607 values = _current_objects(
1608 editor.endpoint, identifier_uri, HAS_LITERAL_VALUE
1609 )
1610 if schemes != [f"{DATACITE_PREFIX}orcid"] or [
1611 normalize_orcid(value) for value in values
1612 ] != [old_value]:
1613 raise RuntimeError(
1614 f"Stale plan: identifier {identifier_uri} no longer represents "
1615 f"ORCID {old_value}"
1616 )
1617 for evidence in _operation_evidence(operation):
1618 br_uri = evidence["br"]
1619 ar_uri = evidence["ar"]
1620 evidence_ra = evidence["ra"]
1621 work_identifier_uri = evidence["work_identifier_uri"]
1622 for uri in (
1623 br_uri,
1624 ar_uri,
1625 evidence_ra,
1626 work_identifier_uri,
1627 ):
1628 _validate_uri(uri)
1629 if work_identifier_uri not in _current_objects(
1630 editor.endpoint, br_uri, HAS_IDENTIFIER
1631 ):
1632 raise RuntimeError(
1633 f"Stale plan: {br_uri} no longer has work identifier "
1634 f"{work_identifier_uri}"
1635 )
1636 work_schemes = _current_objects(
1637 editor.endpoint, work_identifier_uri, USES_IDENTIFIER_SCHEME
1638 )
1639 work_values = _current_objects(
1640 editor.endpoint, work_identifier_uri, HAS_LITERAL_VALUE
1641 )
1642 if work_schemes != [
1643 f"{DATACITE_PREFIX}{evidence['work_identifier_scheme']}"
1644 ] or work_values != [evidence["work_identifier_value"]]:
1645 raise RuntimeError(
1646 f"Stale plan: work identifier {work_identifier_uri} changed"
1647 )
1648 if ar_uri not in _current_objects(
1649 editor.endpoint, br_uri, IS_DOCUMENT_CONTEXT_FOR
1650 ):
1651 raise RuntimeError(
1652 f"Stale plan: {br_uri} no longer contains role {ar_uri}"
1653 )
1654 if _current_objects(editor.endpoint, ar_uri, IS_HELD_BY) != [
1655 evidence_ra
1656 ]:
1657 raise RuntimeError(
1658 f"Stale plan: {ar_uri} is no longer held by {evidence_ra}"
1659 )
1660 expected_next = [evidence["next"]] if evidence["next"] else []
1661 if _current_objects(editor.endpoint, ar_uri, HAS_NEXT) != expected_next:
1662 raise RuntimeError(
1663 f"Stale plan: {ar_uri} hasNext no longer matches the "
1664 "confirmed work evidence"
1665 )
1666 uris.update((ra_uri, identifier_uri))
1667 if action == "replace_identifier":
1668 replacement = normalize_orcid(_operation_string(operation, "new_value"))
1669 if replacement not in replacements:
1670 replacements[replacement] = _find_orcid_identifier(
1671 editor.endpoint, replacement
1672 )
1673 if replacements[replacement]:
1674 uris.add(replacements[replacement])
1675 elif action == "reassign_role":
1676 br_uri = _operation_string(operation, "br")
1677 ar_uri = _operation_string(operation, "ar")
1678 old_ra = _operation_string(operation, "old_value")
1679 new_ra = _operation_string(operation, "new_value")
1680 for uri in (br_uri, ar_uri, old_ra, new_ra):
1681 _validate_uri(uri)
1682 if ar_uri not in _current_objects(
1683 editor.endpoint, br_uri, IS_DOCUMENT_CONTEXT_FOR
1684 ):
1685 raise RuntimeError(
1686 f"Stale plan: {br_uri} no longer contains role {ar_uri}"
1687 )
1688 if _current_objects(editor.endpoint, ar_uri, IS_HELD_BY) != [old_ra]:
1689 raise RuntimeError(
1690 f"Stale plan: {ar_uri} is no longer held by {old_ra}"
1691 )
1692 uris.update((ar_uri, old_ra, new_ra))
1693 elif action == "reorder_chain":
1694 br_uri = _operation_string(operation, "br")
1695 _validate_uri(br_uri)
1696 contributor_uris = _current_objects(
1697 editor.endpoint, br_uri, IS_DOCUMENT_CONTEXT_FOR
1698 )
1699 for link in _operation_links(operation):
1700 ar_uri = link["ar"]
1701 _validate_uri(ar_uri)
1702 if ar_uri not in contributor_uris:
1703 raise RuntimeError(
1704 f"Stale plan: {br_uri} no longer contains role {ar_uri}"
1705 )
1706 current = _current_objects(editor.endpoint, ar_uri, HAS_NEXT)
1707 expected = [link["old_next"]] if link["old_next"] else []
1708 if current != expected:
1709 raise RuntimeError(
1710 f"Stale plan: {ar_uri} hasNext is {current}, expected {expected}"
1711 )
1712 uris.add(ar_uri)
1713 if link["new_next"]:
1714 _validate_uri(link["new_next"])
1715 uris.add(link["new_next"])
1716 return uris, replacements
1719def _apply_operation_group(
1720 editor: MetaEditor, operations: list[dict[str, object]]
1721) -> None:
1722 uris, replacements = _preflight_operations(editor, operations)
1724 g_set = GraphSet(
1725 editor.base_iri,
1726 supplier_prefix=editor.supplier_prefix,
1727 custom_counter_handler=editor.counter_handler,
1728 wanted_label=False,
1729 )
1730 _import_entities(editor, g_set, uris)
1731 created_replacements: dict[str, Identifier] = {}
1732 for operation in operations:
1733 action = _operation_string(operation, "action")
1734 if action in {"detach_identifier", "replace_identifier"}:
1735 ra = _responsible_agent(g_set, _operation_string(operation, "ra"))
1736 identifier_uri = _operation_string(operation, "identifier_uri")
1737 ra.remove_identifier(_identifier(g_set, identifier_uri))
1738 if action == "replace_identifier":
1739 replacement = normalize_orcid(_operation_string(operation, "new_value"))
1740 replacement_uri = replacements[replacement]
1741 if replacement_uri:
1742 replacement_identifier = _identifier(g_set, replacement_uri)
1743 elif replacement in created_replacements:
1744 replacement_identifier = created_replacements[replacement]
1745 else:
1746 replacement_identifier = g_set.add_id(editor.resp_agent)
1747 replacement_identifier.create_orcid(replacement)
1748 created_replacements[replacement] = replacement_identifier
1749 ra.has_identifier(replacement_identifier)
1750 elif action == "reassign_role":
1751 role = _agent_role(g_set, _operation_string(operation, "ar"))
1752 role.remove_is_held_by()
1753 role.is_held_by(
1754 _responsible_agent(g_set, _operation_string(operation, "new_value"))
1755 )
1756 elif action == "reorder_chain":
1757 for link in _operation_links(operation):
1758 role = _agent_role(g_set, link["ar"])
1759 role.remove_next()
1760 if link["new_next"]:
1761 role.has_next(_agent_role(g_set, link["new_next"]))
1762 editor.save(g_set, editor.supplier_prefix)
1765def _validate_approved_operations(operations: list[dict[str, object]]) -> None:
1766 allowed = {
1767 "detach_identifier",
1768 "replace_identifier",
1769 "reassign_role",
1770 "reorder_chain",
1771 }
1772 identifiers = set()
1773 assignments = {}
1774 reorders = {}
1775 for operation in operations:
1776 action = _operation_string(operation, "action")
1777 if action not in allowed:
1778 raise ValueError(f"Unsupported approved operation: {action}")
1779 if action in {"detach_identifier", "replace_identifier"}:
1780 key = (
1781 _operation_string(operation, "ra"),
1782 _operation_string(operation, "identifier_uri"),
1783 )
1784 if key in identifiers:
1785 raise ValueError(f"Conflicting identifier operations for {key}")
1786 identifiers.add(key)
1787 evidence = _operation_evidence(operation)
1788 if not evidence:
1789 raise ValueError(f"Identifier operation has no work evidence: {key}")
1790 if any(item["ra"] != key[0] for item in evidence):
1791 raise ValueError(f"Identifier evidence has another RA: {key}")
1792 if action == "replace_identifier":
1793 old_value = normalize_orcid(_operation_string(operation, "old_value"))
1794 new_value = normalize_orcid(_operation_string(operation, "new_value"))
1795 if old_value == new_value:
1796 raise ValueError(f"ORCID replacement does not change {key}")
1797 if not is_valid_orcid(new_value):
1798 raise ValueError(f"Invalid replacement ORCID: {new_value}")
1799 elif action == "reassign_role":
1800 ar_uri = _operation_string(operation, "ar")
1801 new_ra = _operation_string(operation, "new_value")
1802 if ar_uri in assignments and assignments[ar_uri] != new_ra:
1803 raise ValueError(f"Conflicting role assignments for {ar_uri}")
1804 assignments[ar_uri] = new_ra
1805 else:
1806 br_uri = _operation_string(operation, "br")
1807 links = orjson.dumps(_operation_links(operation))
1808 if br_uri in reorders and reorders[br_uri] != links:
1809 raise ValueError(f"Conflicting chain orders for {br_uri}")
1810 reorders[br_uri] = links
1813def _operation_resources(operation: dict[str, object]) -> set[str]:
1814 action = _operation_string(operation, "action")
1815 if action in {"detach_identifier", "replace_identifier"}:
1816 resources = {
1817 _operation_string(operation, "ra"),
1818 _operation_string(operation, "identifier_uri"),
1819 }
1820 if action == "replace_identifier":
1821 resources.add(
1822 f"orcid:{normalize_orcid(_operation_string(operation, 'new_value'))}"
1823 )
1824 for evidence in _operation_evidence(operation):
1825 resources.update(
1826 {
1827 f"context:{evidence['br']}",
1828 evidence["ar"],
1829 evidence["ra"],
1830 evidence["work_identifier_uri"],
1831 }
1832 )
1833 if evidence["next"]:
1834 resources.add(evidence["next"])
1835 return resources
1836 if action == "reassign_role":
1837 return {
1838 f"context:{_operation_string(operation, 'br')}",
1839 _operation_string(operation, "ar"),
1840 _operation_string(operation, "old_value"),
1841 _operation_string(operation, "new_value"),
1842 }
1843 resources = {f"context:{_operation_string(operation, 'br')}"}
1844 for link in _operation_links(operation):
1845 resources.add(link["ar"])
1846 if link["old_next"]:
1847 resources.add(link["old_next"])
1848 if link["new_next"]:
1849 resources.add(link["new_next"])
1850 return resources
1853def _operation_groups(
1854 operations: list[dict[str, object]],
1855) -> list[tuple[str, list[dict[str, object]]]]:
1856 parents = list(range(len(operations)))
1858 def find(index: int) -> int:
1859 while parents[index] != index:
1860 parents[index] = parents[parents[index]]
1861 index = parents[index]
1862 return index
1864 def union(left: int, right: int) -> None:
1865 left_root = find(left)
1866 right_root = find(right)
1867 if left_root != right_root:
1868 parents[right_root] = left_root
1870 owners = {}
1871 for index, operation in enumerate(operations):
1872 for resource in _operation_resources(operation):
1873 if resource in owners:
1874 union(index, owners[resource])
1875 else:
1876 owners[resource] = index
1878 grouped: dict[int, list[dict[str, object]]] = defaultdict(list)
1879 for index, operation in enumerate(operations):
1880 grouped[find(index)].append(operation)
1881 result = []
1882 for group in grouped.values():
1883 group.sort(key=lambda operation: _operation_string(operation, "operation_id"))
1884 operation_ids = [
1885 _operation_string(operation, "operation_id") for operation in group
1886 ]
1887 result.append((_operation_id("group", *operation_ids), group))
1888 return sorted(result, key=lambda item: item[0])
1891def _write_reindex_sentinel(path: str, plan_path: str) -> None:
1892 with open(path, "w", encoding="utf-8") as stream:
1893 stream.write(
1894 f"{plan_path} changed RDF files on top of the current triplestore "
1895 "snapshot.\nRe-index the triplestore from the RDF files, then delete "
1896 "this file before another correction or merge run. Do not reuse the "
1897 "original duplicate CSV: run duplicate detection again.\n"
1898 )
1901def execute_plan(
1902 config_path: str,
1903 plan_path: str,
1904 review_path: str | None,
1905 resp_agent: str,
1906 progress_path: str,
1907 execution_report_path: str,
1908) -> dict[str, object]:
1909 global _stop_requested
1910 _stop_requested = False
1911 plan_path = os.path.abspath(plan_path)
1912 plan = _read_json_object(plan_path)
1913 if plan["schema_version"] != PLAN_SCHEMA_VERSION:
1914 raise ValueError(f"Unsupported plan schema: {plan['schema_version']}")
1915 if plan["complete"] is not True:
1916 raise ValueError("The correction plan is incomplete and cannot be executed")
1917 if plan["config_sha256"] != _sha256(config_path):
1918 raise ValueError("The meta configuration changed after plan generation")
1919 duplicate_path = cast(str, plan["duplicates"])
1920 if plan["duplicates_sha256"] != _sha256(duplicate_path):
1921 raise ValueError("The duplicate CSV changed after plan generation")
1922 raw_operations = plan["operations"]
1923 if not isinstance(raw_operations, list) or not all(
1924 isinstance(operation, dict) for operation in raw_operations
1925 ):
1926 raise ValueError("Correction plan operations must be a list of objects")
1927 operations = cast(list[dict[str, object]], raw_operations)
1928 selected_review_path = os.path.abspath(
1929 review_path or cast(str, plan["review_file"])
1930 )
1931 approved = read_review_decisions(selected_review_path, operations)
1932 _validate_approved_operations(approved)
1934 sentinel_path = os.path.join(os.path.dirname(plan_path), REINDEX_SENTINEL_FILENAME)
1935 if os.path.exists(sentinel_path):
1936 raise RuntimeError(
1937 f"{sentinel_path} exists. Re-index the triplestore and remove the "
1938 "sentinel before executing this plan."
1939 )
1940 plan_sha256 = _sha256(plan_path)
1941 review_sha256 = _sha256(selected_review_path)
1942 completed = _load_progress(progress_path, plan_sha256, review_sha256)
1943 groups = _operation_groups(approved)
1944 unknown_completed = completed - {group_id for group_id, _ in groups}
1945 if unknown_completed:
1946 raise ValueError(
1947 f"Progress file contains unknown groups: {sorted(unknown_completed)}"
1948 )
1949 attempted_groups = 0
1950 if groups:
1951 editor = MetaEditor(config_path, resp_agent, save_queries=True)
1952 try:
1953 with create_progress() as progress:
1954 task = progress.add_task(
1955 "Applying approved corrections", total=len(groups)
1956 )
1957 for group_id, group_operations in groups:
1958 if group_id in completed:
1959 progress.advance(task)
1960 continue
1961 if _stop_requested:
1962 break
1963 attempted_groups += 1
1964 _apply_operation_group(editor, group_operations)
1965 completed.add(group_id)
1966 _save_progress(progress_path, plan_sha256, review_sha256, completed)
1967 progress.advance(task)
1968 finally:
1969 if attempted_groups:
1970 _write_reindex_sentinel(sentinel_path, plan_path)
1972 complete = len(completed) == len(groups) and not _stop_requested
1973 execution_report: dict[str, object] = {
1974 "schema_version": PLAN_SCHEMA_VERSION,
1975 "plan": plan_path,
1976 "plan_sha256": plan_sha256,
1977 "review_file": selected_review_path,
1978 "review_sha256": review_sha256,
1979 "generated_at": datetime.now(timezone.utc).isoformat(),
1980 "complete": complete,
1981 "approved_operations": len(approved),
1982 "total_groups": len(groups),
1983 "completed_groups": sorted(completed),
1984 "reindex_sentinel": sentinel_path if attempted_groups else None,
1985 }
1986 _write_json(execution_report_path, execution_report)
1987 if complete and os.path.exists(progress_path):
1988 os.remove(progress_path)
1989 return execution_report
1992def main() -> None: # pragma: no cover
1993 parser = argparse.ArgumentParser(
1994 description=(
1995 "Audit duplicate responsible-agent clusters against local role chains, "
1996 "Crossref, DataCite, OpenAlex, and ORCID. Apply only operations approved "
1997 "in the generated review CSV."
1998 ),
1999 formatter_class=RichHelpFormatter,
2000 )
2001 parser.add_argument("-c", "--config", required=True, help="Meta YAML config")
2002 mode = parser.add_mutually_exclusive_group(required=True)
2003 mode.add_argument(
2004 "--dry-run", action="store_true", help="Generate a plan without changing RDF"
2005 )
2006 mode.add_argument("--execute", metavar="PLAN", help="Execute an approved plan")
2007 parser.add_argument(
2008 "--duplicates", help="Duplicate RA CSV produced by find.duplicates"
2009 )
2010 parser.add_argument("--report-file", help="Dry-run JSON plan path")
2011 parser.add_argument(
2012 "--review-file",
2013 help="Review CSV path; defaults to the path stored in the plan on execution",
2014 )
2015 parser.add_argument("--cache-file", help="SQLite API cache path")
2016 parser.add_argument("--mailto", help="Contact email sent to metadata APIs")
2017 parser.add_argument(
2018 "--openalex-api-key",
2019 default=(
2020 os.environ["OPENALEX_API_KEY"] if "OPENALEX_API_KEY" in os.environ else ""
2021 ),
2022 help="OpenAlex API key; defaults to OPENALEX_API_KEY",
2023 )
2024 parser.add_argument(
2025 "--workers",
2026 type=int,
2027 default=min(os.cpu_count() or 1, 16),
2028 help="Threads used for local RDF scanning",
2029 )
2030 parser.add_argument(
2031 "--max-evidence-works",
2032 type=int,
2033 default=5,
2034 help="Maximum works queried per candidate RA",
2035 )
2036 parser.add_argument(
2037 "--all-api",
2038 action="store_true",
2039 help="Query external APIs for every cluster instead of local suspects only",
2040 )
2041 parser.add_argument(
2042 "--refresh-cache", action="store_true", help="Refresh cached API responses"
2043 )
2044 parser.add_argument("-r", "--resp-agent", help="Provenance responsible-agent URI")
2045 parser.add_argument("--progress-file", help="Execution progress JSON path")
2046 parser.add_argument("--execution-report", help="Execution result JSON path")
2047 args = parser.parse_args()
2049 if args.workers < 1:
2050 parser.error("--workers must be positive")
2051 if args.max_evidence_works < 1:
2052 parser.error("--max-evidence-works must be positive")
2054 signal.signal(signal.SIGINT, _handle_signal)
2055 signal.signal(signal.SIGTERM, _handle_signal)
2056 if args.dry_run:
2057 if not args.duplicates or not args.report_file or not args.mailto:
2058 parser.error(
2059 "--duplicates, --report-file, and --mailto are required with --dry-run"
2060 )
2061 review_path = args.review_file or f"{args.report_file}.review.csv"
2062 cache_path = args.cache_file or f"{args.report_file}.cache.sqlite"
2063 report = analyze_duplicate_ras(
2064 config_path=args.config,
2065 duplicate_path=args.duplicates,
2066 report_path=args.report_file,
2067 review_path=review_path,
2068 cache_path=cache_path,
2069 mailto=args.mailto,
2070 workers=args.workers,
2071 max_evidence_works=args.max_evidence_works,
2072 all_api=args.all_api,
2073 refresh_cache=args.refresh_cache,
2074 openalex_api_key=args.openalex_api_key,
2075 )
2076 summary = cast(dict[str, object], report["summary"])
2077 console.print(
2078 f"Plan written to [cyan]{os.path.abspath(args.report_file)}[/cyan]. "
2079 f"Candidate clusters: [cyan]{summary['candidate_clusters']}[/cyan]; "
2080 f"proposed operations: [cyan]{sum(cast(dict[str, int], summary['operation_counts']).values())}[/cyan]."
2081 )
2082 return
2084 if not args.resp_agent:
2085 parser.error("--resp-agent is required with --execute")
2086 _validate_uri(args.resp_agent)
2087 plan_path = cast(str, args.execute)
2088 progress_path = args.progress_file or f"{plan_path}.progress.json"
2089 execution_report_path = args.execution_report or f"{plan_path}.execution.json"
2090 result = execute_plan(
2091 config_path=args.config,
2092 plan_path=plan_path,
2093 review_path=args.review_file,
2094 resp_agent=args.resp_agent,
2095 progress_path=os.path.abspath(progress_path),
2096 execution_report_path=os.path.abspath(execution_report_path),
2097 )
2098 console.print(
2099 f"Execution report written to [cyan]{os.path.abspath(execution_report_path)}[/cyan]. "
2100 f"Completed groups: [cyan]{len(cast(list[str], result['completed_groups']))}[/cyan]."
2101 )
2104if __name__ == "__main__":
2105 main()