Coverage for oc_meta / run / patches / has_next.py: 19%

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

2# 

3# SPDX-License-Identifier: ISC 

4 

5from __future__ import annotations 

6 

7import argparse 

8import csv 

9import os 

10import re 

11import time 

12import unicodedata 

13import xml.etree.ElementTree as ET 

14from collections import defaultdict 

15from datetime import datetime, timezone 

16from typing import Dict, List, Optional, Tuple, cast 

17 

18import orjson 

19import redis 

20import requests 

21import yaml 

22from oc_ocdm.graph import GraphSet 

23from rich_argparse import RichHelpFormatter 

24 

25from oc_ds_converter.crossref.crossref_processing import CrossrefProcessing 

26from oc_ds_converter.datacite.datacite_processing import DataciteProcessing 

27from oc_ds_converter.pubmed.pubmed_processing import PubmedProcessing 

28from oc_ds_converter.oc_idmanager.oc_data_storage.in_memory_manager import ( 

29 InMemoryStorageManager, 

30) 

31from oc_ds_converter.ra_processor import RaProcessor 

32 

33from oc_meta.core.editor import MetaEditor 

34from oc_meta.lib.agent_metadata import ( 

35 AgentMetadata, 

36 JsonObject, 

37 parse_crossref_work, 

38 parse_datacite_work, 

39) 

40from oc_meta.lib.console import create_progress 

41from oc_meta.lib.file_manager import find_rdf_file 

42from oc_meta.run.meta.generate_csv import URI_TYPE_DICT, load_json_from_file 

43 

44HAS_IDENTIFIER = "http://purl.org/spar/datacite/hasIdentifier" 

45USES_ID_SCHEME = "http://purl.org/spar/datacite/usesIdentifierScheme" 

46HAS_LITERAL_VALUE = ( 

47 "http://www.essepuntato.it/2010/06/literalreification/hasLiteralValue" 

48) 

49IS_DOC_CONTEXT_FOR = "http://purl.org/spar/pro/isDocumentContextFor" 

50WITH_ROLE = "http://purl.org/spar/pro/withRole" 

51IS_HELD_BY = "http://purl.org/spar/pro/isHeldBy" 

52HAS_NEXT = "https://w3id.org/oc/ontology/hasNext" 

53FAMILY_NAME = "http://xmlns.com/foaf/0.1/familyName" 

54GIVEN_NAME = "http://xmlns.com/foaf/0.1/givenName" 

55FOAF_NAME = "http://xmlns.com/foaf/0.1/name" 

56DC_TITLE = "http://purl.org/dc/terms/title" 

57FABIO_EXPRESSION = "http://purl.org/spar/fabio/Expression" 

58 

59ROLE_MAP = { 

60 "http://purl.org/spar/pro/author": "author", 

61 "http://purl.org/spar/pro/editor": "editor", 

62 "http://purl.org/spar/pro/publisher": "publisher", 

63} 

64 

65CSV_COLUMNS = [ 

66 "id", 

67 "title", 

68 "author", 

69 "pub_date", 

70 "venue", 

71 "volume", 

72 "issue", 

73 "page", 

74 "type", 

75 "publisher", 

76 "editor", 

77] 

78 

79CROSSREF_BASE = "https://api.crossref.org/works/" 

80DATACITE_BASE = "https://api.datacite.org/dois/" 

81PUBMED_BASE = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils/efetch.fcgi" 

82 

83SESSION = requests.Session() 

84 

85 

86class RedisOrcidIndex: 

87 def __init__(self, host: str, port: int, db: int) -> None: 

88 self._r = redis.Redis(host=host, port=port, db=db, decode_responses=True) 

89 

90 @staticmethod 

91 def _key(doi: str) -> str: 

92 return doi if doi.startswith("doi:") else f"doi:{doi}" 

93 

94 def get_value(self, doi: str) -> Optional[set[str]]: 

95 members = cast("set[str]", self._r.smembers(self._key(doi))) 

96 return members or None 

97 

98 def get_values_batch(self, dois: List[str]) -> Dict[str, set[str]]: 

99 if not dois: 

100 return {} 

101 pipe = self._r.pipeline() 

102 for doi in dois: 

103 pipe.smembers(self._key(doi)) 

