Coverage for oc_meta / lib / finder.py: 90%

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

2# 

3# SPDX-License-Identifier: ISC 

4 

5from __future__ import annotations 

6 

7import multiprocessing 

8from concurrent.futures import ProcessPoolExecutor 

9from functools import partial 

10from typing import TYPE_CHECKING, Dict, List, Tuple, TypedDict 

11 

12import orjson 

13 

14if TYPE_CHECKING: 

15 from rich.progress import Progress 

16from dateutil import parser 

17from oc_ocdm.graph.graph_entity import GraphEntity 

18from oc_ocdm.prov.prov_entity import ProvEntity 

19from oc_ocdm.support import get_resource_number 

20from triplelite import RDFTerm, TripleLite 

21from time_agnostic_library.agnostic_entity import AgnosticEntity 

22from rich.console import Console 

23 

24from oc_meta.constants import ( 

25 QLEVER_BATCH_SIZE, 

26 QLEVER_MAX_WORKERS, 

27 QLEVER_QUERIES_PER_GROUP, 

28) 

29from oc_meta.lib.sparql import execute_sparql_queries 

30 

31_XSD_STRING = "http://www.w3.org/2001/XMLSchema#string" 

32_RDF_TYPE = "http://www.w3.org/1999/02/22-rdf-syntax-ns#type" 

33_DATACITE = "http://purl.org/spar/datacite/" 

34 

35_P_HAS_LITERAL_VALUE = GraphEntity.iri_has_literal_value 

36_P_USES_ID_SCHEME = GraphEntity.iri_uses_identifier_scheme 

37_P_HAS_IDENTIFIER = GraphEntity.iri_has_identifier 

38_P_TITLE = GraphEntity.iri_title 

39_P_NAME = GraphEntity.iri_name 

40_P_FAMILY_NAME = GraphEntity.iri_family_name 

41_P_GIVEN_NAME = GraphEntity.iri_given_name 

42_P_IS_DOC_CONTEXT_FOR = GraphEntity.iri_is_document_context_for 

43_P_HAS_NEXT = GraphEntity.iri_has_next 

44_P_IS_HELD_BY = GraphEntity.iri_is_held_by 

45_P_WITH_ROLE = GraphEntity.iri_with_role 

46_P_EMBODIMENT = GraphEntity.iri_embodiment 

47_P_STARTING_PAGE = GraphEntity.iri_starting_page 

48_P_ENDING_PAGE = GraphEntity.iri_ending_page 

49_P_PUB_DATE = GraphEntity.iri_has_publication_date 

50_P_SEQ_ID = GraphEntity.iri_has_sequence_identifier 

51_P_PART_OF = GraphEntity.iri_part_of 

52_P_TYPE = _RDF_TYPE 

53_T_JOURNAL_VOLUME = GraphEntity.iri_journal_volume 

54_T_JOURNAL_ISSUE = GraphEntity.iri_journal_issue 

55_T_EXPRESSION = GraphEntity.iri_expression 

56_R_AUTHOR = GraphEntity.iri_author 

57_R_EDITOR = GraphEntity.iri_editor 

58_R_PUBLISHER = GraphEntity.iri_publisher 

59 

60 

61class IssueEntry(TypedDict): 

62 id: str 

63 

64 

65class VolumeEntry(TypedDict): 

66 id: str 

67 issue: Dict[str, IssueEntry] 

68 

69 

70class VenueStructure(TypedDict): 

71 issue: Dict[str, IssueEntry] 

72 volume: Dict[str, VolumeEntry] 

73 

74 

75class ResourceFinder: 

76 def __init__( 

77 self, 

78 ts_url: str, 

79 base_iri: str, 

80 settings: dict = dict(), 

81 meta_config_path: str | None = None, 

82 workers: int = 1, 

83 ): 

84 self.ts_url = ts_url 

85 self.base_iri = base_iri[:-1] if base_iri[-1] == "/" else base_iri 

