Multiple doc types

invenio.cfg

@@ -112,6 +112,9 @@ TRUSTED_HOSTS = ['0.0.0.0', 'localhost', '127.0.0.1', 'localhost.cern.ch']
 
 # TODO: Set
 SQLALCHEMY_DATABASE_URI = "postgresql+psycopg2://cds-rdm:cds-rdm@localhost/cds-rdm"
+SQLALCHEMY_ENGINE_OPTIONS={
+  "connect_args": {"options": "-c timezone=UTC"}
+}
 
 # Invenio-App
 # ===========

site/cds_rdm/inspire_harvester/__init__.py

@@ -5,4 +5,4 @@
 # 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.
 
-"""INSPIRE to CDS harvester context module."""
+"""INSPIRE to CDS harvester context module."""

site/cds_rdm/inspire_harvester/load/__init__.py

@@ -0,0 +1,8 @@
+# -*- 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.
+
+"""INSPIRE to CDS harvester context module."""

site/cds_rdm/inspire_harvester/load/draft.py

@@ -0,0 +1,76 @@
+# -*- 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.
+
+"""Draft lifecycle management module."""
+
+from flask import current_app
+from invenio_access.permissions import system_identity
+from invenio_db import db
+from invenio_rdm_records.proxies import current_rdm_records_service
+from invenio_rdm_records.services.errors import ValidationErrorWithMessageAsList
+from invenio_vocabularies.datastreams.errors import WriterError
+from marshmallow import ValidationError
+
+
+class DraftLifecycleManager:
+    """Manages draft creation, editing, versioning, and publishing."""
+
+    def create(self, entry):
+        """Create a new draft from entry data."""
+        return current_rdm_records_service.create(system_identity, data=entry)
+
+    def edit(self, record_pid):
+        """Open an edit draft for an existing published record."""
+        return current_rdm_records_service.edit(system_identity, record_pid)
+
+    def update(self, draft, metadata):
+        """Update draft."""
+        return current_rdm_records_service.update_draft(
+            system_identity, draft.id, metadata
+        )
+
+    def new_version(self, record_pid):
+        """Create a new-version draft from an existing published record."""
+        return current_rdm_records_service.new_version(system_identity, record_pid)
+
+    def add_community(self, draft):
+        """Add the CERN Scientific Community to the draft."""
+        with db.session.begin_nested():
+            community_id = current_app.config["CDS_CERN_SCIENTIFIC_COMMUNITY_ID"]
+            draft_obj = current_rdm_records_service.draft_cls.pid.resolve(
+                draft.id, registered_only=False
+            )
+            draft_obj.parent.communities.add(community_id)
+            draft_obj.parent.communities.default = community_id
+            draft_obj.parent.commit()
+
+    def publish(self, draft_id, logger):
+        """Publish a draft. Deletes the draft on any failure, then raises WriterError."""
+        try:
+            logger.debug(f"Publishing draft {draft_id}")
+            current_rdm_records_service.publish(system_identity, draft_id)
+            logger.info(f"Draft {draft_id} published successfully.")
+        except ValidationError as e:
+            logger.error(
+                f"Failure: draft {draft_id} not published, validation errors: {e}."
+            )
+            current_rdm_records_service.delete_draft(system_identity, draft_id)
+            raise WriterError(
+                f"Failure: draft {draft_id} not published, validation errors: {e}."
+            )
+        except ValidationErrorWithMessageAsList as e:
+            current_rdm_records_service.delete_draft(system_identity, draft_id)
+            raise WriterError(
+                f"Failure: draft {draft_id} not published,"
+                f" validation errors: {e.messages}."
+            )
+        except Exception as e:
+            current_rdm_records_service.delete_draft(system_identity, draft_id)
+            raise WriterError(
+                f"Draft {draft_id} failed publishing"
+                f" because of an unexpected error: {str(e)}."
+            )

site/cds_rdm/inspire_harvester/load/files.py

