Coverage for src/oc_graphenricher/deduplication.py: 95%

1# SPDX-FileCopyrightText: 2021 Gabriele Pisciotta <ga.pisciotta@gmail.com> 

2# SPDX-FileCopyrightText: 2021 Simone Persiani 

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

4# 

5# SPDX-License-Identifier: ISC 

6 

7from __future__ import annotations 

8 

9import logging 

10from collections import defaultdict 

11from dataclasses import replace 

12from typing import TYPE_CHECKING, TypeAlias, TypeVar, cast 

13 

14import Levenshtein 

15import networkx as nx 

16from oc_ocdm.graph.entities.bibliographic.bibliographic_resource import BibliographicResource 

17from oc_ocdm.graph.entities.bibliographic.responsible_agent import ResponsibleAgent 

18from oc_ocdm.graph.entities.identifier import Identifier 

19from oc_ocdm.graph.graph_entity import GraphEntity 

20from oc_ocdm.prov.prov_set import ProvSet 

21 

22from oc_graphenricher._storage import store_graph_set, store_provenance 

23 

24if TYPE_CHECKING: 

25 from collections.abc import Iterable, Mapping 

26 

27 from oc_ocdm.graph.entities.bibliographic.agent_role import AgentRole 

28 from oc_ocdm.graph.graph_set import GraphSet 

29 

30 from oc_graphenricher.storage import Storage 

31 

32LOGGER = logging.getLogger(__name__) 

33NAME_SIMILARITY_THRESHOLD = 0.95 

34Entity = TypeVar("Entity") 

35ManualMergeEntity: TypeAlias = BibliographicResource | ResponsibleAgent | Identifier 

36IdentifierSignature: TypeAlias = tuple[str, str] 

37IdentifierReferenceSources: TypeAlias = defaultdict[Identifier, list[GraphEntity]] 

38MANUAL_BR_ROLE_TYPES = (GraphEntity.iri_author, GraphEntity.iri_editor) 

39 

40 

41class GraphDeduplicator: 

42 def __init__( 

43 self, 

44 graph_set: GraphSet, 

45 storage: Storage | None = None, 

46 *, 

47 debug: bool = False, 

48 merge_similar_named_contributors: bool = False, 

49 preferred_survivors: set[str] | None = None, 

50 ) -> None: 

51 """ 

52 Initialize the graph deduplicator. 

53 

54 The deduplicator merges duplicate entities in a graph set compliant with the OpenCitations Data Model. 

55 

56 :param graph_set: input graph set 

57 :param storage: output storage configuration 

58 :param debug: a bool flag to enable richer output 

59 :param merge_similar_named_contributors: merge contributor roles with similar author names within merged BRs 

60 :param preferred_survivors: entity URIs to keep when they appear in duplicate clusters 

61 """ 

62 self.graph_set = graph_set 

63 self.storage = storage 

64 self.debug = debug 

65 self.merge_similar_named_contributors = merge_similar_named_contributors 

66 self.preferred_survivors = preferred_survivors if preferred_survivors is not None else set() 

67 self.modified_entities: set[str] = set() 

68 self.prov = self.__provenance() 

69 

70 def deduplicate_and_save(self) -> GraphSet: 

71 """ 

72 Deduplicate the graph set and save the graph and provenance. 

73 

74 The process will: 

75 - deduplicate the Responsible Agents (RAs) 

76 - deduplicate the Bibliographic Resources (BRs) 

77 - deduplicate the IDs. 

78 

79 In the end, this process will produce: 

80 - the configured graph output without the duplicates. 

81 - the configured provenance output tracking the changes done. 

82 """ 

83 self.__deduplicate_responsible_agents() 

84 self.__save_and_commit() 

85 self.__deduplicate_bibliographic_resources() 

86 self.__save_and_commit() 

87 self.__deduplicate_identifiers() 

88 self.__save_and_commit() 

89 return self.graph_set 

90 

91 def deduplicate(self) -> GraphSet: 

92 """Deduplicate the graph set without serializing the graph set or provenance.""" 

93 self.deduplicate_responsible_agents() 

94 self.deduplicate_bibliographic_resources() 

95 self.deduplicate_identifiers() 

96 return self.graph_set 

97 

