Request org details to fill in more template fields

nf_core/pipelines/create/basicdetails.py

@@ -10,6 +10,7 @@
 from textual.widgets import Button, Footer, Header, Input, Markdown
 
 from nf_core.pipelines.create.utils import CreateConfig, TextInput, add_hide_class, remove_hide_class
+from nf_core.utils import get_org_url
 
 pipeline_exists_warn = """
 > ⚠️  **The pipeline you are trying to create already exists.**
@@ -22,6 +23,9 @@
 class BasicDetails(Screen):
     """Name, description, author, etc."""
 
+    _auto_org_full_name: str | None = None
+    _auto_org_url: str | None = None
+
     def compose(self) -> ComposeResult:
         yield Header()
         yield Footer()
@@ -58,47 +62,117 @@ def compose(self) -> ComposeResult:
             "Author(s)",
             "Name of the main author / authors",
         )
+        if not self.parent.NFCORE_PIPELINE:
+            yield TextInput(
+                "org_full_name",
+                "Organisation Name",
+                "Display name for the organisation",
+                "nf-core",
+            )
+            yield TextInput(
+                "org_url",
+                "Organisation URL",
+                "Website URL for the organisation",
+                get_org_url("nf-core", self.parent.NFCORE_PIPELINE),
+            )
         yield Markdown(dedent(pipeline_exists_warn), id="exist_warn", classes="hide")
         yield Center(
             Button("Back", id="back", variant="default"),
             Button("Next", id="next", variant="success"),
             classes="cta",
         )
 
-    @on(Input.Changed)
-    @on(Input.Submitted)
-    def show_exists_warn(self):
-        """Check if the pipeline exists on every input change or submitted.
-        If the pipeline exists, show warning message saying that it will be overridden."""
+    def _sync_org_metadata_inputs(self, force: bool = False) -> None:
+        """Keep organisation metadata in sync with the org value until the user overrides it."""
+        if self.parent.NFCORE_PIPELINE:
+            return
+
+        org_input = self._get_input_widget("org")
+        org_full_name_input = self._get_input_widget("org_full_name")
+        org_url_input = self._get_input_widget("org_url")
+        suggested_name = org_input.value or "nf-core"
+        suggested_url = get_org_url(org_input.value or "nf-core", self.parent.NFCORE_PIPELINE)
+
+        if force or org_full_name_input.value in {"", self._auto_org_full_name}:
+            org_full_name_input.value = suggested_name
+
+        if force or org_url_input.value in {"", self._auto_org_url}:
+            org_url_input.value = suggested_url
+
+        self._auto_org_full_name = suggested_name
+        self._auto_org_url = suggested_url
+
+    def _get_input_widget(self, field_id: str) -> Input:
+        """Return the inner Textual input for a named TextInput wrapper."""
+        return self.query_one(f"#{field_id}", TextInput).query_one(Input)
+
+    def _get_config_values(self) -> dict[str, str]:
+        """Collect screen values and inject nf-core defaults for hidden fields."""
         config = {}
         for text_input in self.query("TextInput"):
             this_input = text_input.query_one(Input)
             config[text_input.field_id] = this_input.value
+
+        if self.parent.NFCORE_PIPELINE:
+            # Add nf-core defaults for hidden fields, to avoid errors when creating a pipeline
+            config["org_full_name"] = config["org"]
+            config["org_url"] = get_org_url("nf-core", True)
+
+        return config
+
+    @on(Input.Changed)
+    @on(Input.Submitted)
+    def show_exists_warn(self, event: Input.Changed | Input.Submitted) -> None:
+        """Check if the pipeline exists on every input change or submitted.
+        If the pipeline exists, show warning message saying that it will be overridden."""
+        if event.input is self._get_input_widget("org"):
+            self._sync_org_metadata_inputs()
+        config = self._get_config_values()
         if Path(config["org"] + "-" + config["name"]).is_dir():
             remove_hide_class(self.parent, "exist_warn")
         else:
             add_hide_class(self.parent, "exist_warn")
 
+    @on(Input.Blurred)
+    def restore_empty_org_metadata_on_blur(self, event: Input.Blurred) -> None:
+        """Restore auto-managed org metadata only after the user leaves the field empty."""
+        if self.parent.NFCORE_PIPELINE:
+            return
+
+        org_input = self._get_input_widget("org")
+        for field_id in ("org_full_name", "org_url"):
+            if event.input is self._get_input_widget(field_id):
+                if event.input.value:
+                    break
+                if field_id == "org_full_name":
+                    restored_value = org_input.value or "nf-core"
+                    self._auto_org_full_name = restored_value
+                else:
+                    restored_value = get_org_url(org_input.value or "nf-core", self.parent.NFCORE_PIPELINE)
+                    self._auto_org_url = restored_value
+                event.input.value = restored_value
+                break
+
     def on_screen_resume(self):
         """Hide warn message on screen resume.
         Update displayed value on screen resume."""
         add_hide_class(self.parent, "exist_warn")
         for text_input in self.query("TextInput"):
             if text_input.field_id == "org":
                 text_input.disabled = self.parent.NFCORE_PIPELINE
+        self._sync_org_metadata_inputs(force=True)
 
     @on(Button.Pressed)
     def on_button_pressed(self, event: Button.Pressed) -> None:
         """Save fields to the config."""
-        config = {}
         for text_input in self.query("TextInput"):
             this_input = text_input.query_one(Input)
             validation_result = this_input.validate(this_input.value)
-            config[text_input.field_id] = this_input.value
             if not validation_result.is_valid:
                 text_input.query_one(".validation_msg").update("\n".join(validation_result.failure_descriptions))
             else:
                 text_input.query_one(".validation_msg").update("")
+        config = self._get_config_values()
         try:
             self.parent.TEMPLATE_CONFIG = CreateConfig(**config)
             if event.button.id == "next":

nf_core/pipelines/create/create.py

@@ -194,6 +194,10 @@ def update_config(self, organisation, version, force, outdir):
             self.config.outdir = outdir if outdir else "."
         if self.config.is_nfcore is None or self.config.is_nfcore == "null":
             self.config.is_nfcore = self.config.org == "nf-core"
+        if self.config.org_full_name is None and self.config.org is not None:
+            self.config.org_full_name = self.config.org
+        if self.config.org_url is None and self.config.org is not None:
+            self.config.org_url = nf_core.utils.get_org_url(self.config.org, self.config.is_nfcore)
 
     def obtain_jinja_params_dict(self, features_to_skip: list[str], pipeline_dir: str | Path) -> tuple[dict, list[str]]:
         """Creates a dictionary of parameters for the new pipeline.

