Coverage for oc_meta / run / patches / fix_omid_mismatches.py: 0%
271 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-07-25 10:39 +0000
1#!/usr/bin/python
3# SPDX-FileCopyrightText: 2026 Arcangelo Massari <arcangelo.massari@unibo.it>
4#
5# SPDX-License-Identifier: ISC
7from __future__ import annotations
9import argparse
10import json
11import os
12import signal
13import time
14from collections import defaultdict
15from dataclasses import dataclass, field
16from urllib.parse import quote
18import requests
19import yaml
20from rapidfuzz.distance import Levenshtein
21from rich_argparse import RichHelpFormatter
23from oc_meta.core.editor import MetaEditor
24from oc_meta.run.merge.entities import EntityMerger, MergeRow
25from oc_meta.lib.bibliographic_matching import (
26 CROSSREF_RATE_LIMIT,
27 DATACITE_DOI,
28 DATACITE_HAS_ID,
29 DATACITE_USES_SCHEME,
30 LITERAL_HAS_VALUE,
31 MATCHING_THRESHOLD,
32 compute_matching_score,
33 fetch_crossref_metadata,
34 fetch_triplestore_metadata,
35)
36from oc_meta.lib.console import console, create_progress
37from oc_meta.lib.sparql import execute_sparql
39VENUE_TITLE_THRESHOLD = 0.6
41_stop_requested = False
44def _handle_signal(_signum: int, _frame: object) -> None:
45 global _stop_requested
46 _stop_requested = True
47 console.print("[yellow]Interrupt received, finishing current operation...[/yellow]")
50def _sparql_escape(value: str) -> str:
51 return value.replace("\\", "\\\\").replace('"', '\\"')
54@dataclass
55class MismatchCase:
56 doi_value: str
57 expected_entity: str
58 found_entity: str
59 column: str
60 category: str = ""
61 surviving_entity: str = ""
62 duplicate_entities: list[str] = field(default_factory=list)
63 duplicate_id_entities: list[str] = field(default_factory=list)
64 expected_doi: str = ""
65 found_doi: str = ""
66 action: str = ""
67 reason: str = ""
68 validation: str = ""
69 occurrence_count: int = 1
72def _get_doi(endpoint: str, entity_uri: str) -> str:
73 query = f"""
74 SELECT ?val WHERE {{
75 <{entity_uri}> <{DATACITE_HAS_ID}> ?id .
76 ?id <{DATACITE_USES_SCHEME}> <{DATACITE_DOI}> .
77 ?id <{LITERAL_HAS_VALUE}> ?val .
78 }}
79 LIMIT 1
80 """
81 result = execute_sparql(endpoint, query)
82 bindings = result["results"]["bindings"]
83 return bindings[0]["val"]["value"] if bindings else ""
86def _find_id_entity(endpoint: str, br_uri: str, doi_value: str) -> str:
87 escaped = _sparql_escape(doi_value)
88 query = f"""
89 SELECT ?id_entity WHERE {{
90 <{br_uri}> <{DATACITE_HAS_ID}> ?id_entity .
91 ?id_entity <{DATACITE_USES_SCHEME}> <{DATACITE_DOI}> .
92 ?id_entity <{LITERAL_HAS_VALUE}> "{escaped}" .
93 }}
94 LIMIT 1
95 """
96 result = execute_sparql(endpoint, query)
97 bindings = result["results"]["bindings"]
98 return bindings[0]["id_entity"]["value"] if bindings else ""
101def _find_entity_by_doi(endpoint: str, doi_value: str) -> str:
102 escaped = _sparql_escape(doi_value)
103 query = f"""
104 SELECT ?entity WHERE {{
105 ?id <{LITERAL_HAS_VALUE}> "{escaped}" .
106 ?id <{DATACITE_USES_SCHEME}> <{DATACITE_DOI}> .
107 ?entity <{DATACITE_HAS_ID}> ?id .
108 }}
109 LIMIT 1
110 """
111 result = execute_sparql(endpoint, query)
112 bindings = result["results"]["bindings"]
113 return bindings[0]["entity"]["value"] if bindings else ""
116def load_errors(check_results_path: str) -> list[dict]:
117 with open(check_results_path) as f:
118 data = json.load(f)
119 return [
120 e
121 for e in data["errors"]
122 if e["type"] == "omid_mismatch" and e["schema"] == "doi"
123 ]
126def build_cases(errors: list[dict]) -> list[MismatchCase]:
127 key_to_case: dict[str, MismatchCase] = {}
128 for e in errors:
129 found = e["found_omids"][0]
130 key = f"{e['expected_omid']}|{found}"
131 if key not in key_to_case:
132 key_to_case[key] = MismatchCase(
133 doi_value=e["value"],
134 expected_entity=e["expected_omid"],
135 found_entity=found,
136 column=e["column"],
137 )
138 else:
139 key_to_case[key].occurrence_count += 1
140 return list(key_to_case.values())
143def classify_and_resolve(
144 cases: list[MismatchCase], endpoint: str
145) -> list[MismatchCase]:
146 with create_progress() as progress:
147 task = progress.add_task("Classifying cases", total=len(cases))
149 for case in cases:
150 if _stop_requested:
151 break
153 case.expected_doi = _get_doi(endpoint, case.expected_entity)
154 case.found_doi = _get_doi(endpoint, case.found_entity)
156 if "sj....bjc" in case.doi_value:
157 _classify_bjc(case, endpoint)
158 elif "(asce)" in case.doi_value.lower():
159 _classify_asce(case, endpoint)
160 elif case.expected_doi.endswith("<"):
161 _classify_angle_bracket(case, endpoint)
162 elif case.expected_doi.endswith(";") and not case.found_doi.endswith(";"):
163 case.category = "false_positive"
164 case.action = "false_positive"
165 case.reason = f"Semicolon is part of DOI '{case.expected_doi}'"
166 elif "#" in case.expected_doi and "#" not in case.found_doi:
167 case.category = "false_positive"
168 case.action = "false_positive"
169 case.reason = f"Fragment is part of DOI '{case.expected_doi}'"
170 elif (
171 case.expected_doi == case.doi_value + "."
172 or case.expected_doi == case.doi_value + "..."
173 or case.expected_doi.rstrip(".") == case.doi_value.rstrip(".")
174 ):
175 _classify_trailing_period(case, endpoint)
176 else:
177 case.category = "manual_review"
178 case.action = "manual_review"
179 case.reason = (
180 f"Unrecognized pattern: expected='{case.expected_doi}'"
181 f" found='{case.found_doi}'"
182 )
184 progress.advance(task)
186 return cases
189def _classify_trailing_period(case: MismatchCase, endpoint: str) -> None:
190 case.category = "trailing_period"
191 case.surviving_entity = case.expected_entity
192 case.duplicate_entities = [case.found_entity]
194 found_id = _find_id_entity(endpoint, case.found_entity, case.found_doi)
195 case.duplicate_id_entities = [found_id] if found_id else []
196 case.action = "merge"
197 case.reason = (
198 f"DOI '{case.expected_doi}' (with .) is correct; "
199 f"'{case.found_doi}' (without .) is duplicate"
200 )
203def _classify_bjc(case: MismatchCase, endpoint: str) -> None:
204 case.category = "bjc"
205 correct_doi = case.doi_value.replace("sj....bjc", "sj.bjc")
206 correct_entity = _find_entity_by_doi(endpoint, correct_doi)
208 if not correct_entity:
209 correct_doi_dot = correct_doi + "."
210 correct_entity = _find_entity_by_doi(endpoint, correct_doi_dot)
212 if not correct_entity:
213 case.action = "manual_review"
214 case.reason = f"Correct entity for '{correct_doi}' not found in triplestore"
215 return
217 case.surviving_entity = correct_entity
218 case.duplicate_entities = [case.expected_entity, case.found_entity]
220 exp_id = _find_id_entity(endpoint, case.expected_entity, case.expected_doi)
221 found_id = _find_id_entity(endpoint, case.found_entity, case.found_doi)
222 case.duplicate_id_entities = [x for x in [exp_id, found_id] if x]
223 case.action = "merge"
224 case.reason = (
225 f"BJC: merge corrupted entities into correct entity "
226 f"{correct_entity} (DOI {correct_doi})"
227 )
230def _classify_asce(case: MismatchCase, endpoint: str) -> None:
231 case.category = "asce"
232 case.surviving_entity = case.expected_entity
233 case.duplicate_entities = [case.found_entity]
235 found_id = _find_id_entity(endpoint, case.found_entity, case.found_doi)
236 case.duplicate_id_entities = [found_id] if found_id else []
237 case.action = "merge"
238 case.reason = (
239 f"ASCE: full DOI '{case.expected_doi}' on expected; "
240 f"truncated '{case.found_doi}' on found"
241 )
244def _classify_angle_bracket(case: MismatchCase, endpoint: str) -> None:
245 case.category = "angle_bracket"
246 case.surviving_entity = case.expected_entity
247 case.duplicate_entities = [case.found_entity]
249 found_id = _find_id_entity(endpoint, case.found_entity, case.found_doi)
250 case.duplicate_id_entities = [found_id] if found_id else []
251 case.action = "merge"
252 case.reason = (
253 f"'<' is part of DOI '{case.expected_doi}'; '{case.found_doi}' is truncated"
254 )
257def _doi_resolves(doi: str, mailto: str) -> bool:
258 time.sleep(1.0 / CROSSREF_RATE_LIMIT)
259 try:
260 r = requests.head(
261 f"https://doi.org/{quote(doi, safe='')}",
262 headers={"User-Agent": f"oc_meta/mailto:{mailto}"},
263 allow_redirects=False,
264 timeout=10,
265 )
266 return r.status_code in (301, 302, 303)
267 except requests.RequestException:
268 return False
271def validate_cases(
272 cases: list[MismatchCase], endpoint: str, mailto: str
273) -> list[MismatchCase]:
274 crossref_cache: dict[str, dict | None] = {}
276 with create_progress() as progress:
277 task = progress.add_task("Validating via API", total=len(cases))
279 for case in cases:
280 if _stop_requested:
281 break
283 if case.action != "merge":
284 progress.advance(task)
285 continue
287 surviving_doi = ""
288 if case.category == "bjc":
289 surviving_doi = case.doi_value.replace("sj....bjc", "sj.bjc")
290 else:
291 surviving_doi = case.expected_doi
293 if not _doi_resolves(surviving_doi, mailto):
294 case.action = "manual_review"
295 case.reason += (
296 f" [VALIDATION FAILED: DOI '{surviving_doi}'"
297 f" does not resolve on doi.org]"
298 )
299 case.validation = "doi_not_resolved"
300 progress.advance(task)
301 continue
303 if case.column == "venue":
304 ts_surv = fetch_triplestore_metadata(endpoint, case.surviving_entity)
305 dup_entity = case.duplicate_entities[0]
306 ts_dup = fetch_triplestore_metadata(endpoint, dup_entity)
307 if ts_surv and ts_dup and ts_surv["title"] and ts_dup["title"]:
308 max_len = max(len(ts_surv["title"]), len(ts_dup["title"]))
309 dist = Levenshtein.distance(ts_surv["title"], ts_dup["title"])
310 sim = 1.0 - dist / max_len
311 if sim >= VENUE_TITLE_THRESHOLD:
312 case.validation = f"venue_title_match:{sim:.2f}"
313 else:
314 case.action = "manual_review"
315 case.reason += (
316 f" [VALIDATION FAILED: venue title similarity {sim:.2f}]"
317 )
318 case.validation = f"venue_title_mismatch:{sim:.2f}"
319 else:
320 case.action = "manual_review"
321 case.reason += (
322 " [HELD: no venue titles to confirm the two"
323 " entities are the same journal]"
324 )
325 case.validation = "doi_resolved_no_metadata"
326 else:
327 cr_meta = fetch_crossref_metadata(surviving_doi, crossref_cache, mailto)
328 if cr_meta:
329 best_score = 0.0
330 any_ts_meta = False
331 entities_to_check = [
332 *case.duplicate_entities,
333 case.surviving_entity,
334 ]
335 for ent in entities_to_check:
336 ts_meta = fetch_triplestore_metadata(endpoint, ent)
337 if not ts_meta or not ts_meta["title"]:
338 continue
339 any_ts_meta = True
340 score = compute_matching_score(ts_meta, cr_meta)
341 if score > best_score:
342 best_score = score
344 if not any_ts_meta:
345 case.action = "manual_review"
346 case.reason += (
347 " [HELD: no triplestore metadata to confirm"
348 " the two entities are the same work]"
349 )
350 case.validation = "doi_resolved_no_ts_metadata"
351 elif best_score >= MATCHING_THRESHOLD:
352 case.validation = f"crossref_match:{best_score:.1f}"
353 else:
354 case.action = "manual_review"
355 case.reason += (
356 f" [VALIDATION FAILED: best score"
357 f" {best_score:.1f} < {MATCHING_THRESHOLD}]"
358 )
359 case.validation = f"crossref_mismatch:{best_score:.1f}"
360 else:
361 case.action = "manual_review"
362 case.reason += (
363 " [HELD: DOI not found on Crossref, cannot"
364 " confirm the two entities are the same work]"
365 )
366 case.validation = "doi_resolved_not_on_crossref"
368 progress.advance(task)
370 return cases
373def build_report(cases: list[MismatchCase]) -> dict:
374 actions: dict[str, list[dict]] = defaultdict(list)
375 for case in cases:
376 entry = {
377 "doi_value": case.doi_value,
378 "category": case.category,
379 "expected_entity": case.expected_entity,
380 "expected_doi": case.expected_doi,
381 "found_entity": case.found_entity,
382 "found_doi": case.found_doi,
383 "surviving_entity": case.surviving_entity,
384 "duplicate_entities": case.duplicate_entities,
385 "duplicate_id_entities": case.duplicate_id_entities,
386 "reason": case.reason,
387 "validation": case.validation,
388 "occurrence_count": case.occurrence_count,
389 }
390 actions[case.action].append(entry)
392 categories = defaultdict(int)
393 for case in cases:
394 categories[case.category] += 1
396 return {
397 "summary": {
398 "total_unique_cases": len(cases),
399 "total_occurrences": sum(c.occurrence_count for c in cases),
400 "merge": len(actions.get("merge", [])),
401 "false_positive": len(actions.get("false_positive", [])),
402 "manual_review": len(actions.get("manual_review", [])),
403 "by_category": dict(categories),
404 },
405 "merge": actions.get("merge", []),
406 "false_positive": actions.get("false_positive", []),
407 "manual_review": actions.get("manual_review", []),
408 }
411def _load_progress(path: str) -> set[str]:
412 if os.path.exists(path):
413 with open(path) as f:
414 return set(json.load(f))
415 return set()
418def _save_progress(path: str, completed: set[str]) -> None:
419 with open(path, "w") as f:
420 json.dump(sorted(completed), f)
423def execute_merges(
424 cases: list[MismatchCase],
425 config_path: str,
426 resp_agent: str,
427) -> None:
428 actionable = [c for c in cases if c.action == "merge"]
429 if not actionable:
430 console.print("[green]No merges to execute.[/green]")
431 return
433 rows: list[MergeRow] = [
434 {
435 "surviving_entity": case.surviving_entity,
436 "merged_entities": list(case.duplicate_entities),
437 }
438 for case in actionable
439 ]
440 EntityMerger(config_path, resp_agent).process_rows(rows)
442 console.print(f"Merged: {len(actionable)} cases")
443 console.print(
444 "[bold]The merge wrote RDF files only. Re-index the triplestore from the "
445 "files, then rerun with --cleanup to remove the duplicate identifiers.[/bold]"
446 )
449def execute_cleanup(
450 report_file: str,
451 config_path: str,
452 resp_agent: str,
453 progress_file: str,
454) -> None:
455 with open(report_file) as f:
456 report = json.load(f)
458 entries = [e for e in report["merge"] if e["duplicate_id_entities"]]
459 if not entries:
460 console.print("[green]No identifiers to clean up.[/green]")
461 return
463 editor = MetaEditor(config_path, resp_agent)
464 signal.signal(signal.SIGINT, _handle_signal)
465 signal.signal(signal.SIGTERM, _handle_signal)
467 completed = _load_progress(progress_file)
468 if completed:
469 console.print(f"Resuming: {len(completed)} already processed")
471 with create_progress() as progress:
472 task = progress.add_task("Removing duplicate identifiers", total=len(entries))
473 for entry in entries:
474 if _stop_requested:
475 console.print("[yellow]Interrupted, saving progress...[/yellow]")
476 break
478 entry_key = (
479 f"{entry['surviving_entity']}|{','.join(entry['duplicate_entities'])}"
480 )
481 if entry_key in completed:
482 progress.advance(task)
483 continue
485 for id_ent in entry["duplicate_id_entities"]:
486 editor.delete(entry["surviving_entity"], DATACITE_HAS_ID, id_ent)
487 editor.delete(id_ent)
489 completed.add(entry_key)
490 _save_progress(progress_file, completed)
491 progress.advance(task)
493 if not _stop_requested and os.path.exists(progress_file):
494 os.remove(progress_file)
497def main() -> None: # pragma: no cover
498 parser = argparse.ArgumentParser(
499 description="Fix omid_mismatch errors by merging duplicate entities",
500 formatter_class=RichHelpFormatter,
501 )
502 parser.add_argument(
503 "-c", "--config", required=True, help="Path to meta_config.yaml"
504 )
505 parser.add_argument("--check-results", help="Path to check_results.json")
506 parser.add_argument("-r", "--resp-agent", help="Responsible agent URI")
507 parser.add_argument(
508 "--mailto",
509 help="Email for the Crossref / doi.org polite pool User-Agent",
510 )
511 mode = parser.add_mutually_exclusive_group(required=True)
512 mode.add_argument(
513 "--dry-run",
514 action="store_true",
515 dest="dry_run",
516 help="Report only, no modifications",
517 )
518 mode.add_argument(
519 "--no-dry-run",
520 action="store_true",
521 dest="no_dry_run",
522 help="Merge the duplicates into the RDF files",
523 )
524 mode.add_argument(
525 "--cleanup",
526 action="store_true",
527 help="Remove the duplicate identifiers listed in the report "
528 "(run after re-indexing the triplestore from the merged files)",
529 )
530 parser.add_argument(
531 "--report-file",
532 default="fix_omid_mismatches_report.json",
533 help="Output report path",
534 )
535 parser.add_argument(
536 "--progress-file",
537 default="fix_omid_mismatches_progress.json",
538 help="Progress tracking file",
539 )
540 args = parser.parse_args()
542 if (args.no_dry_run or args.cleanup) and not args.resp_agent:
543 parser.error("--resp-agent is required with --no-dry-run and --cleanup")
545 if args.cleanup:
546 execute_cleanup(
547 args.report_file, args.config, args.resp_agent, args.progress_file
548 )
549 return
551 if not args.check_results or not args.mailto:
552 parser.error(
553 "--check-results and --mailto are required with --dry-run and --no-dry-run"
554 )
556 with open(args.config) as f:
557 settings = yaml.safe_load(f)
559 endpoint = settings["triplestore_url"]
561 console.print("[bold]Loading check_results.json...[/bold]")
562 errors = load_errors(args.check_results)
563 console.print(f" omid_mismatch DOI errors: {len(errors)}")
565 console.print("\n[bold]Building unique cases...[/bold]")
566 cases = build_cases(errors)
567 console.print(f" Unique (expected, found) pairs: {len(cases)}")
569 console.print("\n[bold]Classifying and resolving...[/bold]")
570 cases = classify_and_resolve(cases, endpoint)
572 console.print("\n[bold]Validating surviving DOIs via API...[/bold]")
573 cases = validate_cases(cases, endpoint, args.mailto)
575 report = build_report(cases)
576 with open(args.report_file, "w") as f:
577 json.dump(report, f, indent=2)
579 summary = report["summary"]
580 console.print("\n[bold]Summary:[/bold]")
581 console.print(f" Unique cases: {summary['total_unique_cases']}")
582 console.print(f" Total occurrences: {summary['total_occurrences']}")
583 console.print(f" Merge: {summary['merge']}")
584 console.print(f" False positive: {summary['false_positive']}")
585 console.print(f" Manual review: {summary['manual_review']}")
586 console.print(f" By category: {summary['by_category']}")
587 console.print(f"\nReport: {args.report_file}")
589 if args.no_dry_run:
590 if summary["merge"]:
591 console.print("\n[bold]Applying merges...[/bold]")
592 execute_merges(cases, args.config, args.resp_agent)
593 else:
594 console.print(
595 f"\n[dim]Dry run complete. {summary['merge']} merges pending."
596 f" Use --no-dry-run to apply.[/dim]"
597 )
600if __name__ == "__main__":
601 main()