104 results = cast("list[set[str]]", pipe.execute()) 

105 return {doi: members for doi, members in zip(dois, results) if members} 

106 

107 

108_SHARED_STORAGE: Optional[InMemoryStorageManager] = None 

109_ORCID_INDEX: Optional[RedisOrcidIndex] = None 

110_PROCESSORS: Dict[str, RaProcessor] = {} 

111 

112 

113def _shared_storage() -> InMemoryStorageManager: 

114 global _SHARED_STORAGE 

115 if _SHARED_STORAGE is None: 

116 _SHARED_STORAGE = InMemoryStorageManager() 

117 return _SHARED_STORAGE 

118 

119 

120def setup_orcid_index(host: str, port: int, db: int) -> None: 

121 global _ORCID_INDEX 

122 _ORCID_INDEX = RedisOrcidIndex(host, port, db) 

123 

124 

125def get_processor(source: str) -> RaProcessor: 

126 if source not in _PROCESSORS: 

127 index = cast("str", _ORCID_INDEX) 

128 if source == "crossref": 

129 _PROCESSORS[source] = CrossrefProcessing( 

130 orcid_index=index, 

131 storage_manager=_shared_storage(), 

132 testing=False, 

133 use_orcid_api=False, 

134 use_redis_orcid_index=False, 

135 use_redis_publishers=False, 

136 ) 

137 elif source == "datacite": 

138 _PROCESSORS[source] = DataciteProcessing( 

139 orcid_index=index, 

140 storage_manager=_shared_storage(), 

141 testing=False, 

142 use_orcid_api=False, 

143 use_ror_api=False, 

144 use_viaf_api=False, 

145 use_wikidata_api=False, 

146 ) 

147 elif source == "pubmed": 

148 _PROCESSORS[source] = PubmedProcessing(orcid_index=index) 

149 return _PROCESSORS[source] 

150 

151 

152def get_supplier_prefix(uri: str) -> str | None: 

153 match = re.match(r"^(.+)/([a-z][a-z])/(0[1-9]+0)?([1-9][0-9]*)$", uri) 

154 if match is None: 

155 return None 

156 return match.group(3) 

157 

158 

159def extract_omid_number(uri: str) -> int: 

160 return int(uri.split("/")[-1]) 

161 

162 

163def normalize_name(name: str) -> str: 

164 if not name: 

165 return "" 

166 nfkd = unicodedata.normalize("NFKD", name) 

167 return "".join(c for c in nfkd if not unicodedata.combining(c)).lower().strip() 

168 

169 

170def normalize_orcid(orcid: str) -> str: 

171 if not orcid: 

172 return "" 

173 return ( 

174 orcid.replace("https://orcid.org/", "") 

175 .replace("http://orcid.org/", "") 

176 .strip() 

177 .upper() 

178 ) 

179 

180 

181def find_entity_in_file( 

182 uri: str, rdf_dir: str, dir_split: int, items_per_file: int 

183) -> Optional[dict]: 

184 filepath = find_rdf_file(uri, rdf_dir, dir_split, items_per_file, zip_output=True) 

185 if not os.path.exists(filepath): 

186 return None 

187 data = load_json_from_file(filepath) 

188 for graph in data: 

189 for entity in graph["@graph"]: 

190 if entity["@id"] == uri: 

191 return entity 

192 return None 

193 

194 

195def load_br_identifiers( 

196 br_uri: str, rdf_dir: str, dir_split: int, items_per_file: int 

197) -> Dict[str, str]: 

198 br_entity = find_entity_in_file(br_uri, rdf_dir, dir_split, items_per_file) 

199 if not br_entity or HAS_IDENTIFIER not in br_entity: 

200 return {} 

201 result = {} 

202 for id_ref in br_entity[HAS_IDENTIFIER]: 

203 id_entity = find_entity_in_file( 

204 id_ref["@id"], rdf_dir, dir_split, items_per_file 

205 ) 

206 if not id_entity: 

207 continue 

208 if USES_ID_SCHEME not in id_entity or HAS_LITERAL_VALUE not in id_entity: 

209 continue 

210 scheme = id_entity[USES_ID_SCHEME][0]["@id"].split("/datacite/")[1] 

