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

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 dataclasses import replace 

11from typing import TYPE_CHECKING, TypeAlias, TypeVar, cast 

12 

13import Levenshtein 

14import networkx as nx 

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

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

17from oc_ocdm.graph.entities.identifier import Identifier 

18from oc_ocdm.graph.graph_entity import GraphEntity 

19from oc_ocdm.prov.prov_set import ProvSet 

20 

21from oc_graphenricher._storage import store_graph_set, store_provenance 

22 

23if TYPE_CHECKING: 

24 from collections.abc import Iterable, Mapping 

25 

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

27 from oc_ocdm.graph.graph_set import GraphSet 

28 

29 from oc_graphenricher.storage import Storage 

30 

31LOGGER = logging.getLogger(__name__) 

32NAME_SIMILARITY_THRESHOLD = 0.95 

33Entity = TypeVar("Entity") 

34ManualMergeEntity: TypeAlias = BibliographicResource | ResponsibleAgent | Identifier 

35IdentifierSignature: TypeAlias = tuple[str, str] 

36 

37 

38class GraphDeduplicator: 

39 def __init__( 

40 self, 

41 graph_set: GraphSet, 

42 storage: Storage | None = None, 

43 *, 

44 debug: bool = False, 

45 merge_similar_named_contributors: bool = False, 

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

47 ) -> None: 

48 """ 

49 Initialize the graph deduplicator. 

50 

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

52 

53 :param graph_set: input graph set 

54 :param storage: output storage configuration 

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

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

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

58 """ 

59 self.graph_set = graph_set 

60 self.storage = storage 

61 self.debug = debug 

62 self.merge_similar_named_contributors = merge_similar_named_contributors 

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

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

65 self.prov = self.__provenance() 

66 

67 def deduplicate_and_save(self) -> GraphSet: 

68 """ 

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

70 

71 The process will: 

72 - deduplicate the Responsible Agents (RAs) 

73 - deduplicate the Bibliographic Resources (BRs) 

74 - deduplicate the IDs. 

75 

76 In the end, this process will produce: 

77 - the configured graph output without the duplicates. 

78 - the configured provenance output tracking the changes done. 

79 """ 

80 self.__deduplicate_responsible_agents() 

81 self.__save_and_commit() 

82 self.__deduplicate_bibliographic_resources() 

83 self.__save_and_commit() 

84 self.__deduplicate_identifiers() 

85 self.__save_and_commit() 

86 return self.graph_set 

87 

88 def deduplicate(self) -> GraphSet: 

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

90 self.deduplicate_responsible_agents() 

91 self.deduplicate_bibliographic_resources() 

92 self.deduplicate_identifiers() 

93 return self.graph_set 

94 

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

96 self.__merge_clusters(clusters) 

97 self.graph_set.commit_changes() 

98 return self.graph_set 

99 

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

101 normalized_clusters = self.__manual_merge_clusters(clusters) 

102 if not normalized_clusters: 

103 return 

104 

105 self.__validate_identifier_clusters(normalized_clusters) 

106 associated_ar_ra = self.__get_association_ar_ra() 

107 _, id_to_resources = self.__id_maps() 

108 

109 for surviving_entity, merged_entities in normalized_clusters: 

110 if isinstance(surviving_entity, ResponsibleAgent): 

111 for merged_entity in merged_entities: 

112 self.__merge_responsible_agent( 

113 surviving_entity, 

114 self.__as_responsible_agent(merged_entity), 

115 associated_ar_ra, 

116 ) 

117 elif isinstance(surviving_entity, BibliographicResource): 

118 self.__merge_bibliographic_resources( 

119 surviving_entity, 

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

121 ) 

122 else: 

123 self.__merge_identifiers( 

124 surviving_entity, 

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

126 id_to_resources, 

127 ) 

128 

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

130 

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

132 self.__merge_clusters(clusters) 

133 self.__save_and_commit() 

134 return self.graph_set 

135 

136 def save(self) -> None: 

