From 3ba20baaf7ae6ba2d757749baabb260c094042e0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Z=C3=BCbeyde=20Civelek?= Date: Thu, 25 Jun 2026 12:26:34 +0200 Subject: [PATCH 1/4] add(load): EP approval record migration stream --- cds_migrator_kit/rdm/cli.py | 13 +- cds_migrator_kit/rdm/migration_config.py | 25 + cds_migrator_kit/rdm/records/load/__init__.py | 3 +- .../rdm/records/load/ep_approval_load.py | 759 ++++++++++++++++++ cds_migrator_kit/rdm/records/load/load.py | 24 +- cds_migrator_kit/rdm/records/streams.py | 10 +- .../rdm/records/transform/transform.py | 4 + .../xml_processing/rules/research.py | 30 + 8 files changed, 861 insertions(+), 7 deletions(-) create mode 100644 cds_migrator_kit/rdm/records/load/ep_approval_load.py diff --git a/cds_migrator_kit/rdm/cli.py b/cds_migrator_kit/rdm/cli.py index bcdf811d..1ab8c02a 100644 --- a/cds_migrator_kit/rdm/cli.py +++ b/cds_migrator_kit/rdm/cli.py @@ -22,6 +22,7 @@ CommentsStreamDefinition, ) from cds_migrator_kit.rdm.records.streams import ( # UserStreamDefinition, + RecordEPApprovalStreamDefinition, RecordStreamDefinition, ) from cds_migrator_kit.rdm.stats.runner import RecordStatsRunner @@ -69,12 +70,20 @@ def migration(): "Can also be set per-collection in streams.yaml under transform.workers." ), ) +@click.option( + "--ep-approval", + is_flag=True, + help="Use the EP approval load stream (pre-EP draft snapshots without legacy minting).", +) @with_appcontext -def run(collection, dry_run=False, keep_logs=False, workers=None): +def run(collection, dry_run=False, keep_logs=False, workers=None, ep_approval=False): """Run.""" stream_config = current_app.config["CDS_MIGRATOR_KIT_STREAM_CONFIG"] + stream_definition = ( + RecordEPApprovalStreamDefinition if ep_approval else RecordStreamDefinition + ) runner = Runner( - stream_definitions=[RecordStreamDefinition], + stream_definitions=[stream_definition], # stream_definitions=[UserStreamDefinition], config_filepath=Path(stream_config).absolute(), dry_run=dry_run, diff --git a/cds_migrator_kit/rdm/migration_config.py b/cds_migrator_kit/rdm/migration_config.py index 3d5ad8a4..459e9b52 100644 --- a/cds_migrator_kit/rdm/migration_config.py +++ b/cds_migrator_kit/rdm/migration_config.py @@ -359,6 +359,11 @@ def _(x): # needed to avoid start time failure with lazy strings "validator": always_valid, "datacite": "CDS", }, + "apprn": { + "label": _("Approval Report Number"), + "validator": schemes.is_approval_report_number, + "datacite": "CDS", + }, "aleph": { "label": _("Aleph number"), "validator": schemes.is_aleph, @@ -483,6 +488,9 @@ def resolve_record_pid(pid): "EligibilityHRCirc": ["eligibility-retr-actual", "hr-web-gacepa"], "CERNPeopleEligibility": ["cern-personnel", "eligibility-retr-actual"], "FAPDepRestrFile": ["fap-dep"], + # Access group already added to the record as 506__m when draft created, no need to add: + # https://gitlab.cern.ch/cds-team/cds-legacy/-/blob/master/src/wn-cdsweb/lib/python/invenio/websubmit_functions/EPPHAPP_Test_values.py#L249-266 + "EP Restricted Draft": [], # CERN E-guide restricted docs: https://cds.cern.ch/admin/webaccess/webaccessadmin.py/showroledetails?id_role=69 CERN personnel has view rights } @@ -523,3 +531,20 @@ def resolve_record_pid(pid): # don't generate logs for migration AUDIT_LOGS_ENABLED = False + +### EP Approval configuration only needed for local, it should use cds-rdm config for de/sandbox/prod +# =========================== +CDS_CERN_SCIENTIFIC_COMMUNITY_ID = "78b3c4aa-c4e6-4502-8226-67ba2d347afe" +"""The id of the CERN Scientific community.""" + +CDS_COMMITTEE_APPROVAL_COMMUNITIES = { + "dd13404c-bcd6-4b15-aeef-38d678c61ff1": { + "label": "EP approval", # shown in UI buttons/headings + "referee_group": "cds-ph-ep-publication", # CERN e-group slug + "report_number": { + "prefix": "CERN-EP", # literal prefix, e.g. "CERN-EP" + "include_year": True, # append the current year after prefix + "counter_digits": 3, # zero-padding width, e.g. 3 → "001" + }, + }, +} diff --git a/cds_migrator_kit/rdm/records/load/__init__.py b/cds_migrator_kit/rdm/records/load/__init__.py index 38f8fc5e..4ce3126c 100644 --- a/cds_migrator_kit/rdm/records/load/__init__.py +++ b/cds_migrator_kit/rdm/records/load/__init__.py @@ -7,6 +7,7 @@ """CDS-RDM Migration load package.""" +from .ep_approval_load import CDSEPApprovalRecordServiceLoad from .load import CDSRecordServiceLoad -__all__ = ("CDSRecordServiceLoad",) +__all__ = ("CDSEPApprovalRecordServiceLoad", "CDSRecordServiceLoad") diff --git a/cds_migrator_kit/rdm/records/load/ep_approval_load.py b/cds_migrator_kit/rdm/records/load/ep_approval_load.py new file mode 100644 index 00000000..2e5e7591 --- /dev/null +++ b/cds_migrator_kit/rdm/records/load/ep_approval_load.py @@ -0,0 +1,759 @@ +# -*- coding: utf-8 -*- +# +# Copyright (C) 2026 CERN. +# +# CDS-RDM is free software; you can redistribute it and/or modify it under +# the terms of the MIT License; see LICENSE file for more details. + +"""CDS-RDM migration load module for records with EP approval.""" +import re +from collections import OrderedDict +from copy import deepcopy +from datetime import datetime, timezone +from pathlib import Path + +from cds_rdm.requests.committee_approval import APPRN_PID_TYPE, CommitteeApprovalRequest +from flask import current_app +from invenio_access.permissions import system_identity +from invenio_accounts.models import User +from invenio_db import db +from invenio_db.uow import UnitOfWork +from invenio_drafts_resources.services.records.uow import ParentRecordCommitOp +from invenio_pidstore.errors import PIDAlreadyExists +from invenio_pidstore.models import PersistentIdentifier, PIDStatus +from invenio_rdm_records.proxies import current_rdm_records_service +from invenio_rdm_records.records.api import RDMParent +from invenio_records_resources.services.uow import RecordCommitOp +from invenio_requests.customizations.event_types import LogEventType +from invenio_requests.proxies import current_events_service, current_requests_service +from invenio_requests.resolvers.registry import ResolverRegistry + +from cds_migrator_kit.errors import ManualImportRequired, UnexpectedValue +from cds_migrator_kit.rdm.migration_config import CDS_CERN_SCIENTIFIC_COMMUNITY_ID + +from .load import CDSRecordServiceLoad + +EPPHAPP_FILE_TYPE = "EPPHAPP_FILE" +EP_APPROVAL_WAITING_STATUS = "waiting" +EP_APPROVAL_APPROVED_STATUS = "approved" +EP_APPROVAL_REPORT_NUMBER_PREFIX = "CERN-EP" +EP_APPROVAL_REPORT_NUMBER_RE = re.compile(r"^CERN-EP-\d{4}-\d{3}$") + + +class CDSEPApprovalRecordServiceLoad(CDSRecordServiceLoad): + """Load records with EP approval. + + Splits a legacy record into two RDM records before load: + - a public record with non-EPPHAPP files + - a restricted record with restricted EPPHAPP files + """ + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.ep_approval_metadata = { + "title": None, + "experiment": None, + "resource_type": None, + "report_number": None, + } + self._ep_approval_parsed = None + self._load_flags = self._public_load_flags() + + def _public_load_flags(self): + """Load flags for the public migrated record.""" + return { + "mint_pids": True, + "mint_legacy_recid": True, + "save_original_dump": True, + "clc_sync": True, + "record_state": True, + } + + def _restricted_load_flags(self): + """Load flags for the restricted EPPHAPP snapshot record.""" + return { + "mint_pids": False, + "mint_legacy_recid": False, + "save_original_dump": False, + "clc_sync": False, + "record_state": False, + } + + def _should_log_record_state(self): + return self._load_flags["record_state"] + + def _after_publish_mint_recid(self, record, entry, version): + if self._load_flags["mint_legacy_recid"]: + super()._after_publish_mint_recid(record, entry, version) + + def _after_publish_update_dois(self, identity, record, entry, uow): + if self._load_flags["mint_pids"]: + return super()._after_publish_update_dois(identity, record, entry, uow) + + def _assign_rep_numbers(self, draft): + if self._load_flags["mint_pids"]: + super()._assign_rep_numbers(draft) + + def _save_original_dumped_record(self, entry, recid_state): + if self._load_flags["save_original_dump"]: + super()._save_original_dumped_record(entry, recid_state) + + def _after_load_clc_sync(self, record_state): + if self._load_flags["clc_sync"]: + super()._after_load_clc_sync(record_state) + + def _load(self, entry): + """ + Load the record with EP approval. + Configure the 2 records by separating the files, then: + 1. create the restricted record + 2. create and approve the EP approval request + 3. create the public record and link both with related_identifiers + """ + if not entry: + return + try: + recid = entry.get("record", {}).get("recid") + ep_approval = entry.get("record", {}).get("ep_approval") + if not ep_approval: + raise UnexpectedValue( + message="EP approval request not found", + stage="load", + recid=recid, + priority="critical", + ) + record_json = entry.get("record", {}).get("json", {}) + metadata = record_json.get("metadata", {}) + + # Set the EP approval metadata + self.ep_approval_metadata["resource_type"] = metadata.get("resource_type") + self.ep_approval_metadata["title"] = metadata.get("title") + + # Validate the EP approval data + self._validate_ep_approval(ep_approval, recid) + + # Split the metadata and files + public_entry = self._split_entry(entry, include_epphapp=False) + restricted_entry = self._split_entry(entry, include_epphapp=True) + + # 1. Create restricted record + restricted_record_state = self._load_split_record( + restricted_entry, self._restricted_load_flags(), finalise=False + ) + + # 2. Create and approve EP approval request + self._create_ep_approval(restricted_record_state, legacy_recid=recid) + + # 3. Create public record and link both records + public_record_state = self._load_split_record( + public_entry, + self._public_load_flags(), + finalise=True, + ) + self._link_ep_approval_records( + restricted_record_state, public_record_state, legacy_recid=recid + ) + except (UnexpectedValue, ManualImportRequired) as e: + self.migration_logger.add_log(e, record=entry) + except Exception as e: + exc = ManualImportRequired( + message=str(e), + field="validation", + stage="load", + recid=recid, + priority="warning", + ) + self.migration_logger.add_log(exc, record=entry) + + def _load_split_record(self, entry, load_flags, finalise): + """Load the record.""" + self._load_flags = load_flags + self._finalise_on_load = finalise + return super()._load(entry) + + def _validate_ep_approval(self, ep_approval, legacy_recid): + """Validate EP approval data before creating any records.""" + waiting_entry, approved_entry, report_number = self._parse_ep_approval_history( + ep_approval + ) + + existing = self._existing_ep_approval_request(legacy_recid) + if existing: + raise ManualImportRequired( + message=f"EP approval request {existing['id']} already exists", + stage="load", + priority="critical", + ) + if self._exists_apprn_pid(report_number): + raise ManualImportRequired( + message=f"APPRN PID {report_number} already exists", + stage="load", + priority="critical", + ) + + self._ep_approval_parsed = (waiting_entry, approved_entry, report_number) + + def _create_ep_approval(self, restricted_record_state, legacy_recid): + """Create and approve EP approval request after restricted record exists.""" + waiting_entry, approved_entry, report_number = self._ep_approval_parsed + publication_title = self.ep_approval_metadata["title"] + + if not self.dry_run: + if not restricted_record_state: + raise UnexpectedValue( + message="Restricted record is required for EP approval.", + stage="load", + recid=legacy_recid, + priority="critical", + ) + restricted_recid = restricted_record_state["latest_version"] + restricted_parent = RDMParent.get_record( + restricted_record_state["parent_object_uuid"] + ) + self._create_ep_approval_request( + legacy_recid, + restricted_recid, + restricted_parent, + waiting_entry, + approved_entry, + report_number, + publication_title, + ) + self._mint_apprn_pid( + report_number, restricted_record_state["latest_version_object_uuid"] + ) + + def _link_ep_approval_records( + self, restricted_record_state, public_record_state, legacy_recid + ): + """Write parent metadata and link public/restricted records.""" + _, approved_entry, report_number = self._ep_approval_parsed + approval_iso = self._parse_legacy_datetime( + approved_entry.get("date") + ).isoformat() + + if not self.dry_run: + if not restricted_record_state or not public_record_state: + raise UnexpectedValue( + message="Both public and restricted records are required for EP approval.", + stage="load", + recid=legacy_recid, + priority="critical", + ) + restricted_recid = restricted_record_state["latest_version"] + public_recid = public_record_state["latest_version"] + restricted_parent = RDMParent.get_record( + restricted_record_state["parent_object_uuid"] + ) + public_parent = RDMParent.get_record( + public_record_state["parent_object_uuid"] + ) + + with UnitOfWork() as uow: + self._write_parent_ep_approval( + restricted_parent, + { + "reportnumber": report_number, + "datetime": approval_iso, + "approved_internal_version": restricted_recid, + "approved_public_version": public_recid, + "source_public_version": restricted_recid, + }, + uow, + ) + self._write_parent_ep_approval( + public_parent, + { + "reportnumber": report_number, + "source_internal_version": restricted_recid, + }, + uow, + ) + uow.commit() + # Link the records with related_identifiers + self._append_related_identifier( + public_recid, + restricted_recid, + "isversionof", + self.ep_approval_metadata["resource_type"], + ) + self._append_related_identifier( + restricted_recid, + public_recid, + "isvariantformof", + self.ep_approval_metadata["resource_type"], + ) + + def _parse_ep_approval_history(self, ep_approval): + """Return waiting/approved history entries and the report number.""" + if len(ep_approval) != 2: + raise UnexpectedValue( + message="EP approval history has more/less than 2 entries", + stage="load", + priority="critical", + ) + history = ep_approval or [] + waiting = next( + ( + item + for item in history + if item.get("status") == EP_APPROVAL_WAITING_STATUS + ), + None, + ) + approved = next( + ( + item + for item in history + if item.get("status") == EP_APPROVAL_APPROVED_STATUS + ), + None, + ) + if not waiting: + raise UnexpectedValue( + message="EP approval history has no waiting entry", + stage="load", + priority="critical", + ) + if not approved: + raise UnexpectedValue( + message="EP approval history has no approved entry", + stage="load", + priority="critical", + ) + + report_number = approved.get("ep_report_number") + if not report_number: + raise UnexpectedValue( + message="EP approval approved entry is missing ep_report_number", + stage="load", + priority="critical", + ) + if waiting.get("ep_report_number") != report_number: + raise UnexpectedValue( + message="EP approval waiting entry has different ep_report_number than approved entry", + stage="load", + priority="critical", + ) + self.ep_approval_metadata["report_number"] = report_number + # Check if the submitters are exists + self._resolve_user_by_email(waiting.get("submitted_by"), "submitter") + self._resolve_user_by_email(approved.get("submitted_by"), "approver") + + # Check if record approved after the deadline + waiting_deadline = self._parse_legacy_datetime(waiting.get("deadline")) + approved_date = self._parse_legacy_datetime(approved.get("date")) + created_at = self._parse_legacy_datetime(waiting.get("date")) + if not created_at or not approved_date or not waiting_deadline: + raise UnexpectedValue( + message="EP approval history has missing timestamps", + stage="load", + priority="critical", + ) + if waiting_deadline and approved_date > waiting_deadline: + raise UnexpectedValue( + message="Record approved after the deadline", + stage="load", + priority="critical", + ) + return waiting, approved, report_number + + @staticmethod + def _parse_legacy_datetime(value): + """Parse legacy EP approval timestamps into timezone-aware datetimes.""" + if not value: + return None + for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d"): + try: + return datetime.strptime(value, fmt).replace(tzinfo=timezone.utc) + except ValueError: + continue + return None + + def _get_ep_referee_group(self, restricted_parent): + """Get the EP approval referee group from the restricted record.""" + default_community_id = restricted_parent.get("communities", {}).get("default") + if not default_community_id: + raise UnexpectedValue( + message="Restricted record has no default community for EP approval", + stage="load", + priority="critical", + ) + ep_config = current_app.config.get( + "CDS_COMMITTEE_APPROVAL_COMMUNITIES", {} + ).get(default_community_id) + if not ep_config: + raise UnexpectedValue( + message=( + f"Community {default_community_id} is not enrolled in " + "CDS_COMMITTEE_APPROVAL_COMMUNITIES" + ), + stage="load", + priority="critical", + ) + return ep_config["referee_group"] + + def _resolve_user_by_email(self, email, role): + """Resolve the user by email.""" + if not email: + raise UnexpectedValue( + message=f"EP approval {role} email is missing", + stage="load", + priority="critical", + ) + user = User.query.filter_by(email=email).one_or_none() + if not user: + raise UnexpectedValue( + message=f"EP approval {role} user not found: {email}", + stage="load", + priority="critical", + ) + return {"user": str(user.id)} + + def _existing_ep_approval_request(self, legacy_recid): + """Check if the EP approval request already exists.""" + number = f"lrecid:{legacy_recid}:ep-approval" + results = current_requests_service.search( + system_identity, + params={"q": f'number:"{number}"', "size": 1}, + ) + hits = list(results.hits) + return hits[0] if hits else None + + def _exists_apprn_pid(self, report_number): + """Check if the APPRN PID already exists.""" + existing = PersistentIdentifier.query.filter_by( + pid_type=APPRN_PID_TYPE, + pid_value=report_number, + ).one_or_none() + if existing: + return True + return False + + def _mint_apprn_pid(self, report_number, restricted_version_uuid): + """Mint the APPRN PID.""" + try: + PersistentIdentifier.create( + pid_type=APPRN_PID_TYPE, + pid_value=report_number, + object_type="rec", + object_uuid=str(restricted_version_uuid), + status=PIDStatus.REGISTERED, + ) + except PIDAlreadyExists: + raise ManualImportRequired( + message=f"APPRN PID {report_number} already exists", + stage="load", + priority="critical", + ) + + def _write_parent_ep_approval( + self, + parent, + ep_approval, + uow, + ): + """Write the EP approval metadata to the parent record.""" + pf = parent.get("permission_flags") or {} + pf["committee_approval"] = ep_approval + parent["permission_flags"] = pf + uow.register(ParentRecordCommitOp(parent)) + + def _create_accept_log_event(self, request, approved_entry, uow): + """Create the accept timeline event with the legacy approver as created_by.""" + approver_ref = self._resolve_user_by_email( + approved_entry.get("submitted_by"), + "approver", + ) + + event = current_events_service.record_cls.create( + {}, + request=request.model, + request_id=str(request.id), + type=LogEventType, + ) + event.update({"payload": {"event": "accepted"}}) + event.created_by = ResolverRegistry.resolve_entity_proxy( + approver_ref, raise_=True + ) + + approved_at = self._parse_legacy_datetime(approved_entry.get("date")) + if approved_at: + event.model.created = approved_at + + uow.register(RecordCommitOp(event, indexer=current_events_service.indexer)) + + def _apply_approved_entry_to_request( + self, request, approved_entry, report_number, uow + ): + """Update an existing request to accepted using the legacy approved entry.""" + payload = dict(request.get("payload") or {}) + payload["approved_report_number"] = report_number + request["payload"] = payload + request.status = "accepted" + + approved_at = self._parse_legacy_datetime(approved_entry.get("date")) + if approved_at: + request.model.updated = approved_at + + self._create_accept_log_event(request, approved_entry, uow) + + def _create_ep_approval_request( + self, + legacy_recid, + restricted_recid, + restricted_parent, + waiting_entry, + approved_entry, + report_number, + publication_title, + ): + """Create request from waiting entry, then update it with approved entry.""" + expires_at = self._parse_legacy_datetime(waiting_entry.get("deadline")) + + referee_group = self._get_ep_referee_group(restricted_parent) + with UnitOfWork() as uow: + request_item = current_requests_service.create( + system_identity, + data={ + "title": f'EP approval for "{publication_title}"', + # Use the default + "payload": {}, + }, + request_type=CommitteeApprovalRequest, + receiver={"group": referee_group}, + creator=self._resolve_user_by_email( + waiting_entry.get("submitted_by"), "submitter" + ), + topic={"record": restricted_recid}, + expires_at=expires_at, + uow=uow, + ) + request = request_item._record + request.number = f"lrecid:{legacy_recid}:ep-approval" + request.status = "submitted" + + submitted_at = self._parse_legacy_datetime(waiting_entry.get("date")) + if submitted_at: + request.model.created = submitted_at + + self._apply_approved_entry_to_request( + request, approved_entry, report_number, uow + ) + + uow.register( + RecordCommitOp(request, indexer=current_requests_service.indexer) + ) + uow.commit() + + def _append_related_identifier( + self, record_id, target_id, relation_id, resource_type + ): + """Append the related identifier to the record.""" + draft = current_rdm_records_service.edit(system_identity, id_=record_id) + data = draft.data + related = list(data.get("metadata", {}).get("related_identifiers", [])) + + entry = { + "identifier": target_id, + "scheme": "cds", + "relation_type": {"id": relation_id}, + } + if resource_type: + entry["resource_type"] = resource_type + related.append(entry) + data.setdefault("metadata", {})["related_identifiers"] = related + current_rdm_records_service.update_draft( + system_identity, id_=draft.id, data=data + ) + current_rdm_records_service.publish(system_identity, id_=draft.id) + return True + + def _add_cern_scientific_community(self, entry): + """Add the CERN Scientific community to the public record parent.""" + communities = entry.get("parent", {}).get("json", {}).get("communities", {}) + ids = list(communities.get("ids", [])) + if CDS_CERN_SCIENTIFIC_COMMUNITY_ID not in ids: + ids.append(CDS_CERN_SCIENTIFIC_COMMUNITY_ID) + communities["ids"] = ids + entry.setdefault("parent", {}).setdefault("json", {})[ + "communities" + ] = communities + + def _should_remove_ep_report_number(self, identifier, public_split): + """Return whether an EP report number should be stripped from metadata.""" + if not identifier.startswith(EP_APPROVAL_REPORT_NUMBER_PREFIX): + return False + if public_split: + return True + if EP_APPROVAL_REPORT_NUMBER_RE.match(identifier): + if identifier != self.ep_approval_metadata["report_number"]: + raise UnexpectedValue( + message="EP report number is not the same as the approved entry", + stage="load", + priority="critical", + ) + return True + return False + + def _remove_ep_report_numbers_from_metadata(self, entry, include_epphapp): + """Strip EP report numbers from split record metadata before load.""" + recid = entry.get("record", {}).get("recid") + metadata = entry.get("record", {}).get("json", {}).get("metadata", {}) + identifiers = metadata.get("identifiers", []) + if not identifiers: + return + + kept = [] + removed = [] + for id_entry in identifiers: + if id_entry.get("scheme") != "cdsrn": + kept.append(id_entry) + continue + identifier = id_entry.get("identifier", "") + if self._should_remove_ep_report_number(identifier, not include_epphapp): + removed.append(identifier) + else: + kept.append(id_entry) + + if not removed: + return + + metadata["identifiers"] = kept + split_type = "restricted" if include_epphapp else "public" + self.migration_logger.add_information( + recid, + { + "message": ( + f"Removed EP approval report number(s) from {split_type} " "record." + ), + "value": removed, + }, + ) + + def _remove_doi_pid_from_metadata(self, entry, include_epphapp): + """Strip DOI PID from restricted EPPHAPP split record metadata before load.""" + if not include_epphapp: + return + + recid = entry.get("record", {}).get("recid") + record_json = entry.get("record", {}).get("json", {}) + pids = record_json.get("pids") + + if not pids or "doi" not in pids: + return + + removed = pids.pop("doi") + + self.migration_logger.add_information( + recid, + { + "message": "Removed DOI PID from restricted record.", + "value": removed, + }, + ) + + def _split_entry(self, entry, include_epphapp): + """Return a load entry for the public or restricted EP approval split.""" + split = deepcopy(entry) + split["record"].pop("ep_approval", None) + + new_versions = OrderedDict() + versioned_files = OrderedDict() + previous_signature = None + + for _, version_data in split.get("versions", {}).items(): + current_version_files = OrderedDict() + + for key, file_data in version_data.get("files", {}).items(): + is_epphapp = file_data.get("type") == EPPHAPP_FILE_TYPE + + if include_epphapp != is_epphapp: + continue + + if not include_epphapp and file_data.get("access"): + raise UnexpectedValue( + message=( + "Public split contains restricted files after excluding " + f"EPPHAPP files: {[key]}" + ), + stage="load", + recid=split["record"]["recid"], + priority="critical", + ) + + current_version_files[key] = deepcopy(file_data) + + if not current_version_files: + continue + + versioned_files.update(current_version_files) + + signature = tuple( + sorted( + ( + key, + file_data.get("checksum"), + file_data.get("id_bibdoc"), + file_data.get("version"), + file_data.get("type"), + file_data.get("access"), + ) + for key, file_data in versioned_files.items() + ) + ) + # If the signature is the same, skip the version + if signature == previous_signature: + continue + + previous_signature = signature + + version_access = deepcopy(version_data.get("access", {})) + access_obj = deepcopy(version_access.get("access_obj", {})) + + if include_epphapp: + access_obj["record"] = "restricted" + access_obj["files"] = "restricted" + else: + access_obj["record"] = "public" + access_obj["files"] = "public" + # Remove the meta field from the public version access + version_access.pop("meta", None) + + version_access["access_obj"] = access_obj + + new_version_data = deepcopy(version_data) + new_version_data["files"] = deepcopy(versioned_files) + new_version_data["access"] = version_access + + new_versions[len(new_versions) + 1] = new_version_data + + if not new_versions: + raise UnexpectedValue( + message=( + "No EPPHAPP files found to load for EP approval restricted split" + if include_epphapp + else "No public files found to load for EP approval public split" + ), + stage="load", + recid=split["record"]["recid"], + priority="critical", + ) + + if not include_epphapp: + split["record"]["access"] = "public" + self._add_cern_scientific_community(split) + # Add the approval report number to the public record metadata + _, _, report_number = self._ep_approval_parsed + split["record"]["json"]["metadata"]["identifiers"].append( + { + "identifier": report_number, + "scheme": "apprn", + } + ) + + split["versions"] = new_versions + self._remove_ep_report_numbers_from_metadata(split, include_epphapp) + self._remove_doi_pid_from_metadata(split, include_epphapp) + + return split diff --git a/cds_migrator_kit/rdm/records/load/load.py b/cds_migrator_kit/rdm/records/load/load.py index 361cfde3..19e1b142 100644 --- a/cds_migrator_kit/rdm/records/load/load.py +++ b/cds_migrator_kit/rdm/records/load/load.py @@ -44,6 +44,7 @@ GrantCreationError, ManualImportRequired, RecordFlaggedCuration, + UnexpectedValue, ) @@ -133,6 +134,7 @@ def __init__( self.dry_run = dry_run self.legacy_pids_to_redirect = {} self.clc_sync = False + self._finalise_on_load = True self.collection = collection self.update_new_version_publication_date = update_new_version_publication_date self.create_inclusion_request = create_inclusion_request @@ -697,9 +699,14 @@ def _load_versions(self, entry, uow): record_state_context = self._load_record_state(legacy_recid, records) # Dump the computed record state. This is useful to migrate then the record stats if record_state_context: - self.record_state_logger.add_record_state(record_state_context) + if self._should_log_record_state(): + self.record_state_logger.add_record_state(record_state_context) return record_state_context + def _should_log_record_state(self): + """Whether to persist record state for stats migration.""" + return True + def _dry_load(self, entry): current_rdm_records_service.schema.load( entry["record"]["json"], @@ -835,8 +842,17 @@ def _load(self, entry): del entry["_clc_sync"] try: + ep_approval = entry.get("record", {}).get("ep_approval") + if ep_approval: + raise UnexpectedValue( + message="EP approval records must be loaded with the EP approval stream", + stage="load", + recid=recid, + priority="critical", + ) if self.dry_run: self._dry_load(entry) + recid_state_after_load = None else: with UnitOfWork(db.session) as uow: recid_state_after_load = self._load_versions(entry, uow) @@ -846,8 +862,10 @@ def _load(self, entry): ) self._after_load_clc_sync(recid_state_after_load) uow.commit() - self.migration_logger.finalise_record(recid) - except ManualImportRequired as e: + if self._finalise_on_load: + self.migration_logger.finalise_record(recid) + return recid_state_after_load + except (UnexpectedValue, ManualImportRequired) as e: self.migration_logger.add_log(e, record=entry) except GrantCreationError as e: self.migration_logger.add_log(e, record=entry) diff --git a/cds_migrator_kit/rdm/records/streams.py b/cds_migrator_kit/rdm/records/streams.py index 0a637798..eccfe125 100644 --- a/cds_migrator_kit/rdm/records/streams.py +++ b/cds_migrator_kit/rdm/records/streams.py @@ -11,7 +11,7 @@ from cds_migrator_kit.extract.extract import LegacyExtract from cds_migrator_kit.rdm.records.transform.transform import CDSToRDMRecordTransform -from .load import CDSRecordServiceLoad +from .load import CDSEPApprovalRecordServiceLoad, CDSRecordServiceLoad RecordStreamDefinition = StreamDefinition( name="records", @@ -20,3 +20,11 @@ load_cls=CDSRecordServiceLoad, ) """ETL stream for CDS to RDM records.""" + +RecordEPApprovalStreamDefinition = StreamDefinition( + name="records", + extract_cls=LegacyExtract, + transform_cls=CDSToRDMRecordTransform, + load_cls=CDSEPApprovalRecordServiceLoad, +) +"""ETL stream for CDS to RDM records with EP approval.""" diff --git a/cds_migrator_kit/rdm/records/transform/transform.py b/cds_migrator_kit/rdm/records/transform/transform.py index 0d269c88..f0bb349c 100644 --- a/cds_migrator_kit/rdm/records/transform/transform.py +++ b/cds_migrator_kit/rdm/records/transform/transform.py @@ -144,6 +144,7 @@ def __init__( self.access_grants_view = access_grants_view self.migration_logger = migration_logger self.record_state_logger = record_state_logger + self.ep_approval_request = None super().__init__(partial) def _created(self, entry): @@ -531,7 +532,9 @@ def subjects(json_entry): "custom_fields", "_pids", "internal_notes", + "ep_approval", ] + self.ep_approval_request = json_entry.get("ep_approval", []) keys = deepcopy(list(json_entry.keys())) for item in helper_keys: @@ -801,6 +804,7 @@ def transform(self, entry): "_original_dump": entry, "_request_data": request_data, "_clc_sync": clc_sync, + "ep_approval": self.ep_approval_request, } diff --git a/cds_migrator_kit/rdm/records/transform/xml_processing/rules/research.py b/cds_migrator_kit/rdm/records/transform/xml_processing/rules/research.py index fce97fb6..fddbf611 100644 --- a/cds_migrator_kit/rdm/records/transform/xml_processing/rules/research.py +++ b/cds_migrator_kit/rdm/records/transform/xml_processing/rules/research.py @@ -774,3 +774,33 @@ def resource_type(self, key, value): raise IgnoreKey("resource_type") else: return mapping[best_value] + + +@model.over("ep_approval", "^9031_") +@for_each_value +def ep_approval(self, key, value): + """Translates EP approval status.""" + status = value.get("s", "").strip().lower() + submitted_by = value.get("f", "").strip().lower() + date = value.get("d", "").strip() + deadline = value.get("e", "").strip() + description = value.get("a", "").strip() + ep_report_number = value.get("b", "").strip() + stamp_info = value.get("g", "").strip() + doc_type = value.get("c", "").strip() + if status not in ["waiting", "approved"]: + raise UnexpectedValue(subfield="a", field=key, value=value) + return { + k: v + for k, v in { + "status": status, + "submitted_by": submitted_by, + "date": date, + "deadline": deadline, + "description": description, + "ep_report_number": ep_report_number, + "stamp_info": stamp_info, + "doc_type": doc_type, + }.items() + if v + } From aa864e79acf3b99f12f3fe785358fe8e437d29df Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Z=C3=BCbeyde=20Civelek?= Date: Mon, 20 Jul 2026 14:34:13 +0200 Subject: [PATCH 2/4] change(ep-approve-load): add approval request class --- cds_migrator_kit/rdm/README.md | 21 + .../rdm/records/load/approval_request.py | 328 +++++++++++ .../rdm/records/load/ep_approval_load.py | 553 +++++------------- cds_migrator_kit/rdm/records/load/load.py | 38 +- .../xml_processing/rules/faser_publication.py | 4 + cds_migrator_kit/rdm/streams.yaml | 45 +- .../cds_migrator_kit_records/records.html | 5 + cds_migrator_kit/reports/views.py | 8 + 8 files changed, 571 insertions(+), 431 deletions(-) create mode 100644 cds_migrator_kit/rdm/records/load/approval_request.py diff --git a/cds_migrator_kit/rdm/README.md b/cds_migrator_kit/rdm/README.md index 0d5371fc..c021869c 100644 --- a/cds_migrator_kit/rdm/README.md +++ b/cds_migrator_kit/rdm/README.md @@ -152,6 +152,27 @@ Run the below command to migrate records in the created community from before: invenio migration run ``` +#### EP approval records (`--ep-approval`) + +EP approval records must be migrated in a **separate stream**. Do not mix them with regular records. + +1. **Dump** them filtered by `9031_:EPPHAPP`, for example (FASER): + +```bash +inveniomigrator dump records -q '980__a:ARTICLE or 980__a:PREPRINT and 693:"FASER" not 980:CONFERENCEPAPER not 591__b:"Draft" 9031_:EPPHAPP -980:DELETED -980:HIDDEN -980__c:MIGRATED -980__a:DUMMY' --file-prefix faser-papers-cern-ep --chunk-size=1000 +``` + +2. Configure a dedicated collection entry in `streams.yaml` pointing at that dump. + +3. **Migrate** with the `--ep-approval` flag (dry run first): + +```shell +invenio migration run --collection faser-ep --ep-approval --dry-run +invenio migration run --collection faser-ep --ep-approval +``` + +Without `--ep-approval`, the loader will reject EP approval records. + ### Migrate the statistics for the successfully migrated records When the `invenio migration run` command ends it will produce a `rdm_records_state.json` file which has linked information about the migrated records and the old system. The format will be similar to below: diff --git a/cds_migrator_kit/rdm/records/load/approval_request.py b/cds_migrator_kit/rdm/records/load/approval_request.py new file mode 100644 index 00000000..db4fdc6c --- /dev/null +++ b/cds_migrator_kit/rdm/records/load/approval_request.py @@ -0,0 +1,328 @@ +# -*- coding: utf-8 -*- +# +# Copyright (C) 2026 CERN. +# +# CDS-RDM is free software; you can redistribute it and/or modify it under +# the terms of the MIT License; see LICENSE file for more details. + +"""CDS-RDM EP approval request validation and creation.""" + +from datetime import datetime, timezone + +from cds_rdm.requests.committee_approval import APPRN_PID_TYPE, CommitteeApprovalRequest +from flask import current_app +from invenio_access.permissions import system_identity +from invenio_accounts.models import User +from invenio_db.uow import UnitOfWork +from invenio_pidstore.errors import PIDAlreadyExists +from invenio_pidstore.models import PersistentIdentifier, PIDStatus +from invenio_rdm_records.records.api import RDMParent +from invenio_records_resources.services.uow import RecordCommitOp +from invenio_requests.customizations.event_types import LogEventType +from invenio_requests.proxies import current_events_service, current_requests_service +from invenio_requests.resolvers.registry import ResolverRegistry + +from cds_migrator_kit.errors import ManualImportRequired, UnexpectedValue + +EP_APPROVAL_WAITING_STATUS = "waiting" +EP_APPROVAL_APPROVED_STATUS = "approved" + + +class ApprovalRequest: + """Validate and create a migrated EP committee approval request.""" + + def __init__( + self, + ep_approval, + legacy_recid, + title=None, + resource_type=None, + dry_run=False, + ): + self.ep_approval = ep_approval + self.legacy_recid = legacy_recid + self.title = title + self.resource_type = resource_type + self.dry_run = dry_run + self.waiting_entry = None + self.approved_entry = None + self.report_number = None + self.approved_at = None + + def validate(self): + """Validate EP approval data before creating any records.""" + waiting_entry, approved_entry, report_number = self._parse_history() + + existing = self._existing_request() + if existing: + raise ManualImportRequired( + message=f"EP approval request {existing['id']} already exists", + stage="load", + priority="critical", + ) + if self._exists_apprn_pid(report_number): + raise ManualImportRequired( + message=f"APPRN PID {report_number} already exists", + stage="load", + priority="critical", + ) + + self.waiting_entry = waiting_entry + self.approved_entry = approved_entry + self.report_number = report_number + + def create(self, restricted_record_state): + """Create and approve EP approval request after restricted record exists.""" + if self.dry_run: + return + + if not restricted_record_state: + raise UnexpectedValue( + message="Restricted record is required for EP approval.", + stage="load", + recid=self.legacy_recid, + priority="critical", + ) + + restricted_recid = restricted_record_state["latest_version"] + restricted_parent = RDMParent.get_record( + restricted_record_state["parent_object_uuid"] + ) + self._create_request( + restricted_recid, + restricted_parent, + ) + self._mint_apprn_pid(restricted_record_state["latest_version_object_uuid"]) + + def _parse_history(self): + """Return waiting/approved history entries and the report number.""" + if len(self.ep_approval) != 2: + raise UnexpectedValue( + message="EP approval history has more/less than 2 entries", + stage="load", + priority="critical", + ) + history = self.ep_approval or [] + waiting = next( + ( + item + for item in history + if item.get("status") == EP_APPROVAL_WAITING_STATUS + ), + None, + ) + approved = next( + ( + item + for item in history + if item.get("status") == EP_APPROVAL_APPROVED_STATUS + ), + None, + ) + if not waiting: + raise UnexpectedValue( + message="EP approval history has no waiting entry", + stage="load", + priority="critical", + ) + if not approved: + raise UnexpectedValue( + message="EP approval history has no approved entry", + stage="load", + priority="critical", + ) + + report_number = approved.get("ep_report_number") + if not report_number: + raise UnexpectedValue( + message="EP approval approved entry is missing ep_report_number", + stage="load", + priority="critical", + ) + if waiting.get("ep_report_number") != report_number: + raise UnexpectedValue( + message=( + "EP approval waiting entry has different ep_report_number " + "than approved entry" + ), + stage="load", + priority="critical", + ) + + self._resolve_user_by_email(waiting.get("submitted_by"), "submitter") + self._resolve_user_by_email(approved.get("submitted_by"), "approver") + + waiting_deadline = self.parse_legacy_datetime(waiting.get("deadline")) + approved_date = self.parse_legacy_datetime(approved.get("date")) + self.approved_at = approved_date + created_at = self.parse_legacy_datetime(waiting.get("date")) + if not created_at or not approved_date or not waiting_deadline: + raise UnexpectedValue( + message="EP approval history has missing timestamps", + stage="load", + priority="critical", + ) + + return waiting, approved, report_number + + @staticmethod + def parse_legacy_datetime(value): + """Parse legacy EP approval timestamps into timezone-aware datetimes.""" + if not value: + return None + for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d"): + try: + return datetime.strptime(value, fmt).replace(tzinfo=timezone.utc) + except ValueError: + continue + return None + + def _get_referee_group(self, restricted_parent): + """Get the EP approval referee group from the restricted record.""" + default_community_id = restricted_parent.get("communities", {}).get("default") + if not default_community_id: + raise UnexpectedValue( + message="Restricted record has no default community for EP approval", + stage="load", + priority="critical", + ) + ep_config = current_app.config.get( + "CDS_COMMITTEE_APPROVAL_COMMUNITIES", {} + ).get(default_community_id) + if not ep_config: + raise UnexpectedValue( + message=( + f"Community {default_community_id} is not enrolled in " + "CDS_COMMITTEE_APPROVAL_COMMUNITIES" + ), + stage="load", + priority="critical", + ) + return ep_config["referee_group"] + + @staticmethod + def _resolve_user_by_email(email, role): + """Resolve the user by email.""" + if not email: + raise UnexpectedValue( + message=f"EP approval {role} email is missing", + stage="load", + priority="critical", + ) + user = User.query.filter_by(email=email).one_or_none() + if not user: + raise UnexpectedValue( + message=f"EP approval {role} user not found: {email}", + stage="load", + priority="critical", + ) + return {"user": str(user.id)} + + def _existing_request(self): + """Check if the EP approval request already exists.""" + number = f"lrecid:{self.legacy_recid}:ep-approval" + results = current_requests_service.search( + system_identity, + params={"q": f'number:"{number}"', "size": 1}, + ) + hits = list(results.hits) + return hits[0] if hits else None + + @staticmethod + def _exists_apprn_pid(report_number): + """Check if the APPRN PID already exists.""" + existing = PersistentIdentifier.query.filter_by( + pid_type=APPRN_PID_TYPE, + pid_value=report_number, + ).one_or_none() + return bool(existing) + + def _mint_apprn_pid(self, restricted_version_uuid): + """Mint the APPRN PID.""" + try: + PersistentIdentifier.create( + pid_type=APPRN_PID_TYPE, + pid_value=self.report_number, + object_type="rec", + object_uuid=str(restricted_version_uuid), + status=PIDStatus.REGISTERED, + ) + except PIDAlreadyExists: + raise ManualImportRequired( + message=f"APPRN PID {self.report_number} already exists", + stage="load", + priority="critical", + ) + + def _create_accept_log_event(self, request, uow): + """Create the accept timeline event with the legacy approver as created_by.""" + approver_ref = self._resolve_user_by_email( + self.approved_entry.get("submitted_by"), + "approver", + ) + + event = current_events_service.record_cls.create( + {}, + request=request.model, + request_id=str(request.id), + type=LogEventType, + ) + event.update({"payload": {"event": "accepted"}}) + event.created_by = ResolverRegistry.resolve_entity_proxy( + approver_ref, raise_=True + ) + + approved_at = self.parse_legacy_datetime(self.approved_entry.get("date")) + if approved_at: + event.model.created = approved_at + + uow.register(RecordCommitOp(event, indexer=current_events_service.indexer)) + + def _apply_approved_entry(self, request, uow): + """Update an existing request to accepted using the legacy approved entry.""" + payload = dict(request.get("payload") or {}) + payload["approved_report_number"] = self.report_number + request["payload"] = payload + request.status = "accepted" + + approved_at = self.parse_legacy_datetime(self.approved_entry.get("date")) + if approved_at: + request.model.updated = approved_at + + self._create_accept_log_event(request, uow) + + def _create_request(self, restricted_recid, restricted_parent): + """Create request from waiting entry, then update it with approved entry.""" + expires_at = self.parse_legacy_datetime(self.waiting_entry.get("deadline")) + referee_group = self._get_referee_group(restricted_parent) + + with UnitOfWork() as uow: + request_item = current_requests_service.create( + system_identity, + data={ + "title": f'EP approval for "{self.title}"', + "payload": {}, + }, + request_type=CommitteeApprovalRequest, + receiver={"group": referee_group}, + creator=self._resolve_user_by_email( + self.waiting_entry.get("submitted_by"), "submitter" + ), + topic={"record": restricted_recid}, + expires_at=expires_at, + uow=uow, + ) + request = request_item._record + request.number = f"lrecid:{self.legacy_recid}:ep-approval" + request.status = "submitted" + + submitted_at = self.parse_legacy_datetime(self.waiting_entry.get("date")) + if submitted_at: + request.model.created = submitted_at + + self._apply_approved_entry(request, uow) + + uow.register( + RecordCommitOp(request, indexer=current_requests_service.indexer) + ) + uow.commit() diff --git a/cds_migrator_kit/rdm/records/load/ep_approval_load.py b/cds_migrator_kit/rdm/records/load/ep_approval_load.py index 2e5e7591..43e57f50 100644 --- a/cds_migrator_kit/rdm/records/load/ep_approval_load.py +++ b/cds_migrator_kit/rdm/records/load/ep_approval_load.py @@ -6,41 +6,34 @@ # the terms of the MIT License; see LICENSE file for more details. """CDS-RDM migration load module for records with EP approval.""" +import json import re from collections import OrderedDict from copy import deepcopy -from datetime import datetime, timezone -from pathlib import Path -from cds_rdm.requests.committee_approval import APPRN_PID_TYPE, CommitteeApprovalRequest -from flask import current_app +from cds_rdm.legacy.resolver import get_pid_by_legacy_recid +from cds_rdm.minters import legacy_recid_minter from invenio_access.permissions import system_identity -from invenio_accounts.models import User from invenio_db import db from invenio_db.uow import UnitOfWork from invenio_drafts_resources.services.records.uow import ParentRecordCommitOp -from invenio_pidstore.errors import PIDAlreadyExists -from invenio_pidstore.models import PersistentIdentifier, PIDStatus +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_rdm_records.records.api import RDMParent -from invenio_records_resources.services.uow import RecordCommitOp -from invenio_requests.customizations.event_types import LogEventType -from invenio_requests.proxies import current_events_service, current_requests_service -from invenio_requests.resolvers.registry import ResolverRegistry from cds_migrator_kit.errors import ManualImportRequired, UnexpectedValue from cds_migrator_kit.rdm.migration_config import CDS_CERN_SCIENTIFIC_COMMUNITY_ID +from .approval_request import ApprovalRequest from .load import CDSRecordServiceLoad EPPHAPP_FILE_TYPE = "EPPHAPP_FILE" -EP_APPROVAL_WAITING_STATUS = "waiting" -EP_APPROVAL_APPROVED_STATUS = "approved" EP_APPROVAL_REPORT_NUMBER_PREFIX = "CERN-EP" EP_APPROVAL_REPORT_NUMBER_RE = re.compile(r"^CERN-EP-\d{4}-\d{3}$") -class CDSEPApprovalRecordServiceLoad(CDSRecordServiceLoad): +class CDSEPApprovalRecordServiceLoad(Load): """Load records with EP approval. Splits a legacy record into two RDM records before load: @@ -48,59 +41,32 @@ class CDSEPApprovalRecordServiceLoad(CDSRecordServiceLoad): - a restricted record with restricted EPPHAPP files """ - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) - self.ep_approval_metadata = { - "title": None, - "experiment": None, - "resource_type": None, - "report_number": None, - } - self._ep_approval_parsed = None - self._load_flags = self._public_load_flags() - - def _public_load_flags(self): - """Load flags for the public migrated record.""" - return { - "mint_pids": True, - "mint_legacy_recid": True, - "save_original_dump": True, - "clc_sync": True, - "record_state": True, - } - - def _restricted_load_flags(self): - """Load flags for the restricted EPPHAPP snapshot record.""" - return { - "mint_pids": False, - "mint_legacy_recid": False, - "save_original_dump": False, - "clc_sync": False, - "record_state": False, - } - - def _should_log_record_state(self): - return self._load_flags["record_state"] - - def _after_publish_mint_recid(self, record, entry, version): - if self._load_flags["mint_legacy_recid"]: - super()._after_publish_mint_recid(record, entry, version) - - def _after_publish_update_dois(self, identity, record, entry, uow): - if self._load_flags["mint_pids"]: - return super()._after_publish_update_dois(identity, record, entry, uow) - - def _assign_rep_numbers(self, draft): - if self._load_flags["mint_pids"]: - super()._assign_rep_numbers(draft) - - def _save_original_dumped_record(self, entry, recid_state): - if self._load_flags["save_original_dump"]: - super()._save_original_dumped_record(entry, recid_state) - - def _after_load_clc_sync(self, record_state): - if self._load_flags["clc_sync"]: - super()._after_load_clc_sync(record_state) + def __init__( + self, + db_uri, + data_dir, + tmp_dir, + entries=None, + dry_run=False, + legacy_pids_to_redirect=None, + collection=None, + update_new_version_publication_date=True, + create_inclusion_request=False, + migration_logger=None, + record_state_logger=None, + ): + self.dry_run = dry_run + self.legacy_pids_to_redirect = {} + self.clc_sync = False + self.collection = collection + self.update_new_version_publication_date = update_new_version_publication_date + self.create_inclusion_request = create_inclusion_request + self.migration_logger = migration_logger + self.record_state_logger = record_state_logger + self.approval_request = None + if legacy_pids_to_redirect is not None: + with open(legacy_pids_to_redirect, "r") as fp: + self.legacy_pids_to_redirect = json.load(fp) def _load(self, entry): """ @@ -125,34 +91,71 @@ def _load(self, entry): record_json = entry.get("record", {}).get("json", {}) metadata = record_json.get("metadata", {}) - # Set the EP approval metadata - self.ep_approval_metadata["resource_type"] = metadata.get("resource_type") - self.ep_approval_metadata["title"] = metadata.get("title") - - # Validate the EP approval data - self._validate_ep_approval(ep_approval, recid) + self.approval_request = ApprovalRequest( + ep_approval=ep_approval, + legacy_recid=recid, + title=metadata.get("title"), + resource_type=metadata.get("resource_type"), + dry_run=self.dry_run, + ) + self.approval_request.validate() # Split the metadata and files public_entry = self._split_entry(entry, include_epphapp=False) restricted_entry = self._split_entry(entry, include_epphapp=True) # 1. Create restricted record - restricted_record_state = self._load_split_record( - restricted_entry, self._restricted_load_flags(), finalise=False + restricted_record_service = CDSRecordServiceLoad( + dry_run=self.dry_run, + collection=self.collection, + create_inclusion_request=self.create_inclusion_request, + migration_logger=self.migration_logger, + record_state_logger=self.record_state_logger, + legacy_pids_to_redirect=self.legacy_pids_to_redirect, + _is_final_record=False, ) + restricted_record_state = restricted_record_service._load(restricted_entry) # 2. Create and approve EP approval request - self._create_ep_approval(restricted_record_state, legacy_recid=recid) + self.approval_request.create(restricted_record_state) # 3. Create public record and link both records - public_record_state = self._load_split_record( - public_entry, - self._public_load_flags(), - finalise=True, - ) - self._link_ep_approval_records( - restricted_record_state, public_record_state, legacy_recid=recid - ) + public_record_service = CDSRecordServiceLoad( + dry_run=self.dry_run, + collection=self.collection, + create_inclusion_request=self.create_inclusion_request, + migration_logger=self.migration_logger, + record_state_logger=self.record_state_logger, + legacy_pids_to_redirect=self.legacy_pids_to_redirect, + _is_final_record=True, + ) + public_record_state = public_record_service._load(public_entry) + + if not self.dry_run: + # Link the records with related_identifiers + self._append_related_identifier( + public_record_state["latest_version"], + restricted_record_state["latest_version"], + "isversionof", + self.approval_request.resource_type, + ) + self._append_related_identifier( + restricted_record_state["latest_version"], + public_record_state["latest_version"], + "isvariantformof", + self.approval_request.resource_type, + ) + + # 4. Link the records with related_identifiers + self._link_parent_ep_approvals( + restricted_record_state, public_record_state, legacy_recid=recid + ) + + public_record_state["internal_version"] = restricted_record_state[ + "latest_version" + ] + self.record_state_logger.add_record_state(public_record_state) + self.migration_logger.finalise_record(recid) except (UnexpectedValue, ManualImportRequired) as e: self.migration_logger.add_log(e, record=entry) except Exception as e: @@ -161,76 +164,17 @@ def _load(self, entry): field="validation", stage="load", recid=recid, - priority="warning", - ) - self.migration_logger.add_log(exc, record=entry) - - def _load_split_record(self, entry, load_flags, finalise): - """Load the record.""" - self._load_flags = load_flags - self._finalise_on_load = finalise - return super()._load(entry) - - def _validate_ep_approval(self, ep_approval, legacy_recid): - """Validate EP approval data before creating any records.""" - waiting_entry, approved_entry, report_number = self._parse_ep_approval_history( - ep_approval - ) - - existing = self._existing_ep_approval_request(legacy_recid) - if existing: - raise ManualImportRequired( - message=f"EP approval request {existing['id']} already exists", - stage="load", priority="critical", ) - if self._exists_apprn_pid(report_number): - raise ManualImportRequired( - message=f"APPRN PID {report_number} already exists", - stage="load", - priority="critical", - ) - - self._ep_approval_parsed = (waiting_entry, approved_entry, report_number) - - def _create_ep_approval(self, restricted_record_state, legacy_recid): - """Create and approve EP approval request after restricted record exists.""" - waiting_entry, approved_entry, report_number = self._ep_approval_parsed - publication_title = self.ep_approval_metadata["title"] - - if not self.dry_run: - if not restricted_record_state: - raise UnexpectedValue( - message="Restricted record is required for EP approval.", - stage="load", - recid=legacy_recid, - priority="critical", - ) - restricted_recid = restricted_record_state["latest_version"] - restricted_parent = RDMParent.get_record( - restricted_record_state["parent_object_uuid"] - ) - self._create_ep_approval_request( - legacy_recid, - restricted_recid, - restricted_parent, - waiting_entry, - approved_entry, - report_number, - publication_title, - ) - self._mint_apprn_pid( - report_number, restricted_record_state["latest_version_object_uuid"] - ) + self.migration_logger.add_log(exc, record=entry) - def _link_ep_approval_records( + def _link_parent_ep_approvals( self, restricted_record_state, public_record_state, legacy_recid ): """Write parent metadata and link public/restricted records.""" - _, approved_entry, report_number = self._ep_approval_parsed - approval_iso = self._parse_legacy_datetime( - approved_entry.get("date") - ).isoformat() + approved_entry = self.approval_request.approved_entry + report_number = self.approval_request.report_number + approval_iso = self.approval_request.approved_at.isoformat() if not self.dry_run: if not restricted_record_state or not public_record_state: @@ -270,182 +214,6 @@ def _link_ep_approval_records( uow, ) uow.commit() - # Link the records with related_identifiers - self._append_related_identifier( - public_recid, - restricted_recid, - "isversionof", - self.ep_approval_metadata["resource_type"], - ) - self._append_related_identifier( - restricted_recid, - public_recid, - "isvariantformof", - self.ep_approval_metadata["resource_type"], - ) - - def _parse_ep_approval_history(self, ep_approval): - """Return waiting/approved history entries and the report number.""" - if len(ep_approval) != 2: - raise UnexpectedValue( - message="EP approval history has more/less than 2 entries", - stage="load", - priority="critical", - ) - history = ep_approval or [] - waiting = next( - ( - item - for item in history - if item.get("status") == EP_APPROVAL_WAITING_STATUS - ), - None, - ) - approved = next( - ( - item - for item in history - if item.get("status") == EP_APPROVAL_APPROVED_STATUS - ), - None, - ) - if not waiting: - raise UnexpectedValue( - message="EP approval history has no waiting entry", - stage="load", - priority="critical", - ) - if not approved: - raise UnexpectedValue( - message="EP approval history has no approved entry", - stage="load", - priority="critical", - ) - - report_number = approved.get("ep_report_number") - if not report_number: - raise UnexpectedValue( - message="EP approval approved entry is missing ep_report_number", - stage="load", - priority="critical", - ) - if waiting.get("ep_report_number") != report_number: - raise UnexpectedValue( - message="EP approval waiting entry has different ep_report_number than approved entry", - stage="load", - priority="critical", - ) - self.ep_approval_metadata["report_number"] = report_number - # Check if the submitters are exists - self._resolve_user_by_email(waiting.get("submitted_by"), "submitter") - self._resolve_user_by_email(approved.get("submitted_by"), "approver") - - # Check if record approved after the deadline - waiting_deadline = self._parse_legacy_datetime(waiting.get("deadline")) - approved_date = self._parse_legacy_datetime(approved.get("date")) - created_at = self._parse_legacy_datetime(waiting.get("date")) - if not created_at or not approved_date or not waiting_deadline: - raise UnexpectedValue( - message="EP approval history has missing timestamps", - stage="load", - priority="critical", - ) - if waiting_deadline and approved_date > waiting_deadline: - raise UnexpectedValue( - message="Record approved after the deadline", - stage="load", - priority="critical", - ) - return waiting, approved, report_number - - @staticmethod - def _parse_legacy_datetime(value): - """Parse legacy EP approval timestamps into timezone-aware datetimes.""" - if not value: - return None - for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d"): - try: - return datetime.strptime(value, fmt).replace(tzinfo=timezone.utc) - except ValueError: - continue - return None - - def _get_ep_referee_group(self, restricted_parent): - """Get the EP approval referee group from the restricted record.""" - default_community_id = restricted_parent.get("communities", {}).get("default") - if not default_community_id: - raise UnexpectedValue( - message="Restricted record has no default community for EP approval", - stage="load", - priority="critical", - ) - ep_config = current_app.config.get( - "CDS_COMMITTEE_APPROVAL_COMMUNITIES", {} - ).get(default_community_id) - if not ep_config: - raise UnexpectedValue( - message=( - f"Community {default_community_id} is not enrolled in " - "CDS_COMMITTEE_APPROVAL_COMMUNITIES" - ), - stage="load", - priority="critical", - ) - return ep_config["referee_group"] - - def _resolve_user_by_email(self, email, role): - """Resolve the user by email.""" - if not email: - raise UnexpectedValue( - message=f"EP approval {role} email is missing", - stage="load", - priority="critical", - ) - user = User.query.filter_by(email=email).one_or_none() - if not user: - raise UnexpectedValue( - message=f"EP approval {role} user not found: {email}", - stage="load", - priority="critical", - ) - return {"user": str(user.id)} - - def _existing_ep_approval_request(self, legacy_recid): - """Check if the EP approval request already exists.""" - number = f"lrecid:{legacy_recid}:ep-approval" - results = current_requests_service.search( - system_identity, - params={"q": f'number:"{number}"', "size": 1}, - ) - hits = list(results.hits) - return hits[0] if hits else None - - def _exists_apprn_pid(self, report_number): - """Check if the APPRN PID already exists.""" - existing = PersistentIdentifier.query.filter_by( - pid_type=APPRN_PID_TYPE, - pid_value=report_number, - ).one_or_none() - if existing: - return True - return False - - def _mint_apprn_pid(self, report_number, restricted_version_uuid): - """Mint the APPRN PID.""" - try: - PersistentIdentifier.create( - pid_type=APPRN_PID_TYPE, - pid_value=report_number, - object_type="rec", - object_uuid=str(restricted_version_uuid), - status=PIDStatus.REGISTERED, - ) - except PIDAlreadyExists: - raise ManualImportRequired( - message=f"APPRN PID {report_number} already exists", - stage="load", - priority="critical", - ) def _write_parent_ep_approval( self, @@ -459,93 +227,6 @@ def _write_parent_ep_approval( parent["permission_flags"] = pf uow.register(ParentRecordCommitOp(parent)) - def _create_accept_log_event(self, request, approved_entry, uow): - """Create the accept timeline event with the legacy approver as created_by.""" - approver_ref = self._resolve_user_by_email( - approved_entry.get("submitted_by"), - "approver", - ) - - event = current_events_service.record_cls.create( - {}, - request=request.model, - request_id=str(request.id), - type=LogEventType, - ) - event.update({"payload": {"event": "accepted"}}) - event.created_by = ResolverRegistry.resolve_entity_proxy( - approver_ref, raise_=True - ) - - approved_at = self._parse_legacy_datetime(approved_entry.get("date")) - if approved_at: - event.model.created = approved_at - - uow.register(RecordCommitOp(event, indexer=current_events_service.indexer)) - - def _apply_approved_entry_to_request( - self, request, approved_entry, report_number, uow - ): - """Update an existing request to accepted using the legacy approved entry.""" - payload = dict(request.get("payload") or {}) - payload["approved_report_number"] = report_number - request["payload"] = payload - request.status = "accepted" - - approved_at = self._parse_legacy_datetime(approved_entry.get("date")) - if approved_at: - request.model.updated = approved_at - - self._create_accept_log_event(request, approved_entry, uow) - - def _create_ep_approval_request( - self, - legacy_recid, - restricted_recid, - restricted_parent, - waiting_entry, - approved_entry, - report_number, - publication_title, - ): - """Create request from waiting entry, then update it with approved entry.""" - expires_at = self._parse_legacy_datetime(waiting_entry.get("deadline")) - - referee_group = self._get_ep_referee_group(restricted_parent) - with UnitOfWork() as uow: - request_item = current_requests_service.create( - system_identity, - data={ - "title": f'EP approval for "{publication_title}"', - # Use the default - "payload": {}, - }, - request_type=CommitteeApprovalRequest, - receiver={"group": referee_group}, - creator=self._resolve_user_by_email( - waiting_entry.get("submitted_by"), "submitter" - ), - topic={"record": restricted_recid}, - expires_at=expires_at, - uow=uow, - ) - request = request_item._record - request.number = f"lrecid:{legacy_recid}:ep-approval" - request.status = "submitted" - - submitted_at = self._parse_legacy_datetime(waiting_entry.get("date")) - if submitted_at: - request.model.created = submitted_at - - self._apply_approved_entry_to_request( - request, approved_entry, report_number, uow - ) - - uow.register( - RecordCommitOp(request, indexer=current_requests_service.indexer) - ) - uow.commit() - def _append_related_identifier( self, record_id, target_id, relation_id, resource_type ): @@ -587,7 +268,7 @@ def _should_remove_ep_report_number(self, identifier, public_split): if public_split: return True if EP_APPROVAL_REPORT_NUMBER_RE.match(identifier): - if identifier != self.ep_approval_metadata["report_number"]: + if identifier != self.approval_request.report_number: raise UnexpectedValue( message="EP report number is not the same as the approved entry", stage="load", @@ -661,6 +342,22 @@ def _split_entry(self, entry, include_epphapp): new_versions = OrderedDict() versioned_files = OrderedDict() previous_signature = None + has_epphapp_files = any( + file_data.get("type") == EPPHAPP_FILE_TYPE + for version_data in split.get("versions", {}).values() + for file_data in version_data.get("files", {}).values() + ) + if include_epphapp and not has_epphapp_files: + self.migration_logger.add_information( + split["record"]["recid"], + { + "message": ( + "No EPPHAPP files found; public files used for the " + "restricted record." + ), + "value": "public files", + }, + ) for _, version_data in split.get("versions", {}).items(): current_version_files = OrderedDict() @@ -668,7 +365,11 @@ def _split_entry(self, entry, include_epphapp): for key, file_data in version_data.get("files", {}).items(): is_epphapp = file_data.get("type") == EPPHAPP_FILE_TYPE - if include_epphapp != is_epphapp: + # If there are no EPPHAPP files, use the public files for the + # restricted split as well. + if include_epphapp != is_epphapp and not ( + include_epphapp and not has_epphapp_files + ): continue if not include_epphapp and file_data.get("access"): @@ -741,13 +442,15 @@ def _split_entry(self, entry, include_epphapp): ) if not include_epphapp: - split["record"]["access"] = "public" + # Public record does not need inclusion request + split["record"].pop("_request_data", None) + split["record"]["owned_by"] = "system" + split["parent"]["json"]["access"]["owned_by"] = {"user": "system"} self._add_cern_scientific_community(split) # Add the approval report number to the public record metadata - _, _, report_number = self._ep_approval_parsed split["record"]["json"]["metadata"]["identifiers"].append( { - "identifier": report_number, + "identifier": self.approval_request.report_number, "scheme": "apprn", } ) @@ -757,3 +460,21 @@ def _split_entry(self, entry, include_epphapp): self._remove_doi_pid_from_metadata(split, include_epphapp) return split + + def _cleanup(self, *args, **kwargs): + """Post migration process.""" + for legacy_src_pid, legacy_dest_pid in self.legacy_pids_to_redirect.items(): + if CDSRecordServiceLoad._have_migrated_recid(legacy_src_pid): + continue + try: + parent_dest_pid = get_pid_by_legacy_recid(str(legacy_dest_pid)) + assert str(parent_dest_pid.status) == "R" + legacy_recid_minter(legacy_src_pid, parent_dest_pid.object_uuid) + db.session.commit() + self.migration_logger.finalise_record(legacy_src_pid) + except Exception as exc: + db.session.rollback() + self.migration_logger.add_log( + f"Failed to redirect {legacy_src_pid} to {legacy_dest_pid}: {str(exc)}", + record={"recid": legacy_src_pid}, + ) diff --git a/cds_migrator_kit/rdm/records/load/load.py b/cds_migrator_kit/rdm/records/load/load.py index 19e1b142..1f8b581c 100644 --- a/cds_migrator_kit/rdm/records/load/load.py +++ b/cds_migrator_kit/rdm/records/load/load.py @@ -119,8 +119,9 @@ class CDSRecordServiceLoad(Load): def __init__( self, - db_uri, - data_dir, + db_uri=None, + data_dir=None, + tmp_dir=None, entries=None, dry_run=False, legacy_pids_to_redirect=None, @@ -129,20 +130,24 @@ def __init__( create_inclusion_request=False, migration_logger=None, record_state_logger=None, + _is_final_record=True, ): """Constructor.""" self.dry_run = dry_run self.legacy_pids_to_redirect = {} self.clc_sync = False - self._finalise_on_load = True self.collection = collection self.update_new_version_publication_date = update_new_version_publication_date self.create_inclusion_request = create_inclusion_request self.migration_logger = migration_logger self.record_state_logger = record_state_logger + self._is_final_record = _is_final_record if legacy_pids_to_redirect is not None: - with open(legacy_pids_to_redirect, "r") as fp: - self.legacy_pids_to_redirect = json.load(fp) + if isinstance(legacy_pids_to_redirect, dict): + self.legacy_pids_to_redirect = legacy_pids_to_redirect + else: + with open(legacy_pids_to_redirect, "r") as fp: + self.legacy_pids_to_redirect = json.load(fp) def _prepare(self, entry): """Prepare the record.""" @@ -247,6 +252,8 @@ def _load_record_access(self, draft, access_dict): def _after_publish_update_dois(self, identity, record, entry, uow): """Update migrated DOIs post publish.""" + if not self._is_final_record: + return migrated_pids = entry["record"]["json"]["pids"] for pid_type, identifier in migrated_pids.items(): if pid_type == "doi": @@ -457,6 +464,8 @@ def _after_publish_update_created(self, record, entry, version): def _after_publish_mint_recid(self, record, entry, version): """Mint legacy ids for redirections assigned to the parent.""" + if not self._is_final_record: + return legacy_recid = entry["record"]["recid"] if record._record.versions.index == 1: # it seems more intuitive if we mint the lrecid for parent @@ -583,6 +592,8 @@ def _after_publish(self, identity, published_record, entry, version, uow): # db.session.commit() def _assign_rep_numbers(self, draft): + if not self._is_final_record: + return draft_report_nums = {} for index, id in enumerate(draft.data["metadata"].get("identifiers", [])): if id["scheme"] == "cdsrn": @@ -699,14 +710,10 @@ def _load_versions(self, entry, uow): record_state_context = self._load_record_state(legacy_recid, records) # Dump the computed record state. This is useful to migrate then the record stats if record_state_context: - if self._should_log_record_state(): + if self._is_final_record: self.record_state_logger.add_record_state(record_state_context) return record_state_context - def _should_log_record_state(self): - """Whether to persist record state for stats migration.""" - return True - def _dry_load(self, entry): current_rdm_records_service.schema.load( entry["record"]["json"], @@ -794,6 +801,8 @@ def _save_original_dumped_record(self, entry, recid_state): This is the originally extracted record before any transformation. """ + if not self._is_final_record: + return _original_dump = entry["_original_dump"] _original_dump_model = CDSMigrationLegacyRecord( json=_original_dump, @@ -803,7 +812,8 @@ def _save_original_dumped_record(self, entry, recid_state): ) db.session.add(_original_dump_model) - def _have_migrated_recid(self, recid): + @staticmethod + def _have_migrated_recid(recid): """Check if we have minted `lrecid` pid.""" pid = PersistentIdentifier.query.filter_by( pid_type="lrecid", @@ -818,6 +828,8 @@ def _should_skip_recid(self, recid): return False def _after_load_clc_sync(self, record_state): + if not self._is_final_record: + return if self.clc_sync: sync = CDSToCLCSyncModel( parent_record_pid=record_state["parent_recid"], @@ -845,7 +857,7 @@ def _load(self, entry): ep_approval = entry.get("record", {}).get("ep_approval") if ep_approval: raise UnexpectedValue( - message="EP approval records must be loaded with the EP approval stream", + message="EP approval records must be loaded with the '--ep-approval' flag", stage="load", recid=recid, priority="critical", @@ -862,7 +874,7 @@ def _load(self, entry): ) self._after_load_clc_sync(recid_state_after_load) uow.commit() - if self._finalise_on_load: + if self._is_final_record: self.migration_logger.finalise_record(recid) return recid_state_after_load except (UnexpectedValue, ManualImportRequired) as e: diff --git a/cds_migrator_kit/rdm/records/transform/xml_processing/rules/faser_publication.py b/cds_migrator_kit/rdm/records/transform/xml_processing/rules/faser_publication.py index af7b1ea7..59995a4a 100644 --- a/cds_migrator_kit/rdm/records/transform/xml_processing/rules/faser_publication.py +++ b/cds_migrator_kit/rdm/records/transform/xml_processing/rules/faser_publication.py @@ -47,6 +47,10 @@ def access_grants(self, key, value): "faser-slide", ]: raise UnexpectedValue(subfield="a", field=key, value=value) + raw_identifier = value.get("m") + subject_identifier = StringValue(raw_identifier).parse() + if subject_identifier: + return {str(subject_identifier): "view"} raise IgnoreKey("access_grants") diff --git a/cds_migrator_kit/rdm/streams.yaml b/cds_migrator_kit/rdm/streams.yaml index cfcc70c1..1b80dfa8 100644 --- a/cds_migrator_kit/rdm/streams.yaml +++ b/cds_migrator_kit/rdm/streams.yaml @@ -410,7 +410,11 @@ records: - "" faser-drafts: data_dir: cds_migrator_kit/rdm/data/faser-drafts - restricted: "True" + tmp_dir: cds_migrator_kit/rdm/tmp/faser-drafts + log_dir: cds_migrator_kit/rdm/log/faser-drafts + restricted: "True" + plots: true + create_inclusion_request: true access_grants_view: - faser-all extract: @@ -422,6 +426,10 @@ records: - "33af9368-5bad-45cb-9360-8c9e5dfca09f" faser: data_dir: cds_migrator_kit/rdm/data/faser + tmp_dir: cds_migrator_kit/rdm/tmp/faser + log_dir: cds_migrator_kit/rdm/log/faser + create_inclusion_request: true + plots: true extract: dirpath: cds_migrator_kit/rdm/data/faser/dump/ transform: @@ -431,13 +439,46 @@ records: - "33af9368-5bad-45cb-9360-8c9e5dfca09f" faser-ep: data_dir: cds_migrator_kit/rdm/data/faser + tmp_dir: cds_migrator_kit/rdm/tmp/faser + log_dir: cds_migrator_kit/rdm/log/faser + plots: true extract: dirpath: cds_migrator_kit/rdm/data/faser-ep/dump/ transform: files_dump_dir: cds_migrator_kit/rdm/data/faser/files/ missing_users: cds_migrator_kit/rdm/data/users communities_ids: - - "33af9368-5bad-45cb-9360-8c9e5dfca09f" + - "e5de962b-5e2c-408f-be69-2823fda1f732" + lcd_restr: + data_dir: cds_migrator_kit/rdm/data/lep_exp/lcd + tmp_dir: cds_migrator_kit/rdm/tmp/lep_exp/lcd + log_dir: cds_migrator_kit/rdm/log/lep_exp/lcd + restricted: "True" + create_inclusion_request: true + extract: + dirpath: cds_migrator_kit/rdm/data/lep_exp/lcd/dump/ + transform: + files_dump_dir: cds_migrator_kit/rdm/data/lep_exp/lcd/files/ + missing_users: cds_migrator_kit/rdm/data/users + communities_ids: + - "" + load: + legacy_pids_to_redirect: cds_migrator_kit/rdm/data/lep_exp/lcd/duplicated_pids.json + re29_restr: + data_dir: cds_migrator_kit/rdm/data/lep_exp/re29 + tmp_dir: cds_migrator_kit/rdm/tmp/lep_exp/re29 + log_dir: cds_migrator_kit/rdm/log/lep_exp/re29 + restricted: "True" + create_inclusion_request: true + extract: + dirpath: cds_migrator_kit/rdm/data/lep_exp/re29/dump/ + transform: + files_dump_dir: cds_migrator_kit/rdm/data/lep_exp/re29/files/ + missing_users: cds_migrator_kit/rdm/data/users + communities_ids: + - "" + load: + legacy_pids_to_redirect: cds_migrator_kit/rdm/data/lep_exp/re29/duplicated_pids.json staff_association: data_dir: cds_migrator_kit/rdm/data/staff_association extract: diff --git a/cds_migrator_kit/reports/templates/cds_migrator_kit_records/records.html b/cds_migrator_kit/reports/templates/cds_migrator_kit_records/records.html index 63603246..93ebda11 100644 --- a/cds_migrator_kit/reports/templates/cds_migrator_kit_records/records.html +++ b/cds_migrator_kit/reports/templates/cds_migrator_kit_records/records.html @@ -81,10 +81,15 @@

Records: {{ total }} | Migrated: {{ migrated }} | {% if row["recid"] %} + {% set record = record_states.get(row["recid"], {}) %} View JSON | View record + {% if record.get("internal_version") %} + | + View internal record + {% endif %} {% endif %} diff --git a/cds_migrator_kit/reports/views.py b/cds_migrator_kit/reports/views.py index 7f8f7977..7ae5538f 100644 --- a/cds_migrator_kit/reports/views.py +++ b/cds_migrator_kit/reports/views.py @@ -58,6 +58,13 @@ def results(collection): next_page = False logger = MigrationProgressLogger(collection=collection) record_logs = logger.read_log() + state_logger = RecordStateLogger(collection=collection, keep_logs=True) + state_logger._load_existing_logs() + record_states = { + str(state["legacy_recid"]): state + for state in state_logger._record_states + if state.get("legacy_recid") is not None + } template = "cds_migrator_kit_records/records.html" record_logs = list(record_logs) critical = 0 @@ -97,6 +104,7 @@ def results(collection): prev_page=prev_page, next_page=next_page, paginated_record_logs=paginated_record_logs, + record_states=record_states, ) except FileNotFoundError as e: template = "cds_migrator_kit_records/rectype_missing.html" From f46917bf61173987676a597609a2561b2f722d33 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Z=C3=BCbeyde=20Civelek?= Date: Tue, 21 Jul 2026 10:49:48 +0200 Subject: [PATCH 3/4] add(ep-approval): new entry to split the records --- .../rdm/records/load/ep_approval_entry.py | 312 ++++++++++++++++++ .../rdm/records/load/ep_approval_load.py | 234 +------------ cds_migrator_kit/rdm/streams.yaml | 1 + 3 files changed, 324 insertions(+), 223 deletions(-) create mode 100644 cds_migrator_kit/rdm/records/load/ep_approval_entry.py diff --git a/cds_migrator_kit/rdm/records/load/ep_approval_entry.py b/cds_migrator_kit/rdm/records/load/ep_approval_entry.py new file mode 100644 index 00000000..365b3734 --- /dev/null +++ b/cds_migrator_kit/rdm/records/load/ep_approval_entry.py @@ -0,0 +1,312 @@ +# -*- coding: utf-8 -*- +# +# Copyright (C) 2026 CERN. +# +# CDS-RDM is free software; you can redistribute it and/or modify it under +# the terms of the MIT License; see LICENSE file for more details. + +"""Build public and restricted load entries for EP approval records.""" +import re +from collections import OrderedDict +from copy import deepcopy + +from cds_migrator_kit.errors import UnexpectedValue +from cds_migrator_kit.rdm.migration_config import CDS_CERN_SCIENTIFIC_COMMUNITY_ID + +EPPHAPP_FILE_TYPE = "EPPHAPP_FILE" +EP_APPROVAL_REPORT_NUMBER_PREFIX = "CERN-EP" +EP_APPROVAL_REPORT_NUMBER_RE = re.compile(r"^CERN-EP-\d{4}-\d{3}$") + + +class MetadataEntry: + """Build a load entry for the public or restricted EP approval split.""" + + def __init__(self, entry, approval_request, migration_logger): + self.entry = entry + self.approval_request = approval_request + self.migration_logger = migration_logger + + def identifiers(self, identifiers): + """Return identifiers for this split.""" + raise NotImplementedError + + def build(self): + """Return a load entry with split files and modified metadata.""" + split = deepcopy(self.entry) + split["record"].pop("ep_approval", None) + split["versions"] = self._build_versions(split) + self._apply_metadata(split) + self._apply_entry_modifications(split) + return split + + def _apply_metadata(self, split): + metadata = split["record"]["json"]["metadata"] + metadata["identifiers"] = self.identifiers(metadata.get("identifiers", [])) + self._remove_doi_pid(split) + + def _apply_entry_modifications(self, split): + """Apply record/parent level modifications.""" + + def _log_removed_identifiers(self, removed, split_type): + recid = self.entry.get("record", {}).get("recid") + self.migration_logger.add_information( + recid, + { + "message": ( + f"Removed EP approval report number(s) from {split_type} " "record." + ), + "value": removed, + }, + ) + + def _remove_doi_pid(self, split): + """Remove DOI PID from record.""" + pass + + def _build_versions(self, split): + """Return versioned files for this split; override in subclasses.""" + raise NotImplementedError + + @staticmethod + def _version_signature(versioned_files): + return tuple( + sorted( + ( + key, + file_data.get("checksum"), + file_data.get("id_bibdoc"), + file_data.get("version"), + file_data.get("type"), + file_data.get("access"), + ) + for key, file_data in versioned_files.items() + ) + ) + + +class PublicEntry(MetadataEntry): + """Build the public EP approval split entry.""" + + def _build_versions(self, split): + new_versions = OrderedDict() + versioned_files = OrderedDict() + previous_signature = None + + for _, version_data in split.get("versions", {}).items(): + current_version_files = OrderedDict() + + for key, file_data in version_data.get("files", {}).items(): + if file_data.get("type") == EPPHAPP_FILE_TYPE: + continue + + if file_data.get("access"): + raise UnexpectedValue( + message=( + "Public split contains restricted files after excluding " + f"EPPHAPP files: {[key]}" + ), + stage="load", + recid=split["record"]["recid"], + priority="critical", + ) + + current_version_files[key] = deepcopy(file_data) + + if not current_version_files: + continue + + versioned_files.update(current_version_files) + + signature = self._version_signature(versioned_files) + # If the signature is the same, skip the version. + if signature == previous_signature: + continue + + previous_signature = signature + + version_access = deepcopy(version_data.get("access", {})) + access_obj = deepcopy(version_access.get("access_obj", {})) + access_obj["record"] = "public" + access_obj["files"] = "public" + version_access.pop("meta", None) + version_access["access_obj"] = access_obj + + new_version_data = deepcopy(version_data) + new_version_data["files"] = deepcopy(versioned_files) + new_version_data["access"] = version_access + + new_versions[len(new_versions) + 1] = new_version_data + + if not new_versions: + raise UnexpectedValue( + message="No public files found to load for EP approval public split", + stage="load", + recid=split["record"]["recid"], + priority="critical", + ) + + return new_versions + + def identifiers(self, identifiers): + kept = [] + removed = [] + for id_entry in identifiers: + if id_entry.get("scheme") != "cdsrn": + kept.append(id_entry) + continue + identifier = id_entry.get("identifier", "") + if identifier.startswith(EP_APPROVAL_REPORT_NUMBER_PREFIX): + removed.append(identifier) + else: + kept.append(id_entry) + + kept.append( + { + "identifier": self.approval_request.report_number, + "scheme": "apprn", + } + ) + + if removed: + self._log_removed_identifiers(removed, "public") + + return kept + + def _apply_entry_modifications(self, split): + split["record"].pop("_request_data", None) + split["record"]["owned_by"] = "system" + split["parent"]["json"]["access"]["owned_by"] = {"user": "system"} + self._add_cern_scientific_community(split) + + def _add_cern_scientific_community(self, entry): + communities = entry.get("parent", {}).get("json", {}).get("communities", {}) + ids = list(communities.get("ids", [])) + if CDS_CERN_SCIENTIFIC_COMMUNITY_ID not in ids: + ids.append(CDS_CERN_SCIENTIFIC_COMMUNITY_ID) + communities["ids"] = ids + entry.setdefault("parent", {}).setdefault("json", {})[ + "communities" + ] = communities + + +class RestrictedEntry(MetadataEntry): + """Build the restricted EP approval split entry.""" + + def _has_epphapp_files(self, split): + return any( + file_data.get("type") == EPPHAPP_FILE_TYPE + for version_data in split.get("versions", {}).values() + for file_data in version_data.get("files", {}).values() + ) + + def _build_versions(self, split): + new_versions = OrderedDict() + versioned_files = OrderedDict() + previous_signature = None + has_epphapp_files = self._has_epphapp_files(split) + + if not has_epphapp_files: + self.migration_logger.add_information( + split["record"]["recid"], + { + "message": ( + "No EPPHAPP files found; public files used for the " + "restricted record." + ), + "value": "public files", + }, + ) + + for _, version_data in split.get("versions", {}).items(): + current_version_files = OrderedDict() + + for key, file_data in version_data.get("files", {}).items(): + is_epphapp = file_data.get("type") == EPPHAPP_FILE_TYPE + + # If draft file exists, use that otherwise use the public files. + if not is_epphapp and has_epphapp_files: + continue + + current_version_files[key] = deepcopy(file_data) + + if not current_version_files: + continue + + versioned_files.update(current_version_files) + + signature = self._version_signature(versioned_files) + if signature == previous_signature: + continue + + previous_signature = signature + + version_access = deepcopy(version_data.get("access", {})) + access_obj = deepcopy(version_access.get("access_obj", {})) + access_obj["record"] = "restricted" + access_obj["files"] = "restricted" + version_access["access_obj"] = access_obj + + new_version_data = deepcopy(version_data) + new_version_data["files"] = deepcopy(versioned_files) + new_version_data["access"] = version_access + + new_versions[len(new_versions) + 1] = new_version_data + + if not new_versions: + raise UnexpectedValue( + message=("No files found to load for EP approval restricted split"), + stage="load", + recid=split["record"]["recid"], + priority="critical", + ) + + return new_versions + + def identifiers(self, identifiers): + kept = [] + removed = [] + for id_entry in identifiers: + if id_entry.get("scheme") != "cdsrn": + kept.append(id_entry) + continue + identifier = id_entry.get("identifier", "") + if not identifier.startswith(EP_APPROVAL_REPORT_NUMBER_PREFIX): + kept.append(id_entry) + continue + # Remove CERN-EP-YYYY-NNN but keep CERN-EP-DRAFT report number + if EP_APPROVAL_REPORT_NUMBER_RE.match(identifier): + if identifier != self.approval_request.report_number: + raise UnexpectedValue( + message=( + "EP report number is not the same as the approved entry" + ), + stage="load", + priority="critical", + ) + removed.append(identifier) + else: + kept.append(id_entry) + + if removed: + self._log_removed_identifiers(removed, "restricted") + + return kept + + def _remove_doi_pid(self, split): + """Remove DOI PID from restricted record.""" + recid = split.get("record", {}).get("recid") + record_json = split.get("record", {}).get("json", {}) + pids = record_json.get("pids") + + if not pids or "doi" not in pids: + return + + removed = pids.pop("doi") + + self.migration_logger.add_information( + recid, + { + "message": "Removed DOI PID from restricted record.", + "value": removed, + }, + ) diff --git a/cds_migrator_kit/rdm/records/load/ep_approval_load.py b/cds_migrator_kit/rdm/records/load/ep_approval_load.py index 43e57f50..c7550ad4 100644 --- a/cds_migrator_kit/rdm/records/load/ep_approval_load.py +++ b/cds_migrator_kit/rdm/records/load/ep_approval_load.py @@ -7,9 +7,6 @@ """CDS-RDM migration load module for records with EP approval.""" import json -import re -from collections import OrderedDict -from copy import deepcopy from cds_rdm.legacy.resolver import get_pid_by_legacy_recid from cds_rdm.minters import legacy_recid_minter @@ -23,15 +20,11 @@ from invenio_rdm_records.records.api import RDMParent from cds_migrator_kit.errors import ManualImportRequired, UnexpectedValue -from cds_migrator_kit.rdm.migration_config import CDS_CERN_SCIENTIFIC_COMMUNITY_ID from .approval_request import ApprovalRequest +from .ep_approval_entry import PublicEntry, RestrictedEntry from .load import CDSRecordServiceLoad -EPPHAPP_FILE_TYPE = "EPPHAPP_FILE" -EP_APPROVAL_REPORT_NUMBER_PREFIX = "CERN-EP" -EP_APPROVAL_REPORT_NUMBER_RE = re.compile(r"^CERN-EP-\d{4}-\d{3}$") - class CDSEPApprovalRecordServiceLoad(Load): """Load records with EP approval. @@ -101,8 +94,16 @@ def _load(self, entry): self.approval_request.validate() # Split the metadata and files - public_entry = self._split_entry(entry, include_epphapp=False) - restricted_entry = self._split_entry(entry, include_epphapp=True) + public_entry = PublicEntry( + entry, + approval_request=self.approval_request, + migration_logger=self.migration_logger, + ).build() + restricted_entry = RestrictedEntry( + entry, + approval_request=self.approval_request, + migration_logger=self.migration_logger, + ).build() # 1. Create restricted record restricted_record_service = CDSRecordServiceLoad( @@ -154,8 +155,6 @@ def _load(self, entry): public_record_state["internal_version"] = restricted_record_state[ "latest_version" ] - self.record_state_logger.add_record_state(public_record_state) - self.migration_logger.finalise_record(recid) except (UnexpectedValue, ManualImportRequired) as e: self.migration_logger.add_log(e, record=entry) except Exception as e: @@ -250,217 +249,6 @@ def _append_related_identifier( current_rdm_records_service.publish(system_identity, id_=draft.id) return True - def _add_cern_scientific_community(self, entry): - """Add the CERN Scientific community to the public record parent.""" - communities = entry.get("parent", {}).get("json", {}).get("communities", {}) - ids = list(communities.get("ids", [])) - if CDS_CERN_SCIENTIFIC_COMMUNITY_ID not in ids: - ids.append(CDS_CERN_SCIENTIFIC_COMMUNITY_ID) - communities["ids"] = ids - entry.setdefault("parent", {}).setdefault("json", {})[ - "communities" - ] = communities - - def _should_remove_ep_report_number(self, identifier, public_split): - """Return whether an EP report number should be stripped from metadata.""" - if not identifier.startswith(EP_APPROVAL_REPORT_NUMBER_PREFIX): - return False - if public_split: - return True - if EP_APPROVAL_REPORT_NUMBER_RE.match(identifier): - if identifier != self.approval_request.report_number: - raise UnexpectedValue( - message="EP report number is not the same as the approved entry", - stage="load", - priority="critical", - ) - return True - return False - - def _remove_ep_report_numbers_from_metadata(self, entry, include_epphapp): - """Strip EP report numbers from split record metadata before load.""" - recid = entry.get("record", {}).get("recid") - metadata = entry.get("record", {}).get("json", {}).get("metadata", {}) - identifiers = metadata.get("identifiers", []) - if not identifiers: - return - - kept = [] - removed = [] - for id_entry in identifiers: - if id_entry.get("scheme") != "cdsrn": - kept.append(id_entry) - continue - identifier = id_entry.get("identifier", "") - if self._should_remove_ep_report_number(identifier, not include_epphapp): - removed.append(identifier) - else: - kept.append(id_entry) - - if not removed: - return - - metadata["identifiers"] = kept - split_type = "restricted" if include_epphapp else "public" - self.migration_logger.add_information( - recid, - { - "message": ( - f"Removed EP approval report number(s) from {split_type} " "record." - ), - "value": removed, - }, - ) - - def _remove_doi_pid_from_metadata(self, entry, include_epphapp): - """Strip DOI PID from restricted EPPHAPP split record metadata before load.""" - if not include_epphapp: - return - - recid = entry.get("record", {}).get("recid") - record_json = entry.get("record", {}).get("json", {}) - pids = record_json.get("pids") - - if not pids or "doi" not in pids: - return - - removed = pids.pop("doi") - - self.migration_logger.add_information( - recid, - { - "message": "Removed DOI PID from restricted record.", - "value": removed, - }, - ) - - def _split_entry(self, entry, include_epphapp): - """Return a load entry for the public or restricted EP approval split.""" - split = deepcopy(entry) - split["record"].pop("ep_approval", None) - - new_versions = OrderedDict() - versioned_files = OrderedDict() - previous_signature = None - has_epphapp_files = any( - file_data.get("type") == EPPHAPP_FILE_TYPE - for version_data in split.get("versions", {}).values() - for file_data in version_data.get("files", {}).values() - ) - if include_epphapp and not has_epphapp_files: - self.migration_logger.add_information( - split["record"]["recid"], - { - "message": ( - "No EPPHAPP files found; public files used for the " - "restricted record." - ), - "value": "public files", - }, - ) - - for _, version_data in split.get("versions", {}).items(): - current_version_files = OrderedDict() - - for key, file_data in version_data.get("files", {}).items(): - is_epphapp = file_data.get("type") == EPPHAPP_FILE_TYPE - - # If there are no EPPHAPP files, use the public files for the - # restricted split as well. - if include_epphapp != is_epphapp and not ( - include_epphapp and not has_epphapp_files - ): - continue - - if not include_epphapp and file_data.get("access"): - raise UnexpectedValue( - message=( - "Public split contains restricted files after excluding " - f"EPPHAPP files: {[key]}" - ), - stage="load", - recid=split["record"]["recid"], - priority="critical", - ) - - current_version_files[key] = deepcopy(file_data) - - if not current_version_files: - continue - - versioned_files.update(current_version_files) - - signature = tuple( - sorted( - ( - key, - file_data.get("checksum"), - file_data.get("id_bibdoc"), - file_data.get("version"), - file_data.get("type"), - file_data.get("access"), - ) - for key, file_data in versioned_files.items() - ) - ) - # If the signature is the same, skip the version - if signature == previous_signature: - continue - - previous_signature = signature - - version_access = deepcopy(version_data.get("access", {})) - access_obj = deepcopy(version_access.get("access_obj", {})) - - if include_epphapp: - access_obj["record"] = "restricted" - access_obj["files"] = "restricted" - else: - access_obj["record"] = "public" - access_obj["files"] = "public" - # Remove the meta field from the public version access - version_access.pop("meta", None) - - version_access["access_obj"] = access_obj - - new_version_data = deepcopy(version_data) - new_version_data["files"] = deepcopy(versioned_files) - new_version_data["access"] = version_access - - new_versions[len(new_versions) + 1] = new_version_data - - if not new_versions: - raise UnexpectedValue( - message=( - "No EPPHAPP files found to load for EP approval restricted split" - if include_epphapp - else "No public files found to load for EP approval public split" - ), - stage="load", - recid=split["record"]["recid"], - priority="critical", - ) - - if not include_epphapp: - # Public record does not need inclusion request - split["record"].pop("_request_data", None) - split["record"]["owned_by"] = "system" - split["parent"]["json"]["access"]["owned_by"] = {"user": "system"} - self._add_cern_scientific_community(split) - # Add the approval report number to the public record metadata - split["record"]["json"]["metadata"]["identifiers"].append( - { - "identifier": self.approval_request.report_number, - "scheme": "apprn", - } - ) - - split["versions"] = new_versions - self._remove_ep_report_numbers_from_metadata(split, include_epphapp) - self._remove_doi_pid_from_metadata(split, include_epphapp) - - return split - def _cleanup(self, *args, **kwargs): """Post migration process.""" for legacy_src_pid, legacy_dest_pid in self.legacy_pids_to_redirect.items(): diff --git a/cds_migrator_kit/rdm/streams.yaml b/cds_migrator_kit/rdm/streams.yaml index 1b80dfa8..b9d5b0cb 100644 --- a/cds_migrator_kit/rdm/streams.yaml +++ b/cds_migrator_kit/rdm/streams.yaml @@ -441,6 +441,7 @@ records: data_dir: cds_migrator_kit/rdm/data/faser tmp_dir: cds_migrator_kit/rdm/tmp/faser log_dir: cds_migrator_kit/rdm/log/faser + create_inclusion_request: true plots: true extract: dirpath: cds_migrator_kit/rdm/data/faser-ep/dump/ From 2bffd96b95d5d7cec29693e3e635d5ed056341f9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Z=C3=BCbeyde=20Civelek?= Date: Tue, 21 Jul 2026 11:29:53 +0200 Subject: [PATCH 4/4] add(tests): ep approval split entry --- tests/cds-rdm/test_ep_approval_entry.py | 562 ++++++++++++++++++++++++ 1 file changed, 562 insertions(+) create mode 100644 tests/cds-rdm/test_ep_approval_entry.py diff --git a/tests/cds-rdm/test_ep_approval_entry.py b/tests/cds-rdm/test_ep_approval_entry.py new file mode 100644 index 00000000..d0d18470 --- /dev/null +++ b/tests/cds-rdm/test_ep_approval_entry.py @@ -0,0 +1,562 @@ +# -*- coding: utf-8 -*- +# +# Copyright (C) 2026 CERN. +# +# CDS-RDM is free software; you can redistribute it and/or modify it under +# the terms of the MIT License; see LICENSE file for more details. + +"""Tests for EP approval entry splitting (PublicEntry / RestrictedEntry).""" + +from collections import OrderedDict +from copy import deepcopy +from unittest.mock import MagicMock + +import pytest + +from cds_migrator_kit.errors import UnexpectedValue +from cds_migrator_kit.rdm.migration_config import CDS_CERN_SCIENTIFIC_COMMUNITY_ID +from cds_migrator_kit.rdm.records.load.ep_approval_entry import ( + EPPHAPP_FILE_TYPE, + PublicEntry, + RestrictedEntry, +) + +RECID = "12345" +APPROVED_REPORT_NUMBER = "CERN-EP-2020-001" +DRAFT_REPORT_NUMBER = "CERN-EP-DRAFT-TEST-2020-001" +PUBLIC_FILE_KEY = "main.pdf" +DRAFT_FILE_KEY = "draft.pdf" + + +def _make_approval_request(report_number=APPROVED_REPORT_NUMBER): + ar = MagicMock() + ar.report_number = report_number + ar.resource_type = {"id": "publication-article"} + return ar + + +def _make_migration_logger(): + return MagicMock() + + +def _public_file(key=PUBLIC_FILE_KEY, checksum="aaa", version=1, id_bibdoc=100): + return { + "key": key, + "checksum": checksum, + "version": version, + "id_bibdoc": id_bibdoc, + "access": "", + "type": "Main", + "creation_date": "2020-01-15", + } + + +def _epphapp_file(key=DRAFT_FILE_KEY, checksum="bbb", version=1, id_bibdoc=200): + return { + "key": key, + "checksum": checksum, + "version": version, + "id_bibdoc": id_bibdoc, + "access": "EP Restricted Draft", + "type": EPPHAPP_FILE_TYPE, + "creation_date": "2020-01-10", + } + + +def _make_entry( + versions, + recid=RECID, + identifiers=None, + has_doi=False, + report_number=APPROVED_REPORT_NUMBER, +): + """Build a minimal entry dict for testing.""" + if identifiers is None: + identifiers = [ + {"identifier": recid, "scheme": "cds"}, + {"scheme": "cdsrn", "identifier": report_number}, + {"scheme": "cdsrn", "identifier": DRAFT_REPORT_NUMBER}, + ] + + record_json = { + "metadata": { + "creators": [ + { + "person_or_org": { + "type": "personal", + "family_name": "Smith", + "given_name": "Alice", + }, + "affiliations": [{"name": "Example University"}], + } + ], + "title": "Example publication title", + "resource_type": {"id": "publication-article"}, + "description": "Example description.", + "publication_date": "2020-01-15", + "identifiers": identifiers, + }, + } + if has_doi: + record_json["pids"] = { + "doi": { + "identifier": "10.1234/example-doi", + "provider": "external", + } + } + + return { + "record": { + "recid": recid, + "json": record_json, + "ep_approval": [ + { + "status": "waiting", + "ep_report_number": report_number, + }, + { + "status": "approved", + "ep_report_number": report_number, + }, + ], + "owned_by": "uploader", + "_request_data": {"placeholder": True}, + }, + "parent": { + "json": { + "access": {"owned_by": {"user": "uploader"}}, + "communities": {"ids": ["example-community"]}, + } + }, + "versions": versions, + } + + +def _versions_with_epphapp(): + """Four legacy versions: draft file changes, public file stays the same.""" + return OrderedDict( + [ + ( + 1, + { + "files": { + DRAFT_FILE_KEY: _epphapp_file(checksum="draft-v1", version=1), + PUBLIC_FILE_KEY: _public_file(checksum="main-v1"), + }, + "publication_date": "2020-01-10", + "access": { + "access_obj": {"record": None, "files": "restricted"}, + "meta": "EP Restricted Draft", + }, + }, + ), + ( + 2, + { + "files": { + DRAFT_FILE_KEY: _epphapp_file(checksum="draft-v2", version=2), + PUBLIC_FILE_KEY: _public_file(checksum="main-v1"), + }, + "publication_date": "2020-01-12", + "access": { + "access_obj": {"record": None, "files": "restricted"}, + "meta": "EP Restricted Draft", + }, + }, + ), + ( + 3, + { + "files": { + DRAFT_FILE_KEY: _epphapp_file(checksum="draft-v3", version=3), + PUBLIC_FILE_KEY: _public_file(checksum="main-v1"), + }, + "publication_date": "2020-01-14", + "access": { + "access_obj": {"record": None, "files": "restricted"}, + "meta": "EP Restricted Draft", + }, + }, + ), + ( + 4, + { + "files": { + DRAFT_FILE_KEY: _epphapp_file(checksum="draft-v4", version=4), + PUBLIC_FILE_KEY: _public_file(checksum="main-v1"), + }, + "publication_date": "2020-01-15", + "access": { + "access_obj": {"record": None, "files": "restricted"}, + "meta": "EP Restricted Draft", + }, + }, + ), + ] + ) + + +def _versions_public_only(): + """Single version with only public files.""" + return OrderedDict( + [ + ( + 1, + { + "files": { + "document.pdf": _public_file( + key="document.pdf", checksum="doc-v1", id_bibdoc=300 + ), + }, + "publication_date": "2020-02-01", + "access": {"access_obj": {"record": None, "files": None}}, + }, + ), + ] + ) + + +class TestPublicEntryVersions: + """Test that PublicEntry filters out EPPHAPP files and deduplicates versions.""" + + def test_public_excludes_epphapp_files(self): + entry = _make_entry(_versions_with_epphapp()) + result = PublicEntry( + entry, _make_approval_request(), _make_migration_logger() + ).build() + + for _, vdata in result["versions"].items(): + for key, fdata in vdata["files"].items(): + assert ( + fdata["type"] != EPPHAPP_FILE_TYPE + ), f"EPPHAPP file {key} should not appear in public split" + + def test_public_deduplicates_identical_versions(self): + entry = _make_entry(_versions_with_epphapp()) + result = PublicEntry( + entry, _make_approval_request(), _make_migration_logger() + ).build() + + assert len(result["versions"]) == 1 + + def test_public_access_is_public(self): + entry = _make_entry(_versions_with_epphapp()) + result = PublicEntry( + entry, _make_approval_request(), _make_migration_logger() + ).build() + + for _, vdata in result["versions"].items(): + assert vdata["access"]["access_obj"]["record"] == "public" + assert vdata["access"]["access_obj"]["files"] == "public" + + def test_public_raises_when_no_public_files(self): + versions = OrderedDict( + [ + ( + 1, + { + "files": { + DRAFT_FILE_KEY: _epphapp_file(checksum="draft-only"), + }, + "publication_date": "2020-01-10", + "access": { + "access_obj": {"record": None, "files": "restricted"}, + }, + }, + ), + ] + ) + entry = _make_entry(versions) + with pytest.raises(UnexpectedValue, match="No public files found"): + PublicEntry( + entry, _make_approval_request(), _make_migration_logger() + ).build() + + def test_public_raises_on_restricted_files(self): + versions = OrderedDict( + [ + ( + 1, + { + "files": { + "restricted.pdf": { + "key": "restricted.pdf", + "checksum": "restricted-v1", + "version": 1, + "id_bibdoc": 999, + "access": "restricted", + "type": "Main", + "creation_date": "2020-01-01", + }, + }, + "publication_date": "2020-01-01", + "access": {"access_obj": {"record": None, "files": None}}, + }, + ), + ] + ) + entry = _make_entry(versions) + with pytest.raises(UnexpectedValue, match="restricted files"): + PublicEntry( + entry, _make_approval_request(), _make_migration_logger() + ).build() + + def test_public_multiple_distinct_versions(self): + versions = OrderedDict( + [ + ( + 1, + { + "files": { + "paper.pdf": _public_file( + key="paper.pdf", checksum="paper-v1", id_bibdoc=300 + ), + }, + "publication_date": "2020-01-01", + "access": {"access_obj": {"record": None, "files": None}}, + }, + ), + ( + 2, + { + "files": { + "paper.pdf": _public_file( + key="paper.pdf", + checksum="paper-v2", + version=2, + id_bibdoc=300, + ), + }, + "publication_date": "2020-02-01", + "access": {"access_obj": {"record": None, "files": None}}, + }, + ), + ] + ) + entry = _make_entry(versions) + result = PublicEntry( + entry, _make_approval_request(), _make_migration_logger() + ).build() + + assert len(result["versions"]) == 2 + + +class TestRestrictedEntryVersions: + """Test that RestrictedEntry keeps EPPHAPP files when present.""" + + def test_restricted_keeps_only_epphapp_when_present(self): + entry = _make_entry(_versions_with_epphapp()) + result = RestrictedEntry( + entry, _make_approval_request(), _make_migration_logger() + ).build() + + for _, vdata in result["versions"].items(): + for key, fdata in vdata["files"].items(): + assert ( + fdata["type"] == EPPHAPP_FILE_TYPE + ), f"Non-EPPHAPP file {key} should not appear in restricted split" + + def test_restricted_keeps_all_changing_epphapp_versions(self): + entry = _make_entry(_versions_with_epphapp()) + result = RestrictedEntry( + entry, _make_approval_request(), _make_migration_logger() + ).build() + + assert len(result["versions"]) == 4 + + def test_restricted_access_is_restricted(self): + entry = _make_entry(_versions_with_epphapp()) + result = RestrictedEntry( + entry, _make_approval_request(), _make_migration_logger() + ).build() + + for _, vdata in result["versions"].items(): + assert vdata["access"]["access_obj"]["record"] == "restricted" + assert vdata["access"]["access_obj"]["files"] == "restricted" + + def test_restricted_uses_public_files_when_no_epphapp(self): + entry = _make_entry(_versions_public_only()) + logger = _make_migration_logger() + result = RestrictedEntry(entry, _make_approval_request(), logger).build() + + assert len(result["versions"]) == 1 + assert "document.pdf" in result["versions"][1]["files"] + logger.add_information.assert_called() + + def test_restricted_raises_when_no_files_at_all(self): + versions = OrderedDict( + [ + ( + 1, + { + "files": {}, + "publication_date": "2020-01-01", + "access": {"access_obj": {"record": None, "files": None}}, + }, + ), + ] + ) + entry = _make_entry(versions) + with pytest.raises(UnexpectedValue, match="No files found"): + RestrictedEntry( + entry, _make_approval_request(), _make_migration_logger() + ).build() + + +class TestPublicEntryIdentifiers: + """Test identifier handling in the public split.""" + + def test_public_removes_cern_ep_report_numbers(self): + entry = _make_entry(_versions_with_epphapp()) + result = PublicEntry( + entry, _make_approval_request(), _make_migration_logger() + ).build() + + identifiers = result["record"]["json"]["metadata"]["identifiers"] + cdsrn_values = {i["identifier"] for i in identifiers if i["scheme"] == "cdsrn"} + + assert APPROVED_REPORT_NUMBER not in cdsrn_values + assert any( + i["scheme"] == "apprn" and i["identifier"] == APPROVED_REPORT_NUMBER + for i in identifiers + ) + + def test_public_keeps_non_ep_cdsrn(self): + identifiers = [ + {"identifier": RECID, "scheme": "cds"}, + {"scheme": "cdsrn", "identifier": APPROVED_REPORT_NUMBER}, + {"scheme": "cdsrn", "identifier": "OTHER-RN-001"}, + ] + entry = _make_entry(_versions_with_epphapp(), identifiers=identifiers) + result = PublicEntry( + entry, _make_approval_request(), _make_migration_logger() + ).build() + + cdsrn_ids = [ + i + for i in result["record"]["json"]["metadata"]["identifiers"] + if i["scheme"] == "cdsrn" + ] + assert len(cdsrn_ids) == 1 + assert cdsrn_ids[0]["identifier"] == "OTHER-RN-001" + + +class TestRestrictedEntryIdentifiers: + """Test identifier handling in the restricted split.""" + + def test_restricted_removes_matching_cern_ep_rn(self): + entry = _make_entry(_versions_with_epphapp()) + result = RestrictedEntry( + entry, _make_approval_request(), _make_migration_logger() + ).build() + + cdsrn_values = { + i["identifier"] + for i in result["record"]["json"]["metadata"]["identifiers"] + if i["scheme"] == "cdsrn" + } + + assert APPROVED_REPORT_NUMBER not in cdsrn_values + + def test_restricted_keeps_draft_report_number(self): + entry = _make_entry(_versions_with_epphapp()) + result = RestrictedEntry( + entry, _make_approval_request(), _make_migration_logger() + ).build() + + cdsrn_values = { + i["identifier"] + for i in result["record"]["json"]["metadata"]["identifiers"] + if i["scheme"] == "cdsrn" + } + + assert DRAFT_REPORT_NUMBER in cdsrn_values + + def test_restricted_raises_on_mismatched_report_number(self): + identifiers = [ + {"identifier": RECID, "scheme": "cds"}, + {"scheme": "cdsrn", "identifier": "CERN-EP-2020-999"}, + ] + entry = _make_entry(_versions_with_epphapp(), identifiers=identifiers) + with pytest.raises(UnexpectedValue, match="not the same"): + RestrictedEntry( + entry, _make_approval_request(), _make_migration_logger() + ).build() + + def test_restricted_removes_doi_pid(self): + entry = _make_entry(_versions_with_epphapp(), has_doi=True) + result = RestrictedEntry( + entry, _make_approval_request(), _make_migration_logger() + ).build() + + assert "doi" not in result["record"]["json"].get("pids", {}) + + +class TestPublicEntryModifications: + """Test record/parent level modifications on the public split.""" + + def test_public_removes_request_data(self): + entry = _make_entry(_versions_with_epphapp()) + result = PublicEntry( + entry, _make_approval_request(), _make_migration_logger() + ).build() + + assert "_request_data" not in result["record"] + + def test_public_sets_owned_by_system(self): + entry = _make_entry(_versions_with_epphapp()) + result = PublicEntry( + entry, _make_approval_request(), _make_migration_logger() + ).build() + + assert result["record"]["owned_by"] == "system" + assert result["parent"]["json"]["access"]["owned_by"] == {"user": "system"} + + def test_public_adds_cern_scientific_community(self): + entry = _make_entry(_versions_with_epphapp()) + result = PublicEntry( + entry, _make_approval_request(), _make_migration_logger() + ).build() + + assert CDS_CERN_SCIENTIFIC_COMMUNITY_ID in ( + result["parent"]["json"]["communities"]["ids"] + ) + + def test_public_does_not_duplicate_community(self): + entry = _make_entry(_versions_with_epphapp()) + entry["parent"]["json"]["communities"]["ids"] = [ + "example-community", + CDS_CERN_SCIENTIFIC_COMMUNITY_ID, + ] + result = PublicEntry( + entry, _make_approval_request(), _make_migration_logger() + ).build() + + community_ids = result["parent"]["json"]["communities"]["ids"] + assert community_ids.count(CDS_CERN_SCIENTIFIC_COMMUNITY_ID) == 1 + + +class TestEntryImmutability: + """Ensure build() deep-copies and does not mutate the original entry.""" + + def test_public_build_does_not_mutate_original(self): + entry = _make_entry(_versions_with_epphapp()) + original = deepcopy(entry) + PublicEntry(entry, _make_approval_request(), _make_migration_logger()).build() + + assert ( + entry["record"]["json"]["metadata"]["identifiers"] + == original["record"]["json"]["metadata"]["identifiers"] + ) + + def test_restricted_build_does_not_mutate_original(self): + entry = _make_entry(_versions_with_epphapp()) + original = deepcopy(entry) + RestrictedEntry( + entry, _make_approval_request(), _make_migration_logger() + ).build() + + assert ( + entry["record"]["json"]["metadata"]["identifiers"] + == original["record"]["json"]["metadata"]["identifiers"] + )