86 self.graph = TripleLite( 

87 reverse_index_predicates=frozenset(self._PO_S_INDEXED_PREDICATES) 

88 ) 

89 self.meta_config_path = meta_config_path 

90 self.meta_settings = settings 

91 self.virtuoso_full_text_search = ( 

92 settings["virtuoso_full_text_search"] 

93 if settings and "virtuoso_full_text_search" in settings 

94 else False 

95 ) 

96 self.workers = workers 

97 

98 _PO_S_INDEXED_PREDICATES = {_P_HAS_LITERAL_VALUE, _P_HAS_IDENTIFIER, _P_PART_OF} 

99 

100 def add_triple(self, s: str, p: str, o: str, o_datatype: str = "") -> None: 

101 if o_datatype: 

102 term = RDFTerm("literal", o, o_datatype) 

103 elif o.startswith("http"): 

104 term = RDFTerm("uri", o) 

105 else: 

106 term = RDFTerm("literal", o, _XSD_STRING) 

107 self.graph.add((s, p, term)) 

108 

109 def __contains__(self, uri: str) -> bool: 

110 return self.graph.has_subject(uri) 

111 

112 def _get_objects(self, subject: str, predicate: str) -> list[str]: 

113 return [t.value for t in self.graph.objects(subject, predicate)] 

114 

115 def _get_all_po(self, subject: str) -> dict[str, list[str]]: 

116 result: dict[str, list[str]] = {} 

117 for p, o in self.graph.predicate_objects(subject): 

118 result.setdefault(p, []).append(o.value) 

119 return result 

120 

121 def _get_subjects(self, predicate: str, obj: str) -> set[str]: 

122 if predicate == _P_HAS_LITERAL_VALUE: 

123 term = RDFTerm("literal", obj, _XSD_STRING) 

124 else: 

125 term = RDFTerm("uri", obj) 

126 return set(self.graph.subjects(predicate, term)) 

127 

128 # _______________________________BR_________________________________ # 

129 

130 def _find_id_uri(self, schema: str, value: str) -> str | None: 

131 schema_uri = _DATACITE + schema 

132 for id_uri in self._get_subjects(_P_HAS_LITERAL_VALUE, value): 

133 schemes = self._get_objects(id_uri, _P_USES_ID_SCHEME) 

134 if schemes[0] == schema_uri: 

135 return id_uri 

136 return None 

137 

138 def _collect_entity_ids( 

139 self, entity_uri: str, exclude_id_uri: str | None = None 

140 ) -> list[tuple[str, str]]: 

141 result: list[tuple[str, str]] = [] 

142 for id_uri in self._get_objects(entity_uri, _P_HAS_IDENTIFIER): 

143 if id_uri == exclude_id_uri: 

144 continue 

145 po = self._get_all_po(id_uri) 

146 schemes = po.get(_P_USES_ID_SCHEME, []) 

147 literals = po.get(_P_HAS_LITERAL_VALUE, []) 

148 if not schemes or not literals: 

149 raise ValueError(f"Identifier {id_uri} missing schema or literal value") 

150 full_id = f"{schemes[0].replace(_DATACITE, '')}:{literals[0]}" 

151 result.append((id_uri.replace(f"{self.base_iri}/", ""), full_id)) 

152 return result 

153 

154 def retrieve_br_from_id( 

155 self, schema: str, value: str 

156 ) -> List[Tuple[str, str, list]]: 

157 id_uri = self._find_id_uri(schema, value) 

158 if not id_uri: 

159 return [] 

160 metaid_id_list = [ 

161 (id_uri.replace(f"{self.base_iri}/", ""), f"{schema}:{value}") 

162 ] 

163 result_list = [] 

164 for entity_uri in self._get_subjects(_P_HAS_IDENTIFIER, id_uri): 

165 title = "" 

166 titles = self._get_objects(entity_uri, _P_TITLE) 

167 if titles: 

168 title = titles[0] 

169 other_ids = self._collect_entity_ids(entity_uri, exclude_id_uri=id_uri) 