137 """ 

138 Serialize the graph set into the specified RDF file. 

139 

140 Serialize the provenance in another specified RDF file. 

141 """ 

142 storage = self.__storage() 

143 store_graph_set(self.graph_set, storage) 

144 store_provenance(self.prov, storage) 

145 

146 def __save_and_commit(self) -> None: 

147 self.save() 

148 self.graph_set.commit_changes() 

149 self.modified_entities.clear() 

150 

151 def deduplicate_responsible_agents(self) -> None: 

152 """ 

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

154 

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

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

157 """ 

158 self.__deduplicate_responsible_agents() 

159 self.graph_set.commit_changes() 

160 

161 def __deduplicate_responsible_agents(self) -> None: 

162 associated_ar_ra = self.__get_association_ar_ra() 

163 clusters = self.__sorted_clusters( 

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

165 ) 

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

167 

168 for cluster_index, cluster in enumerate(clusters): 

169 entity_first_raw, other_entities_raw = self.__ordered_entities(cluster) 

170 entity_first = self.__as_responsible_agent(entity_first_raw) 

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

172 for other_entity_raw in other_entities_raw: 

173 self.__merge_responsible_agent( 

174 entity_first, 

175 self.__as_responsible_agent(other_entity_raw), 

176 associated_ar_ra, 

177 ) 

178 

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

180 

181 def deduplicate_bibliographic_resources(self) -> None: 

182 """ 

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

184 

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

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

187 """ 

188 self.__deduplicate_bibliographic_resources() 

189 self.graph_set.commit_changes() 

190 

191 def __deduplicate_bibliographic_resources(self) -> None: 

192 clusters = self.__sorted_clusters( 

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

194 ) 

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

196 

197 for cluster_index, cluster in enumerate(clusters): 

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

199 self.__merge_br_cluster(cluster) 

200 

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

202 

203 def deduplicate_identifiers(self) -> None: 

204 """ 

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

206 

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

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

209 """ 

210 self.__deduplicate_identifiers() 

211 self.graph_set.commit_changes() 

212 

213 def __deduplicate_identifiers(self) -> None: 

214 literal_to_id, id_to_resources = self.__id_maps() 

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

216 if len(identifiers) > 1: 

217 self.__merge_identifier_group(literal, identifiers, id_to_resources) 

218 

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

220 

221 def __merge_graph( 

222 self, 

223 entities: Iterable[ResponsibleAgent | BibliographicResource], 

224 debug_message: str, 

225 ) -> nx.Graph: 

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

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

228 for entity in entities: 

229 for identifier in entity.get_identifiers(): 

230 scheme = identifier.get_scheme() 

231 literal_value = identifier.get_literal_value() 

232 if scheme is None or literal_value is None: 

233 continue 

234 identifiers.setdefault(scheme, {}) 

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

236 if entity_first is None: 

237 identifiers[scheme][literal_value] = entity 

238 else: 

239 merge_graph.add_edge(entity_first, entity) 

240 self.__debug( 

241 debug_message, 

242 entity.res, 

243 entity_first.res, 

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

245 literal_value, 

246 ) 

247 return merge_graph 

248 

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

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

251 