211 value = id_entity[HAS_LITERAL_VALUE][0]["@value"] 

212 result[scheme] = value 

213 return result 

214 

215 

216def load_br_core( 

217 br_uri: str, rdf_dir: str, dir_split: int, items_per_file: int 

218) -> Dict[str, str]: 

219 br_entity = find_entity_in_file(br_uri, rdf_dir, dir_split, items_per_file) 

220 if not br_entity: 

221 return {"type": "", "title": ""} 

222 title = "" 

223 if DC_TITLE in br_entity: 

224 title = br_entity[DC_TITLE][0]["@value"] 

225 br_type = "" 

226 if "@type" in br_entity: 

227 for t in br_entity["@type"]: 

228 if t == FABIO_EXPRESSION: 

229 continue 

230 mapped = URI_TYPE_DICT.get(t, "") 

231 if mapped: 

232 br_type = mapped 

233 break 

234 return {"type": br_type, "title": title} 

235 

236 

237def load_ra_info( 

238 ra_uri: str, rdf_dir: str, dir_split: int, items_per_file: int 

239) -> dict: 

240 ra_entity = find_entity_in_file(ra_uri, rdf_dir, dir_split, items_per_file) 

241 if not ra_entity: 

242 return {"family_name": None, "given_name": None, "name": None, "orcid": None} 

243 family = None 

244 given = None 

245 name = None 

246 orcid = None 

247 if FAMILY_NAME in ra_entity: 

248 family = ra_entity[FAMILY_NAME][0]["@value"] 

249 if GIVEN_NAME in ra_entity: 

250 given = ra_entity[GIVEN_NAME][0]["@value"] 

251 if FOAF_NAME in ra_entity: 

252 name = ra_entity[FOAF_NAME][0]["@value"] 

253 if HAS_IDENTIFIER in ra_entity: 

254 for id_ref in ra_entity[HAS_IDENTIFIER]: 

255 id_entity = find_entity_in_file( 

256 id_ref["@id"], rdf_dir, dir_split, items_per_file 

257 ) 

258 if not id_entity: 

259 continue 

260 if USES_ID_SCHEME not in id_entity or HAS_LITERAL_VALUE not in id_entity: 

261 continue 

262 scheme = id_entity[USES_ID_SCHEME][0]["@id"].split("/datacite/")[1] 

263 if scheme == "orcid": 

264 orcid = id_entity[HAS_LITERAL_VALUE][0]["@value"] 

265 break 

266 return {"family_name": family, "given_name": given, "name": name, "orcid": orcid} 

267 

268 

269def load_all_ars_for_br_role( 

270 br_uri: str, role_type: str, rdf_dir: str, dir_split: int, items_per_file: int 

271) -> List[dict]: 

272 br_entity = find_entity_in_file(br_uri, rdf_dir, dir_split, items_per_file) 

273 if not br_entity or IS_DOC_CONTEXT_FOR not in br_entity: 

274 return [] 

275 result = [] 

276 for ar_ref in br_entity[IS_DOC_CONTEXT_FOR]: 

277 ar_uri = ar_ref["@id"] 

278 ar_entity = find_entity_in_file(ar_uri, rdf_dir, dir_split, items_per_file) 

279 if not ar_entity or WITH_ROLE not in ar_entity: 

280 continue 

281 role_uri = ar_entity[WITH_ROLE][0]["@id"] 

282 ar_role = ROLE_MAP.get(role_uri, "unknown") 

283 if ar_role != role_type: 

284 continue 

285 ra_uri = None 

286 if IS_HELD_BY in ar_entity: 

287 ra_uri = ar_entity[IS_HELD_BY][0]["@id"] 

288 has_next = [] 

289 if HAS_NEXT in ar_entity: 

290 has_next = [item["@id"] for item in ar_entity[HAS_NEXT]] 

291 ra_info = {"family_name": None, "given_name": None, "name": None, "orcid": None} 

292 if ra_uri: 

293 ra_info = load_ra_info(ra_uri, rdf_dir, dir_split, items_per_file) 

294 ra_name = "" 

295 if ra_info["family_name"] or ra_info["given_name"]: 

296 parts = [ra_info["family_name"] or "", ra_info["given_name"] or ""] 

297 ra_name = f"{parts[0]}, {parts[1]}" 