170 result_list.append( 

171 ( 

172 entity_uri.replace(f"{self.base_iri}/", ""), 

173 title, 

174 metaid_id_list + other_ids, 

175 ) 

176 ) 

177 return result_list 

178 

179 def retrieve_br_from_meta( 

180 self, metaid: str 

181 ) -> Tuple[str, List[Tuple[str, str]], bool]: 

182 metaid_uri = f"{self.base_iri}/{metaid}" 

183 po = self._get_all_po(metaid_uri) 

184 if not po: 

185 return "", [], False 

186 title = "" 

187 titles = po.get(_P_TITLE, []) 

188 if titles: 

189 title = titles[0] 

190 identifiers = self._collect_entity_ids(metaid_uri) 

191 return title, identifiers, True 

192 

193 # _______________________________ID_________________________________ # 

194 

195 def retrieve_metaid_from_id(self, schema: str, value: str) -> str | None: 

196 id_uri = self._find_id_uri(schema, value) 

197 if id_uri: 

198 return id_uri.replace(f"{self.base_iri}/", "") 

199 return None 

200 

201 def retrieve_metaid_from_merged_entity( 

202 self, metaid_uri: str, prov_config: str 

203 ) -> str | None: 

204 """ 

205 It looks for MetaId in the provenance. If the input entity was deleted due to a merge, this function returns the target entity. Otherwise, it returns None. 

206 

207 :params metaid_uri: a MetaId URI 

208 :type metaid_uri: str 

209 :params prov_config: the path of the configuration file required by time-agnostic-library 

210 :type prov_config: str 

211 :returns str | None: -- It returns the MetaID associated with the target entity after a merge. If there was no merge, it returns None. 

212 """ 

213 metaval: str | None = None 

214 with open(prov_config, "rb") as f: 

215 prov_config_dict = orjson.loads(f.read()) 

216 agnostic_meta = AgnosticEntity( 

217 res=metaid_uri, 

218 config=prov_config_dict, 

219 include_related_objects=False, 

220 include_merged_entities=False, 

221 include_reverse_relations=False, 

222 ) 

223 agnostic_meta_history = agnostic_meta.get_history(include_prov_metadata=True) 

224 meta_history_data = agnostic_meta_history[0][metaid_uri] 

225 if meta_history_data: 

226 meta_history_metadata = agnostic_meta_history[1][metaid_uri] 

227 penultimate_snapshot = sorted( 

228 meta_history_metadata.items(), 

229 key=lambda x: parser.parse(x[1]["generatedAtTime"]).replace( 

230 tzinfo=None 

231 ), 

232 reverse=True, 

233 )[1][0] 

234 query_if_it_was_merged = f""" 

235 SELECT DISTINCT ?se 

236 WHERE {{ 

237 ?se a <{ProvEntity.iri_entity}>; 

238 <{ProvEntity.iri_was_derived_from}> <{penultimate_snapshot}>. 

239 }} 

240 """ 

241 prov_endpoint = prov_config_dict["provenance"]["triplestore_urls"][0] 

242 results = execute_sparql_queries(prov_endpoint, [query_if_it_was_merged])[0] 

243 merged_entities = [ 

244 se for se in results if metaid_uri not in se["se"]["value"] 

245 ] 

246 if merged_entities: 

247 merged_entity_uri = merged_entities[0]["se"]["value"] 

248 merged_entity_uri = merged_entity_uri.split("/prov/")[0] 

249 metaval = merged_entity_uri.split("/")[-1] 

250 return metaval 

251 

252 # _______________________________RA_________________________________ # 

253 def retrieve_ra_from_meta( 

254 self, metaid: str 

255 ) -> Tuple[str, List[Tuple[str, str]], bool]: 

256 metaid_uri = f"{self.base_iri}/{metaid}" 

257 po = self._get_all_po(metaid_uri) 

258 if not po: 

259 return "", [], False 

260 family_names = po.get(_P_FAMILY_NAME, []) 

261 given_names = po.get(_P_GIVEN_NAME, []) 

