Coverage for oc_meta / run / patches / fix_duplicate_part_of.py: 90%
280 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 json
9import multiprocessing
10import os
11import signal
12from collections import defaultdict
13from concurrent.futures import ProcessPoolExecutor, as_completed
14from dataclasses import dataclass
15from zipfile import ZipFile
17import orjson
18import yaml
19from oc_ocdm.graph import GraphSet
20from oc_ocdm.support import get_prefix
21from rich_argparse import RichHelpFormatter
23from oc_meta.core.editor import MetaEditor
24from oc_meta.lib.console import console, create_progress
25from oc_meta.lib.file_manager import collect_files, collect_zip_files, find_rdf_file
27FRBR_PART_OF = "http://purl.org/vocab/frbr/core#partOf"
28HAS_IDENTIFIER = "http://purl.org/spar/datacite/hasIdentifier"
29USES_ID_SCHEME = "http://purl.org/spar/datacite/usesIdentifierScheme"
30HAS_LITERAL_VALUE = (
31 "http://www.essepuntato.it/2010/06/literalreification/hasLiteralValue"
32)
33DCTERMS_TITLE = "http://purl.org/dc/terms/title"
34FABIO_EXPRESSION = "http://purl.org/spar/fabio/Expression"
36BATCH_SIZE = 100
38_stop_requested = False
41def _worker_init() -> None:
42 signal.signal(signal.SIGINT, signal.SIG_IGN)
45def _handle_signal(_signum: int, _frame: object) -> None:
46 global _stop_requested
47 _stop_requested = True
48 console.print("[yellow]Interrupt received, finishing current entity...[/yellow]")
51def _iter_entities(files: list, zip_output: bool):
52 for fpath in files:
53 if zip_output:
54 with ZipFile(fpath, "r") as zf:
55 data = orjson.loads(zf.read(zf.namelist()[0]))
56 else:
57 with open(fpath, "rb") as f:
58 data = orjson.loads(f.read())
59 for graph in data:
60 for entity in graph.get("@graph", []):
61 yield entity
64def _read_entity(
65 uri: str,
66 rdf_dir: str,
67 dir_split: int,
68 items_per_file: int,
69 zip_output: bool,
70) -> dict | None:
71 fpath = find_rdf_file(uri, rdf_dir, dir_split, items_per_file, zip_output)
72 if not os.path.exists(fpath):
73 return None
74 if zip_output:
75 with ZipFile(fpath, "r") as zf:
76 data = orjson.loads(zf.read(zf.namelist()[0]))
77 else:
78 with open(fpath, "rb") as f:
79 data = orjson.loads(f.read())
80 for graph in data:
81 for entity in graph.get("@graph", []):
82 if entity["@id"] == uri:
83 return entity
84 return None
87@dataclass
88class ResolvedCase:
89 br_uri: str
90 correct_part_of: str | None
91 to_remove: list[str]
92 method: str
93 reason: str
96# ── Phase 1: Scan ──
99def _scan_br_batch(files: list[str], zip_output: bool) -> list[tuple[str, list[str]]]:
100 results: list[tuple[str, list[str]]] = []
101 for entity in _iter_entities(files, zip_output):
102 if FRBR_PART_OF in entity:
103 parents = entity[FRBR_PART_OF]
104 if len(parents) > 1:
105 results.append((entity["@id"], [p["@id"] for p in parents]))
106 return results
109def scan_duplicate_part_of(
110 rdf_dir: str,
111 zip_output: bool,
112 workers: int = 4,
113 batch_size: int = BATCH_SIZE,
114) -> list[tuple[str, list[str]]]:
115 br_dir = os.path.join(rdf_dir, "br")
116 if zip_output:
117 br_files = collect_zip_files(br_dir, only_data=True)
118 else:
119 br_files = collect_files(br_dir, "*.json", lambda p: "prov" not in p)
121 batches = [
122 br_files[i : i + batch_size] for i in range(0, len(br_files), batch_size)
123 ]
124 all_results: list[tuple[str, list[str]]] = []
126 ctx = multiprocessing.get_context("forkserver")
127 with create_progress() as progress:
128 task = progress.add_task("Scanning BR files", total=len(br_files))
129 executor = ProcessPoolExecutor(
130 max_workers=workers, initializer=_worker_init, mp_context=ctx
131 )
132 try:
133 futures = {
134 executor.submit(_scan_br_batch, batch, zip_output): batch
135 for batch in batches
136 }
137 for future in as_completed(futures):
138 all_results.extend(future.result())
139 progress.advance(task, len(futures[future]))
140 finally:
141 executor.shutdown(wait=True)
143 return all_results
146# ── Phase 2: Build chain map (targeted file reads) ──
149def _get_title(entity: dict) -> str:
150 titles = entity.get(DCTERMS_TITLE, [])
151 if titles:
152 return titles[0].get("@value", "")
153 return ""
156def _get_types(entity: dict) -> frozenset[str]:
157 return frozenset(entity.get("@type", []))
160def _type_label(types: frozenset[str]) -> str:
161 return ", ".join(
162 sorted(t.rsplit("/", 1)[-1] for t in types if t != FABIO_EXPRESSION)
163 )
166def build_chain_map(
167 container_uris: set[str],
168 rdf_dir: str,
169 dir_split: int,
170 items_per_file: int,
171 zip_output: bool,
172) -> tuple[dict[str, list[str]], dict[str, tuple[str, frozenset[str]]]]:
173 chain_map: dict[str, list[str]] = {}
174 entity_meta: dict[str, tuple[str, frozenset[str]]] = {}
175 needed = set(container_uris)
176 depth = 0
178 with create_progress() as progress:
179 task = progress.add_task("Building chain map", total=len(needed))
180 while needed:
181 depth += 1
182 file_to_uris: dict[str, set[str]] = defaultdict(set)
183 for uri in needed:
184 fpath = find_rdf_file(
185 uri, rdf_dir, dir_split, items_per_file, zip_output
186 )
187 file_to_uris[fpath].add(uri)
189 next_needed: set[str] = set()
190 for fpath, uris in file_to_uris.items():
191 if not os.path.exists(fpath):
192 progress.advance(task, len(uris))
193 continue
194 for entity in _iter_entities([fpath], zip_output):
195 eid = entity["@id"]
196 if eid in uris:
197 entity_meta[eid] = (_get_title(entity), _get_types(entity))
198 if FRBR_PART_OF in entity:
199 parents = [p["@id"] for p in entity[FRBR_PART_OF]]
200 chain_map[eid] = parents
201 for p in parents:
202 if p not in chain_map and p not in needed:
203 next_needed.add(p)
204 progress.advance(task, len(uris))
206 if next_needed:
207 progress.update(
208 task,
209 description=f"Building chain map (depth {depth + 1})",
210 total=(progress.tasks[task].total or 0) + len(next_needed),
211 )
212 needed = next_needed
214 return chain_map, entity_meta
217# ── Phase 3: Classify & Resolve ──
220def _follow_to_venue(uri: str, chain_map: dict[str, list[str]]) -> str:
221 visited: set[str] = set()
222 current = uri
223 while current in chain_map:
224 if current in visited:
225 break
226 visited.add(current)
227 parents = chain_map[current]
228 if len(parents) != 1:
229 break
230 current = parents[0]
231 return current
234def resolve_cases(
235 raw_cases: list[tuple[str, list[str]]],
236 chain_map: dict[str, list[str]],
237 entity_meta: dict[str, tuple[str, frozenset[str]]],
238) -> list[ResolvedCase]:
239 resolved: list[ResolvedCase] = []
240 for br_uri, container_uris in raw_cases:
241 venues: list[str] = []
242 for c_uri in container_uris:
243 venues.append(_follow_to_venue(c_uri, chain_map))
245 venue_set = set(venues)
246 if len(venue_set) == 1:
247 sorted_containers = sorted(container_uris)
248 resolved.append(
249 ResolvedCase(
250 br_uri=br_uri,
251 correct_part_of=sorted_containers[0],
252 to_remove=sorted_containers[1:],
253 method="same_venue",
254 reason=f"All chains converge to {venues[0]}",
255 )
256 )
257 continue
259 venue_keys: set[tuple[str, frozenset[str]]] = set()
260 for v_uri in venue_set:
261 if v_uri in entity_meta:
262 title, types = entity_meta[v_uri]
263 normalized = " ".join(title.strip().lower().split())
264 venue_keys.add((normalized, types))
265 else:
266 venue_keys.add((v_uri, frozenset()))
268 if len(venue_keys) == 1:
269 sorted_containers = sorted(container_uris)
270 resolved.append(
271 ResolvedCase(
272 br_uri=br_uri,
273 correct_part_of=sorted_containers[0],
274 to_remove=sorted_containers[1:],
275 method="equivalent_venues",
276 reason=(
277 f"Chains reach different URIs "
278 f"({', '.join(sorted(venue_set))}) "
279 f"but same title/type"
280 ),
281 )
282 )
283 else:
284 venue_descs = []
285 for v_uri in sorted(venue_set):
286 if v_uri in entity_meta:
287 title, types = entity_meta[v_uri]
288 venue_descs.append(f"'{title}' ({_type_label(types)})")
289 else:
290 venue_descs.append(v_uri)
291 resolved.append(
292 ResolvedCase(
293 br_uri=br_uri,
294 correct_part_of=None,
295 to_remove=[],
296 method="manual_review",
297 reason=(
298 f"Chains reach different venues: {' vs '.join(venue_descs)}"
299 ),
300 )
301 )
302 return resolved
305# ── Enrich manual review cases ──
308def _batch_read_entities(
309 uris: set[str],
310 rdf_dir: str,
311 dir_split: int,
312 items_per_file: int,
313 zip_output: bool,
314) -> dict[str, dict]:
315 file_to_uris: dict[str, set[str]] = defaultdict(set)
316 for uri in uris:
317 file_to_uris[
318 find_rdf_file(uri, rdf_dir, dir_split, items_per_file, zip_output)
319 ].add(uri)
321 result: dict[str, dict] = {}
322 for fpath, target_uris in file_to_uris.items():
323 if not os.path.exists(fpath):
324 continue
325 for entity in _iter_entities([fpath], zip_output):
326 if entity["@id"] in target_uris:
327 result[entity["@id"]] = entity
328 return result
331def enrich_manual_review(
332 manual_cases: list[ResolvedCase],
333 raw_case_map: dict[str, list[str]],
334 entity_meta: dict[str, tuple[str, frozenset[str]]],
335 rdf_dir: str,
336 dir_split: int,
337 items_per_file: int,
338 zip_output: bool,
339) -> list[dict]:
340 with create_progress() as progress:
341 task = progress.add_task("Enriching manual cases", total=3)
343 br_uris = {res.br_uri for res in manual_cases}
344 br_entities = _batch_read_entities(
345 br_uris, rdf_dir, dir_split, items_per_file, zip_output
346 )
347 progress.advance(task)
349 id_uris: set[str] = set()
350 for br_entity in br_entities.values():
351 for id_ref in br_entity.get(HAS_IDENTIFIER, []):
352 id_uris.add(id_ref["@id"])
353 id_entities = _batch_read_entities(
354 id_uris, rdf_dir, dir_split, items_per_file, zip_output
355 )
356 progress.advance(task)
358 enriched = []
359 for res in manual_cases:
360 br_ids: list[str] = []
361 br_entity = br_entities.get(res.br_uri)
362 if br_entity:
363 for id_ref in br_entity.get(HAS_IDENTIFIER, []):
364 id_entity = id_entities.get(id_ref["@id"])
365 if id_entity:
366 scheme_list = id_entity.get(USES_ID_SCHEME, [])
367 value_list = id_entity.get(HAS_LITERAL_VALUE, [])
368 if scheme_list and value_list:
369 scheme = scheme_list[0]["@id"].rsplit("/", 1)[-1]
370 value = value_list[0]["@value"]
371 br_ids.append(f"{scheme}:{value}")
373 candidates = []
374 for c_uri in raw_case_map[res.br_uri]:
375 title, types = entity_meta.get(c_uri, ("", frozenset()))
376 candidates.append(
377 {"uri": c_uri, "title": title, "type": _type_label(types)}
378 )
380 enriched.append(
381 {
382 "br_uri": res.br_uri,
383 "identifiers": br_ids,
384 "candidates": candidates,
385 "reason": res.reason,
386 }
387 )
388 progress.advance(task)
389 return enriched
392# ── Fix ──
395def fix_br_part_of(
396 editor: MetaEditor,
397 br_uri: str,
398 correct_part_of_uri: str,
399 incorrect_part_of_uris: list[str],
400) -> None:
401 supplier_prefix = get_prefix(br_uri)
402 g_set = GraphSet(
403 editor.base_iri,
404 supplier_prefix=supplier_prefix,
405 custom_counter_handler=editor.counter_handler,
406 wanted_label=False,
407 )
409 file_paths: set[str] = set()
410 for uri in [br_uri, correct_part_of_uri] + incorrect_part_of_uris:
411 file_paths.add(
412 find_rdf_file(
413 uri,
414 editor.base_dir,
415 editor.dir_split,
416 editor.n_file_item,
417 editor.zip_output_rdf,
418 )
419 )
421 for fp in file_paths:
422 imported = editor.reader.load(fp)
423 if imported is not None:
424 editor.reader.import_entities_from_graph(g_set, imported, editor.resp_agent)
426 br_entity = g_set.get_entity(br_uri)
427 correct_container = g_set.get_entity(correct_part_of_uri)
428 assert br_entity is not None, f"BR not found: {br_uri}"
429 assert correct_container is not None, f"Container not found: {correct_part_of_uri}"
431 br_entity.remove_is_part_of() # type: ignore[attr-defined]
432 br_entity.is_part_of(correct_container) # type: ignore[attr-defined]
434 editor.save(g_set, supplier_prefix)
437# ── Orphan check ──
440def _check_orphan_batch(
441 files: list[str], zip_output: bool, target_uris: list[str]
442) -> set[str]:
443 target_set = set(target_uris)
444 referenced: set[str] = set()
445 for entity in _iter_entities(files, zip_output):
446 if FRBR_PART_OF in entity:
447 for parent in entity[FRBR_PART_OF]:
448 pid = parent["@id"]
449 if pid in target_set:
450 referenced.add(pid)
451 return referenced
454def check_orphans(
455 removed_uris: list[str],
456 rdf_dir: str,
457 zip_output: bool,
458 workers: int = 4,
459 batch_size: int = BATCH_SIZE,
460) -> list[str]:
461 if not removed_uris:
462 return []
464 removed_set = set(removed_uris)
465 br_dir = os.path.join(rdf_dir, "br")
466 if zip_output:
467 br_files = collect_zip_files(br_dir, only_data=True)
468 else:
469 br_files = collect_files(br_dir, "*.json", lambda p: "prov" not in p)
471 referenced: set[str] = set()
472 batches = [
473 br_files[i : i + batch_size] for i in range(0, len(br_files), batch_size)
474 ]
476 ctx = multiprocessing.get_context("forkserver")
477 with create_progress() as progress:
478 task = progress.add_task("Checking orphans", total=len(br_files))
479 executor = ProcessPoolExecutor(
480 max_workers=workers, initializer=_worker_init, mp_context=ctx
481 )
482 try:
483 futures = {
484 executor.submit(
485 _check_orphan_batch, batch, zip_output, list(removed_set)
486 ): batch
487 for batch in batches
488 }
489 for future in as_completed(futures):
490 referenced.update(future.result())
491 progress.advance(task, len(futures[future]))
492 finally:
493 executor.shutdown(wait=True)
495 return sorted(uri for uri in removed_uris if uri not in referenced)
498# ── Progress ──
501def _load_progress(path: str) -> set[str]:
502 if not os.path.exists(path):
503 return set()
504 with open(path) as f:
505 return set(json.load(f))
508def _save_progress(path: str, completed: set[str]) -> None:
509 with open(path, "w") as f:
510 json.dump(sorted(completed), f)
513# ── Report ──
516def _build_report(
517 resolved: list[ResolvedCase],
518 manual_enriched: list[dict],
519 orphans: list[str],
520) -> dict:
521 fixed = []
522 for res in resolved:
523 if res.correct_part_of is not None:
524 fixed.append(
525 {
526 "br_uri": res.br_uri,
527 "kept_part_of": res.correct_part_of,
528 "removed_part_of": res.to_remove,
529 "method": res.method,
530 "reason": res.reason,
531 }
532 )
534 return {
535 "summary": {
536 "total_affected": len(resolved),
537 "auto_fixed": len(fixed),
538 "manual_review": len(manual_enriched),
539 "orphaned_containers": len(orphans),
540 },
541 "fixed": fixed,
542 "manual_review": manual_enriched,
543 "orphaned_containers": orphans,
544 }
547# ── CLI ──
550def main() -> None: # pragma: no cover
551 parser = argparse.ArgumentParser(
552 description="Fix duplicate frbr:partOf values on bibliographic resources",
553 formatter_class=RichHelpFormatter,
554 )
555 parser.add_argument(
556 "-c", "--config", required=True, help="Path to meta_config.yaml"
557 )
558 parser.add_argument("-r", "--resp-agent", help="Responsible agent URI")
559 parser.add_argument(
560 "--dry-run", action="store_true", help="Report only, no modifications"
561 )
562 parser.add_argument(
563 "-w", "--workers", type=int, default=4, help="Parallel scan workers"
564 )
565 parser.add_argument(
566 "-b", "--batch-size", type=int, default=BATCH_SIZE, help="Files per batch"
567 )
568 parser.add_argument(
569 "--progress-file", default="fix_duplicate_part_of_progress.json"
570 )
571 parser.add_argument("--report-file", default="fix_duplicate_part_of_report.json")
572 args = parser.parse_args()
574 if not args.dry_run and not args.resp_agent:
575 parser.error("--resp-agent is required when not using --dry-run")
577 with open(args.config) as f:
578 settings = yaml.safe_load(f)
580 rdf_dir = os.path.join(settings["base_output_dir"], "rdf") + os.sep
581 dir_split = settings["dir_split_number"]
582 items_per_file = settings["items_per_file"]
583 zip_output = settings["zip_output_rdf"]
585 # Phase 1: Scan
586 console.print("[bold]Scanning RDF files for duplicate partOf...[/bold]")
587 raw_cases = scan_duplicate_part_of(
588 rdf_dir, zip_output, args.workers, args.batch_size
589 )
590 console.print(f"Found {len(raw_cases)} BRs with duplicate partOf")
592 if not raw_cases:
593 console.print("[green]No duplicate partOf found.[/green]")
594 return
596 # Phase 2: Build chain map
597 all_container_uris = {uri for _, containers in raw_cases for uri in containers}
598 console.print(
599 f"[bold]Building chain map for {len(all_container_uris)} containers...[/bold]"
600 )
601 chain_map, entity_meta = build_chain_map(
602 all_container_uris, rdf_dir, dir_split, items_per_file, zip_output
603 )
605 # Phase 3: Classify & resolve
606 resolved = resolve_cases(raw_cases, chain_map, entity_meta)
607 auto_fix = [r for r in resolved if r.correct_part_of is not None]
608 manual = [r for r in resolved if r.correct_part_of is None]
609 console.print(f"Auto-fixable: {len(auto_fix)}, Manual review: {len(manual)}")
611 # Enrich manual review cases for report
612 raw_case_map = {br: containers for br, containers in raw_cases}
613 manual_enriched = enrich_manual_review(
614 manual,
615 raw_case_map,
616 entity_meta,
617 rdf_dir,
618 dir_split,
619 items_per_file,
620 zip_output,
621 )
623 if args.dry_run:
624 report = _build_report(resolved, manual_enriched, [])
625 with open(args.report_file, "w") as f:
626 json.dump(report, f, indent=2)
627 console.print(f"\n[bold]Dry run report written to {args.report_file}[/bold]")
628 for res in manual:
629 console.print(f" [yellow]REVIEW[/yellow] {res.br_uri}: {res.reason}")
630 return
632 # Phase 4: Fix
633 editor = MetaEditor(args.config, args.resp_agent)
634 signal.signal(signal.SIGINT, _handle_signal)
635 signal.signal(signal.SIGTERM, _handle_signal)
637 completed = _load_progress(args.progress_file)
638 if completed:
639 console.print(f"Resuming: {len(completed)} already processed")
641 fixed_count = 0
642 failed_count = 0
644 with create_progress() as progress:
645 task = progress.add_task("Fixing partOf", total=len(auto_fix))
646 for res in auto_fix:
647 if _stop_requested:
648 console.print("[yellow]Interrupted, saving progress...[/yellow]")
649 break
650 if res.br_uri in completed:
651 progress.advance(task)
652 continue
653 try:
654 assert res.correct_part_of is not None
655 fix_br_part_of(editor, res.br_uri, res.correct_part_of, res.to_remove)
656 completed.add(res.br_uri)
657 _save_progress(args.progress_file, completed)
658 fixed_count += 1
659 except Exception as e:
660 console.print(f"[red]Error fixing {res.br_uri}: {e}[/red]")
661 failed_count += 1
662 progress.advance(task)
664 console.print(f"Fixed: {fixed_count}, Failed: {failed_count}")
666 # Phase 5: Orphan check
667 all_removed = [
668 uri for res in resolved if res.correct_part_of for uri in res.to_remove
669 ]
670 console.print("[bold]Checking for orphaned containers...[/bold]")
671 orphans = check_orphans(
672 all_removed, rdf_dir, zip_output, args.workers, args.batch_size
673 )
674 console.print(f"Orphaned containers: {len(orphans)}")
676 # Report
677 report = _build_report(resolved, manual_enriched, orphans)
678 with open(args.report_file, "w") as f:
679 json.dump(report, f, indent=2)
680 console.print(f"Report written to {args.report_file}")
682 if not _stop_requested and failed_count == 0 and os.path.exists(args.progress_file):
683 os.remove(args.progress_file)
686if __name__ == "__main__":
687 main()