298 elif ra_info["name"]: 

299 ra_name = ra_info["name"] 

300 result.append( 

301 { 

302 "ar": ar_uri, 

303 "ra": ra_uri, 

304 "ra_name": ra_name, 

305 "ra_family": ra_info["family_name"], 

306 "ra_given": ra_info["given_name"], 

307 "ra_orcid": ra_info["orcid"], 

308 "has_next": has_next, 

309 } 

310 ) 

311 return result 

312 

313 

314def _strip_orcid_url(orcid: str) -> str: 

315 if not orcid: 

316 return orcid 

317 return orcid.replace("https://orcid.org/", "").replace("http://orcid.org/", "") 

318 

319 

320def _api_agents(agents: list[AgentMetadata]) -> list[dict[str, object]]: 

321 return [ 

322 { 

323 "family": agent["family"], 

324 "given": agent["given"], 

325 "name": agent["name"], 

326 "orcid": agent["orcid"], 

327 "position": agent["position"], 

328 } 

329 for agent in agents 

330 ] 

331 

332 

333def fetch_crossref(doi: str) -> Optional[dict]: 

334 resp = SESSION.get(CROSSREF_BASE + doi, timeout=30) 

335 if resp.status_code == 404: 

336 return None 

337 resp.raise_for_status() 

338 data = cast(JsonObject, resp.json()) 

339 parsed = parse_crossref_work(data, doi) 

340 publisher_identifiers = parsed["publisher_identifiers"] 

341 return { 

342 "author": _api_agents(parsed["author"]), 

343 "editor": _api_agents(parsed["editor"]), 

344 "publisher": parsed["publisher"], 

345 "publisher_crossref_id": ( 

346 publisher_identifiers[0]["value"] if publisher_identifiers else None 

347 ), 

348 "source": "crossref", 

349 } 

350 

351 

352def fetch_datacite(doi: str) -> Optional[dict]: 

353 resp = SESSION.get(DATACITE_BASE + doi, timeout=30) 

354 if resp.status_code == 404: 

355 return None 

356 resp.raise_for_status() 

357 parsed = parse_datacite_work(cast(JsonObject, resp.json()), doi) 

358 return { 

359 "author": _api_agents(parsed["author"]), 

360 "editor": _api_agents(parsed["editor"]), 

361 "publisher": parsed["publisher"], 

362 "publisher_crossref_id": None, 

363 "source": "datacite", 

364 } 

365 

366 

367def fetch_pubmed(pmid: str) -> Optional[dict]: 

368 params = {"db": "pubmed", "id": pmid, "rettype": "xml", "retmode": "xml"} 

369 resp = SESSION.get(PUBMED_BASE, params=params, timeout=30) 

370 resp.raise_for_status() 

371 root = ET.fromstring(resp.content) 

372 article = root.find(".//Article") 

373 if article is None: 

374 return None 

375 authors = [] 

376 author_list = article.find("AuthorList") 

377 if author_list is not None: 

378 for i, author_elem in enumerate(author_list.findall("Author")): 

379 orcid = None 

380 for ident in author_elem.findall("Identifier"): 

381 if ident.get("Source") == "ORCID" and ident.text is not None: 

382 orcid = _strip_orcid_url(ident.text) 

383 break 

384 authors.append( 

385 { 

386 "family": author_elem.findtext("LastName", ""), 

387 "given": author_elem.findtext("ForeName", ""), 

388 "name": author_elem.findtext("CollectiveName", ""), 

389 "orcid": orcid, 

390 "position": i, 

391 } 

392 ) 

393 return { 

394 "author": authors, 

395 "editor": [], 

396 "publisher": "", 

397 "publisher_crossref_id": None, 

398 "source": "pubmed", 

399 } 

400 

401 

402def fetch_api_data(identifiers: Dict[str, str]) -> Tuple[Optional[dict], str]: 

403 if "doi" in identifiers: 

404 doi = identifiers["doi"] 

405 try: 

406 result = fetch_crossref(doi) 

407 if result: 

408 return result, f"doi:{doi}" 

409 except requests.RequestException: 

410 pass 

411 try: 

412 result = fetch_datacite(doi) 

413 if result: 

414 return result, f"doi:{doi}" 

