Create more elaborate base class for curation plugins

hermes.toml

@@ -8,6 +8,9 @@ sources = [ "cff", "toml", "file_exists" ] # ordered priority (first one is most
 [harvest.file_exists.search_patterns]
 community = ["contributing.md", "governance.md"]
 
+[curate]
+method = "accept"
+
 [deposit]
 target = "invenio_rdm"
 

pyproject.toml

@@ -59,6 +59,9 @@ cff = "hermes.commands.harvest.cff:CffHarvestPlugin"
 codemeta = "hermes.commands.harvest.codemeta:CodeMetaHarvestPlugin"
 file_exists = "hermes.commands.harvest.file_exists:FileExistsHarvestPlugin"
 
+[project.entry-points."hermes.curate"]
+accept = "hermes.commands.curate.accept:AcceptCuratePlugin"
+
 [project.entry-points."hermes.deposit"]
 file = "hermes.commands.deposit.file:FileDepositPlugin"
 invenio = "hermes.commands.deposit.invenio:InvenioDepositPlugin"

src/hermes/commands/base.py

@@ -162,7 +162,10 @@ def __call__(self, args: argparse.Namespace):
 
 
 class HermesPlugin(abc.ABC):
-    """Base class for all HERMES plugins."""
+    """Base class for all HERMES plugins.
+
+    Objects of this class are callables.
+    """
 
     settings_class: Optional[Type] = None
 

src/hermes/commands/curate/accept.py

@@ -0,0 +1,36 @@
+# SPDX-FileCopyrightText: 2022 German Aerospace Center (DLR), 2025 Helmholtz-Zentrum Dresden-Rossendorf (HZDR)
+#
+# SPDX-License-Identifier: Apache-2.0
+
+# SPDX-FileContributor: Michael Meinel
+# SPDX-FileContributor: David Pape
+
+import os
+import shutil
+
+from hermes.commands.curate.base import BaseCuratePlugin
+
+
+class AcceptCuratePlugin(BaseCuratePlugin):
+    """Accept plugin for the curation step.
+
+    This plugin creates a positive curation result, i.e. it accepts the produced
+    metadata as correct and lets the execution continue without human intervention. It
+    also copies the metadata produced in the process step to the "curate" directory.
+    """
+
+    def is_publication_approved(self):
+        """Simulate positive curation result."""
+        return True
+
+    def process_decision_positive(self):
+        """In case of positive curation result, copy files to next step."""
+        process_output = (
+            self.ctx.hermes_dir / "process" / (self.ctx.hermes_name + ".json")
+        )
+
+        os.makedirs(self.ctx.hermes_dir / "curate", exist_ok=True)
+        shutil.copy(
+            process_output,
+            self.ctx.hermes_dir / "curate" / (self.ctx.hermes_name + ".json"),
+        )

src/hermes/commands/curate/base.py

@@ -1,28 +1,105 @@
-# SPDX-FileCopyrightText: 2022 German Aerospace Center (DLR)
+# SPDX-FileCopyrightText: 2022 German Aerospace Center (DLR), 2025 Helmholtz-Zentrum Dresden-Rossendorf (HZDR)
 #
 # SPDX-License-Identifier: Apache-2.0
 
 # SPDX-FileContributor: Michael Meinel
+# SPDX-FileContributor: David Pape
 
+from abc import abstractmethod
 import argparse
-import os
-import shutil
+import json
 import sys
 
 from pydantic import BaseModel
 
-from hermes.commands.base import HermesCommand
+from hermes.commands.base import HermesCommand, HermesPlugin
 from hermes.model.context import CodeMetaContext
+from hermes.model.errors import HermesValidationError
+from hermes.model.path import ContextPath
 
 
 class _CurateSettings(BaseModel):
-    """Generic deposition settings."""
+    """Generic curation settings."""
 
-    pass
+    #: Parameter by which the plugin is selected. By default, the accept plugin is used.
+    method: str = "accept"
+
+
+class BaseCuratePlugin(HermesPlugin):
+    """Base class for curation plugins."""
+
+    def __init__(self, command, ctx):
+        self.command = command
+        self.ctx = ctx
+
+    def __call__(self, command: HermesCommand) -> None:
+        """Entry point of the callable.
+
+        This method runs the main logic of the plugin. It calls the other methods of the
+        object in the correct order. Depending on the result of
+        ``is_publication_approved`` the corresponding ``process_decision_*()`` method is
+        called, based on the curation decision.
+        """
+        self.prepare()
+        self.validate()
+        self.create_report()
+        if self.is_publication_approved():
+            self.process_decision_positive()
+        else:
+            self.process_decision_negative()
+
+    def prepare(self):
+        """Prepare the plugin.
+
+        This method may be used to perform preparatory tasks such as configuration
+        checks, token permission checks, loading of resources, etc.
+        """
+        pass
+
+    def validate(self):
+        """Validate the metadata.
+
+        This method performs the validation of the metadata from the data model.
+        """
+        pass
+
+    def create_report(self):
+        """Create a curation report.
+
+        This method is responsible for creating any number of reports about the curation
+        process. These reports may be machine-readable, human-readable, or both.
+        """
+        pass
+
+    @abstractmethod
+    def is_publication_approved(self) -> bool:
+        """Return the publication decision made through the curation process.
+
+        If publication is allowed, this method must return ``True``, otherwise
+        ``False``.
+        """
+        pass
+
+    def process_decision_positive(self):
+        """Process a positive curation decision.
+
+        This method is called if a positive publication decision was made in the
+        curation process.
+        """
+        pass
+
+    def process_decision_negative(self):
+        """Process a negative curation decision.
+
+        This method is called if a negative publication decision was made in the
+        curation process. By default, a ``RuntimeError`` is raised, halting the
+        execution.
+        """
+        raise RuntimeError("Curation declined further processing")
 
 
 class HermesCurateCommand(HermesCommand):
-    """ Curate the unified metadata before deposition. """
+    """Curate the processed metadata before deposition."""
 
     command_name = "curate"
     settings_class = _CurateSettings
@@ -31,17 +108,31 @@ def init_command_parser(self, command_parser: argparse.ArgumentParser) -> None:
         pass
 
     def __call__(self, args: argparse.Namespace) -> None:
-
-        self.log.info("# Metadata curation")
+        self.args = args
+        plugin_name = self.settings.method
 
         ctx = CodeMetaContext()
-        process_output = ctx.hermes_dir / 'process' / (ctx.hermes_name + ".json")
-
-        if not process_output.is_file():
+        process_output = ctx.get_cache("process", ctx.hermes_name)
+        if not process_output.exists():
             self.log.error(
                 "No processed metadata found. Please run `hermes process` before curation."
             )
             sys.exit(1)
 
-        os.makedirs(ctx.hermes_dir / 'curate', exist_ok=True)
-        shutil.copy(process_output, ctx.hermes_dir / 'curate' / (ctx.hermes_name + '.json'))
+        curate_path = ContextPath("curate")
+        with open(process_output) as process_output_fh:
+            ctx.update(curate_path, json.load(process_output_fh))
+
+        try:
+            plugin_func = self.plugins[plugin_name](self, ctx)
+
+        except KeyError as e:
+            self.log.error("Plugin '%s' not found.", plugin_name)
+            self.errors.append(e)
+
+        try:
+            plugin_func(self)
+
+        except HermesValidationError as e:
+            self.log.error("Error while executing %s: %s", plugin_name, e)
+            self.errors.append(e)