262 names = po.get(_P_NAME, []) 

263 full_name = self._construct_full_name( 

264 names[0] if names else "", 

265 family_names[0] if family_names else "", 

266 given_names[0] if given_names else "", 

267 ) 

268 identifiers = self._collect_entity_ids(metaid_uri) 

269 return full_name, identifiers, True 

270 

271 def retrieve_ra_from_id( 

272 self, schema: str, value: str 

273 ) -> List[Tuple[str, str, list]]: 

274 id_uri = self._find_id_uri(schema, value) 

275 if not id_uri: 

276 return [] 

277 metaid_id_list: List[Tuple[str, str]] = [ 

278 (id_uri.replace(f"{self.base_iri}/", ""), f"{schema}:{value}") 

279 ] 

280 result_list = [] 

281 for entity_uri in self._get_subjects(_P_HAS_IDENTIFIER, id_uri): 

282 po = self._get_all_po(entity_uri) 

283 names = po.get(_P_NAME, []) 

284 family_names = po.get(_P_FAMILY_NAME, []) 

285 given_names = po.get(_P_GIVEN_NAME, []) 

286 full_name = self._construct_full_name( 

287 names[0] if names else "", 

288 family_names[0] if family_names else "", 

289 given_names[0] if given_names else "", 

290 ) 

291 other_ids = self._collect_entity_ids(entity_uri, exclude_id_uri=id_uri) 

292 result_list.append( 

293 ( 

294 entity_uri.replace(f"{self.base_iri}/", ""), 

295 full_name, 

296 metaid_id_list + other_ids, 

297 ) 

298 ) 

299 return result_list 

300 

301 def _construct_full_name(self, name: str, family_name: str, given_name: str) -> str: 

302 if name and not family_name and not given_name: 

303 return name 

304 elif not name and family_name and not given_name: 

305 return f"{family_name}," 

306 elif not name and not family_name and given_name: 

307 return f", {given_name}" 

308 elif not name and family_name and given_name: 

309 return f"{family_name}, {given_name}" 

310 else: 

311 return "" 

312 

313 def retrieve_ra_sequence_from_br_meta( 

314 self, metaid: str, col_name: str 

315 ) -> List[Dict[str, tuple]]: 

316 if col_name == "author": 

317 role_str = _R_AUTHOR 

318 elif col_name == "editor": 

319 role_str = _R_EDITOR 

320 else: 

321 role_str = _R_PUBLISHER 

322 

323 metaid_uri = f"{self.base_iri}/{metaid}" 

324 dict_ar: dict[str, dict[str, str]] = {} 

325 

326 for ar_uri in self._get_objects(metaid_uri, _P_IS_DOC_CONTEXT_FOR): 

327 ar_po = self._get_all_po(ar_uri) 

328 roles = ar_po.get(_P_WITH_ROLE, []) 

329 if role_str in roles: 

330 role_value = ar_uri.replace(f"{self.base_iri}/", "") 

331 next_list = ar_po.get(_P_HAS_NEXT, []) 

332 next_role = ( 

333 next_list[0].replace(f"{self.base_iri}/", "") if next_list else "" 

334 ) 

335 held_by = ar_po.get(_P_IS_HELD_BY, []) 

336 ra = held_by[0].replace(f"{self.base_iri}/", "") if held_by else None 

337 if ra is not None: 

338 dict_ar[role_value] = {"next": next_role, "ra": ra} 

339 

340 all_roles = set(dict_ar.keys()) 

341 roles_with_next = set( 

342 details["next"] for details in dict_ar.values() if details["next"] 

343 ) 

344 start_role_candidates = all_roles - roles_with_next 

345 

346 if len(all_roles) == 0: 

347 return [] 

348 

349 if len(start_role_candidates) == 0: 

350 sorted_ars = sorted( 

351 all_roles, key=lambda ar: get_resource_number(f"{self.base_iri}/{ar}") 

352 ) 

353 start_role_candidates = {sorted_ars[0]} 

354 

