diff --git a/site/cds_rdm/inspire_harvester/load/matcher.py b/site/cds_rdm/inspire_harvester/load/matcher.py index 105f6036..12ac2e15 100644 --- a/site/cds_rdm/inspire_harvester/load/matcher.py +++ b/site/cds_rdm/inspire_harvester/load/matcher.py @@ -7,7 +7,6 @@ """Record matching module.""" -from collections import OrderedDict from dataclasses import dataclass, field from typing import List, Optional @@ -26,6 +25,99 @@ class MatchResult: matched_ids: List[str] = field(default_factory=list) +@dataclass(frozen=True) +class FilterCandidate: + """Search filter tried as part of the record-matching priority chain.""" + + value: Optional[str] + + @property + def query(self) -> List[dsl.Q]: + """Build the search query for this candidate.""" + raise NotImplementedError + + +@dataclass(frozen=True) +class ParentMatchFilter(FilterCandidate): + """Match the CDS-RDM parent identifier supplied by INSPIRE.""" + + @property + def query(self): + """Build the parent identifier query.""" + return [dsl.Q("term", **{"parent.id": self.value})] + + +@dataclass(frozen=True) +class CDSIdentifierMatchFilter(FilterCandidate): + """Match a legacy CDS identifier.""" + + @property + def query(self): + """Build the legacy CDS identifier query.""" + return [ + dsl.Q("term", **{"metadata.identifiers.scheme": "cds"}), + dsl.Q("term", **{"metadata.identifiers.identifier": self.value}), + ] + + +@dataclass(frozen=True) +class DOIMatchFilter(FilterCandidate): + """Match a DOI.""" + + @property + def query(self): + """Build the DOI query.""" + return [dsl.Q("term", **{"pids.doi.identifier.keyword": self.value})] + + +@dataclass(frozen=True) +class InspireIdentifierMatchFilter(FilterCandidate): + """Match an INSPIRE identifier.""" + + @property + def query(self): + """Build the INSPIRE identifier query.""" + return [ + dsl.Q("term", **{"metadata.related_identifiers.scheme": "inspire"}), + dsl.Q( + "term", + **{"metadata.related_identifiers.identifier": self.value}, + ), + ] + + +@dataclass(frozen=True) +class ArxivIdentifierMatchFilter(FilterCandidate): + """Match an arXiv identifier.""" + + @property + def query(self): + """Build the arXiv identifier query.""" + return [ + dsl.Q("term", **{"metadata.related_identifiers.scheme": "arxiv"}), + dsl.Q( + "term", + **{"metadata.related_identifiers.identifier": self.value}, + ), + ] + + +@dataclass(frozen=True) +class ReportNumberMatchFilter(FilterCandidate): + """Match a CDS report number.""" + + @property + def query(self): + """Build the CDS report number query.""" + return [ + dsl.Q("term", **{"metadata.related_identifiers.scheme": "cdsrn"}), + dsl.Q( + "term", + **{"metadata.related_identifiers.identifier": self.value}, + ), + ] + + class RecordMatcher: """Finds existing CDS records that match an incoming INSPIRE entry.""" @@ -33,12 +125,14 @@ def match(self, stream_entry, inspire_id, logger) -> MatchResult: """Search for existing records using a priority-ordered filter chain.""" entry = stream_entry.entry ctx = entry["_inspire_ctx"] - filters_priority = self._build_filter_priority(entry, inspire_id, ctx["cds_id"]) + filter_candidates = self._build_filter_priority( + entry, inspire_id, ctx["cds_id"] + ) result = None - for filter_key, filter_data in filters_priority.items(): - if filter_data["value"]: - combined_filter = dsl.Q("bool", filter=filter_data["filter"]) - logger.debug(f"Searching for existing records: {filter_data['filter']}") + for candidate in filter_candidates: + if candidate.value: + combined_filter = dsl.Q("bool", filter=candidate.query) + logger.debug(f"Searching for existing records: {candidate.query}") result = current_rdm_records_service.search( system_identity, extra_filter=combined_filter ) @@ -66,59 +160,19 @@ def _retrieve_identifier(self, identifiers, scheme) -> Optional[str]: None, ) - def _build_filter_priority(self, entry, inspire_id, cdsrdm_id) -> OrderedDict: - """Build ordered filter dict for priority-based record lookup.""" + def _build_filter_priority(self, entry, inspire_id, cdsrdm_id): + """Build the priority-ordered record match candidates.""" doi = entry.get("pids", {}).get("doi", {}).get("identifier") related_identifiers = entry["metadata"].get("related_identifiers", []) cds_id = self._retrieve_identifier(related_identifiers, "cds") arxiv_id = self._retrieve_identifier(related_identifiers, "arxiv") report_number = self._retrieve_identifier(related_identifiers, "cdsrn") - return OrderedDict( - cds_pid={ - # INSPIRE stores parent PID - "filter": [dsl.Q("term", **{"parent.id": cdsrdm_id})], - "value": cdsrdm_id, - }, - cds_identifiers={ - "filter": [ - dsl.Q("term", **{"metadata.identifiers.scheme": "cds"}), - dsl.Q("term", **{"metadata.identifiers.identifier": cds_id}), - ], - "value": cds_id, - }, - doi={ - "filter": [dsl.Q("term", **{"pids.doi.identifier.keyword": doi})], - "value": doi, - }, - inspire_id={ - "filter": [ - dsl.Q("term", **{"metadata.related_identifiers.scheme": "inspire"}), - dsl.Q( - "term", - **{"metadata.related_identifiers.identifier": inspire_id}, - ), - ], - "value": inspire_id, - }, - arxiv_filters={ - "filter": [ - dsl.Q("term", **{"metadata.related_identifiers.scheme": "arxiv"}), - dsl.Q( - "term", - **{"metadata.related_identifiers.identifier": arxiv_id}, - ), - ], - "value": arxiv_id, - }, - report_number_filters={ - "filter": [ - dsl.Q("term", **{"metadata.related_identifiers.scheme": "cdsrn"}), - dsl.Q( - "term", - **{"metadata.related_identifiers.identifier": report_number}, - ), - ], - "value": report_number, - }, - ) + return [ + ParentMatchFilter(value=cdsrdm_id), + CDSIdentifierMatchFilter(value=cds_id), + DOIMatchFilter(value=doi), + InspireIdentifierMatchFilter(value=inspire_id), + ArxivIdentifierMatchFilter(value=arxiv_id), + ReportNumberMatchFilter(value=report_number), + ] diff --git a/site/cds_rdm/inspire_harvester/transform/splitter.py b/site/cds_rdm/inspire_harvester/transform/splitter.py index 25e36c5e..b1286b62 100644 --- a/site/cds_rdm/inspire_harvester/transform/splitter.py +++ b/site/cds_rdm/inspire_harvester/transform/splitter.py @@ -1,8 +1,6 @@ """Splits a multi-doc-type INSPIRE record into per-source sub-records.""" -from copy import deepcopy - -from cds_rdm.inspire_harvester.logger import Logger, hlog +from cds_rdm.inspire_harvester.logger import Logger from cds_rdm.inspire_harvester.transform.config import mapper_policy from cds_rdm.inspire_harvester.transform.context import MetadataSerializationContext from cds_rdm.inspire_harvester.transform.resource_types import ( @@ -10,8 +8,6 @@ ) from cds_rdm.inspire_harvester.utils import assert_unique_ids, deep_merge_all -# Doc types that belong to the arXiv/preprint stream -_PREPRINT_DOC_TYPES = frozenset({"report", "note", "activity report"}) _ARXIV_SOURCES = {"arxiv"} @@ -67,11 +63,15 @@ def split(self): meta = self.inspire_record["metadata"] doc_types = meta.get("document_type", []) + resource_types = set() for doc_type in doc_types: self.logger.debug(f"Mapping {doc_type} to version.") resource_type = INSPIRE_DOCUMENT_TYPE_MAPPING[doc_type] self.logger.info(f"Mapped {doc_type} to {resource_type}.") + if resource_type in resource_types: + continue + resource_types.add(resource_type) if resource_type is not self.main_res_type: version_ctx = MetadataSerializationContext( resource_type=resource_type, diff --git a/site/tests/inspire_harvester/test_transformer.py b/site/tests/inspire_harvester/test_transformer.py index c208130e..82eb6bbf 100644 --- a/site/tests/inspire_harvester/test_transformer.py +++ b/site/tests/inspire_harvester/test_transformer.py @@ -37,6 +37,7 @@ RelatedIdentifiersMapper, ) from cds_rdm.inspire_harvester.transform.resource_types import ResourceType +from cds_rdm.inspire_harvester.transform.splitter import InspireVersionSplitter from cds_rdm.inspire_harvester.transform.transform_entry import Inspire2RDM @@ -233,6 +234,34 @@ def test_transform_document_type_multiple(running_app): assert len(errors) == 0 +@patch("cds_rdm.inspire_harvester.transform.splitter.Logger") +def test_splitter_creates_one_version_for_duplicate_resource_types(mock_logger): + """Test document types mapping to the same resource create one version.""" + record = { + "id": "12345", + "metadata": { + "document_type": ["thesis", "report", "activity report"], + "documents": [{"source": "arxiv"}, {"source": "publisher"}], + }, + } + ctx = MetadataSerializationContext( + resource_type=ResourceType.THESIS, inspire_id="12345" + ) + mapper = Mock(id="resource-type") + mapper.apply.side_effect = lambda record, ctx, logger: { + "metadata": {"resource_type": {"id": ctx.resource_type.value}} + } + policy = Mock() + policy.build_for.return_value = [mapper] + + versions = InspireVersionSplitter(record, ctx, None, policy=policy).split() + + assert versions == [ + {"metadata": {"resource_type": {"id": ResourceType.REPORT.value}}} + ] + policy.build_for.assert_called_once_with(ResourceType.REPORT) + + def test_transform_document_type_unmapped(running_app): """Test ResourceTypeDetector with unmapped type.""" from cds_rdm.inspire_harvester.transform.resource_types import ResourceTypeDetector