Coverage for oc_meta/run/patches/fix_corrupted

1#!/usr/bin/python

3# SPDX-FileCopyrightText: 2026 Arcangelo Massari <arcangelo.massari@unibo.it>

5# SPDX-License-Identifier: ISC

7from __future__ import annotations

9import argparse

10import json

11import os

12import signal

13from collections import defaultdict

14from dataclasses import dataclass, field

16import yaml

17from rich_argparse import RichHelpFormatter

19from oc_meta.core.editor import MetaEditor

20from oc_meta.run.merge.entities import EntityMerger, MergeRow

21from oc_meta.lib.bibliographic_matching import (

22 DATACITE_DOI,

23 DATACITE_HAS_ID,

24 DATACITE_USES_SCHEME,

25 LITERAL_HAS_VALUE,

26 MATCHING_THRESHOLD,

27 compute_matching_score,

28 fetch_crossref_metadata,

29 fetch_triplestore_metadata,

30)

31from oc_meta.lib.console import console, create_progress

32from oc_meta.lib.sparql import execute_sparql

34_stop_requested = False

37def _handle_signal(_signum: int, _frame: object) -> None:

38 global _stop_requested

39 _stop_requested = True

40 console.print("[yellow]Interrupt received, finishing current operation...[/yellow]")

43def _sparql_escape(value: str) -> str:

44 return value.replace("\\", "\\\\").replace('"', '\\"')

47@dataclass

48class CorrectionCase:

49 truncated_doi: str

50 candidate_doi: str

51 duplicate_entity: str

52 surviving_entity: str

53 duplicate_id_entity: str | None = None

54 action: str = ""

55 matching_score: float = 0.0

56 is_1_to_n: bool = False

57 all_expected_omids: list[str] = field(default_factory=list)

58 reason: str = ""

61def extract_sici_mismatch_errors(errors: list[dict]) -> list[dict]:

62 return [

63 e

64 for e in errors

65 if e["type"] == "omid_mismatch"

66 and e["schema"] == "doi"

67 and e["value"].lower().endswith("co;2-")

68 ]

71def build_sici_cases(errors: list[dict]) -> list[CorrectionCase]:

72 found_to_errors: dict[str, list[dict]] = defaultdict(list)

73 for error in errors:

74 found_to_errors[error["found_omids"][0]].append(error)

76 cases = []

77 for found_omid, group in found_to_errors.items():

78 first = group[0]

79 all_expected = [e["expected_omid"] for e in group]

80 cases.append(

81 CorrectionCase(

82 truncated_doi=first["value"],

83 candidate_doi=first["value"] + "#",

84 duplicate_entity=found_omid,

85 surviving_entity=first["expected_omid"],

86 is_1_to_n=len(group) > 1,

87 all_expected_omids=all_expected,

88 reason=f"{len(group)} expected_omid(s)" if len(group) > 1 else "",

89 )

90 )

91 return cases

94def _find_id_entity(endpoint: str, br_uri: str, doi_value: str) -> str | None:

95 escaped = _sparql_escape(doi_value)

96 query = f"""

97 SELECT ?id_entity WHERE {{

98 <{br_uri}> <{DATACITE_HAS_ID}> ?id_entity .

99 ?id_entity <{DATACITE_USES_SCHEME}> <{DATACITE_DOI}> .

100 ?id_entity <{LITERAL_HAS_VALUE}> "{escaped}" .

101 }}

102 LIMIT 1

103 """

104 result = execute_sparql(endpoint, query)

105 bindings = result["results"]["bindings"]

106 return bindings[0]["id_entity"]["value"] if bindings else None

107

108

109def determine_actions(

110 cases: list[CorrectionCase],

111 endpoint: str,

112 mailto: str,

113) -> list[CorrectionCase]:

114 crossref_cache: dict[str, dict | None] = {}

115

116 with create_progress() as progress:

117 task = progress.add_task(

118 "Fetching metadata and computing scores", total=len(cases)

119 )

120

121 for case in cases:

122 if _stop_requested:

123 break

124

