Init Plugin integration

Author: nheeb

src/hermes/commands/init/base.py

@@ -7,14 +7,15 @@
 import os
 import re
 import sys
+import traceback
+import requests
+import toml
+
 from dataclasses import dataclass
 from enum import Enum, auto
 from importlib import metadata
 from pathlib import Path
 from urllib.parse import urljoin, urlparse
-
-import requests
-import toml
 from pydantic import BaseModel
 from requests import HTTPError
 
@@ -174,6 +175,18 @@ def __init__(self, parser: argparse.ArgumentParser):
             "deposit_extra_files": "",
             "push_branch": "main"
         }
+        self.hermes_toml_data = {
+            "harvest": {
+                "sources": ["cff"]
+            },
+            "deposit": {
+                "target": "invenio_rdm",
+                "invenio_rdm": {
+                    "site_url": "",
+                    "access_right": "open"
+                 }
+            }
+        }
         self.plugin_relevant_commands = ["harvest", "deposit"]
         self.builtin_plugins: dict[str: HermesPlugin] = get_builtin_plugins(self.plugin_relevant_commands)
         self.selected_plugins: list[marketplace.PluginInfo] = []
@@ -210,40 +223,51 @@ def __call__(self, args: argparse.Namespace) -> None:
             if args.template_branch != "":
                 self.template_branch = args.template_branch
 
-        # Test if init is valid in current folder
-        self.test_initialization()
+        try:
+            # Test if init is valid in current folder
+            self.test_initialization()
+
+            sc.echo(f"Starting to initialize HERMES in {self.folder_info.absolute_path}\n")
+            sc.max_steps = 8
 
-        sc.echo(f"Starting to initialize HERMES in {self.folder_info.absolute_path}")
-        sc.max_steps = 8
+            sc.next_step("Configure HERMES plugins")
+            self.choose_plugins()
+            self.integrate_plugins()
 
-        sc.next_step("Configure HERMES plugins")
-        self.choose_plugins()
+            sc.next_step("Configure deposition platform and setup method")
+            self.choose_deposit_platform()
+            self.integrate_deposit_platform()
+            self.choose_setup_method()
 
-        sc.next_step("Configure deposition platform and setup method")
-        self.choose_deposit_platform()
-        self.choose_setup_method()
+            sc.next_step("Configure HERMES behaviour")
+            self.choose_push_branch()
+            self.choose_deposit_files()
 
-        sc.next_step("Configure HERMES behaviour")
-        self.choose_push_branch()
-        self.choose_deposit_files()
+            sc.next_step("Create hermes.toml file")
+            self.create_hermes_toml()
 
-        sc.next_step("Create hermes.toml file")
-        self.create_hermes_toml()
+            sc.next_step("Create CITATION.cff file")
+            self.create_citation_cff()
 
-        sc.next_step("Create CITATION.cff file")
-        self.create_citation_cff()
+            sc.next_step("Create git CI files")
+            self.update_gitignore()
+            self.create_ci_template()
 
-        sc.next_step("Create git CI files")
-        self.update_gitignore()
-        self.create_ci_template()
+            sc.next_step("Connect with deposition platform")
+            self.connect_deposit_platform()
 
-        sc.next_step("Connect with deposition platform")
-        self.connect_deposit_platform()
+            sc.next_step("Connect with git hoster")
+            self.configure_git_project()
 
-        sc.next_step("Connect with git hoster")
-        self.configure_git_project()
+            sc.echo("\nHERMES is now initialized and ready to be used.\n",
+                    formatting=sc.Formats.OKGREEN+sc.Formats.BOLD)
 
-        sc.echo("\nHERMES is now initialized and ready to be used.\n", formatting=sc.Formats.OKGREEN+sc.Formats.BOLD)
+        except Exception as e:
+            # More useful message on error
+            sc.echo(f"An error occurred during execution of HERMES init: {e}",
+                    formatting=sc.Formats.FAIL+sc.Formats.BOLD)
+            sc.debug_info(traceback.format_exc())
+            sys.exit(2)
 
     def test_initialization(self) -> None:
         """Test if init is possible and wanted. If not: sys.exit()"""
@@ -284,6 +308,7 @@ def test_initialization(self) -> None:
         if self.git_remote:
             self.git_remote_url = git_info.get_remote_url(self.git_remote)
             self.git_hoster = get_git_hoster_from_url(self.git_remote_url)
