From 3bb75c282289d29c657323202242108a7ba4dbf6 Mon Sep 17 00:00:00 2001 From: Karolina Przerwa Date: Fri, 17 Jul 2026 15:14:16 +0200 Subject: [PATCH 1/2] fix(research): find reviewers in users table --- cds_migrator_kit/rdm/records/load/load.py | 146 +++++++++++------- .../xml_processing/rules/research.py | 14 +- cds_migrator_kit/rdm/streams.yaml | 5 + 3 files changed, 106 insertions(+), 59 deletions(-) diff --git a/cds_migrator_kit/rdm/records/load/load.py b/cds_migrator_kit/rdm/records/load/load.py index 1afd28f0..0c305142 100644 --- a/cds_migrator_kit/rdm/records/load/load.py +++ b/cds_migrator_kit/rdm/records/load/load.py @@ -32,9 +32,9 @@ from invenio_records_resources.services.uow import RecordCommitOp from invenio_requests.customizations.event_types import ( LogEventType, - ReviewersUpdatedType, ) from invenio_requests.proxies import current_events_service, current_requests_service +from invenio_requests.records.models import RequestMetadata from marshmallow import ValidationError from psycopg2.errors import UniqueViolation from sqlalchemy.exc import IntegrityError @@ -46,6 +46,52 @@ ) +def _is_email(value): + """Return True if the reviewer value looks like an email address.""" + return "@" in value + + +def _parse_reviewer_name(name): + """Split a 'Family, Given' or 'Given Family' string into (family, given). + + ``request_reviewers`` (906__p) stores names as "Given Family" (comma + already resolved), but legacy data can also arrive as "Family, Given". + """ + name = name.strip() + if "," in name: + family, _, given = name.partition(",") + return family.strip(), given.strip() + parts = name.split() + if len(parts) > 1: + return parts[-1], " ".join(parts[:-1]) + return name, "" + + +def find_reviewer(reviewer): + """Resolve a reviewer string (email or name) to a User. + + :param reviewer: email address, or a "Family, Given"/"Given Family" name. + :raises UnexpectedValue: if no matching user is found. + """ + reviewer = reviewer.strip() + if _is_email(reviewer): + user = User.query.filter_by(email=reviewer).one_or_none() + else: + family_name, given_name = _parse_reviewer_name(reviewer) + query = User.query.filter( + db.func.lower(User._user_profile["family_name"].as_string()) + == family_name.lower() + ) + if given_name: + query = query.filter( + db.func.lower(User._user_profile["given_name"].as_string()) + == given_name.lower() + ) + user = query.one_or_none() + + return user + + def import_legacy_files(filepath): """Download file from legacy.""" if current_app.config["CDS_MIGRATOR_KIT_ENV"] == "local": @@ -406,15 +452,21 @@ def _after_publish_mint_recid(self, record, entry, version): # but then we get a double redirection legacy_recid_minter(legacy_recid, record._record.parent.model.id) - def _after_publish_add_inclusion_request(self, request_data, record, entry): + def _after_publish_add_inclusion_request(self, request_data, record, entry, uow): """Create community inclusion request after publish.""" + legacy_recid = entry["record"]["recid"] + request_number = f"lrecid:{legacy_recid}" + + # Defensive/idempotency guard: skip if a request for this record was + # already committed by a previous (partial) load attempt, otherwise + # re-creating it would violate the unique constraint on `number`. + if RequestMetadata.query.filter_by(number=request_number).first() is not None: + return status = request_data.get("status", "accepted") reviewer_names = request_data.get("reviewers", []) - reviewers = User.query.filter(User._displayname.in_(reviewer_names)).all() - # TODO: check if reviewers are missing and log a warning + reviewers = [find_reviewer(name) for name in reviewer_names] - legacy_recid = entry["record"]["recid"] created_at = datetime.datetime.fromisoformat(record["created"]) parent = record._record.parent @@ -438,59 +490,43 @@ def create_event(request_model, payload, event_type, user): uow.register(RecordCommitOp(event, indexer=current_events_service.indexer)) - with UnitOfWork() as uow: - request_item = current_requests_service.create( - system_identity, - data={"title": record["metadata"]["title"]}, - request_type=CommunityInclusion, - receiver=receiver, - creator=creator, - topic={"record": record.id}, - uow=uow, - ) - - request = request_item._record - request.status = "submitted" - request.number = f"lrecid:{legacy_recid}" - request.model.created = created_at - - if reviewers: - reviewers_payload = [{"user": str(r.id)} for r in reviewers] - request.reviewers = reviewers_payload - - for reviewer in reviewers: - create_event( - request.model, - { - "event": "reviewers_updated", - "content": _("added a reviewer"), - "reviewers": [{"user": str(reviewer.id)}], - }, - event_type=ReviewersUpdatedType, - user="system", - ) - - if status: - request.status = status - if status == "accepted": - parent_to_request_relation = ( - parent.communities._m2m_model_cls.query.filter_by( - record_id=parent.id, community_id=community.id - ).one() - ) - parent_to_request_relation.request_id = request.id + request_item = current_requests_service.create( + system_identity, + data={"title": record["metadata"]["title"]}, + request_type=CommunityInclusion, + receiver=receiver, + creator=creator, + topic={"record": record.id}, + uow=uow, + ) - create_event( - request.model, - {"event": status}, - event_type=LogEventType, - user="system", + request = request_item._record + request.status = "submitted" + request.number = request_number + request.model.created = created_at + + if reviewers: + reviewers_payload = [{"user": str(r.id) if r else "-1"} for r in reviewers] + request.reviewers = reviewers_payload + + if status: + request.status = status + if status == "accepted": + parent_to_request_relation = ( + parent.communities._m2m_model_cls.query.filter_by( + record_id=parent.id, community_id=community.id + ).one() ) + parent_to_request_relation.request_id = request.id - uow.register( - RecordCommitOp(request, indexer=current_requests_service.indexer) + create_event( + request.model, + {"event": status}, + event_type=LogEventType, + user="system", ) - uow.commit() + + uow.register(RecordCommitOp(request, indexer=current_requests_service.indexer)) def _after_publish_update_files_created(self, record, entry, version): """Update the created date of the files post publish.""" @@ -520,7 +556,7 @@ def _after_publish(self, identity, published_record, entry, version, uow): if self.create_inclusion_request and request_data: self._after_publish_add_inclusion_request( - request_data, published_record, entry + request_data, published_record, entry, uow ) # db.session.commit() 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 dd23fa21..fce97fb6 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 @@ -49,7 +49,9 @@ def _sub(v, code): def isbn(self, key, value): _custom_fields = self.get("custom_fields", {}) _isbn = StringValue(value.get("a", "")).parse() + _isbn_u = StringValue(value.get("u", "")).parse() + _isbn = _isbn or _isbn_u if _isbn: try: _isbn = normalize_isbn(_isbn) @@ -455,13 +457,17 @@ def organisation(self, key, value): @for_each_value def request_reviewers(self, key, value): name = StringValue(value.get("p", "")).parse().strip() + email = StringValue(value.get("m", "")).parse().strip() - if "," in name: - last, first = (part.strip() for part in name.split(",", 1)) + if email: + reviewer = email else: - last, first = name, "" + if "," in name: + last, first = (part.strip() for part in name.split(",", 1)) + else: + last, first = name, "" - reviewer = " ".join(part for part in (first, last) if part) + reviewer = " ".join(part for part in (first, last) if part) if reviewer: request_data = self.setdefault("request_data", {}) diff --git a/cds_migrator_kit/rdm/streams.yaml b/cds_migrator_kit/rdm/streams.yaml index c12a32ea..9c37262e 100644 --- a/cds_migrator_kit/rdm/streams.yaml +++ b/cds_migrator_kit/rdm/streams.yaml @@ -5,6 +5,7 @@ records: tmp_dir: cds_migrator_kit/rdm/tmp/lep_exp/aleph log_dir: cds_migrator_kit/rdm/log/lep_exp/aleph plots: true + create_inclusion_request: true extract: dirpath: cds_migrator_kit/rdm/data/lep_exp/aleph/dump/ transform: @@ -36,6 +37,7 @@ records: tmp_dir: cds_migrator_kit/rdm/tmp/lep_exp/l3 log_dir: cds_migrator_kit/rdm/log/lep_exp/l3 plots: true + create_inclusion_request: true extract: dirpath: cds_migrator_kit/rdm/data/lep_exp/l3/dump/ transform: @@ -51,6 +53,7 @@ records: tmp_dir: cds_migrator_kit/rdm/tmp/lep_exp/opal log_dir: cds_migrator_kit/rdm/log/lep_exp/opal plots: true + create_inclusion_request: true extract: dirpath: cds_migrator_kit/rdm/data/lep_exp/opal/dump/ transform: @@ -66,6 +69,7 @@ records: tmp_dir: cds_migrator_kit/rdm/tmp/lep_exp/delphi log_dir: cds_migrator_kit/rdm/log/lep_exp/delphi plots: true + create_inclusion_request: true extract: dirpath: cds_migrator_kit/rdm/data/lep_exp/delphi/dump/ transform: @@ -82,6 +86,7 @@ records: log_dir: cds_migrator_kit/rdm/log/lep_exp/delphi_priv restricted: "True" plots: true + create_inclusion_request: true extract: dirpath: cds_migrator_kit/rdm/data/lep_exp/delphi_priv/dump/ transform: From b732264162ffcef9e219a5c75410db2194751741 Mon Sep 17 00:00:00 2001 From: Karolina Przerwa Date: Fri, 17 Jul 2026 16:42:16 +0200 Subject: [PATCH 2/2] fix(submitters): precreate also reviewers accounts if missing --- .../rdm/records/transform/transform.py | 20 ++- cds_migrator_kit/rdm/users/log.py | 46 ++++++ cds_migrator_kit/rdm/users/runner.py | 5 + .../xml_processing/models/submitter.py | 14 +- .../xml_processing/rules/reviewers.py | 52 +++++++ cds_migrator_kit/users/load.py | 131 ++++++++++++++++++ cds_migrator_kit/users/transform.py | 3 +- scripts/opensearch-init.sh | 97 +++++++++++++ scripts/postgres-init.sh | 71 ++++++++++ setup.cfg | 3 + tests/cds-rdm/test_publications_rules.py | 20 ++- 11 files changed, 446 insertions(+), 16 deletions(-) create mode 100644 cds_migrator_kit/rdm/users/log.py create mode 100644 cds_migrator_kit/rdm/users/transform/xml_processing/rules/reviewers.py create mode 100755 scripts/opensearch-init.sh create mode 100755 scripts/postgres-init.sh diff --git a/cds_migrator_kit/rdm/records/transform/transform.py b/cds_migrator_kit/rdm/records/transform/transform.py index 990d5cfc..0d269c88 100644 --- a/cds_migrator_kit/rdm/records/transform/transform.py +++ b/cds_migrator_kit/rdm/records/transform/transform.py @@ -654,17 +654,23 @@ def field_journal(record_json): journal = record_json.get("custom_fields", {}).get("journal:journal", {}) if journal: if not journal.get("title"): - raise UnexpectedValue( - subfield="a", - value=journal, - field="journal", - message="Title is missing in journal field", - stage="vocabulary match", + raise RecordFlaggedCuration( + message="found partial journal field, to be checked", + stage="transform", + field="773", ) return journal return {} _cf = json_entry.get("custom_fields", {}) + try: + journal = field_journal(json_entry) + except RecordFlaggedCuration as e: + self.migration_logger.add_information( + json_entry["recid"], + {"message": e.message, "value": e.value}, + ) + journal = {} custom_fields = { "cern:experiments": [], "cern:departments": [], @@ -678,7 +684,7 @@ def field_journal(record_json): "cern:committees": _cf.get("cern:committees"), "cern:oa_funding_model": _cf.get("cern:oa_funding_model"), "thesis:thesis": _cf.get("thesis:thesis", {}), - "journal:journal": field_journal(json_entry), + "journal:journal": journal, "imprint:imprint": _cf.get("imprint:imprint", {}), "meeting:meeting": _cf.get("meeting:meeting", {}), } diff --git a/cds_migrator_kit/rdm/users/log.py b/cds_migrator_kit/rdm/users/log.py new file mode 100644 index 00000000..69aa0705 --- /dev/null +++ b/cds_migrator_kit/rdm/users/log.py @@ -0,0 +1,46 @@ +# -*- 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 submitter/reviewer accounts migration logger module.""" + +import logging + + +class SubmitterLogger: + """Migrator submitter/reviewer accounts logger.""" + + @classmethod + def initialize(cls, log_dir): + """Attach file and stream handlers to the submitter-migrator logger.""" + formatter = logging.Formatter( + fmt="%(asctime)s %(levelname)-8s %(message)s", datefmt="%Y-%m-%d %H:%M:%S" + ) + logger = logging.getLogger("submitter-migrator") + logger.setLevel(logging.INFO) + + # info and above to file + fh = logging.FileHandler(log_dir / "info.log") + fh.setLevel(logging.INFO) + fh.setFormatter(formatter) + logger.addHandler(fh) + + # errors to their own file + fh = logging.FileHandler(log_dir / "error.log") + fh.setLevel(logging.ERROR) + fh.setFormatter(formatter) + logger.addHandler(fh) + + # info to stream/stdout + sh = logging.StreamHandler() + sh.setFormatter(formatter) + sh.setLevel(logging.INFO) + logger.addHandler(sh) + + @classmethod + def get_logger(cls): + """Get migration logger.""" + return logging.getLogger("submitter-migrator") diff --git a/cds_migrator_kit/rdm/users/runner.py b/cds_migrator_kit/rdm/users/runner.py index a42c9f94..b18bc02d 100644 --- a/cds_migrator_kit/rdm/users/runner.py +++ b/cds_migrator_kit/rdm/users/runner.py @@ -14,6 +14,7 @@ from cds_migrator_kit.rdm.affiliations.log import AffiliationsLogger from cds_migrator_kit.rdm.users.api import CDSMigrationUserAPI +from cds_migrator_kit.rdm.users.log import SubmitterLogger from .transform import people_marc21, users_migrator_marc21 @@ -55,6 +56,9 @@ def __init__(self, stream_definition, dirpath, missing_users_dir, log_dir, dry_r """Constructor.""" self.log_dir = Path(log_dir) self.log_dir.mkdir(parents=True, exist_ok=True) + + SubmitterLogger.initialize(self.log_dir) + self.stream = Stream( stream_definition.name, extract=stream_definition.extract_cls(dirpath), @@ -65,6 +69,7 @@ def __init__(self, stream_definition, dirpath, missing_users_dir, log_dir, dry_r dry_run=dry_run, missing_users_dir=missing_users_dir, user_api_cls=CDSMigrationUserAPI, + logger=SubmitterLogger.get_logger(), ), ) diff --git a/cds_migrator_kit/rdm/users/transform/xml_processing/models/submitter.py b/cds_migrator_kit/rdm/users/transform/xml_processing/models/submitter.py index 689dbfa4..a10886e2 100644 --- a/cds_migrator_kit/rdm/users/transform/xml_processing/models/submitter.py +++ b/cds_migrator_kit/rdm/users/transform/xml_processing/models/submitter.py @@ -19,6 +19,12 @@ """CDS-RDM Summer student model.""" +# Registers the 906__ reviewers rule onto base_model. Must be imported here, +# before SubmitterModel is constructed below: `bases=(base_model,)` takes a +# snapshot of base_model.rules at construction time, and this model's own +# entry_point_group is only collected lazily on first use - too late for a +# rule registered there to make it into that snapshot. +import cds_migrator_kit.rdm.users.transform.xml_processing.rules.reviewers # noqa: E402,F401 from cds_migrator_kit.transform.overdo import CdsOverdo from cds_migrator_kit.transform.xml_processing.models.base import model as base_model @@ -237,7 +243,7 @@ class SubmitterModel(CdsOverdo): "916__z", # issue number (bulletin) "925__a", # https://cds.cern.ch/record/1032351/export/hm?ln=en "925__b", # https://cds.cern.ch/record/1032351/export/hm?ln=en - "906__m", # email info + # "906__m", # reviewer's email, used to find/recreate their account "962__b", # https://cds.cern.ch/record/450847/export/hm?ln=en "962__k", # https://cds.cern.ch/record/450847/export/hm?ln=en "962__n", # https://cds.cern.ch/record/450847/export/hm?ln=en @@ -334,7 +340,7 @@ class SubmitterModel(CdsOverdo): "8564_u", # exclude files but include links (filter by domain) "8564_x", # Files system field "8564_y", # Files - "906__p", # names, is it supervisor? + # "906__p", # reviewer name(s), used to find/recreate their account "916__n", "916__s", "916__w", @@ -356,7 +362,7 @@ class SubmitterModel(CdsOverdo): submitter_model = SubmitterModel( - # use base rules - we don't need any other rule than 859 + # use base rules (859__f submitter) plus the reviewers rules (903__m/p) bases=(base_model,), - entry_point_group="cds_migrator_kit.migrator.rules.base", + entry_point_group="cds_migrator_kit.migrator.rules.submitter", ) diff --git a/cds_migrator_kit/rdm/users/transform/xml_processing/rules/reviewers.py b/cds_migrator_kit/rdm/users/transform/xml_processing/rules/reviewers.py new file mode 100644 index 00000000..b44a24b3 --- /dev/null +++ b/cds_migrator_kit/rdm/users/transform/xml_processing/rules/reviewers.py @@ -0,0 +1,52 @@ +# -*- 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 request reviewer accounts migration rules. + +Mirrors the 859__f "submitter" rule (see +cds_migrator_kit/transform/xml_processing/rules/base.py) which is used to +find/recreate the record owner's account: this extracts the reviewers +listed in 906__m/906__p, later used to find or recreate their accounts. +""" + +from dojson.errors import IgnoreKey + +from cds_migrator_kit.errors import UnexpectedValue +from cds_migrator_kit.transform.xml_processing.models.base import model + + +@model.over("reviewers", "^906__") +def record_reviewer(self, key, value): + """Translate request reviewers. + + 906__m holds the reviewer's email directly, same as 859__f does for + the submitter. 906__p instead holds one or more "Family name, Given + name" strings - multiple reviewers can be packed into a single + occurrence, joined by a literal "\\n" - used as a fallback to later + match or recreate the reviewer's account when no email is available. + """ + reviewers = self.get("reviewers", []) + + email = value.get("m") + if type(email) is tuple: + raise UnexpectedValue(field=key, subfield="m", value=email) + if email: + email = email.strip().lower() + if email not in reviewers: + reviewers.append(email) + + name = value.get("p") + if type(name) is tuple: + raise UnexpectedValue(field=key, subfield="p", value=name) + if name: + for single_name in name.split("\\n"): + single_name = single_name.strip() + if single_name and single_name not in reviewers: + reviewers.append(single_name) + + self["reviewers"] = reviewers + raise IgnoreKey("reviewers") diff --git a/cds_migrator_kit/users/load.py b/cds_migrator_kit/users/load.py index c486c4aa..c8826b89 100644 --- a/cds_migrator_kit/users/load.py +++ b/cds_migrator_kit/users/load.py @@ -14,6 +14,7 @@ from flask import current_app from invenio_accounts.models import User +from invenio_db import db from invenio_rdm_migrator.load.base import Load from sqlalchemy.exc import NoResultFound @@ -44,6 +45,7 @@ def __init__( def _load(self, entry): """Load users.""" self._owner(entry) + self._reviewers(entry) def _validate(self, entry): """Validate data before loading.""" @@ -54,6 +56,26 @@ def _validate(self, entry): def _owner(self, json_entry): """Fetch or create owner.""" email = json_entry.get("submitter") + return self._find_or_create_by_email(email) + + def _reviewers(self, json_entry): + """Fetch or create request reviewer accounts (906__m/906__p). + + 906__m gives an email directly, handled the same way as the + submitter (859__f). 906__p only gives a "Family name, Given name" + string with no email: it is first matched against an existing + account by profile name, then falls back to the same people.csv + lookup used for owners (by name this time), and as a last resort + a CERN-style email is made up so an account can still be created. + """ + for reviewer in json_entry.get("reviewers", []): + if "@" in reviewer: + self._find_or_create_by_email(reviewer) + else: + self._find_or_create_reviewer_by_name(reviewer) + + def _find_or_create_by_email(self, email): + """Fetch or create a user account by email.""" if not email: return try: @@ -64,6 +86,115 @@ def _owner(self, json_entry): user_id = self._create_owner(email) return user_id + def _parse_reviewer_name(self, name): + """Split a "Family name, Given name" or "Given name Family name" + string into (family_name, given_name).""" + if "," in name: + family_name, _, given_name = name.partition(",") + return family_name.strip(), given_name.strip() + parts = name.split() + family_name = parts[-1] if parts else name.strip() + given_name = " ".join(parts[:-1]) + return family_name, given_name + + def _find_reviewer_by_name(self, family_name, given_name): + """Match a reviewer name to an existing account via their profile.""" + query = User.query.filter( + db.func.lower(User._user_profile["family_name"].as_string()) + == family_name.lower() + ) + if given_name: + query = query.filter( + db.func.lower(User._user_profile["given_name"].as_string()) + == given_name.lower() + ) + return query.one_or_none() + + def _find_person_email_by_name(self, family_name, given_name): + """Look up a reviewer's email in the people collection dump by name. + + Same source as _create_owner's `get_person`, but keyed by name + since a name-only reviewer (906__p) has no email to search by. + """ + missing_users_dump = os.path.join( + self.missing_users_dir, self.missing_users_filename + ) + with open(missing_users_dump) as csv_file: + for row in csv.reader(csv_file): + surname, given_names = row[2].strip().lower(), row[3].strip().lower() + if surname != family_name.lower(): + continue + if given_name and given_names != given_name.lower(): + continue + return row[0] + return None + + def _fabricate_email(self, family_name, given_name): + """Make up a plausible CERN email when none can be found anywhere, + following the firstname.lastname@cern.ch convention.""" + given = re.sub(r"[^a-z]", "", given_name.lower()) + family = re.sub(r"[^a-z]", "", family_name.lower()) + local_part = f"{given}.{family}" if given else family + return f"{local_part}@cern.ch" + + def _find_or_create_reviewer_by_name(self, name): + """Resolve a "Family name, Given name" reviewer with no email. + + Tries, in order: (1) an existing account matched by profile name, + (2) an email looked up by name in the people collection dump, and + (3) a fabricated CERN-style email as a last resort, so a reviewer + with no email information can still get an account created - the + same way an owner does, just starting from a name instead. + """ + family_name, given_name = self._parse_reviewer_name(name) + + user = self._find_reviewer_by_name(family_name, given_name) + if user is not None: + return user.id + + if self.dry_run: + return None + + email = self._find_person_email_by_name(family_name, given_name) + if not email: + email = self._fabricate_email(family_name, given_name) + self.logger.warning( + f"Reviewer '{name}' has no email and no matching person was " + f"found in the people collection - using a made-up email: " + f"{email}" + ) + + user_id = self._find_or_create_by_email(email) + if user_id: + self._ensure_reviewer_profile_name(user_id, family_name, given_name) + return user_id + + def _ensure_reviewer_profile_name(self, user_id, family_name, given_name): + """Make sure family_name/given_name are set on the profile. + + MigrationUserAPI.create_user() only ever sets `full_name`, but + find_reviewer() (cds_migrator_kit/rdm/records/load/load.py) matches + reviewers by `family_name`/`given_name` - without this, a + just-created reviewer account would still be unmatchable by name + later on. Only fills in missing values, never overwrites an + existing (e.g. already CERN-synced) profile. + """ + user = User.query.get(user_id) + if user is None: + return + profile = dict(user.user_profile or {}) + changed = False + if not profile.get("family_name"): + profile["family_name"] = family_name + changed = True + if given_name and not profile.get("given_name"): + profile["given_name"] = given_name + changed = True + if changed: + user.user_profile = profile + db.session.add(user) + db.session.commit() + def _create_owner(self, email_addr): """Create owner from legacy data. diff --git a/cds_migrator_kit/users/transform.py b/cds_migrator_kit/users/transform.py index 150e0139..ec38a06d 100644 --- a/cds_migrator_kit/users/transform.py +++ b/cds_migrator_kit/users/transform.py @@ -37,7 +37,8 @@ def _transform(self, entry): timestamp, json_data = record_dump.latest_revision email = json_data.get("submitter") - return {"submitter": email} + reviewers = json_data.get("reviewers", []) + return {"submitter": email, "reviewers": reviewers} except Exception as e: cli_logger.exception(e) diff --git a/scripts/opensearch-init.sh b/scripts/opensearch-init.sh new file mode 100755 index 00000000..4b97766e --- /dev/null +++ b/scripts/opensearch-init.sh @@ -0,0 +1,97 @@ +#!/usr/bin/env bash +# Dump/restore OpenSearch index *data* to/from local JSON files. +# +# Mappings/settings/aliases are NOT dumped or restored from file: they are +# defined by the Invenio code (index templates) and are recreated with +# `invenio index init`, which is the source of truth - like a DB schema +# migration. Restoring a stale dumped mapping instead has caused field-type +# mismatches (e.g. a keyword field coming back as text), breaking +# aggregations/facets with a "Misconfigured search" error. +# +# Usage: +# scripts/opensearch-init.sh dump [output_dir] +# scripts/opensearch-init.sh restore [input_dir] +set -euo pipefail + +HOST="${OPENSEARCH_HOST:-http://127.0.0.1:9200}" +ACTION="${1:-}" +DIR="${2:-dumps}" + +# Dump filenames look like "-.json", +# where is a long numeric id multielasticdump appends per run +# (not part of the real index/alias name). Physical index names get a new +# numeric suffix each time `invenio index init` (re)creates them, so on +# restore we resolve each dump file back to whatever alias it belongs to, +# then to whichever real index that alias currently points to. +SKIP_PATTERNS='-percolators\.json$|^top_queries-' + +usage() { + echo "Usage: $0 {dump|restore} [dir]" >&2 + exit 1 +} + +[[ "$ACTION" == "dump" || "$ACTION" == "restore" ]] || usage + +if ! command -v multielasticdump &>/dev/null; then + echo "multielasticdump not found. Install it with: npm install -g multielasticdump" >&2 + exit 1 +fi +if ! command -v elasticdump &>/dev/null; then + echo "elasticdump not found. Install it with: npm install -g elasticdump" >&2 + exit 1 +fi + +resolve_real_index() { + local alias_guess="$1" + curl -s "$HOST/_alias/$alias_guess" | python3 -c " +import json, sys +try: + d = json.load(sys.stdin) + print(list(d.keys())[0]) +except Exception: + print('') +" +} + +case "$ACTION" in + dump) + mkdir -p "$DIR" + echo "Dumping index data (no mapping/settings/alias) from $HOST to $DIR ..." + multielasticdump --direction=dump --input="$HOST" --output="$DIR" --includeType=data + ;; + restore) + [[ -d "$DIR" ]] || { echo "Dump directory not found: $DIR" >&2; exit 1; } + + echo "Rebuilding indices from current Invenio mappings ..." + invenio index destroy --force --yes-i-know + invenio index init + + echo "Restoring data from $DIR into the freshly-mapped indices ..." + cd "$DIR" + for f in *.json; do + [[ -e "$f" ]] || continue + grep -qE "$SKIP_PATTERNS" <<<"$f" && { echo "Skipping $f (not alias-managed)"; continue; } + [[ "$(stat -f%z "$f" 2>/dev/null || stat -c%s "$f")" -eq 0 ]] && continue + + base="${f%.json}" + alias_guess="$(sed -E 's/-[0-9]{6,}$//' <<<"$base")" + real_index="$(resolve_real_index "$alias_guess")" + + if [[ -z "$real_index" ]]; then + # Not an aliased index (e.g. stats/events indices use a literal + # name) - fall back to the same name if it already exists. + if curl -sf -o /dev/null "$HOST/$alias_guess"; then + real_index="$alias_guess" + else + echo "Skipping $f (no matching alias or index for $alias_guess)" + continue + fi + fi + + echo "=== Restoring $f -> $real_index ===" + elasticdump --input="$f" --output="$HOST/$real_index" --type=data + done + ;; +esac + +echo "Done." diff --git a/scripts/postgres-init.sh b/scripts/postgres-init.sh new file mode 100755 index 00000000..c92a0788 --- /dev/null +++ b/scripts/postgres-init.sh @@ -0,0 +1,71 @@ +#!/usr/bin/env bash +# Dump/restore the Postgres database to/from a local SQL file. +# Auto-detects the running Postgres container - override with DB_CONTAINER +# if you have more than one running. +# Usage: +# scripts/postgres-init.sh dump [dump_file] +# scripts/postgres-init.sh restore +set -euo pipefail + +DB_USER="${POSTGRES_USER:-cds-rdm}" +DB_NAME="${POSTGRES_DB:-cds-rdm}" +ACTION="${1:-}" +DUMP_FILE="${2:-}" + +usage() { + echo "Usage: $0 {dump|restore} " >&2 + exit 1 +} + +[[ "$ACTION" == "dump" || "$ACTION" == "restore" ]] || usage + +if [[ "$ACTION" == "dump" && -z "$DUMP_FILE" ]]; then + DUMP_FILE="dump_$(date +%Y%m%d_%H%M%S).sql" +fi +[[ -n "$DUMP_FILE" ]] || usage + +detect_container() { + local matches + # Match by image name prefix (e.g. "postgres:14", "postgres:16-alpine"), + # not docker's --filter ancestor=, which requires an exact repo:tag match. + matches="$(docker ps --format '{{.ID}}\t{{.Image}}\t{{.Names}}' | awk -F'\t' '$2 ~ /^postgres(:|$)/')" + if [[ -z "$matches" ]]; then + echo "No running Postgres container found (looked for image 'postgres*')." >&2 + echo "Start it, or set DB_CONTAINER explicitly." >&2 + exit 1 + fi + if [[ "$(echo "$matches" | wc -l)" -gt 1 ]]; then + echo "Multiple Postgres containers found, using the first one:" >&2 + echo "$matches" >&2 + echo "Set DB_CONTAINER to pick a different one." >&2 + fi + echo "$matches" | head -n1 | cut -f1 +} + +CONTAINER_ID="${DB_CONTAINER:-$(detect_container)}" +CONTAINER_NAME="$(docker inspect --format '{{.Name}}' "$CONTAINER_ID" 2>/dev/null | sed 's|^/||')" +echo "Using Postgres container: $CONTAINER_ID (${CONTAINER_NAME:-unknown})" + +case "$ACTION" in + dump) + echo "Dumping database '$DB_NAME' to $DUMP_FILE ..." + docker exec "$CONTAINER_ID" pg_dump -U "$DB_USER" -d "$DB_NAME" >"$DUMP_FILE" + ;; + restore) + [[ -f "$DUMP_FILE" ]] || { echo "Dump file not found: $DUMP_FILE" >&2; exit 1; } + + if [[ "${FORCE:-}" != "1" ]]; then + read -r -p "This will DESTROY the current database '$DB_NAME' and restore it from $DUMP_FILE. Continue? [y/N] " reply + [[ "$reply" =~ ^[Yy]$ ]] || { echo "Aborted."; exit 1; } + fi + + echo "Recreating database schema ..." + invenio db destroy --yes-i-know + invenio db init + + echo "Restoring $DUMP_FILE into database '$DB_NAME' ..." + docker exec -i "$CONTAINER_ID" psql -U "$DB_USER" -d "$DB_NAME" <"$DUMP_FILE" + ;; +esac + +echo "Done." diff --git a/setup.cfg b/setup.cfg index 9afe9c47..42f92ef1 100644 --- a/setup.cfg +++ b/setup.cfg @@ -92,6 +92,9 @@ cds_migrator_kit.migrator.models = faser_publication = cds_migrator_kit.rdm.records.transform.models.faser_publication:faser_publication_model cds_migrator_kit.migrator.rules.base = base = cds_migrator_kit.transform.xml_processing.rules.base +cds_migrator_kit.migrator.rules.submitter = + base = cds_migrator_kit.transform.xml_processing.rules.base + reviewers = cds_migrator_kit.rdm.users.transform.xml_processing.rules.reviewers cds_migrator_kit.migrator.rdm.rules.base = base = cds_migrator_kit.rdm.records.transform.xml_processing.rules.base cds_migrator_kit.migrator.rdm.rules.publication = diff --git a/tests/cds-rdm/test_publications_rules.py b/tests/cds-rdm/test_publications_rules.py index c2bcf084..7fcf4dd7 100644 --- a/tests/cds-rdm/test_publications_rules.py +++ b/tests/cds-rdm/test_publications_rules.py @@ -565,7 +565,9 @@ def test_962_title_appended_when_no_matching_session(self): assert meetings == [{"title": "unrelated meeting"}] def test_962_book_with_different_artid_is_not_duplicate(self): - """A mismatched book leaves titleless journal data rejected by transform.""" + """A mismatched book leaves titleless journal data flagged for curation.""" + from unittest.mock import MagicMock + from cds_migrator_kit.rdm.records.transform.transform import ( CDSToRDMRecordEntry, ) @@ -573,7 +575,7 @@ def test_962_book_with_different_artid_is_not_duplicate(self): related_identifiers, ) - record = {"custom_fields": {}} + record = {"custom_fields": {}, "recid": 123456} record["custom_fields"] = journal( record, "773__", @@ -590,8 +592,18 @@ def test_962_book_with_different_artid_is_not_duplicate(self): assert record["custom_fields"]["meeting:meeting"] == [] record["resource_type"] = "publication-other" - with pytest.raises(UnexpectedValue, match="Title is missing in journal field"): - CDSToRDMRecordEntry()._custom_fields(record, {"metadata": {}}) + migration_logger = MagicMock() + custom_fields = CDSToRDMRecordEntry( + migration_logger=migration_logger + )._custom_fields(record, {"metadata": {}}) + + # titleless journal data is dropped (falsy values are filtered out), + # not raised as a hard error + assert "journal:journal" not in custom_fields + migration_logger.add_information.assert_called_once() + recid, info = migration_logger.add_information.call_args[0] + assert recid == 123456 + assert "found partial journal field" in info["message"] def test_matching_962_removes_temporary_773_c(self): """Matching 962__k consumes the temporary value stored from 773__c."""