252 def __manual_merge_clusters( 

253 self, 

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

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

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

257 survivors: set[str] = set() 

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

259 container_uris = self.__container_uris_with_members() 

260 

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

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

263 survivor_type = self.__merge_entity_type(survivor) 

264 merged_entities = [ 

265 self.__manual_merged_entity( 

266 survivor_uri, 

267 survivor_type, 

268 merged_uri, 

269 survivors, 

270 merged_to_survivor, 

271 ) 

272 for merged_uri in merged_uris 

273 ] 

274 

275 if not merged_entities: 

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

277 raise ValueError(message) 

278 self.__validate_container_type_compatibility(survivor, merged_entities, container_uris) 

279 normalized_clusters.append((survivor, merged_entities)) 

280 

281 return normalized_clusters 

282 

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

284 members: set[str] = set() 

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

286 container = br.get_is_part_of() 

287 if container is not None: 

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

289 return members 

290 

291 def __validate_container_type_compatibility( 

292 self, 

293 survivor: ManualMergeEntity, 

294 merged_entities: Iterable[ManualMergeEntity], 

295 container_uris: set[str], 

296 ) -> None: 

297 if not isinstance(survivor, BibliographicResource): 

298 return 

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

300 for merged_entity in merged_entities: 

301 merged = self.__as_bibliographic_resource(merged_entity) 

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

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

304 continue 

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

306 message = ( 

307 "Cannot merge bibliographic resources with incompatible types: " 

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

309 ) 

310 raise ValueError(message) 

311 

312 def __validate_identifier_clusters( 

313 self, 

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

315 ) -> None: 

316 for surviving_entity, merged_entities in normalized_clusters: 

317 if not isinstance(surviving_entity, Identifier): 

318 continue 

319 

320 surviving_signature = self.__identifier_signature(surviving_entity) 

321 for merged_entity in merged_entities: 

322 merged_identifier = self.__as_identifier(merged_entity) 

323 merged_signature = self.__identifier_signature(merged_identifier) 

324 if merged_signature != surviving_signature: 

325 message = ( 

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

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

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

329 ) 

330 raise ValueError(message) 

331 

332 @staticmethod 

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

334 scheme = identifier.get_scheme() 

335 literal = identifier.get_literal_value() 

336 if scheme is None or literal is None: 

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

338 raise ValueError(message) 

339 return scheme, literal 

340 

341 def __manual_survivor( 

342 self, 

343 survivor_uri: str, 

344 merged_uris: Iterable[str], 

345 survivors: set[str], 

346 merged_to_survivor: dict[str, str], 

347 ) -> ManualMergeEntity: 

348 if survivor_uri in merged_to_survivor: 

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

350 raise ValueError(message) 

351 if survivor_uri in survivors: 

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

353 raise ValueError(message) 

354 if isinstance(merged_uris, str): 

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

356 raise TypeError(message) 

357 

358 survivors.add(survivor_uri) 

359 return self.__merge_entity(survivor_uri) 

360 

361 def __manual_merged_entity( 

362 self, 

363 survivor_uri: str, 

364 survivor_type: str, 

365 merged_uri: str, 

366 survivors: set[str], 

367 merged_to_survivor: dict[str, str], 

368 ) -> ManualMergeEntity: 

369 if merged_uri == survivor_uri: 

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

371 raise ValueError(message) 

372 if merged_uri in survivors: 

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

374 raise ValueError(message) 

375 if merged_uri in merged_to_survivor: 

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

377 raise ValueError(message) 

378 

379 merged_entity = self.__merge_entity(merged_uri) 

380 merged_entity_type = self.__merge_entity_type(merged_entity) 

381 if merged_entity_type != survivor_type: 

382 message = ( 

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

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

385 ) 

386 raise ValueError(message) 

387 

388 merged_to_survivor[merged_uri] = survivor_uri 

389 return merged_entity 

390 

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

392 entity = self.graph_set.get_entity(uri) 

393 if entity is None: 

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

395 raise ValueError(message) 

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

397 return entity 

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

399 raise ValueError(message) 

400 

401 @staticmethod 

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

403 if isinstance(entity, ResponsibleAgent): 

404 return "ra" 

405 if isinstance(entity, BibliographicResource): 

406 return "br" 

407 return "id" 

408 

409 def __merge_responsible_agent( 

410 self, 

411 entity_first: ResponsibleAgent, 

412 other_entity: ResponsibleAgent, 

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

414 ) -> None: 

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

416 entity_first.merge(other_entity, prefer_self=True) 

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

418 for ar in associated_ars: 

419 ar.is_held_by(entity_first) 

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

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

422 self.__deduplicate_contributors_for_bibliographic_resources( 

423 self.__bibliographic_resources_for_contributors(associated_ars), 

424 ) 

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

426 

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

428 entity_first_raw, other_entities_raw = self.__ordered_entities(cluster) 

429 self.__merge_bibliographic_resources( 

430 self.__as_bibliographic_resource(entity_first_raw), 

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

432 ) 

433 

434 def __merge_bibliographic_resources( 

435 self, 

436 entity_first: BibliographicResource, 

437 other_entities: Iterable[BibliographicResource], 

438 ) -> None: 

439 for other_entity in other_entities: 

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

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

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

443 publisher_first = self.__get_publisher(entity_first) 

444 entity_first_partofs = self.__get_part_of(entity_first) 

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

446 self.__merge_publisher(publisher_first, other_entity) 

447 entity_first.merge(other_entity, prefer_self=True) 

448 self.__deduplicate_contributors_for_bibliographic_resources([entity_first]) 

449 

450 def __merge_containers( 

451 self, 

452 entity_first_partofs: list[BibliographicResource], 

453 partofs: list[BibliographicResource], 

454 ) -> None: 

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

456 for container in partofs: 

457 for type_iri in self.__specific_types(container): 

458 second_by_type.setdefault(type_iri, container) 

459 

460 parent_merged = True 

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

462 second_partof = self.__matching_container(first_partof, second_by_type) 

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

464 continue 

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

466 first_partof.merge(second_partof, prefer_self=True) 

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

468 else: 

469 parent_merged = False 

470 

471 @staticmethod 

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

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

474 

475 def __matching_container( 

476 self, 

477 container: BibliographicResource, 

478 second_by_type: dict[str, BibliographicResource], 

479 ) -> BibliographicResource | None: 

480 for type_iri in self.__specific_types(container): 

481 if type_iri in second_by_type: 

482 return second_by_type[type_iri] 

483 return None 

484 

485 def __containers_are_equivalent( 

486 self, 

487 first: BibliographicResource, 

488 second: BibliographicResource, 

489 ) -> bool: 

490 first_signatures = self.__identifier_signatures(first) 

491 second_signatures = self.__identifier_signatures(second) 

492 if first_signatures and second_signatures: 

493 return not first_signatures.isdisjoint(second_signatures) 

494 

495 first_number = first.get_number() 

496 second_number = second.get_number() 

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

498 return first_number == second_number 

499 

500 first_title = first.get_title() 

501 second_title = second.get_title() 

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

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

504 

505 return False 

506 

507 @staticmethod 

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

509 signatures: set[IdentifierSignature] = set() 

510 for identifier in entity.get_identifiers(): 

511 scheme = identifier.get_scheme() 

512 literal = identifier.get_literal_value() 

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

514 signatures.add((scheme, literal)) 

515 return signatures 

516 

517 @staticmethod 

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

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

520 

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

522 publisher = self.__get_publisher(entity) 

523 if ( 

524 publisher is not None 

525 and publisher_first is not None 

526 and publisher != publisher_first 

527 and self.__publishers_are_equivalent(publisher_first, publisher) 

528 ): 

529 publisher_first.merge(publisher, prefer_self=True) 

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

531 

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

533 first_agent = first.get_is_held_by() 

534 second_agent = second.get_is_held_by() 

535 if first_agent is None or second_agent is None: 

536 return False 

537 first_signatures = self.__identifier_signatures(first_agent) 

538 second_signatures = self.__identifier_signatures(second_agent) 

539 if first_signatures and second_signatures: 

540 return not first_signatures.isdisjoint(second_signatures) 

541 first_name = self.__agent_full_name(first_agent) 

542 second_name = self.__agent_full_name(second_agent) 

543 if first_name and second_name: 

544 return first_name == second_name 

545 return False 

546 

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

548 name = agent.get_name() 

549 if name is not None: 

550 return self.__normalized_text(name) 

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

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

553 

554 def __bibliographic_resources_for_contributors( 

555 self, 

556 contributors: Iterable[AgentRole], 

557 ) -> list[BibliographicResource]: 

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

559 if not contributor_uris: 

560 return [] 

561 return [ 

562 br 

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

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

565 ] 

566 

567 def __deduplicate_contributors_for_bibliographic_resources( 

568 self, 

569 bibliographic_resources: Iterable[BibliographicResource], 

570 ) -> None: 

571 for bibliographic_resource in bibliographic_resources: 

572 already_merged = self.__merge_same_ra_contributors(bibliographic_resource) 

573 if self.merge_similar_named_contributors: 

574 self.__merge_similar_named_contributors(bibliographic_resource, already_merged) 

575 self.__remove_contributors_without_ra(bibliographic_resource) 

576 

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

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

579 for contributor in self.__author_contributors(entity_first): 

580 responsible_agent = contributor.get_is_held_by() 

581 role_type = contributor.get_role_type() 

582 if responsible_agent is None or role_type is None: 

583 continue 

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

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

586 

587 already_merged: set[AgentRole] = set() 

588 for contributors in contributors_by_agent_role.values(): 

589 ordered_contributors = sorted(contributors, key=str) 

590 surviving_contributor = ordered_contributors[0] 

591 for merged_contributor in ordered_contributors[1:]: 

592 self.__debug( 

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

594 merged_contributor, 

595 entity_first, 

596 ) 

597 self.__merge_contributor(entity_first, surviving_contributor, merged_contributor) 

598 already_merged.add(surviving_contributor) 

599 already_merged.add(merged_contributor) 

600 return already_merged 

601 

602 def __merge_similar_named_contributors( 

603 self, 

604 entity_first: BibliographicResource, 

605 already_merged: set[AgentRole], 

606 ) -> None: 

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

608 contributors = [ 

609 contributor 

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

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

612 ] 

613 merged_contributor_uris: set[str] = set() 

614 for ar1_index, ar1 in enumerate(contributors): 

615 if str(ar1.res) in merged_contributor_uris: 

616 continue 

617 ar1_name = self.__agent_name(ar1) 

618 for ar2 in contributors[ar1_index + 1 :]: 

619 if str(ar2.res) in merged_contributor_uris or ar1.get_role_type() != ar2.get_role_type(): 

620 continue 

621 ar2_name = self.__agent_name(ar2) 

622 name_similarity = self.__name_similarity(ar1_name, ar2_name) 

623 if name_similarity > NAME_SIMILARITY_THRESHOLD: 

624 self.__merge_contributor(entity_first, ar1, ar2) 

625 merged_contributor_uris.add(str(ar2.res)) 

626 self.__debug( 

627 "\tRemoving agent role %s from bibliographic resource %s because it merged to %s", 

628 ar2, 

629 entity_first, 

630 ar1, 

631 ) 

632 

633 def __remove_contributors_without_ra(self, entity_first: BibliographicResource) -> None: 

634 for ar in self.__author_contributors(entity_first): 

635 if ar.to_be_deleted or ar.get_is_held_by() is None: 

636 entity_first.remove_contributor(ar) 

637 

638 def __merge_contributor( 

639 self, 

640 entity_first: BibliographicResource, 

641 surviving_contributor: AgentRole, 

642 merged_contributor: AgentRole, 

643 ) -> None: 

644 surviving_responsible_agent = surviving_contributor.get_is_held_by() 

645 surviving_contributor.merge(merged_contributor) 

646 if surviving_responsible_agent is not None: 

647 surviving_contributor.is_held_by(surviving_responsible_agent) 

648 entity_first.remove_contributor(merged_contributor) 

649 if str(surviving_contributor.res) not in {str(ar.res) for ar in entity_first.get_contributors()}: 

650 entity_first.has_contributor(surviving_contributor) 

651 

652 def __author_contributors(self, br: BibliographicResource) -> list[AgentRole]: 

653 return [ 

654 contributor 

655 for contributor in br.get_contributors() 

656 if contributor.get_role_type() != GraphEntity.iri_publisher 

657 ] 

658 

659 def __agent_name(self, ar: AgentRole) -> str: 

660 responsible_agent = ar.get_is_held_by() 

661 if responsible_agent is None: 

662 return "" 

663 given_name = responsible_agent.get_given_name() 

664 family_name = responsible_agent.get_family_name() 

665 name_parts = [] 

666 if given_name is not None: 

667 name_parts.append(given_name) 

668 if family_name is not None: 

669 name_parts.append(family_name) 

670 return " ".join(name_parts) 

671 

672 @staticmethod 

673 def __name_similarity(left: str, right: str) -> float: 

674 left_normalized = " ".join(left.casefold().split()) 

675 right_normalized = " ".join(right.casefold().split()) 

676 if left_normalized == "" or right_normalized == "": 

677 return 0.0 

678 max_length = max(len(left_normalized), len(right_normalized)) 

679 return 1 - Levenshtein.distance(left_normalized, right_normalized) / max_length 

680 

681 def __id_maps( 

682 self, 

683 ) -> tuple[ 

684 dict[str, list[Identifier]], 

685 dict[Identifier, list[BibliographicResource | ResponsibleAgent]], 

686 ]: 

687 literal_to_id: dict[str, list[Identifier]] = {} 

688 id_to_resources: dict[Identifier, list[BibliographicResource | ResponsibleAgent]] = { 

689 identifier: [] for identifier in self.graph_set.get_id() 

690 } 

691 entities: list[BibliographicResource | ResponsibleAgent] = list(self.graph_set.get_br()) 

692 entities.extend(list(self.graph_set.get_ra())) 

693 

694 for entity in entities: 

695 for identifier in entity.get_identifiers(): 

696 scheme = identifier.get_scheme() 

697 value = identifier.get_literal_value() 

698 if scheme is None or value is None: 

699 continue 

700 literal = f"{scheme}#{value}" 

701 id_to_resources[identifier].append(entity) 

702 literal_to_id.setdefault(literal, []).append(identifier) 

703 return literal_to_id, id_to_resources 

704 

705 def __merge_identifier_group( 

706 self, 

707 literal: str, 

708 identifiers: list[Identifier], 

709 id_to_resources: dict[Identifier, list[BibliographicResource | ResponsibleAgent]], 

710 ) -> None: 

711 schema, value = literal.split("#", maxsplit=1) 

712 merged_identifier, other_identifiers = self.__ordered_entities(identifiers) 

713 self.__debug( 

714 "[dedup-ID] Will merge %s identifiers into %s because they share literal %s and schema %s", 

715 len(identifiers) - 1, 

716 merged_identifier, 

717 value, 

718 schema, 

719 ) 

720 for actual_id in other_identifiers: 

721 self.__merge_identifier(merged_identifier, actual_id, id_to_resources) 

722 

723 def __merge_identifiers( 

724 self, 

725 surviving_identifier: Identifier, 

726 merged_identifiers: Iterable[Identifier], 

727 id_to_resources: dict[Identifier, list[BibliographicResource | ResponsibleAgent]], 

728 ) -> None: 

729 for merged_identifier in merged_identifiers: 

730 self.__merge_identifier(surviving_identifier, merged_identifier, id_to_resources) 

731 

732 def __merge_identifier( 

733 self, 

734 surviving_identifier: Identifier, 

735 merged_identifier: Identifier, 

736 id_to_resources: dict[Identifier, list[BibliographicResource | ResponsibleAgent]], 

737 ) -> None: 

738 surviving_identifier.merge(merged_identifier) 

739 self.__replace_identifier(merged_identifier, surviving_identifier, id_to_resources[merged_identifier]) 

740 merged_identifier.mark_as_to_be_deleted() 

741 

742 def __ordered_entities(self, entities: Iterable[Entity]) -> tuple[Entity, list[Entity]]: 

743 entities_by_key = {str(entity): entity for entity in entities}