Publisher: filter creators based on task

client/ayon_core/pipeline/create/context.py

@@ -37,6 +37,7 @@
 
 # Changes of instances and context are send as tuple of 2 information
 UpdateData = collections.namedtuple("UpdateData", ["instance", "changes"])
+_NOT_SET = object()
 
 
 class UnavailableSharedData(Exception):
@@ -1401,6 +1402,11 @@ def __init__(
         self._current_folder_path = None
         self._current_task_name = None
         self._current_workfile_path = None
+        self._current_project_settings = None
+
+        self._current_folder_entity = _NOT_SET
+        self._current_task_entity = _NOT_SET
+        self._current_task_type = _NOT_SET
 
         self._current_project_anatomy = None
 
@@ -1571,6 +1577,64 @@ def get_current_task_name(self):
 
         return self._current_task_name
 
+    def get_current_task_type(self):
+        """Task type which was used as current context on context reset.
+
+        Returns:
+            Union[str, None]: Task type.
+
+        """
+        if self._current_task_type is _NOT_SET:
+            task_type = None
+            task_entity = self.get_current_task_entity()
+            if task_entity:
+                task_type = task_entity["taskType"]
+            self._current_task_type = task_type
+        return self._current_task_type
+
+    def get_current_folder_entity(self):
+        """Folder entity for current context folder.
+
+        Returns:
+            Union[dict[str, Any], None]: Folder entity.
+
+        """
+        if self._current_folder_entity is not _NOT_SET:
+            return copy.deepcopy(self._current_folder_entity)
+        folder_entity = None
+        folder_path = self.get_current_folder_path()
+        if folder_path:
+            project_name = self.get_current_project_name()
+            folder_entity = ayon_api.get_folder_by_path(
+                project_name, folder_path
+            )
+        self._current_folder_entity = folder_entity
+        return copy.deepcopy(self._current_folder_entity)
+
+    def get_current_task_entity(self):
+        """Task entity for current context task.
+
+        Returns:
+            Union[dict[str, Any], None]: Task entity.
+
+        """
+        if self._current_task_entity is not _NOT_SET:
+            return copy.deepcopy(self._current_task_entity)
+        task_entity = None
+        task_name = self.get_current_task_name()
+        if task_name:
+            folder_entity = self.get_current_folder_entity()
+            if folder_entity:
+                project_name = self.get_current_project_name()
+                task_entity = ayon_api.get_task_by_name(
+                    project_name,
+                    folder_id=folder_entity["id"],
+                    task_name=task_name
+                )
+        self._current_task_entity = task_entity
+        return copy.deepcopy(self._current_task_entity)
+
+
     def get_current_workfile_path(self):
         """Workfile path which was opened on context reset.
 
@@ -1592,6 +1656,12 @@ def get_current_project_anatomy(self):
                 self._current_project_name)
         return self._current_project_anatomy
 
+    def get_current_project_settings(self):
+        if self._current_project_settings is None:
+            self._current_project_settings = get_project_settings(
+                self.get_current_project_name())
+        return self._current_project_settings
+
     @property
     def context_has_changed(self):
         """Host context has changed.
@@ -1718,7 +1788,12 @@ def reset_current_context(self):
         self._current_task_name = task_name
         self._current_workfile_path = workfile_path
 
+        self._current_folder_entity = _NOT_SET
+        self._current_task_entity = _NOT_SET
+        self._current_task_type = _NOT_SET
+
         self._current_project_anatomy = None
+        self._current_project_settings = None
 
     def reset_plugins(self, discover_publish_plugins=True):
         """Reload plugins.
@@ -1772,7 +1847,7 @@ def _reset_publish_plugins(self, discover_publish_plugins):
 
     def _reset_creator_plugins(self):
         # Prepare settings
-        project_settings = get_project_settings(self.project_name)
+        project_settings = self.get_current_project_settings()
 
         # Discover and prepare creators
         creators = {}

client/ayon_core/tools/publisher/control.py

@@ -8,6 +8,7 @@
 import shutil
 import inspect
 from abc import ABCMeta, abstractmethod
+import re
 
 import six
 import arrow
@@ -39,6 +40,7 @@
 )
 from ayon_core.pipeline.publish import get_publish_instance_label
 from ayon_core.tools.common_models import HierarchyModel
+from ayon_core.lib.profiles_filtering import filter_profiles
 
 # Define constant for plugin orders offset
 PLUGIN_ORDER_OFFSET = 0.5
@@ -1686,6 +1688,15 @@ def _publish_plugins(self):
         """Publish plugins."""
         return self._create_context.publish_plugins
 
+    def _get_current_project_settings(self):
+        """Current project settings.
+
+        Returns:
+            dict
+        """
+
+        return self._create_context.get_current_project_settings()
+
     # Hierarchy model
     def get_folder_items(self, project_name, sender=None):
         return self._hierarchy_model.get_folder_items(project_name, sender)
@@ -1827,8 +1838,13 @@ def _reset_plugins(self):
     def _collect_creator_items(self):
         # TODO add crashed initialization of create plugins to report
         output = {}
+        allowed_creator_pattern = self._get_allowed_creators_pattern()
         for identifier, creator in self._create_context.creators.items():
             try:
+                if (not self._is_label_allowed(
+                        creator.label, allowed_creator_pattern)):
+                    self.log.debug(f"{creator.label} not allowed for context")
+                    continue
                 output[identifier] = CreatorItem.from_creator(creator)
             except Exception:
                 self.log.error(
@@ -1839,6 +1855,60 @@ def _collect_creator_items(self):
 
         return output
 
+    def _get_allowed_creators_pattern(self):
+        """Provide regex pattern for configured creator labels in this context
+
+        If no profile matches current context, it shows all creators.
+        Support usage of regular expressions for configured values.
+        Returns:
+            (re.Pattern)[optional]: None or regex compiled patterns
+                into single one ('Render|Image.*')
+        """
+
+        task_type = self._create_context.get_current_task_type()
+        project_settings = self._get_current_project_settings()
+
+        filter_creator_profiles = (
+            project_settings
+            ["core"]
+            ["tools"]
+            ["creator"]
+            ["filter_creator_profiles"]
+        )
+        filtering_criteria = {
+            "task_names": self.current_task_name,
+            "task_types": task_type,
+            "host_names": self._create_context.host_name
+        }
+        profile = filter_profiles(
+            filter_creator_profiles,
+            filtering_criteria,
+            logger=self.log
+        )
+
+        allowed_creator_pattern = None
+        if profile:
+            allowed_creator_labels = {
+                label
+                for label in profile["creator_labels"]
+                if label
+            }
+            self.log.debug(f"Only allowed `{allowed_creator_labels}` creators")
+            allowed_creator_pattern = (
+                re.compile("|".join(allowed_creator_labels)))
+        return allowed_creator_pattern
+
+    def _is_label_allowed(self, label, allowed_labels_regex):
+        """Implement regex support for allowed labels.
+
+        Args:
+            label (str): Label of creator - shown in Publisher
+            allowed_labels_regex (re.Pattern): compiled regular expression
+        """
+        if not allowed_labels_regex:
+            return True
+        return bool(allowed_labels_regex.match(label))
+
     def _reset_instances(self):
         """Reset create instances."""
         if self._resetting_instances:

server/settings/tools.py

@@ -35,6 +35,28 @@ class ProductNameProfile(BaseSettingsModel):
     template: str = SettingsField("", title="Template")
 
 
+class FilterCreatorProfile(BaseSettingsModel):
+    """Provide list of allowed Creator identifiers for context"""
+
+    _layout = "expanded"
+    host_names: list[str] = SettingsField(
+        default_factory=list, title="Host names"
+    )
+    task_types: list[str] = SettingsField(
+        default_factory=list,
+        title="Task types",
+        enum_resolver=task_types_enum
+    )
+    task_names: list[str] = SettingsField(
+        default_factory=list,
+        title="Task names")
+    creator_labels: list[str] = SettingsField(
+        default_factory=list,
+        title="Allowed Creator Labels",
+        description="Copy creator label from Publisher, regex supported."
+    )
+
+
 class CreatorToolModel(BaseSettingsModel):
     # TODO this was dynamic dictionary '{name: task_names}'
     product_types_smart_select: list[ProductTypeSmartSelectModel] = (
@@ -48,6 +70,13 @@ class CreatorToolModel(BaseSettingsModel):
         title="Product name profiles"
     )
 
+    filter_creator_profiles: list[FilterCreatorProfile] = SettingsField(
+        default_factory=list,
+        title="Filter creator profiles",
+        description="Allowed list of creator labels that will be only shown if "
+                    "profile matches context."
+    )
+
     @validator("product_types_smart_select")
     def validate_unique_name(cls, value):
         ensure_unique_names(value)
@@ -420,7 +449,8 @@ class GlobalToolsModel(BaseSettingsModel):
                 "tasks": [],
                 "template": "SK_{folder[name]}{variant}"
             }
-        ]
+        ],
+        "filter_creator_profiles": []
     },
     "Workfiles": {
         "workfile_template_profiles": [