fix harvester document type splitting

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,19 +25,114 @@ 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."""
 
     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),
+        ]

site/cds_rdm/inspire_harvester/transform/splitter.py

@@ -1,17 +1,13 @@
 """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 (
     INSPIRE_DOCUMENT_TYPE_MAPPING,
 )
 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,

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