From e516496cd87791bee2e76da5604303467845274d Mon Sep 17 00:00:00 2001 From: Mohammed Taha Khan Date: Tue, 21 Jul 2026 15:25:38 +0200 Subject: [PATCH] fix(harvester): merge duplicate contributors --- .../inspire_harvester/update/engine.py | 6 +- .../update/fields/creatibutors.py | 109 ++++++++++++++---- 2 files changed, 92 insertions(+), 23 deletions(-) diff --git a/site/cds_rdm/inspire_harvester/update/engine.py b/site/cds_rdm/inspire_harvester/update/engine.py index cbfa023b..ee10bc49 100644 --- a/site/cds_rdm/inspire_harvester/update/engine.py +++ b/site/cds_rdm/inspire_harvester/update/engine.py @@ -96,12 +96,12 @@ def update(self, current, incoming, ctx, logger): warnings.extend(res.warnings) audit.extend(res.audit) - if conflicts or warnings: - self.log_conflicts(conflicts, logger) - logger.debug(str(audit)) if self.fail_on_conflict and conflicts: raise UpdateEngineConflict(conflicts) + if conflicts or warnings: + self.log_conflicts(conflicts, logger) + return UpdateResult(updated=updated, conflicts=conflicts, audit=audit) diff --git a/site/cds_rdm/inspire_harvester/update/fields/creatibutors.py b/site/cds_rdm/inspire_harvester/update/fields/creatibutors.py index 8cdc286d..b05bf3ec 100644 --- a/site/cds_rdm/inspire_harvester/update/fields/creatibutors.py +++ b/site/cds_rdm/inspire_harvester/update/fields/creatibutors.py @@ -33,7 +33,7 @@ def __init__(self, strict=True): self.strict = strict def _union_affiliations(self, cur_list, inc_list): - """Union affiliations by name. + """Union affiliations by identifier or name. Exact duplicates are skipped. If an incoming affiliation differs from an existing one only in punctuation, the existing entry is updated in-place @@ -42,21 +42,23 @@ def _union_affiliations(self, cur_list, inc_list): cur_list = cur_list or [] inc_list = inc_list or [] - out = list(copy.deepcopy(cur_list)) - # Track exact names and normalised names for existing entries. - exact_seen = { - a.get("name") for a in out if isinstance(a, dict) and a.get("name") - } + out = [] + exact_seen = set() + id_seen = set() # Map normalised name → index in out for fuzzy lookup. - norm_index = { - _normalize_affiliation_name(a.get("name")): i - for i, a in enumerate(out) - if isinstance(a, dict) and a.get("name") - } + norm_index = {} - for a in inc_list: + for a in [*cur_list, *inc_list]: if not isinstance(a, dict): continue + affiliation_id = a.get("id") + if affiliation_id: + if affiliation_id in id_seen: + continue + out.append(copy.deepcopy(a)) + id_seen.add(affiliation_id) + continue + nm = a.get("name") if not nm: out.append(copy.deepcopy(a)) @@ -92,10 +94,57 @@ def _key(self, creator: dict): (p.get("name") or "").lower(), ) + def _have_conflicting_values(self, first, second, ignored=()): + """Check whether two entries disagree on any meaningful value. + + Empty values are allowed because the populated entry can fill them during + the merge. Fields in ``ignored`` are combined by their own merge logic. + """ + empty_values = (None, "", [], {}) + for key in set(first) | set(second): + if key in ignored: + continue + first_value = first.get(key) + second_value = second.get(key) + if ( + first_value not in empty_values + and second_value not in empty_values + and first_value != second_value + ): + return True + return False + + def _duplicates_are_compatible(self, creators): + """Check that every matching creator can be safely merged into one.""" + # Compare every pair so a conflict cannot be missed when one entry is empty. + for index, first in enumerate(creators): + for second in creators[index + 1 :]: + # Affiliations and personal data are checked or merged separately. + if self._have_conflicting_values( + first, second, ignored=("affiliations", "person_or_org") + ): + return False + + first_person = first.get("person_or_org") or {} + second_person = second.get("person_or_org") or {} + # Identifiers are combined later, so only compare personal details. + if self._have_conflicting_values( + first_person, second_person, ignored=("identifiers",) + ): + return False + + return True + def _merge_creator(self, cur, inc): """Merge a single current creator entry with its incoming counterpart.""" merged = copy.deepcopy(inc) + for key, value in cur.items(): + if key in ("affiliations", "person_or_org"): + continue + if key not in merged or merged[key] in (None, "", [], {}): + merged[key] = copy.deepcopy(value) + # Affiliations: union, never remove if "affiliations" in cur or "affiliations" in inc: merged["affiliations"] = self._union_affiliations( @@ -120,6 +169,7 @@ def _merge_creator(self, cur, inc): key = (i.get("scheme"), i.get("identifier")) if key not in seen: mp.setdefault("identifiers", []).append(copy.deepcopy(i)) + seen.add(key) merged["person_or_org"] = mp return merged @@ -159,26 +209,45 @@ def update(self, current, incoming, path, ctx): ) else: updated_list.append(copy.deepcopy(inc)) + index.setdefault(k, []).append(len(updated_list) - 1) audit.append(f"{path}: appended creator {k}") continue if len(matches) > 1: - conflicts.append( - UpdateConflict( - path=path, - kind="ambiguous_match", - message="Multiple creators match incoming", - incoming=inc, + matched_creators = [updated_list[i] for i in matches] + if not self._duplicates_are_compatible(matched_creators): + conflicts.append( + UpdateConflict( + path=path, + kind="ambiguous_match", + message="Multiple creators match incoming", + incoming=inc, + ) + ) + continue + + idx = matches[0] + merged = updated_list[idx] + for duplicate_idx in matches[1:]: + merged = self._merge_creator( + merged, updated_list[duplicate_idx] ) + updated_list[idx] = self._merge_creator(merged, inc) + + for duplicate_idx in matches[1:]: + updated_list[duplicate_idx] = None + index[k] = [idx] + audit.append( + f"{path}: merged {len(matches)} duplicate creators {k}" ) continue idx = matches[0] - updated_list[idx] = self._merge_creator(cur_list[idx], inc) + updated_list[idx] = self._merge_creator(updated_list[idx], inc) audit.append(f"{path}: merged creator {k}") updated = copy.deepcopy(current) - set_path(updated, path, updated_list) + set_path(updated, path, [item for item in updated_list if item is not None]) return UpdateResult( updated=updated, conflicts=conflicts, warnings=warnings, audit=audit )