diff --git a/cds_migrator_kit/rdm/records/load/entities/ep_migration_entry_load.py b/cds_migrator_kit/rdm/records/load/entities/ep_migration_entry_load.py index 8a3908e0..2dcdc8af 100644 --- a/cds_migrator_kit/rdm/records/load/entities/ep_migration_entry_load.py +++ b/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"): diff --git a/cds_migrator_kit/rdm/records/load/load.py b/cds_migrator_kit/rdm/records/load/load.py index 6fad6305..da23e869 100644 --- a/cds_migrator_kit/rdm/records/load/load.py +++ b/cds_migrator_kit/rdm/records/load/load.py @@ -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,15 +111,130 @@ def _have_migrated_recid(recid): ).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: CDSRDM first, then repository.cern urls, then CERN DOI + # suffixes. Looked up in that order so a structured id wins over a url. + pat = cds_rdm_regexp.pattern + cdsrdm_ids = set() + for item in identifiers: + scheme = item.get("scheme") or item.get("schema") or "" + value = item.get("identifier") or item.get("value") + if scheme.upper() == "CDSRDM" and value and cds_rdm_regexp.fullmatch(value): + cdsrdm_ids.add(value) + + url_ids = set() + for url in retrieve_identifiers(identifiers, "url"): + url_ids.update( + re.findall(rf"repository\.cern/(?:api/)?records/({pat})", url, re.I) + ) + + doi_suffix_ids = { + 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", cdsrdm_ids, "cdsrdm") + or self._one_pid("recid", url_ids, "url") + or self._one_pid("recid", doi_suffix_ids, "doi") + 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.""" @@ -117,7 +242,7 @@ def _load(self, entry: MigrationEntry): return recid = entry["record"].recid - if self._should_skip_recid(recid): + if self._should_skip_recid(entry): return record_load = RecordLoad(