fix(load): skip hand-submitted CDS records on migrate

cds_migrator_kit/rdm/records/load/entities/ep_migration_entry_load.py

@@ -54,7 +54,7 @@ def _load(self, entry: MigrationEntry):
             return
 
         recid = entry["record"].recid
-        if self._should_skip_recid(recid):
+        if self._should_skip_recid(entry):
             return
 
         if not entry.get("ep_approval"):

cds_migrator_kit/rdm/records/load/load.py

@@ -1,3 +1,3 @@
 # -*- coding: utf-8 -*-
 #
 # Copyright (C) 2022 CERN.
@@ -8,18 +8,28 @@
 """CDS-RDM migration load module."""
 
 import json
+import re
 
 from cds_rdm.clc_sync.models import CDSToCLCSyncModel
+from cds_rdm.inspire_harvester.load.matcher import ArxivIdentifierMatchFilter
+from cds_rdm.inspire_harvester.utils import retrieve_identifiers
 from cds_rdm.legacy.models import CDSMigrationLegacyRecord
 from cds_rdm.legacy.resolver import get_pid_by_legacy_recid
 from cds_rdm.minters import legacy_recid_minter
+from cds_rdm.schemes import cds_rdm_regexp
+from flask import current_app
+from invenio_access.permissions import system_identity
 from invenio_db import db
 from invenio_db.uow import ModelCommitOp, UnitOfWork
 from invenio_i18n import _
+from invenio_pidstore.errors import PIDDoesNotExistError
 from invenio_pidstore.models import PersistentIdentifier
 from invenio_rdm_migrator.load.base import Load
+from invenio_rdm_records.proxies import current_rdm_records_service
 from invenio_records.systemfields.relations import InvalidRelationValue
+from invenio_search.engine import dsl
 from marshmallow import ValidationError
+from sqlalchemy.orm.exc import NoResultFound
 
 from cds_migrator_kit.errors import (
     CDSMigrationException,
@@ -101,23 +111,149 @@
         ).one_or_none()
         return pid is not None
 
-    def _should_skip_recid(self, recid):
-        """Check if recid should be skipped."""
+    def _should_skip_recid(self, entry: MigrationEntry):
+        """Skip if this legacy recid is already on new CDS."""
+        recid = entry["record"].recid
         if recid in self.legacy_pids_to_redirect or self._have_migrated_recid(recid):
             self.migration_logger.add_information(
                 recid, state={"message": "Record already migrated", "value": recid}
             )
             self.migration_logger.finalise_record(recid)
             return True
-        return False
+        try:
+            existing = self._existing_cds_record(entry)
+        except ManualImportRequired as exc:
+            self.migration_logger.add_log(exc, record=entry)
+            return True
+        if not existing:
+            return False
+        if not self.dry_run:
+            legacy_recid_minter(recid, existing._record.parent.model.id)
+            db.session.commit()
+        self.migration_logger.add_information(
+            recid,
+            {
+                "message": "Record already submitted on new CDS",
+                "value": existing.id,
+            },
+        )
+        self.migration_logger.finalise_record(recid)
+        return True
+
+    def _one_pid(self, pid_type, values, field):
+        """Return the record for these pids if exactly one parent matches."""
+        by_parent = {}
+        for value in dict.fromkeys(v for v in values if v):
+            for pid in PersistentIdentifier.query.filter_by(
+                pid_type=pid_type, pid_value=value, object_type="rec"
+            ):
+                recid = (
+                    pid
+                    if pid_type == "recid"
+                    else PersistentIdentifier.query.filter_by(
+                        object_uuid=pid.object_uuid,
+                        object_type="rec",
+                        pid_type="recid",
+                    ).one_or_none()
+                )
+                if not recid:
+                    continue
+                try:
+                    record = current_rdm_records_service.read_latest(
+                        system_identity, id_=recid.pid_value
+                    )
+                except (PIDDoesNotExistError, NoResultFound):
+                    continue
+                by_parent[record._record.parent.pid.pid_value] = record
+        if len(by_parent) > 1:
+            raise ManualImportRequired(
+                message="Multiple existing CDS records match this legacy record",
+                field=field,
+                stage="load",
+                value=", ".join(sorted(by_parent)),
+                priority="warning",
+            )
+        return next(iter(by_parent.values()), None)
+
+    def _existing_cds_record(self, entry: MigrationEntry):
+        """Find a hand-submitted CDS record this dump would duplicate."""
+        body = entry["record"].body
+        metadata = body.get("metadata", {})
+        prefix = current_app.config["DATACITE_PREFIX"]
+        identifiers = metadata.get("identifiers", []) + metadata.get(
+            "related_identifiers", []
+        )
+        dois = list(retrieve_identifiers(identifiers, "doi"))
+        doi = body.get("pids", {}).get("doi", {}).get("identifier")
+        if doi:
+            dois.append(doi)
+
+        # New-CDS ids from related urls, structured CDSRDM, and CERN DOI suffixes.
+        pat = cds_rdm_regexp.pattern
+        rdm_ids = set()
+        for url in retrieve_identifiers(identifiers, "url"):
+            rdm_ids.update(
+                re.findall(
+                    rf"repository\.cern/(?:api/)?records/({pat})", url, re.I
+                )
+            )
+        raw = entry["record"].raw_dump_entry
+        raw = raw if isinstance(raw, dict) else {}
+        items = list(identifiers)
+        for key in (
+            "identifiers",
+            "related_identifiers",
+            "external_system_identifiers",
+        ):
+            items.extend(raw.get(key) or [])
+        for item in items:
+            if not isinstance(item, dict):
+                continue
+            scheme = item.get("scheme") or item.get("schema") or ""
+            value = item.get("identifier") or item.get("value")
+            if (
+                scheme.upper() == "CDSRDM"
+                and isinstance(value, str)
+                and cds_rdm_regexp.fullmatch(value)
+            ):
+                rdm_ids.add(value)
+        rdm_ids.update(
+            d.split("/", 1)[1] for d in dois if d.startswith(f"{prefix}/")
+        )
+
+        arxivs = list(retrieve_identifiers(identifiers, "arxiv"))
+        cores = [
+            v.split(":", 1)[-1] if v.lower().startswith("arxiv:") else v
+            for v in arxivs
+        ]
+        # Includes external DOIs and arXiv DataCite DOIs (10.48550/…).
+        dois.extend(f"10.48550/arXiv.{c}" for c in cores)
+
+        record = self._one_pid("recid", rdm_ids, "cdsrdm") or self._one_pid(
+            "doi", dois, "doi"
+        )
+        if record or not arxivs:
+            return record
+
+        candidate = ArxivIdentifierMatchFilter(
+            values=list(dict.fromkeys([*arxivs, *cores]))
+        )
+        result = current_rdm_records_service.search(
+            system_identity,
+            extra_filter=dsl.Q("bool", filter=candidate.query),
+            params={"size": 25},
+        )
+        return self._one_pid(
+            "recid", [hit["parent"]["id"] for hit in result.hits], "arxiv"
+        )
 
     def _load(self, entry: MigrationEntry):
         """Use the services to load the entry."""
         if not entry:
             return
 
         recid = entry["record"].recid
-        if self._should_skip_recid(recid):
+        if self._should_skip_recid(entry):
             return
 
         record_load = RecordLoad(