355 if len(start_role_candidates) != 1: 

356 chains = [] 

357 for start_candidate in start_role_candidates: 

358 current_role = start_candidate 

359 chain: list[dict[str, tuple]] = [] 

360 visited_roles: set[str] = set() 

361 while current_role and current_role not in visited_roles: 

362 visited_roles.add(current_role) 

363 if current_role in dict_ar: 

364 ra_info = self.retrieve_ra_from_meta( 

365 dict_ar[current_role]["ra"] 

366 )[0:2] 

367 ra_tuple = ra_info + (dict_ar[current_role]["ra"],) 

368 chain.append({current_role: ra_tuple}) 

369 current_role = dict_ar[current_role]["next"] 

370 else: 

371 break 

372 chains.append(chain) 

373 chains.sort( 

374 key=lambda chain: ( 

375 -len(chain), 

376 get_resource_number(f"{self.base_iri}/{list(chain[0].keys())[0]}"), 

377 ) 

378 ) 

379 try: 

380 ordered_ar_list = chains[0] 

381 except Exception as e: 

382 print( 

383 f"\nWarning: Error processing BR: {metaid} for column: {col_name}" 

384 ) 

385 print(f"dict_ar: {dict_ar}") 

386 print(f"All roles: {all_roles}") 

387 print(f"Start role candidates: {start_role_candidates}") 

388 print(f"Roles with next: {roles_with_next}") 

389 print(f"Error: {str(e)}") 

390 return [] 

391 else: 

392 start_role = start_role_candidates.pop() 

393 ordered_ar_list: list[dict[str, tuple]] = [] 

394 current_role = start_role 

395 visited_roles: set[str] = set() 

396 while current_role and current_role not in visited_roles: 

397 visited_roles.add(current_role) 

398 if current_role in dict_ar: 

399 ra_info = self.retrieve_ra_from_meta(dict_ar[current_role]["ra"])[ 

400 0:2 

401 ] 

402 ra_tuple = ra_info + (dict_ar[current_role]["ra"],) 

403 ordered_ar_list.append({current_role: ra_tuple}) 

404 current_role = dict_ar[current_role]["next"] 

405 else: 

406 break 

407 

408 return ordered_ar_list 

409 

410 def retrieve_re_from_br_meta(self, metaid: str) -> Tuple[str, str] | None: 

411 metaid_uri = f"{self.base_iri}/{metaid}" 

412 re_uris = self._get_objects(metaid_uri, _P_EMBODIMENT) 

413 if not re_uris: 

414 return None 

415 re_full_uri = re_uris[0] 

416 re_metaid = re_full_uri.replace(f"{self.base_iri}/", "") 

417 re_po = self._get_all_po(re_full_uri) 

418 starting_pages = re_po.get(_P_STARTING_PAGE, []) 

419 ending_pages = re_po.get(_P_ENDING_PAGE, []) 

420 starting_page = starting_pages[0] if starting_pages else None 

421 ending_page = ending_pages[0] if ending_pages else None 

422 pages = "" 

423 if starting_page and ending_page: 

424 pages = f"{starting_page}-{ending_page}" 

425 elif starting_page: 

426 pages = f"{starting_page}-{starting_page}" 

427 elif ending_page: 

428 pages = f"{ending_page}-{ending_page}" 

429 return re_metaid, pages 

430 

431 def retrieve_br_info_from_meta(self, metaid: str) -> dict: 

432 venue_type_strs = { 

433 GraphEntity.iri_archival_document, 

434 GraphEntity.iri_journal, 

435 GraphEntity.iri_book, 

436 GraphEntity.iri_book_series, 

437 GraphEntity.iri_series, 

438 GraphEntity.iri_academic_proceedings, 

439 GraphEntity.iri_proceedings_series, 

440 GraphEntity.iri_reference_book, 

441 _T_EXPRESSION, 

442 } 

443 

444 def extract_identifiers(entity_uri: str) -> list[str]: 

445 identifiers = [f"omid:{entity_uri.replace(f'{self.base_iri}/', '')}"] 