415 except requests.RequestException: 

416 pass 

417 if "pmid" in identifiers: 

418 pmid = identifiers["pmid"] 

419 try: 

420 time.sleep(0.34) 

421 result = fetch_pubmed(pmid) 

422 if result: 

423 return result, f"pmid:{pmid}" 

424 except requests.RequestException: 

425 pass 

426 return None, "" 

427 

428 

429def match_ars_to_api(ar_infos: List[dict], api_entries: List[dict]) -> List[str]: 

430 if not api_entries: 

431 return [] 

432 orcid_to_pos = {} 

433 for entry in api_entries: 

434 if entry["orcid"]: 

435 orcid_to_pos[normalize_orcid(entry["orcid"])] = entry["position"] 

436 name_to_positions = defaultdict(list) 

437 for entry in api_entries: 

438 if entry["family"]: 

439 name_to_positions[normalize_name(entry["family"])].append(entry["position"]) 

440 ar_to_position: Dict[str, int] = {} 

441 used_positions: set = set() 

442 sorted_ars = sorted(ar_infos, key=lambda a: extract_omid_number(a["ar"])) 

443 for ar in sorted_ars: 

444 if ar["ra_orcid"]: 

445 norm = normalize_orcid(ar["ra_orcid"]) 

446 if norm in orcid_to_pos: 

447 pos = orcid_to_pos[norm] 

448 if pos not in used_positions: 

449 ar_to_position[ar["ar"]] = pos 

450 used_positions.add(pos) 

451 for ar in sorted_ars: 

452 if ar["ar"] in ar_to_position: 

453 continue 

454 if ar["ra_family"]: 

455 norm = normalize_name(ar["ra_family"]) 

456 if norm in name_to_positions: 

457 for pos in name_to_positions[norm]: 

458 if pos not in used_positions: 

459 ar_to_position[ar["ar"]] = pos 

460 used_positions.add(pos) 

461 break 

462 ordered = sorted(ar_to_position.items(), key=lambda x: x[1]) 

463 return [uri for uri, _ in ordered] 

464 

465 

466def match_publisher_ars(ar_infos: List[dict], api_publisher: str) -> List[str]: 

467 if not api_publisher: 

468 return [] 

469 norm_api = normalize_name(api_publisher) 

470 for ar in sorted(ar_infos, key=lambda a: extract_omid_number(a["ar"])): 

471 if ar["ra_name"] and normalize_name(ar["ra_name"]) == norm_api: 

472 return [ar["ar"]] 

473 return [] 

474 

475 

476def _agents_list(api_data: dict) -> List[dict]: 

477 agents: List[dict] = [] 

478 for role in ("author", "editor"): 

479 for entry in api_data[role]: 

480 agents.append( 

481 { 

482 "family": entry["family"], 

483 "given": entry["given"], 

484 "name": entry["name"], 

485 "orcid": entry["orcid"], 

486 "role": role, 

487 } 

488 ) 

489 return agents 

490 

491 

492def build_csv_row( 

493 identifier: str, role_type: str, api_data: dict, br_core: Dict[str, str] 

494) -> dict: 

495 row = {"id": identifier, "type": br_core["type"], "title": br_core["title"]} 

496 if role_type in ("author", "editor"): 

497 bare_id = identifier.split(":", 1)[1] if ":" in identifier else identifier 

498 processor = get_processor(api_data["source"]) 

499 if api_data["source"] == "crossref": 

500 processor.prefetch_doi_orcid_index([bare_id]) # type: ignore[attr-defined] 

501 authors, editors = processor.get_agents_strings_list( 

502 bare_id, _agents_list(api_data) 

503 ) 

504 row[role_type] = "; ".join(authors if role_type == "author" else editors) 

505 elif role_type == "publisher": 

506 publisher_name = api_data["publisher"] 

507 crossref_id = api_data["publisher_crossref_id"] 

508 if crossref_id: 

509 row["publisher"] = f"{publisher_name} [crossref:{crossref_id}]" 

510 else: 

511 row["publisher"] = publisher_name 

512 return row 

513 

514 

515def generate_csv(corrections: List[dict], output_path: str) -> None: 

516 ready = [c for c in corrections if c["status"] == "ready"] 