nf_core/pipelines/create/utils.py

@@ -4,6 +4,7 @@
 from contextvars import ContextVar
 from pathlib import Path
 from typing import Any
+from urllib.parse import urlparse
 
 import yaml
 from pydantic import ConfigDict, ValidationError, ValidationInfo, field_validator
@@ -65,14 +66,23 @@ def name_nospecialchars(cls, v: str, info: ValidationInfo) -> str:
                 raise ValueError("Must not contain special characters. Only '-' or '_' are allowed.")
         return v
 
-    @field_validator("org", "description", "author", "version", "outdir")
+    @field_validator("org", "org_full_name", "org_url", "description", "author", "version", "outdir")
     @classmethod
     def notempty(cls, v: str) -> str:
         """Check that string values are not empty."""
         if v.strip() == "":
             raise ValueError("Cannot be left empty.")
         return v
 
+    @field_validator("org_url")
+    @classmethod
+    def valid_org_url(cls, v: str) -> str:
+        """Check that the organisation URL is a valid HTTP(S) URL."""
+        parsed_url = urlparse(v)
+        if parsed_url.scheme not in {"http", "https"} or not parsed_url.netloc:
+            raise ValueError("Must be a valid http(s) URL.")
+        return v
+
     @field_validator("version")
     @classmethod
     def version_nospecialchars(cls, v: str) -> str:

nf_core/pipelines/rocrate.py

@@ -17,10 +17,13 @@
 from rocrate.model.person import Person
 from rocrate.rocrate import ROCrate as BaseROCrate
 
-from nf_core.utils import Pipeline
+from nf_core.utils import Pipeline, get_docs_url, get_org_url, load_tools_config
 
 log = logging.getLogger(__name__)
 
+DEFAULT_TOPICS = ["nf-core", "nextflow"]
+TOPICS_REQUEST_TIMEOUT_SECONDS = 10
+
 
 # To identify bots, we look for names that contain "[bot]" or end with "-bot" or "_bot", case-insensitive
 BOT_PATTERNS = re.compile(r"\[bot\]|(-bot|_bot)$", re.IGNORECASE)
@@ -88,6 +91,12 @@ def __init__(self, pipeline_dir: Path, version="") -> None:
         self.crate: rocrate.rocrate.ROCrate
         self.pipeline_obj = Pipeline(self.pipeline_dir)
         self.pipeline_obj._load()
+        self.tools_config = None
+
+        try:
+            _, self.tools_config = load_tools_config(self.pipeline_dir)
+        except (AssertionError, FileNotFoundError, UserWarning) as error:
+            log.debug(f"Could not load `.nf-core.yml` for RO-Crate: {self.pipeline_dir}. {error}")
 
         setup_requests_cachedir()
 