446 for id_uri in self._get_objects(entity_uri, _P_HAS_IDENTIFIER): 

447 id_po = self._get_all_po(id_uri) 

448 schemes = id_po.get(_P_USES_ID_SCHEME, []) 

449 literals = id_po.get(_P_HAS_LITERAL_VALUE, []) 

450 if schemes and literals: 

451 scheme = schemes[0].replace(_DATACITE, "") 

452 identifiers.append(f"{scheme}:{literals[0]}") 

453 return identifiers 

454 

455 def check_venue(entity_uri: str) -> str | None: 

456 entity_types = self._get_objects(entity_uri, _P_TYPE) 

457 if any(t in venue_type_strs for t in entity_types): 

458 titles = self._get_objects(entity_uri, _P_TITLE) 

459 if titles: 

460 venue_ids = extract_identifiers(entity_uri) 

461 return f"{titles[0]} [{' '.join(venue_ids)}]" 

462 return None 

463 

464 metaid_uri = ( 

465 f"{self.base_iri}/{metaid}" if self.base_iri not in metaid else metaid 

466 ) 

467 po = self._get_all_po(metaid_uri) 

468 res_dict: dict = { 

469 "pub_date": "", 

470 "type": "", 

471 "page": self.retrieve_re_from_br_meta(metaid), 

472 "issue": "", 

473 "volume": "", 

474 "venue": "", 

475 } 

476 

477 pub_dates = po.get(_P_PUB_DATE, []) 

478 if pub_dates: 

479 res_dict["pub_date"] = pub_dates[0] 

480 

481 types = po.get(_P_TYPE, []) 

482 for t in types: 

483 if t != _T_EXPRESSION and t.startswith("http"): 

484 res_dict["type"] = self._type_it(t) 

485 break 

486 

487 seq_ids = po.get(_P_SEQ_ID, []) 

488 if seq_ids: 

489 entity_types = types 

490 if _T_JOURNAL_ISSUE in entity_types: 

491 res_dict["issue"] = seq_ids[0] 

492 elif _T_JOURNAL_VOLUME in entity_types: 

493 res_dict["volume"] = seq_ids[0] 

494 

495 for container_uri in po.get(_P_PART_OF, []): 

496 container_po = self._get_all_po(container_uri) 

497 container_types = container_po.get(_P_TYPE, []) 

498 

499 if _T_JOURNAL_ISSUE in container_types: 

500 container_seqs = container_po.get(_P_SEQ_ID, []) 

501 if container_seqs: 

502 res_dict["issue"] = container_seqs[0] 

503 elif _T_JOURNAL_VOLUME in container_types: 

504 container_seqs = container_po.get(_P_SEQ_ID, []) 

505 if container_seqs: 

506 res_dict["volume"] = container_seqs[0] 

507 else: 

508 venue_str = check_venue(container_uri) 

509 if venue_str: 

510 res_dict["venue"] = venue_str 

511 

512 for inner_uri in container_po.get(_P_PART_OF, []): 

513 inner_po = self._get_all_po(inner_uri) 

514 inner_types = inner_po.get(_P_TYPE, []) 

515 

516 if _T_JOURNAL_VOLUME in inner_types: 

517 inner_seqs = inner_po.get(_P_SEQ_ID, []) 

518 if inner_seqs: 

519 res_dict["volume"] = inner_seqs[0] 

520 else: 

521 venue_str = check_venue(inner_uri) 

522 if venue_str: 

523 res_dict["venue"] = venue_str 

524 

525 for venue_uri in inner_po.get(_P_PART_OF, []): 

526 titles = self._get_objects(venue_uri, _P_TITLE) 

527 if titles: 

528 venue_ids = extract_identifiers(venue_uri) 

529 res_dict["venue"] = f"{titles[0]} [{' '.join(venue_ids)}]" 

530 

531 return res_dict 

532 