125 crossref_meta = fetch_crossref_metadata(

126 case.candidate_doi, crossref_cache, mailto

127 )

128 if crossref_meta is None:

129 case.action = "manual_review"

130 case.reason += " Candidate DOI not found on Crossref."

131 progress.advance(task)

132 continue

133

134 ts_meta = fetch_triplestore_metadata(endpoint, case.duplicate_entity)

135 if not ts_meta:

136 case.action = "manual_review"

137 case.reason += " No metadata found in triplestore for entity."

138 progress.advance(task)

139 continue

140

141 case.matching_score = compute_matching_score(ts_meta, crossref_meta)

142

143 if case.matching_score < MATCHING_THRESHOLD:

144 case.action = "manual_review"

145 case.reason += f" Matching score {case.matching_score:.1f} below threshold {MATCHING_THRESHOLD}."

146 progress.advance(task)

147 continue

148

149 _resolve_id_entity(case, endpoint)

150 case.action = "merge"

151 progress.advance(task)

152

153 return cases

154

155

156def _resolve_id_entity(case: CorrectionCase, endpoint: str) -> None:

157 if not case.duplicate_id_entity:

158 case.duplicate_id_entity = _find_id_entity(

159 endpoint, case.duplicate_entity, case.truncated_doi

160 )

161

162

163def build_report(cases: list[CorrectionCase]) -> dict:

164 actions: dict[str, list[dict]] = defaultdict(list)

165 for case in cases:

166 entry = {

167 "truncated_doi": case.truncated_doi,

168 "correct_doi": case.candidate_doi,

169 "duplicate_entity": case.duplicate_entity,

170 "duplicate_id_entity": case.duplicate_id_entity,

171 "surviving_entity": case.surviving_entity,

172 "matching_score": round(case.matching_score, 2),

173 "reason": case.reason,

174 }

175 if case.is_1_to_n:

176 entry["all_expected_omids"] = case.all_expected_omids

177 actions[case.action].append(entry)

178

179 merge_cases = [c for c in cases if c.action == "merge"]

180 entities_removed = sum(1 + len(c.all_expected_omids[1:]) for c in merge_cases)

181

182 return {

183 "summary": {

184 "total": len(cases),

185 "auto_merge": len(actions.get("merge", [])),

186 "manual_review": len(actions.get("manual_review", [])),

187 "entities_removed": entities_removed,

188 },

189 "merge": actions.get("merge", []),

190 "manual_review": actions.get("manual_review", []),

191 }

192

193

194def _load_progress(path: str) -> set[str]:

195 if os.path.exists(path):

196 with open(path) as f:

197 return set(json.load(f))

198 return set()

199

200

201def _save_progress(path: str, completed: set[str]) -> None:

202 with open(path, "w") as f:

203 json.dump(sorted(completed), f)

204

205

206def execute_actions(

207 cases: list[CorrectionCase],

208 config_path: str,

209 resp_agent: str,

210) -> None:

211 actionable = [c for c in cases if c.action == "merge"]

212 if not actionable:

213 console.print("[green]No actions to execute.[/green]")

214 return

215

216 rows: list[MergeRow] = [

217 {

218 "surviving_entity": case.surviving_entity,

219 "merged_entities": [case.duplicate_entity, *case.all_expected_omids[1:]],

220 }

221 for case in actionable

222 ]

223 EntityMerger(config_path, resp_agent).process_rows(rows)

224

225 console.print(f"Merged: {len(actionable)} cases")

226 console.print(

227 "[bold]The merge wrote RDF files only. Re-index the triplestore from the "

228 "files, then rerun with --cleanup to remove the corrupted identifiers.[/bold]"

229 )

230

231

232def execute_cleanup(

233 report_file: str,

234 config_path: str,

235 resp_agent: str,

236 progress_file: str,

237) -> None:

238 with open(report_file) as f:

239 report = json.load(f)

240

241 entries = [e for e in report["merge"] if e["duplicate_id_entity"]]

242 if not entries:

243 console.print("[green]No identifiers to clean up.[/green]")