+
         # Abort with no remote
         else:
             sc.echo("Your git project does not have a remote. It is recommended for HERMES to "
@@ -311,26 +336,12 @@ def test_initialization(self) -> None:
                 sys.exit()
 
     def create_hermes_toml(self) -> None:
-        """Creates the hermes.toml file based on a dictionary"""
-        deposit_url = DepositPlatformUrls.get(self.deposit_platform)
-        default_values = {
-            "harvest": {
-                "sources": ["cff"]
-            },
-            "deposit": {
-                "target": "invenio_rdm",
-                "invenio_rdm": {
-                    "site_url": deposit_url,
-                    "access_right": "open"
-                 }
-            }
-        }
-
+        """Creates the hermes.toml file based on a self.hermes_toml_data"""
         if (not self.folder_info.has_hermes_toml) \
                 or sc.confirm("Do you want to replace your `hermes.toml` with a new one?", default=True):
             with open('hermes.toml', 'w') as toml_file:
                 # noinspection PyTypeChecker
-                toml.dump(default_values, toml_file)
+                toml.dump(self.hermes_toml_data, toml_file)
             sc.echo("`hermes.toml` was created.", formatting=sc.Formats.OKGREEN)
 
     def create_citation_cff(self) -> None:
@@ -384,7 +395,6 @@ def create_ci_template(self) -> None:
         """Downloads and configures the ci workflow files using templates from the chosen template branch."""
         match self.git_hoster:
             case GitHoster.GitHub:
-                # TODO Replace this later with the link to the real templates (not the feature branch)
                 template_url = self.get_template_url("TEMPLATE_hermes_github_to_zenodo.yml")
                 ci_file_folder = ".github/workflows"
                 ci_file_name = "hermes_github.yml"
@@ -429,10 +439,11 @@ def configure_ci_template(self, ci_file_path) -> None:
         parameters = list(set(re.findall(r'{%(.*?)%}', content)))
         for parameter in parameters:
             if parameter in self.ci_parameters:
-                content = content.replace(f'{{%{parameter}%}}', self.ci_parameters[parameter])
+                value = str(self.ci_parameters[parameter])
+                content = content.replace(f'{{%{parameter}%}}', value)
             else:
-                sc.echo(f"Warning: CI File Parameter {{%{parameter}%}} was not set.",
-                        formatting=sc.Formats.WARNING)
+                sc.debug_info(f"CI File Parameter {{%{parameter}%}} was not set.", formatting=sc.Formats.WARNING)
+                content = content.replace(f'{{%{parameter}%}}', '')
         with open(ci_file_path, 'w') as file:
             file.write(content)
 
@@ -572,6 +583,11 @@ def choose_deposit_platform(self) -> None:
         )
         self.deposit_platform = deposit_platform_list[deposit_platform_index]
 