533 _IRI_TO_TYPE = { 

534 GraphEntity.iri_archival_document: "archival document", 

535 GraphEntity.iri_book: "book", 

536 GraphEntity.iri_book_chapter: "book chapter", 

537 GraphEntity.iri_part: "book part", 

538 GraphEntity.iri_expression_collection: "book section", 

539 GraphEntity.iri_book_series: "book series", 

540 GraphEntity.iri_book_set: "book set", 

541 GraphEntity.iri_data_file: "data file", 

542 GraphEntity.iri_thesis: "dissertation", 

543 GraphEntity.iri_journal: "journal", 

544 GraphEntity.iri_journal_article: "journal article", 

545 GraphEntity.iri_journal_issue: "journal issue", 

546 GraphEntity.iri_journal_volume: "journal volume", 

547 GraphEntity.iri_proceedings_paper: "proceedings article", 

548 GraphEntity.iri_academic_proceedings: "proceedings", 

549 GraphEntity.iri_reference_book: "reference book", 

550 GraphEntity.iri_reference_entry: "reference entry", 

551 GraphEntity.iri_series: "series", 

552 GraphEntity.iri_report_document: "report", 

553 GraphEntity.iri_specification_document: "standard", 

554 } 

555 

556 @staticmethod 

557 def _type_it(br_type: str) -> str: 

558 return ResourceFinder._IRI_TO_TYPE.get(br_type, "") 

559 

560 def retrieve_publisher_from_br_metaid(self, metaid: str): 

561 metaid_uri = f"{self.base_iri}/{metaid}" 

562 publisher_ar_uris: set[str] = set() 

563 

564 def find_publisher_ars(entity_uri: str) -> None: 

565 for ar_uri in self._get_objects(entity_uri, _P_IS_DOC_CONTEXT_FOR): 

566 roles = self._get_objects(ar_uri, _P_WITH_ROLE) 

567 if _R_PUBLISHER in roles: 

568 publisher_ar_uris.add(ar_uri) 

569 

570 find_publisher_ars(metaid_uri) 

571 for parent_uri in self._get_objects(metaid_uri, _P_PART_OF): 

572 find_publisher_ars(parent_uri) 

573 for grandparent_uri in self._get_objects(parent_uri, _P_PART_OF): 

574 find_publisher_ars(grandparent_uri) 

575 

576 publishers_output = [] 

577 for ar_uri in publisher_ar_uris: 

578 pub_identifiers: List[str] = [] 

579 pub_name: str | None = None 

580 for ra_uri in self._get_objects(ar_uri, _P_IS_HELD_BY): 

581 pub_identifiers.append(ra_uri.replace(f"{self.base_iri}/", "omid:")) 

582 ra_po = self._get_all_po(ra_uri) 

583 names = ra_po.get(_P_NAME, []) 

584 if names: 

585 pub_name = names[0] 

586 for id_uri in ra_po.get(_P_HAS_IDENTIFIER, []): 

587 id_po = self._get_all_po(id_uri) 

588 schemes = id_po.get(_P_USES_ID_SCHEME, []) 

589 literals = id_po.get(_P_HAS_LITERAL_VALUE, []) 

590 if schemes and literals: 

591 pub_identifiers.append( 

592 f"{schemes[0].replace(_DATACITE, '')}:{literals[0]}" 

593 ) 

594 if pub_name is not None: 

595 pub_full = f"{pub_name} [{' '.join(pub_identifiers)}]" 

596 else: 

597 pub_full = f"[{' '.join(pub_identifiers)}]" 

598 publishers_output.append(pub_full) 

599 return "; ".join(publishers_output) 

600 

601 def get_everything_about_res( 

602 self, 

603 metavals: set, 

604 identifiers: set, 

605 vvis: set, 

606 max_depth: int = 10, 

607 progress: Progress | None = None, 

608 ) -> None: 

609 BATCH_SIZE = QLEVER_BATCH_SIZE 

610 MAX_WORKERS = min(self.workers, QLEVER_MAX_WORKERS) 

611 

612 def batch_process(input_set, batch_size): 