244 return

245

246 editor = MetaEditor(config_path, resp_agent)

247 signal.signal(signal.SIGINT, _handle_signal)

248 signal.signal(signal.SIGTERM, _handle_signal)

249

250 completed = _load_progress(progress_file)

251 if completed:

252 console.print(f"Resuming: {len(completed)} already processed")

253

254 with create_progress() as progress:

255 task = progress.add_task("Removing corrupted identifiers", total=len(entries))

256 for entry in entries:

257 if _stop_requested:

258 console.print("[yellow]Interrupted, saving progress...[/yellow]")

259 break

260

261 if entry["duplicate_entity"] in completed:

262 progress.advance(task)

263 continue

264

265 editor.delete(

266 entry["surviving_entity"],

267 DATACITE_HAS_ID,

268 entry["duplicate_id_entity"],

269 )

270 editor.delete(entry["duplicate_id_entity"])

271

272 completed.add(entry["duplicate_entity"])

273 _save_progress(progress_file, completed)

274 progress.advance(task)

275

276 if not _stop_requested and os.path.exists(progress_file):

277 os.remove(progress_file)

278

279

280def main() -> None: # pragma: no cover

281 parser = argparse.ArgumentParser(

282 description="Fix DOIs corrupted by oc_ds_converter suffix_regex bug",

283 formatter_class=RichHelpFormatter,

284 )

285 parser.add_argument(

286 "-c", "--config", required=True, help="Path to meta_config.yaml"

287 )

288 parser.add_argument(

289 "--check-results",

290 help="Path to check_results.json",

291 )

292 parser.add_argument("-r", "--resp-agent", help="Responsible agent URI")

293 parser.add_argument(

294 "--mailto",

295 help="Email for the Crossref polite pool User-Agent",

296 )

297 mode = parser.add_mutually_exclusive_group(required=True)

298 mode.add_argument(

299 "--dry-run",

300 action="store_true",

301 dest="dry_run",

302 help="Report only, no modifications",

303 )

304 mode.add_argument(

305 "--no-dry-run",

306 action="store_true",

307 dest="no_dry_run",

308 help="Merge the duplicates into the RDF files",

309 )

310 mode.add_argument(

311 "--cleanup",

312 action="store_true",

313 help="Remove the corrupted identifiers listed in the report "

314 "(run after re-indexing the triplestore from the merged files)",

315 )

316 parser.add_argument(

317 "--report-file",

318 default="fix_corrupted_dois_report.json",

319 help="Output report path",

320 )

321 parser.add_argument(

322 "--progress-file",

323 default="fix_corrupted_dois_progress.json",

324 help="Progress tracking file for resumability",

325 )

326 args = parser.parse_args()

327

328 if (args.no_dry_run or args.cleanup) and not args.resp_agent:

329 parser.error("--resp-agent is required with --no-dry-run and --cleanup")

330

331 if args.cleanup:

332 execute_cleanup(

333 args.report_file, args.config, args.resp_agent, args.progress_file

334 )

335 return

336

337 if not args.check_results or not args.mailto:

338 parser.error(

339 "--check-results and --mailto are required with --dry-run and --no-dry-run"

340 )

341

342 with open(args.config) as f:

343 settings = yaml.safe_load(f)

344

345 endpoint = settings["triplestore_url"]

346

347 console.print("[bold]Loading check_results.json...[/bold]")

348 with open(args.check_results) as f:

349 check_data = json.load(f)

350

351 sici_errors = extract_sici_mismatch_errors(check_data["errors"])

352 console.print(f" SICI mismatch errors: {len(sici_errors)}")

353

354 console.print("\n[bold]Building correction cases...[/bold]")

355 cases = build_sici_cases(sici_errors)

356 console.print(f" Unique cases: {len(cases)}")

357

358 if not cases:

359 console.print("[green]No correction cases found.[/green]")

360 return

361

362 console.print("\n[bold]Fetching metadata and computing matching scores...[/bold]")

Coverage for oc_meta / run / patches / fix_corrupted_dois.py: 0%

140 statements