98 def merge_clusters(self, clusters: Mapping[str, Iterable[str]]) -> GraphSet: 

99 self.__merge_clusters(clusters) 

100 self.graph_set.commit_changes() 

101 return self.graph_set 

102 

103 def __merge_clusters(self, clusters: Mapping[str, Iterable[str]]) -> None: 

104 normalized_clusters = self.__manual_merge_clusters(clusters) 

105 if not normalized_clusters: 

106 return 

107 

108 self.__validate_identifier_clusters(normalized_clusters) 

109 associated_ar_ra = self.__get_association_ar_ra() 

110 identifier_reference_sources: IdentifierReferenceSources | None = None 

111 

112 for surviving_entity, merged_entities in self.__ordered_clusters(normalized_clusters): 

113 self.__ensure_cluster_entities_alive(surviving_entity, merged_entities) 

114 if isinstance(surviving_entity, ResponsibleAgent): 

115 for merged_entity in merged_entities: 

116 self.__merge_responsible_agent( 

117 surviving_entity, 

118 self.__as_responsible_agent(merged_entity), 

119 associated_ar_ra, 

120 ) 

121 elif isinstance(surviving_entity, BibliographicResource): 

122 self.__merge_manual_bibliographic_resources( 

123 surviving_entity, 

124 [self.__as_bibliographic_resource(merged_entity) for merged_entity in merged_entities], 

125 ) 

126 else: 

127 if identifier_reference_sources is None: 

128 identifier_reference_sources = self.__identifier_reference_sources() 

129 self.__merge_identifiers( 

130 surviving_entity, 

131 [self.__as_identifier(merged_entity) for merged_entity in merged_entities], 

132 identifier_reference_sources, 

133 ) 

134 

135 self.modified_entities.update(self.prov.generate_provenance()) 

136 