517 grouped: Dict[Tuple[str, str], dict] = {} 

518 for c in ready: 

519 key = (c["br"], c["identifier"]) 

520 if key not in grouped: 

521 grouped[key] = {col: "" for col in CSV_COLUMNS} 

522 grouped[key]["id"] = c["identifier"] 

523 for col in CSV_COLUMNS: 

524 value = c["csv_row"].get(col, "") 

525 if value: 

526 grouped[key][col] = value 

527 

528 output_dir = os.path.dirname(os.path.abspath(output_path)) 

529 if output_dir: 

530 os.makedirs(output_dir, exist_ok=True) 

531 with open(output_path, "w", encoding="utf-8", newline="") as f: 

532 writer = csv.DictWriter(f, fieldnames=CSV_COLUMNS) 

533 writer.writeheader() 

534 for row in grouped.values(): 

535 writer.writerow(row) 

536 

537 print(f"CSV for Meta saved to {output_path} ({len(grouped)} rows)") 

538 

539 

540def _ar_summary(ar: dict) -> dict: 

541 return { 

542 "ar": ar["ar"], 

543 "ra": ar["ra"], 

544 "ra_name": ar["ra_name"], 

545 "has_next": ar["has_next"], 

546 } 

547 

548 

549def dry_run( 

550 config_path: str, 

551 anomaly_path: str, 

552 output_path: str, 

553 csv_output: Optional[str], 

554 orcid_redis_host: str, 

555 orcid_redis_port: int, 

556 orcid_redis_db: int, 

557) -> None: 

558 setup_orcid_index(orcid_redis_host, orcid_redis_port, orcid_redis_db) 

559 with open(config_path, encoding="utf-8") as f: 

560 settings = yaml.safe_load(f) 

561 rdf_dir = os.path.join(settings["output_rdf_dir"], "rdf") 

562 dir_split = settings["dir_split_number"] 

563 items_per_file = settings["items_per_file"] 

564 

565 with open(anomaly_path, "rb") as f: 

566 report = orjson.loads(f.read()) 

567 

568 groups: Dict[Tuple[str, str], List[dict]] = defaultdict(list) 

569 for anomaly in report["anomalies"]: 

570 key = (anomaly["br"], anomaly["role_type"]) 

571 groups[key].append(anomaly) 

572 

573 corrections = [] 

574 summary = { 

575 "total_brs": len(groups), 

576 "api_resolved": 0, 

577 "no_identifiers": 0, 

578 "api_error": 0, 

579 "manual_review_needed": 0, 

580 } 

581 

582 with create_progress() as progress: 

583 task = progress.add_task("Analyzing anomalies", total=len(groups)) 

584 for (br_uri, role_type), anomalies in groups.items(): 

585 anomaly_types = list({a["anomaly_type"] for a in anomalies}) 

586 ar_infos = load_all_ars_for_br_role( 

587 br_uri, role_type, rdf_dir, dir_split, items_per_file 

588 ) 

589 ar_summaries = [_ar_summary(ar) for ar in ar_infos] 

590 all_ar_uris = [ar["ar"] for ar in ar_infos] 

591 

592 if role_type == "unknown": 

593 summary["manual_review_needed"] += 1 

594 corrections.append( 

595 { 

596 "br": br_uri, 

597 "role_type": role_type, 

598 "anomalies": anomaly_types, 

599 "source": None, 

600 "identifier": None, 

601 "current_ars": ar_summaries, 

602 "csv_row": {}, 

603 "delete_ars": [], 

604 "operations": [], 

605 "status": "manual_review", 

606 } 

607 ) 

608 progress.update(task, advance=1) 

609 continue 

610 

611 identifiers = load_br_identifiers( 

612 br_uri, rdf_dir, dir_split, items_per_file 

613 ) 

614 

615 if not identifiers: 

616 summary["no_identifiers"] += 1 

617 corrections.append( 

618 { 

619 "br": br_uri, 

620 "role_type": role_type, 

621 "anomalies": anomaly_types, 

622 "source": None, 

623 "identifier": None, 

624 "current_ars": ar_summaries, 

625 "csv_row": {}, 

626 "delete_ars": [], 

627 "operations": [], 

628 "status": "no_identifiers", 

629 } 

630 ) 