+    def integrate_deposit_platform(self) -> None:
+        """Makes changes to the toml data or something else based on the chosen deposit platform."""
+        deposit_url = DepositPlatformUrls.get(self.deposit_platform)
+        self.hermes_toml_data["deposit"]["invenio_rdm"]["site_url"] = deposit_url
+
     def choose_setup_method(self) -> None:
         """User chooses his desired setup method: Either preferring automatic (if available) or manual."""
         setup_method_index = sc.choose(
@@ -615,18 +631,42 @@ def choose_plugins(self):
                 sc.echo("The following plugins are available for installation:")
                 for info in plugins_available:
                     sc.echo(str(info), formatting=sc.Formats.WARNING, no_log=True)
+                    if info.abstract:
+                        sc.echo("-> " + info.abstract, formatting=sc.Formats.ITALIC+sc.Formats.WARNING, no_log=True)
                 sc.echo("")
             else:
                 self.selected_plugins = plugins_selected
                 break
-            choice = sc.choose("Do you want to add a plugin?", ["No"] + [p.name for p in plugins_available])
+            no_text = "No further plugins needed"
+            choice = sc.choose("Do you want to add a plugin?",
+                               [no_text] + [f"Add {p.name}" for p in plugins_available])
             if choice == 0:
                 self.selected_plugins = plugins_selected
                 break
             else:
                 chosen_plugin = plugins_available.pop(choice - 1)
                 plugins_selected.append(chosen_plugin)
 
+    def integrate_plugins(self):
+        """
+        Plugin installation is added to the ci-parameters.
+        Also for now we use this method to do custom plugin installation steps.
+        """
+        for plugin_info in self.selected_plugins:
+            if not plugin_info.is_valid():
+                sc.echo(f"Could not install plugin: {plugin_info.name}", formatting=sc.Formats.FAIL)
+                continue
+            pip_install = plugin_info.get_pip_install_command()
+            self.ci_parameters["pip_install_plugins_github"] = \
+                self.ci_parameters.get("pip_install_plugins_github", "") + "      - run: " + pip_install + "\n"
+            self.ci_parameters["pip_install_plugins_gitlab"] = \
+                self.ci_parameters.get("pip_install_plugins_gitlab", "") + "    - " + pip_install + "\n"
+            match plugin_info.name:
+                case "hermes-plugin-python":
+                    self.hermes_toml_data["harvest"]["sources"].append("toml")
+                case "hermes-plugin-git":
+                    self.hermes_toml_data["harvest"]["sources"].append("git")
+
     def no_git_setup(self, start_question: str = "") -> None:
         """Makes the init for a gitless project (basically just creating hermes.toml)"""
         if start_question == "":

src/hermes/commands/init/util/slim_click.py

@@ -61,6 +61,9 @@ def get_log_type(self, default: int = logging.INFO) -> int:
             return logging.INFO
         return default
 
+    def wrap_around(self, text: str) -> str:
+        return self.get_ansi() + text + Formats.ENDC.get_ansi()
+
 
 def echo(text: str, formatting: Formats = Formats.EMPTY, log_as: int = logging.NOTSET, no_log: bool = False):
     """
@@ -74,10 +77,12 @@ def echo(text: str, formatting: Formats = Formats.EMPTY, log_as: int = logging.N
     if AUTO_LOG_ON_ECHO and log_as == logging.NOTSET and text != "":
         log_as = formatting.get_log_type(logging.INFO)
     # Add text to log if there is a logger
-    if log_as != logging.NOTSET and default_file_logger and no_log == False:
+    if log_as != logging.NOTSET and default_file_logger and not no_log:
         default_file_logger.log(log_as, text)
     # Format the text for the console
     if formatting != Formats.EMPTY:
+        if Formats.ENDC.get_ansi() in text:
+            text = text.replace(Formats.ENDC.get_ansi(), Formats.ENDC.get_ansi() + formatting.get_ansi())
         text = f"{formatting.get_ansi()}{text}{Formats.ENDC.get_ansi()}"
     # Print it
     if (log_as != logging.DEBUG) or PRINT_DEBUG:
@@ -165,7 +170,9 @@ def next_step(description: str):
 
 def create_console_hyperlink(url: str, word: str) -> str:
     """Use this to have a consistent display of hyperlinks."""
-    return f"\033]8;;{url}\033\\{word}\033]8;;\033\\" if USE_FANCY_HYPERLINKS else f"{word} ({url})"
+    if USE_FANCY_HYPERLINKS:
+        return f"\033]8;;{url}\033\\{word}\033]8;;\033\\"
+    return f"{word} ({url})"
 
 
 class ColorLogHandler(logging.Handler):

src/hermes/commands/marketplace.py

@@ -11,6 +11,7 @@
 from pydantic import BaseModel, ConfigDict, Field
 from pydantic.alias_generators import to_camel
 
+from hermes.commands.init.util import slim_click
 from hermes.utils import hermes_doi, hermes_user_agent
 
 MARKETPLACE_URL = "https://hermes.software-metadata.pub/marketplace"
@@ -99,14 +100,43 @@ def _sort_plugins_by_step(plugins: list[SchemaOrgSoftwareApplication]) -> dict[s
 def _plugin_loc(_plugin: SchemaOrgSoftwareApplication) -> str:
     return "builtin" if _plugin.is_part_of == schema_org_hermes else (_plugin.url or "")
 
+
 class PluginInfo:
+    """
+    This class contains all the information about a plugin which are needed for the init-Command.
+    """
     def __init__(self):
         self.name: str = ""
         self.location: str = ""
         self.step: str = ""
         self.builtin: bool = True
+        self.install_url: str = ""
+        self.abstract: str = ""
+
     def __str__(self):
-        return f"[{self.step}] {self.name} ({self.location})"
+        step_text = f"[{self.step}]"
+        return f"{step_text} {slim_click.Formats.BOLD.wrap_around(self.name)} ({self.location})"
+
+    def get_pip_install_command(self) -> str:
+        """
+        Returns the pip install command which can be used to install the plugin.
+        Tries to extract the project name from the install_url (PyPI-URL) if possible.
+        Otherwise, it tries to use the location (Git-Project-URL) for the pip install command.
+        """
+        if self.install_url and self.install_url.startswith("https://pypi.org/project/"):
+            project_name = self.install_url.rstrip("/").removeprefix("https://pypi.org/project/")
+            return f"pip install {project_name}"
+        if self.location and self.location.startswith(("https://", "git@", "ssh://")):
+            git_url = self.location.rstrip("/")
+            return f"pip install git+{git_url}"
+        return ""
+
+    def is_valid(self) -> bool:
+        """
+        Returns True if the plugin can be installed. Maybe we'll check the actual repository here later
+        to make sure that other things are valid too.
+        """
+        return self.get_pip_install_command() != ""
 
 
 def get_plugin_infos() -> list[PluginInfo]:
@@ -124,6 +154,8 @@ def get_plugin_infos() -> list[PluginInfo]:
                 info.step = step
                 info.location = _plugin_loc(plugin)
                 info.builtin = plugin.is_part_of == schema_org_hermes
+                info.install_url = plugin.install_url
+                info.abstract = plugin.abstract
                 infos.append(info)
     return infos