137 def __ordered_clusters( 

138 self, 

139 normalized_clusters: list[tuple[ManualMergeEntity, list[ManualMergeEntity]]], 

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

141 # A merge can delete entities other clusters name: a BR/RA merge 

142 # deduplicates identifiers by scheme and literal (one of the two dies) 

143 # and the container cascade of a BR merge deletes ancestor containers. 

144 # Identifier clusters therefore run first, and BR clusters run top-down 

145 # by container depth, so every cluster is merged before anything can 

146 # delete its entities. 

147 return sorted(normalized_clusters, key=self.__cluster_rank) 

148 

149 def __cluster_rank( 

150 self, 

151 cluster: tuple[ManualMergeEntity, list[ManualMergeEntity]], 

152 ) -> tuple[int, int]: 

153 surviving_entity, merged_entities = cluster 

154 if isinstance(surviving_entity, Identifier): 

155 return (0, 0) 

156 if isinstance(surviving_entity, ResponsibleAgent): 

157 return (1, 0) 

158 depth = max( 

159 len(self.__get_part_of(self.__as_bibliographic_resource(entity))) 

160 for entity in (surviving_entity, *merged_entities) 

161 ) 

162 return (2, depth) 

163 

164 def __ensure_cluster_entities_alive( 

165 self, 

166 surviving_entity: ManualMergeEntity, 

167 merged_entities: list[ManualMergeEntity], 

168 ) -> None: 

169 for entity in (surviving_entity, *merged_entities): 

170 if entity.to_be_deleted: 

171 message = ( 

172 f"Entity {entity.res} is deleted ({self.__deletion_cause(entity)}) " 

173 f"and cannot take part in the merge cluster of {surviving_entity.res}." 

174 ) 

175 raise ValueError(message) 

176 

177 def __deletion_cause(self, entity: ManualMergeEntity) -> str: 

178 target = str(entity.res) 

179 for candidate in self.graph_set.res_to_entity.values(): 

180 if any(str(merged.res) == target for merged in candidate.merge_list): 

181 return f"merged into {candidate.res}" 

182 return "marked for deletion" 

183 

184 def merge_clusters_and_save(self, clusters: Mapping[str, Iterable[str]]) -> GraphSet: 

185 self.__merge_clusters(clusters) 

186 self.__save_and_commit() 

187 return self.graph_set 

188 

189 def save(self) -> None: 

190 """ 

191 Serialize the graph set into the specified RDF file. 

192 

193 Serialize the provenance in another specified RDF file. 

194 """ 

195 storage = self.__storage() 

196 store_graph_set(self.graph_set, storage) 

197 store_provenance(self.prov, storage) 

198 

199 def __save_and_commit(self) -> None: 

200 self.save() 

201 self.graph_set.commit_changes() 

202 self.modified_entities.clear() 

203 

204 def deduplicate_responsible_agents(self) -> None: 

205 """ 

206 Discover Responsible Agents (RAs) that share the same identifier literal. 

207 

208 The process creates a graph of duplicate entities, merges each connected component into one RA, updates Agent 

209 Role references, generates provenance and commits pending changes in the graph set. 

210 """ 

211 self.__deduplicate_responsible_agents() 

212 self.graph_set.commit_changes() 

213 

214 def __deduplicate_responsible_agents(self) -> None: 

215 associated_ar_ra = self.__get_association_ar_ra() 

216 clusters = self.__sorted_clusters( 

217 self.__merge_graph(self.graph_set.get_ra(), "[dedup-RA] Will merge %s and %s due to %s:%s in common"), 

218 ) 

219 LOGGER.info("[dedup-RA] Number of clusters: %s", len(clusters)) 

220 

221 for cluster_index, cluster in enumerate(clusters): 

222 entity_first_raw, other_entities_raw = self.__ordered_entities(cluster) 

223 entity_first = self.__as_responsible_agent(entity_first_raw) 

224 self.__debug("[dedup-RA] Merging cluster #%s, with %s entities", cluster_index, len(cluster)) 

225 for other_entity_raw in other_entities_raw: 

226 self.__merge_responsible_agent( 

227 entity_first, 

228 self.__as_responsible_agent(other_entity_raw), 

229 associated_ar_ra, 

230 ) 

231 

232 self.modified_entities.update(self.prov.generate_provenance()) 

233 

234 def deduplicate_bibliographic_resources(self) -> None: 

235 """ 

236 Discover Bibliographic Resources (BRs) that share the same identifier literal. 

237 

238 The process creates a graph of duplicate BRs, merges each connected component into one BR, merges containers and 

239 publishers where possible, generates provenance and commits pending changes in the graph set. 

240 """ 

241 self.__deduplicate_bibliographic_resources() 

242 self.graph_set.commit_changes() 

243 

244 def __deduplicate_bibliographic_resources(self) -> None: 

245 clusters = self.__sorted_clusters( 

246 self.__merge_graph(self.graph_set.get_br(), "[dedup-BR] Will merge %s into %s due to %s:%s in common"), 

247 ) 

248 LOGGER.info("[dedup-BR] Number of clusters: %s", len(clusters)) 

249 

250 for cluster_index, cluster in enumerate(clusters): 

251 self.__debug("[dedup-BR] Merging cluster #%s, with %s entities", cluster_index, len(cluster)) 

252 self.__merge_br_cluster(cluster) 

253 

254 self.modified_entities.update(self.prov.generate_provenance()) 

255 

256 def deduplicate_identifiers(self) -> None: 

257 """ 

258 Discover duplicate IDs related to Bibliographic Resources and Responsible Agents. 

259 

260 IDs are duplicates when they share the same schema and literal. The process merges duplicates into one ID, 

261 substitutes references with the merged ID, generates provenance and commits pending changes in the graph set. 

262 """ 

263 self.__deduplicate_identifiers() 

264 self.graph_set.commit_changes() 

265 

266 def __deduplicate_identifiers(self) -> None: 

267 literal_to_id = self.__identifier_groups() 

268 identifier_reference_sources: IdentifierReferenceSources | None = None 

269 for literal, identifiers in literal_to_id.items(): 

270 if len(identifiers) > 1: 

271 if identifier_reference_sources is None: 

272 identifier_reference_sources = self.__identifier_reference_sources() 

273 self.__merge_identifier_group(literal, identifiers, identifier_reference_sources) 

274 

275 self.modified_entities.update(self.prov.generate_provenance()) 

276 

277 def __merge_graph( 

278 self, 

279 entities: Iterable[ResponsibleAgent | BibliographicResource], 

280 debug_message: str, 

281 ) -> nx.Graph: 

282 merge_graph: nx.Graph = nx.Graph() 

283 identifiers: dict[str, dict[str, ResponsibleAgent | BibliographicResource]] = {} 

284 for entity in entities: 

285 for identifier in entity.get_identifiers(): 

286 scheme = identifier.get_scheme() 

287 literal_value = identifier.get_literal_value() 

288 if scheme is None or literal_value is None: 

289 continue 

290 identifiers.setdefault(scheme, {}) 

291 entity_first = identifiers[scheme].get(literal_value) 

292 if entity_first is None: 

293 identifiers[scheme][literal_value] = entity 

294 else: 

295 merge_graph.add_edge(entity_first, entity) 

296 self.__debug( 

297 debug_message, 

298 entity.res, 

299 entity_first.res, 

300 scheme.split("/")[-1], 

301 literal_value, 

302 ) 

303 return merge_graph 

304 

305 def __sorted_clusters(self, merge_graph: nx.Graph) -> list[set[ResponsibleAgent | BibliographicResource]]: 

306 return sorted(nx.connected_components(merge_graph), key=len, reverse=True) 

307 

308 def __manual_merge_clusters( 

309 self, 

310 clusters: Mapping[str, Iterable[str]], 

311 ) -> list[tuple[ManualMergeEntity, list[ManualMergeEntity]]]: 

312 normalized_clusters: list[tuple[ManualMergeEntity, list[ManualMergeEntity]]] = [] 

313 survivors: set[str] = set() 

314 merged_to_survivor: dict[str, str] = {} 

315 container_uris = self.__container_uris_with_members() 

316 

317 for survivor_uri, merged_uris in clusters.items(): 

318 survivor = self.__manual_survivor(survivor_uri, merged_uris, survivors, merged_to_survivor) 

319 survivor_type = self.__merge_entity_type(survivor) 

320 merged_entities = [ 

321 self.__manual_merged_entity( 

322 survivor_uri, 

323 survivor_type, 

324 merged_uri, 

325 survivors, 

326 merged_to_survivor, 

327 ) 

328 for merged_uri in merged_uris 

329 ] 

330 

331 if not merged_entities: 

332 message = f"Merge cluster for survivor {survivor_uri} must include at least one merged entity." 

333 raise ValueError(message) 

334 self.__validate_container_type_compatibility(survivor, merged_entities, container_uris) 

335 normalized_clusters.append((survivor, merged_entities)) 

336 

337 return normalized_clusters 

338 

339 def __container_uris_with_members(self) -> set[str]: 

340 members: set[str] = set() 

341 for br in self.graph_set.get_br(): 

342 container = br.get_is_part_of() 

343 if container is not None: 

344 members.add(str(container.res)) 

345 return members 

346 

347 def __validate_container_type_compatibility( 

348 self, 

349 survivor: ManualMergeEntity, 

350 merged_entities: Iterable[ManualMergeEntity], 

351 container_uris: set[str], 

352 ) -> None: 

353 if not isinstance(survivor, BibliographicResource): 

354 return 

355 survivor_types = set(self.__specific_types(survivor)) 

356 for merged_entity in merged_entities: 

357 merged = self.__as_bibliographic_resource(merged_entity) 

358 merged_types = set(self.__specific_types(merged)) 

359 if not survivor_types or not merged_types or not survivor_types.isdisjoint(merged_types): 

360 continue 

361 if str(survivor.res) in container_uris or str(merged.res) in container_uris: 

362 message = ( 

363 "Cannot merge bibliographic resources with incompatible types: " 

364 f"{survivor.res} is {sorted(survivor_types)}, {merged.res} is {sorted(merged_types)}." 

365 ) 

366 raise ValueError(message) 

367 

368 def __validate_identifier_clusters( 

369 self, 

370 normalized_clusters: Iterable[tuple[ManualMergeEntity, list[ManualMergeEntity]]], 

371 ) -> None: 

372 for surviving_entity, merged_entities in normalized_clusters: 

373 if not isinstance(surviving_entity, Identifier): 

374 continue 

375 

376 surviving_signature = self.__identifier_signature(surviving_entity) 

377 for merged_entity in merged_entities: 

378 merged_identifier = self.__as_identifier(merged_entity) 

379 merged_signature = self.__identifier_signature(merged_identifier) 

380 if merged_signature != surviving_signature: 

381 message = ( 

382 "Cannot merge identifiers with different scheme/literal: " 

383 f"{surviving_entity.res} has {surviving_signature[0]}#{surviving_signature[1]}, " 

384 f"{merged_identifier.res} has {merged_signature[0]}#{merged_signature[1]}." 

385 ) 

386 raise ValueError(message) 

387 

388 @staticmethod 

389 def __identifier_signature(identifier: Identifier) -> IdentifierSignature: 

390 scheme = identifier.get_scheme() 

391 literal = identifier.get_literal_value() 

392 if scheme is None or literal is None: 

393 message = f"Cannot merge identifier {identifier.res} without scheme or literal." 

394 raise ValueError(message) 

395 return scheme, literal 

396 

397 def __manual_survivor( 

398 self, 

399 survivor_uri: str, 

400 merged_uris: Iterable[str], 

401 survivors: set[str], 

402 merged_to_survivor: dict[str, str], 

403 ) -> ManualMergeEntity: 

404 if survivor_uri in merged_to_survivor: 

405 message = f"Entity {survivor_uri} cannot be both survivor and merged entity." 

406 raise ValueError(message) 

407 if survivor_uri in survivors: 

408 message = f"Duplicate survivor in merge clusters: {survivor_uri}." 

409 raise ValueError(message) 

410 if isinstance(merged_uris, str): 

411 message = f"Merge cluster for survivor {survivor_uri} must be an iterable of entity URIs." 

412 raise TypeError(message) 

413 

414 survivors.add(survivor_uri) 

415 return self.__merge_entity(survivor_uri) 

416 

417 def __manual_merged_entity( 

418 self, 

419 survivor_uri: str, 

420 survivor_type: str, 

421 merged_uri: str, 

422 survivors: set[str], 

423 merged_to_survivor: dict[str, str], 

424 ) -> ManualMergeEntity: 

425 if merged_uri == survivor_uri: 

426 message = f"Entity {survivor_uri} cannot be merged into itself." 

427 raise ValueError(message) 

428 if merged_uri in survivors: 

429 message = f"Entity {merged_uri} cannot be both survivor and merged entity." 

430 raise ValueError(message) 

431 if merged_uri in merged_to_survivor: 

432 message = f"Entity {merged_uri} is already assigned to survivor {merged_to_survivor[merged_uri]}." 

433 raise ValueError(message) 

434 

435 merged_entity = self.__merge_entity(merged_uri) 

436 merged_entity_type = self.__merge_entity_type(merged_entity) 

437 if merged_entity_type != survivor_type: 

438 message = ( 

439 f"Merge cluster for survivor {survivor_uri} mixes entity types: " 

440 f"{survivor_type} and {merged_entity_type}." 

441 ) 

442 raise ValueError(message) 

443 

444 merged_to_survivor[merged_uri] = survivor_uri 

445 return merged_entity 

446 

447 def __merge_entity(self, uri: str) -> ManualMergeEntity: 

448 entity = self.graph_set.get_entity(uri) 

449 if entity is None: 

450 message = f"Entity not found in GraphSet: {uri}." 

451 raise ValueError(message) 

452 if isinstance(entity, BibliographicResource | ResponsibleAgent | Identifier): 

453 return entity 

454 message = f"Entity {uri} has unsupported merge type: {type(entity).__name__}." 

455 raise ValueError(message) 

456 

457 @staticmethod 

458 def __merge_entity_type(entity: ManualMergeEntity) -> str: 

459 if isinstance(entity, ResponsibleAgent): 

460 return "ra" 

461 if isinstance(entity, BibliographicResource): 

462 return "br" 

463 return "id" 

464 

465 def __merge_responsible_agent( 

466 self, 

467 entity_first: ResponsibleAgent, 

468 other_entity: ResponsibleAgent, 

469 associated_ar_ra: dict[ResponsibleAgent, list[AgentRole]], 

470 ) -> None: 

471 self.__debug("\tMerging responsible agent %s in responsible agent %s", other_entity, entity_first) 

472 entity_first.merge(other_entity, prefer_self=True) 

473 associated_ars = associated_ar_ra.get(other_entity, []) 

474 for ar in associated_ars: 

475 ar.is_held_by(entity_first) 

476 self.__debug("\tUnset %s as helded by of %s", other_entity, ar) 

477 self.__debug("\tSet %s as helded by of %s", entity_first, ar) 

478 self.__deduplicate_contributors_for_bibliographic_resources( 

479 self.__bibliographic_resources_for_contributors(associated_ars), 

480 ) 

481 self.__debug("\tMarking to delete: %s", other_entity) 

482 

483 def __merge_br_cluster(self, cluster: set[ResponsibleAgent | BibliographicResource]) -> None: 

484 entity_first_raw, other_entities_raw = self.__ordered_entities(cluster) 

485 self.__merge_bibliographic_resources( 

486 self.__as_bibliographic_resource(entity_first_raw), 

487 [self.__as_bibliographic_resource(other_entity) for other_entity in other_entities_raw], 

488 ) 

489 

490 def __merge_bibliographic_resources( 

491 self, 

492 entity_first: BibliographicResource, 

493 other_entities: Iterable[BibliographicResource], 

494 ) -> None: 

495 for other_entity in other_entities: 

496 # Recompute the survivor's publisher and part-of chain on every iteration: an 

497 # earlier merge may have filled a previously missing publisher or container, and 

498 # the next equivalent one must merge into it instead of surviving as a duplicate. 

499 publisher_first = self.__get_publisher(entity_first) 

500 entity_first_partofs = self.__get_part_of(entity_first) 

501 self.__merge_containers(entity_first_partofs, self.__get_part_of(other_entity)) 

502 self.__merge_publisher(publisher_first, other_entity) 

503 entity_first.merge(other_entity, prefer_self=True) 

504 self.__deduplicate_contributors_for_bibliographic_resources([entity_first]) 

505 

506 def __merge_manual_bibliographic_resources( 

507 self, 

508 entity_first: BibliographicResource, 

509 other_entities: Iterable[BibliographicResource], 

510 ) -> None: 

511 merged_entities = list(other_entities) 

512 self.__discard_merged_author_editor_roles(entity_first, merged_entities) 

513 for other_entity in merged_entities: 

514 for first_partof, second_partof in self.__container_merge_pairs( 

515 self.__get_part_of(entity_first), 

516 self.__get_part_of(other_entity), 

517 ): 

518 self.__discard_merged_author_editor_roles(first_partof, [second_partof]) 

519 self.__merge_bibliographic_resources(entity_first, [other_entity]) 

520 

521 def __merge_containers( 

522 self, 

523 entity_first_partofs: list[BibliographicResource], 

524 partofs: list[BibliographicResource], 

525 ) -> None: 

526 for first_partof, second_partof in self.__container_merge_pairs(entity_first_partofs, partofs): 

527 self.__merge_publisher(self.__get_publisher(first_partof), second_partof) 

528 first_partof.merge(second_partof, prefer_self=True) 

529 self.__deduplicate_contributors_for_bibliographic_resources([first_partof]) 

530 self.__debug("\tMerging container %s in container %s", second_partof, first_partof) 

531 

532 def __container_merge_pairs( 

533 self, 

534 entity_first_partofs: list[BibliographicResource], 

535 partofs: list[BibliographicResource], 

536 ) -> list[tuple[BibliographicResource, BibliographicResource]]: 

537 second_by_type: dict[str, BibliographicResource] = {} 

538 for container in partofs: 

539 for type_iri in self.__specific_types(container): 

540 second_by_type.setdefault(type_iri, container) 

541 

542 pairs: list[tuple[BibliographicResource, BibliographicResource]] = [] 

543 parent_merged = True 

544 for first_partof in reversed(entity_first_partofs): # walk from the top of the hierarchy downward 

545 second_partof = self.__matching_container(first_partof, second_by_type) 

546 if second_partof is None or first_partof.res == second_partof.res: 

547 continue 

548 if parent_merged and self.__containers_are_equivalent(first_partof, second_partof): 

549 pairs.append((first_partof, second_partof)) 

550 else: 

551 parent_merged = False 

552 return pairs 

553 

554 @staticmethod 

555 def __specific_types(container: BibliographicResource) -> list[str]: 

556 return [type_iri for type_iri in container.get_types() if type_iri != GraphEntity.iri_expression] 

557 

558 def __matching_container( 

559 self, 

560 container: BibliographicResource, 

561 second_by_type: dict[str, BibliographicResource], 

562 ) -> BibliographicResource | None: 

563 for type_iri in self.__specific_types(container): 

564 if type_iri in second_by_type: 

565 return second_by_type[type_iri] 

566 return None 

567 

568 def __containers_are_equivalent( 

569 self, 

570 first: BibliographicResource, 

571 second: BibliographicResource, 

572 ) -> bool: 

573 first_signatures = self.__identifier_signatures(first) 

574 second_signatures = self.__identifier_signatures(second) 

575 if first_signatures and second_signatures: 

576 return not first_signatures.isdisjoint(second_signatures) 

577 

578 first_number = first.get_number() 

579 second_number = second.get_number() 

580 if first_number is not None and second_number is not None: 

581 return first_number == second_number 

582 

583 first_title = first.get_title() 

584 second_title = second.get_title() 

585 if first_title is not None and second_title is not None: 

586 return self.__normalized_text(first_title) == self.__normalized_text(second_title) 

587 

588 return False 

589 

590 @staticmethod 

591 def __identifier_signatures(entity: BibliographicResource | ResponsibleAgent) -> set[IdentifierSignature]: 

592 signatures: set[IdentifierSignature] = set() 

593 for identifier in entity.get_identifiers(): 

594 scheme = identifier.get_scheme() 

595 literal = identifier.get_literal_value() 

596 if scheme is not None and literal is not None: 

597 signatures.add((scheme, literal)) 

598 return signatures 

599 

600 @staticmethod 

601 def __normalized_text(value: str) -> str: 

602 return " ".join(value.casefold().split()) 

603 

604 def __merge_publisher(self, publisher_first: AgentRole | None, entity: BibliographicResource) -> None: 

605 publisher = self.__get_publisher(entity) 

606 if ( 

607 publisher is not None 

608 and publisher_first is not None 

609 and publisher != publisher_first 

610 and self.__publishers_are_equivalent(publisher_first, publisher) 

611 ): 

612 publisher_first.merge(publisher, prefer_self=True) 

613 self.__debug("\tMerging publisher %s in publisher %s", publisher, publisher_first) 

614 

615 def __publishers_are_equivalent(self, first: AgentRole, second: AgentRole) -> bool: 

616 first_agent = first.get_is_held_by() 

617 second_agent = second.get_is_held_by() 

618 if first_agent is None or second_agent is None: 

619 return False 

620 # The responsible agent behind a publisher role is often outside the loaded 

621 # closure, so its identifiers and name are unavailable; two roles held by the 

622 # same agent URI are the same publisher regardless of what data is loaded. 

623 if str(first_agent.res) == str(second_agent.res): 

624 return True 

625 first_signatures = self.__identifier_signatures(first_agent) 

626 second_signatures = self.__identifier_signatures(second_agent) 

627 if first_signatures and second_signatures: 

628 return not first_signatures.isdisjoint(second_signatures) 

629 first_name = self.__agent_full_name(first_agent) 

630 second_name = self.__agent_full_name(second_agent) 

631 if first_name and second_name: 

632 return first_name == second_name 

633 return False 

634 

635 def __agent_full_name(self, agent: ResponsibleAgent) -> str: 

636 name = agent.get_name() 

637 if name is not None: 

638 return self.__normalized_text(name) 

639 parts = [part for part in (agent.get_given_name(), agent.get_family_name()) if part is not None] 

640 return self.__normalized_text(" ".join(parts)) if parts else "" 

641 

642 def __bibliographic_resources_for_contributors( 

643 self, 

644 contributors: Iterable[AgentRole], 

645 ) -> list[BibliographicResource]: 

646 contributor_uris = {str(contributor.res) for contributor in contributors} 

647 if not contributor_uris: 

648 return [] 

649 return [ 

650 br 

651 for br in sorted(self.graph_set.get_br(), key=str) 

652 if any(str(contributor.res) in contributor_uris for contributor in br.get_contributors()) 

653 ] 

654 

655 def __deduplicate_contributors_for_bibliographic_resources( 

656 self, 

657 bibliographic_resources: Iterable[BibliographicResource], 

658 ) -> None: 

659 for bibliographic_resource in bibliographic_resources: 

660 already_merged = self.__merge_same_ra_contributors(bibliographic_resource) 

661 if self.merge_similar_named_contributors: 

662 self.__merge_similar_named_contributors(bibliographic_resource, already_merged) 

663 self.__remove_contributors_without_ra(bibliographic_resource) 

664 

665 def __discard_merged_author_editor_roles( 

666 self, 

667 entity_first: BibliographicResource, 

668 merged_entities: list[BibliographicResource], 

669 ) -> None: 

670 for role_type in MANUAL_BR_ROLE_TYPES: 

671 self.__keep_single_role_chain(entity_first, merged_entities, role_type) 

672 

673 def __keep_single_role_chain( 

674 self, 

675 entity_first: BibliographicResource, 

676 merged_entities: list[BibliographicResource], 

677 role_type: str, 

678 ) -> None: 

679 donor = None 

680 if not self.__has_contributor_role(entity_first, role_type): 

681 donor = self.__richest_role_donor(merged_entities, role_type) 

682 for merged_entity in merged_entities: 

683 if merged_entity is donor: 

684 continue 

685 for contributor in list(merged_entity.get_contributors()): 

686 if contributor.get_role_type() == role_type: 

687 merged_entity.remove_contributor(contributor) 

688 contributor.mark_as_to_be_deleted() 

689 

690 def __richest_role_donor( 

691 self, 

692 merged_entities: list[BibliographicResource], 

693 role_type: str, 

694 ) -> BibliographicResource | None: 

695 donors = [entity for entity in merged_entities if self.__contributor_role_count(entity, role_type) > 0] 

696 if not donors: 

697 return None 

698 return min(donors, key=lambda entity: (-self.__contributor_role_count(entity, role_type), str(entity.res))) 

699 

700 def __has_contributor_role(self, entity: BibliographicResource, role_type: str) -> bool: 

701 return any(contributor.get_role_type() == role_type for contributor in entity.get_contributors()) 

702 

703 def __contributor_role_count(self, entity: BibliographicResource, role_type: str) -> int: 

704 return sum(1 for contributor in entity.get_contributors() if contributor.get_role_type() == role_type) 

705 

706 def __merge_same_ra_contributors(self, entity_first: BibliographicResource) -> set[AgentRole]: 

707 # Group every contributor role, publishers included, by (RA, role type). A 

708 # direct RA merge or a container merge can leave two publisher roles held by 

709 # the same agent on one resource, and __merge_publisher only reconciles the 

710 # single publisher returned by __get_publisher, so the extras must collapse 

711 # here alongside authors and editors. 

712 contributors_by_agent_role: dict[tuple[str, str], list[AgentRole]] = {} 

713 for contributor in entity_first.get_contributors(): 

714 responsible_agent = contributor.get_is_held_by() 

715 role_type = contributor.get_role_type() 

716 if responsible_agent is None or role_type is None: 

717 continue 

718 key = (str(responsible_agent.res), role_type) 

719 contributors_by_agent_role.setdefault(key, []).append(contributor) 

720 

721 already_merged: set[AgentRole] = set() 

722 for contributors in contributors_by_agent_role.values(): 

723 ordered_contributors = sorted(contributors, key=str) 

724 surviving_contributor = ordered_contributors[0] 

725 for merged_contributor in ordered_contributors[1:]: 

726 self.__debug( 

727 "\tRemoving agent role %s from bibliographic resource %s because both point to the same RA", 

728 merged_contributor, 

729 entity_first, 

730 ) 

731 self.__merge_contributor(entity_first, surviving_contributor, merged_contributor) 

732 already_merged.add(surviving_contributor) 

733 already_merged.add(merged_contributor) 

734 return already_merged 

735 

736 def __merge_similar_named_contributors( 

737 self, 

738 entity_first: BibliographicResource, 

739 already_merged: set[AgentRole], 

740 ) -> None: 

741 already_merged_uris = {str(contributor.res) for contributor in already_merged} 

742 contributors = [ 

743 contributor 

744 for contributor in sorted(self.__author_contributors(entity_first), key=str) 

745 if str(contributor.res) not in already_merged_uris and contributor.get_role_type() is not None 

746 ] 

747 merged_contributor_uris: set[str] = set() 

748 for ar1_index, ar1 in enumerate(contributors):