631 progress.update(task, advance=1) 

632 continue 

633 

634 api_data, identifier_str = fetch_api_data(identifiers) 

635 

636 if not api_data: 

637 summary["api_error"] += 1 

638 id_str = next((f"{k}:{v}" for k, v in identifiers.items()), None) 

639 corrections.append( 

640 { 

641 "br": br_uri, 

642 "role_type": role_type, 

643 "anomalies": anomaly_types, 

644 "source": None, 

645 "identifier": id_str, 

646 "current_ars": ar_summaries, 

647 "csv_row": {}, 

648 "delete_ars": [], 

649 "operations": [], 

650 "status": "api_error", 

651 } 

652 ) 

653 progress.update(task, advance=1) 

654 continue 

655 

656 extra_fields = {} 

657 if role_type in ("author", "editor"): 

658 api_entries = api_data[role_type] 

659 extra_fields[f"api_{role_type}s"] = api_entries 

660 api_matched = match_ars_to_api(ar_infos, api_entries) 

661 extra_fields["api_matched_ars"] = api_matched 

662 has_api_data = len(api_entries) > 0 

663 else: 

664 api_publisher = api_data["publisher"] 

665 extra_fields["api_publisher"] = api_publisher 

666 api_matched = match_publisher_ars(ar_infos, api_publisher) 

667 extra_fields["api_matched_ars"] = api_matched 

668 has_api_data = bool(api_publisher) 

669 

670 if has_api_data: 

671 status = "ready" 

672 summary["api_resolved"] += 1 

673 else: 

674 status = "manual_review" 

675 summary["manual_review_needed"] += 1 

676 

677 operations = [] 

678 if status == "ready": 

679 for ar_uri in all_ar_uris: 

680 operations.append({"action": "remove_next", "ar": ar_uri}) 

681 for ar_uri in all_ar_uris: 

682 operations.append( 

683 {"action": "delete_ar", "ar": ar_uri, "br": br_uri} 

684 ) 

685 

686 if status == "ready": 

687 br_core = load_br_core(br_uri, rdf_dir, dir_split, items_per_file) 

688 csv_row = build_csv_row(identifier_str, role_type, api_data, br_core) 

689 else: 

690 csv_row = {} 

691 

692 correction = { 

693 "br": br_uri, 

694 "role_type": role_type, 

695 "anomalies": anomaly_types, 

696 "source": api_data["source"], 

697 "identifier": identifier_str, 

698 **extra_fields, 

699 "current_ars": ar_summaries, 

700 "csv_row": csv_row, 

701 "delete_ars": all_ar_uris if status == "ready" else [], 

702 "operations": operations, 

703 "status": status, 

704 } 

705 corrections.append(correction) 

706 progress.update(task, advance=1) 

707 

708 plan = { 

709 "generated": datetime.now(timezone.utc).isoformat(), 

710 "config": os.path.abspath(config_path), 

711 "summary": summary, 

712 "corrections": corrections, 

713 } 

714 

715 output_dir = os.path.dirname(os.path.abspath(output_path)) 

716 if output_dir: 

717 os.makedirs(output_dir, exist_ok=True) 

718 with open(output_path, "wb") as f: 

719 f.write(orjson.dumps(plan, option=orjson.OPT_INDENT_2)) 

720 

721 print(f"Correction plan saved to {output_path}") 

722 print(f" Total groups: {summary['total_brs']}") 

723 print(f" API resolved: {summary['api_resolved']}") 

724 print(f" No identifiers: {summary['no_identifiers']}") 

725 print(f" API errors: {summary['api_error']}") 

726 print(f" Manual review: {summary['manual_review_needed']}") 

727 

728 if csv_output: 

729 generate_csv(corrections, csv_output) 

730 

731 

732def apply_corrections_for_br( 

733 editor: MetaEditor, br_uri: str, corrections: List[dict] 

734) -> None: 

735 supplier_prefix = get_supplier_prefix(br_uri) 

736 assert supplier_prefix is not None 

737 g_set = GraphSet( 

738 editor.base_iri, 

739 supplier_prefix=supplier_prefix, 

740 custom_counter_handler=editor.counter_handler, 

741 wanted_label=False, 

742 ) 

743 entities_to_import: list[str] = [br_uri]