613 """Generator to split input data into smaller batches if batch_size is not None.""" 

614 if batch_size is None: 

615 yield input_set 

616 else: 

617 for i in range(0, len(input_set), batch_size): 

618 yield input_set[i : i + batch_size] 

619 

620 task_metavals = None 

621 task_identifiers = None 

622 task_vvis = None 

623 if progress: 

624 if metavals: 

625 task_metavals = progress.add_task( 

626 " [dim]Resolving OMIDs[/dim]", total=len(metavals) 

627 ) 

628 if identifiers: 

629 task_identifiers = progress.add_task( 

630 " [dim]Resolving identifiers[/dim]", total=len(identifiers) 

631 ) 

632 if vvis: 

633 task_vvis = progress.add_task( 

634 " [dim]Resolving VVI[/dim]", total=len(vvis) 

635 ) 

636 

637 max_depth_reached = 0 

638 

639 def process_batch_parallel(subjects, cur_depth, visited_subjects): 

640 nonlocal max_depth_reached 

641 if not subjects or (max_depth and cur_depth > max_depth): 

642 return 

643 

644 new_subjects = subjects - visited_subjects 

645 if not new_subjects: 

646 return 

647 

648 if cur_depth > max_depth_reached: 

649 max_depth_reached = cur_depth 

650 

651 visited_subjects.update(new_subjects) 

652 

653 if progress and task_traversal is not None: 

654 progress.update( 

655 task_traversal, 

656 description=f" [dim]Graph traversal (depth {cur_depth}/{max_depth}, {len(visited_subjects):,} subjects)[/dim]", 

657 ) 

658 

659 subject_list = list(new_subjects) 

660 batches = list(batch_process(subject_list, BATCH_SIZE)) 

661 batch_queries = [] 

662 ts_url = self.ts_url 

663 

664 for batch in batches: 

665 query = f""" 

666 SELECT ?s ?p ?o 

667 WHERE {{ 

668 VALUES ?s {{ {" ".join([f"<{s}>" for s in batch])} }} 

669 ?s ?p ?o. 

670 }}""" 

671 batch_queries.append(query) 

672 

673 next_subjects = set() 

674 if len(batch_queries) > 1 and MAX_WORKERS > 1: 

675 queries_per_worker = max(1, len(batch_queries) // MAX_WORKERS) 

676 query_groups = [ 

677 batch_queries[i : i + queries_per_worker] 

678 for i in range(0, len(batch_queries), queries_per_worker) 

679 ] 

680 worker = partial(execute_sparql_queries, ts_url) 

681 with ProcessPoolExecutor( 

682 max_workers=min(len(query_groups), MAX_WORKERS), 

683 mp_context=multiprocessing.get_context("forkserver"), 

684 ) as executor: 

685 grouped_results = list(executor.map(worker, query_groups)) 

686 results = [item for sublist in grouped_results for item in sublist] 

687 else: 

688 results = ( 

689 execute_sparql_queries(endpoint_url=ts_url, queries=batch_queries) 

690 if batch_queries 

691 else [] 

692 ) 

693 

694 _skip_preds = {_P_TYPE, _P_WITH_ROLE, _P_USES_ID_SCHEME} 

695 for result in results: 

696 for row in result: 

697 s_str = row["s"]["value"] 

698 p_str = row["p"]["value"] 

699 o_binding = row["o"] 

700 o_str = o_binding["value"] 

701 o_datatype = ( 

702 o_binding.get("datatype", "") 

703 if o_binding["type"] in ("literal", "typed-literal") 

704 else "" 

705 ) 

706 self.add_triple(s_str, p_str, o_str, o_datatype=o_datatype) 

707 if o_binding["type"] == "uri" and p_str not in _skip_preds: 

708 next_subjects.add(o_str) 

709 

710 process_batch_parallel(next_subjects, cur_depth + 1, visited_subjects) 

711 

712 def get_initial_subjects_from_metavals(metavals): 

713 """Convert metavals to a set of subjects."""