@@ -0,0 +1,210 @@
+# -*- 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.
+
+"""File synchronization module."""
+
+import time
+from dataclasses import dataclass
+from io import BytesIO
+from typing import List
+
+import requests
+from invenio_access.permissions import system_identity
+from invenio_rdm_records.proxies import current_rdm_records_service
+from invenio_records_resources.services.errors import FileKeyNotFoundError
+from invenio_vocabularies.datastreams.errors import WriterError
+
+
+@dataclass
+class RetryConfig:
+    """Configuration for file fetch retries."""
+
+    max_retries: int = 3
+    retry_delay: int = 60  # seconds; only applied on network exceptions
+
+
+@dataclass
+class FileDiff:
+    """Diff between existing and new file sets, keyed by checksum."""
+
+    to_add: List[str]  # checksums of new files to upload
+    to_delete: List[str]  # checksums of files to remove
+    existing: List[str]
+
+
+class FileSynchronizer:
+    """Handles file I/O, diffing, uploading, and deletion for draft records."""
+
+    def __init__(self, retry_config: RetryConfig = None):
+        """Constructor."""
+        self.retry_config = retry_config or RetryConfig()
+
+    def compute_diff(self, existing_files, new_files) -> FileDiff:
+        """Return the set difference between existing and new file checksums."""
+        existing_checksums = [value["checksum"] for value in existing_files.values()]
+        new_checksums = [value["checksum"] for value in new_files.values()]
+
+        return FileDiff(
+            to_add=list(set(new_checksums) - set(existing_checksums)),
+            to_delete=list(set(existing_checksums) - set(new_checksums)),
+            existing=list(set(existing_checksums)),
+        )
+
+    def fetch(self, url, logger) -> BytesIO:
+        """Fetch file content from URL.
+
+        Raises WriterError after exhausting retries.
+        """
+        max_retries = self.retry_config.max_retries
+        retry_delay = self.retry_config.retry_delay
+
+        logger.debug(f"File URL: {url}")
+        attempt = 0
+        while attempt < max_retries:
+            attempt += 1
+            try:
+                logger.debug(
+                    f"Attempt {attempt}/{max_retries} - HEAD request to: {url}"
+                )
+                head = requests.head(url, allow_redirects=True)
+                resolved_url = head.url
+                logger.info(f"Get file, URL: {resolved_url}.")
+                response = requests.get(resolved_url, stream=True)
+                logger.debug(f"Response status code: {response.status_code}")
+                if response.status_code == 200:
+                    logger.debug("Success: File retrieved.")
+                    return BytesIO(response.content)
+                else:
+                    logger.warning(
+                        f"Retrieving file request failed. "
+                        f"Attempt {attempt}/{max_retries} "
+                        f"Error {response.status_code}."
+                        f" URL: {resolved_url}."
+                    )
+            except Exception as e:
+                logger.warning(
+                    f"Attempt {attempt}/{max_retries} failed with exception: {e}"
+                )
+                logger.debug("Retrying in 1 minute...")
+                time.sleep(retry_delay)
+
+        logger.error(
+            f"Retrieving file request failed. Max retries {max_retries} reached."
+            f" URL: {url}."
+        )
+        raise WriterError(
+            f"Failed to fetch file from {url} after {max_retries} retries."
+        )
+
+    def check_files_should_update(self, record, incoming_record, logger):
+        """Check if files should be updated."""
+        if not record:
+            return True
+        record_dict = record.to_dict()
+        existing_files = record_dict["files"]["entries"]
+        new_files = incoming_record["files"].get("entries", {})
+        logger.info(
+            f"Existing files count: {len(existing_files)},"
+            f" New files count: {len(new_files)}"
+        )
+
+        diff = self.compute_diff(existing_files, new_files)
+        logger.debug(f"Existing files' checksums: {diff.existing}.")
+        logger.debug(f"New files' checksums: {diff.to_add}.")
+        should_update_files = bool(new_files) and bool(diff.to_add or diff.to_delete)
+
+        return should_update_files
+
+    def sync(self, draft, record, incoming_record, logger, import_files=True):
+        """Sync files on a draft: delete removed files, upload added files."""
+        existing_files = {}
+        if not record:
+            should_import_files = False
+        else:
+            should_import_files = (
+                record
+                and import_files
+                and record.data.get("files", {}).get("enabled", False)
+            )
+            record_dict = record.to_dict()
+            existing_files = record_dict["files"]["entries"]
+        if should_import_files:
+            current_rdm_records_service.import_files(system_identity, draft.id)
+            logger.debug(
+                f"Imported files to {draft.id} from previous version: {record.id}"
+            )
+
+        new_files = incoming_record["files"].get("entries", {})
+        diff = self.compute_diff(existing_files, new_files)
+        should_update_files = bool(new_files) and bool(diff.to_add or diff.to_delete)
+        if should_update_files:
+            for filename, file_data in existing_files.items():
+                if file_data["checksum"] in diff.to_delete:
+                    logger.debug(f"Delete file: {filename}")
+                    current_rdm_records_service.draft_files.delete_file(
+                        system_identity, draft.id, filename
+                    )
+
+            logger.info(f"{len(diff.to_delete)} files successfully deleted.")
+
+            logger.debug("Creating new files")
+            for key, file in new_files.items():
+                if file["checksum"] in diff.to_add:
+                    logger.debug(f"Processing new file: {key}")
+                    inspire_url = file.get("source_url")
+                    file_content = self.fetch(inspire_url, logger)
+                    self._upload_file(draft, file, file_content, logger)
+            logger.info(f"{len(new_files)} files successfully created.")
+
+    def _upload_file(self, draft, file_data, file_content, logger):
+        """Initialize, upload, and commit a single file to the draft."""
+        logger.debug(f"Filename: '{file_data['key']}'.")
+        service = current_rdm_records_service
+        inspire_checksum = file_data["checksum"]
+        new_checksum = None
+
+        try:
+            if inspire_checksum is None:
+                # this can happen when we get the file directly from arxiv.
+                # unfortunately, arxiv does not expose checksums
+                del file_data["checksum"]
+            file_data_to_init = {
+                k: v for k, v in file_data.items() if k != "source_url"
+            }
+            service.draft_files.init_files(
+                system_identity, draft.id, [file_data_to_init]
+            )
+            logger.debug(f"Filename: '{file_data['key']}' initialized successfully.")
+
+            service.draft_files.set_file_content(
+                system_identity, draft.id, file_data["key"], file_content
+            )
+            logger.debug(
+                f"Filename: '{file_data['key']}' content set successfully. Commit file..."
+            )
+
+            result = service.draft_files.commit_file(
+                system_identity, draft.id, file_data["key"]
+            )
+            new_checksum = result.data["checksum"]
+            logger.debug(
+                f"Filename: '{file_data['key']}' committed."
+                f" File checksum: {result.data['checksum']}."
+            )
+
+            if inspire_checksum:
+                assert inspire_checksum == new_checksum
+        except AssertionError:
+            logger.error(
+                f"Files checksums don't match."
+                f" Delete file: '{file_data['key']}' from draft."
+            )
+            service.draft_files.delete_file(system_identity, draft.id, file_data["key"])
+            raise WriterError(
+                f"File {file_data['key']} checksum mismatch."
+                f" Expected: {inspire_checksum}, got: {new_checksum}."
+            )