@@ -143,6 +152,61 @@ def create_rocrate(self, json_path: None | Path = None, zip_path: None | Path =
 
         return True
 
+    def _get_template_parameter(self, param_name: str) -> str | None:
+        if self.tools_config and getattr(self.tools_config, "template", None):
+            org_name = getattr(self.tools_config.template, param_name, None)
+            if org_name:
+                return org_name
+        return None
+
+    def _get_pipeline_org(self) -> str:
+        return self._get_template_parameter("org") or "nf-core"
+
+    def _get_pipeline_org_full_name(self) -> str:
+        return self._get_template_parameter("org_full_name") or self._get_pipeline_org()
+
+    def _get_pipeline_org_url(self) -> str:
+        return self._get_template_parameter("org_url") or get_org_url(self._get_pipeline_org())
+
+    def _get_main_entity_url(self) -> str:
+        """Return a stable URL for the selected workflow revision."""
+        return get_docs_url(
+            self._get_pipeline_org_url(),
+            self.crate.name,
+            self.crate.name.replace(self._get_pipeline_org() + "/", ""),
+            self.version,
+            None,
+        )
+
+    def _get_remote_workflow_topics(self) -> list[str]:
+        """Fetch workflow topics from the organisation pipelines index when available."""
+        assert self.pipeline_obj.pipeline_name is not None  # mypy
+        topics = DEFAULT_TOPICS.copy()
+
+        try:
+            response = requests.get(
+                f"{self._get_pipeline_org_url()}/pipelines.json",
+                timeout=TOPICS_REQUEST_TIMEOUT_SECONDS,
+            )
+            response.raise_for_status()
+        except requests.exceptions.RequestException as error:
+            log.debug("Could not fetch workflow topics for RO-Crate from %s: %s", self._get_pipeline_org_url(), error)
+            return topics
+
+        full_pipeline_name = f"{self._get_pipeline_org()}/{self.pipeline_obj.pipeline_name}"
+        try:
+            payload = response.json()
+            remote_workflows = payload["remote_workflows"]
+            for remote_wf in remote_workflows:
+                if remote_wf["full_name"] == full_pipeline_name or remote_wf["name"] == self.pipeline_obj.pipeline_name:
+                    topics.extend(remote_wf["topics"])
+                    break
+
+        except (ValueError, KeyError, TypeError) as error:
+            log.error("Could not parse pipelines index from %s: %s", response.url, error)
+
+        return topics
+
     def make_workflow_rocrate(self) -> None:
         """
         Create an RO Crate for a pipeline
@@ -191,9 +255,9 @@ def make_workflow_rocrate(self) -> None:
         except FileNotFoundError:
             log.error(f"Could not find LICENSE file in {self.pipeline_dir}")
 
-        self.crate.add_jsonld(
-            {"@id": "https://nf-co.re/", "@type": "Organization", "name": "nf-core", "url": "https://nf-co.re/"}
-        )
+        org_full_name = self._get_pipeline_org_full_name()
+        org_url = self._get_pipeline_org_url()
+        self.crate.add_jsonld({"@id": org_url, "@type": "Organization", "name": org_full_name, "url": org_url})
 
         # Set metadata for main entity file
         self.set_main_entity("main.nf")
@@ -213,25 +277,14 @@ def set_main_entity(self, main_entity_filename: str):
         self.crate.mainEntity.append_to(
             "dateModified", str(datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ")), compact=True
         )
-        self.crate.mainEntity.append_to("sdPublisher", {"@id": "https://nf-co.re/"}, compact=True)
-        url = "dev" if self.version.endswith("dev") else self.version
-        self.crate.mainEntity.append_to(
-            "url", f"https://nf-co.re/{self.crate.name.replace('nf-core/', '')}/{url}/", compact=True
-        )
+        self.crate.mainEntity.append_to("sdPublisher", {"@id": self._get_pipeline_org_url()}, compact=True)
+        self.crate.mainEntity.append_to("url", self._get_main_entity_url(), compact=True)
         self.crate.mainEntity.append_to("version", self.version, compact=True)
 
         # remove duplicate entries for version
         self.crate.mainEntity["version"] = list(set(self.crate.mainEntity["version"]))
 
-        # get keywords from nf-core website
-        remote_workflows = requests.get("https://nf-co.re/pipelines.json").json()["remote_workflows"]
-        # go through all remote workflows and find the one that matches the pipeline name
-        topics = ["nf-core", "nextflow"]
-        for remote_wf in remote_workflows:
-            assert self.pipeline_obj.pipeline_name is not None  # mypy
-            if remote_wf["name"] == self.pipeline_obj.pipeline_name.replace("nf-core/", ""):
-                topics = topics + remote_wf["topics"]
-                break
+        topics = self._get_remote_workflow_topics()
 
         log.debug(f"Adding topics: {topics}")
         self.crate.mainEntity.append_to("keywords", topics)