Allow CSV ingest to create new shots.

client/ayon_traypublisher/plugins/create/create_csv_ingest.py

@@ -181,6 +181,8 @@ def __init__(
         self.variant = variant
         self.product_type = product_type
         self.repre_items: List[RepreItem] = []
+        self.has_promised_context = False
+        self.parents = None
         self._unique_name = None
         self._pre_product_name = None
 
@@ -249,6 +251,7 @@ class IngestCSV(TrayPublishCreator):
     # settings for this creator
     columns_config = {}
     representations_config = {}
+    folder_creation_config = {}
 
     def get_instance_attr_defs(self):
         return [
@@ -388,6 +391,79 @@ def _resolve_repre_path(
 
         return filepath
 
+    def _get_folder_type_from_regex_settings(self, folder_name: str) -> str:
+        """ Get the folder type that matches the regex settings.
+
+        Args:
+            folder_name (str): The folder name.
+
+        Returns:
+            str. The folder type to use.
+        """
+        for folder_setting in self.folder_creation_config["folder_type_regexes"]:
+            if re.match(folder_setting["regex"], folder_name):
+                folder_type = folder_setting["folder_type"]
+                return folder_type
+
+        return self.folder_creation_config["folder_create_type"]
+
+    def _compute_parents_data(self, project_name: str, product_item: ProductItem) -> list:
+        """ Compute parent data when new hierarchy has to be created during the
+            publishing process.
+
+        Args:
+            project_name (str): The project name.
+            product_item (ProductItem): The product item to inspect.
+
+        Returns:
+            list. The parent list if any
+
+        Raise:
+            ValueError: When provided folder_path parent do not exist.
+        """
+        parent_folder_names = product_item.folder_path.lstrip("/").split("/")
+        # Rename name of folder itself
+        parent_folder_names.pop(-1)
+        if not parent_folder_names:
+            return []
+
+        parent_paths = []
+        parent_path = ""
+        for name in parent_folder_names:
+            path = f"{parent_path}/{name}"
+            parent_paths.append(path)
+            parent_path = path
+
+        folders_by_path = {
+            folder["path"]: folder
+            for folder in ayon_api.get_folders(
+                project_name,
+                folder_paths=parent_paths,
+                fields={"folderType", "path"}
+            )
+        }
+        parent_data = []
+        for path in parent_paths:
+            folder_entity = folders_by_path.get(path)
+            name = path.rsplit("/", 1)[-1]
+
+            # Folder exists, retrieve data from existing.
+            if folder_entity:
+                folder_type = folder_entity["folderType"]
+
+            # Define folder type from settings.
+            else:
+                folder_type = self._get_folder_type_from_regex_settings(name)
+
+            item = {
+                "entity_name": name,
+                "folder_type": folder_type,
+            }
+            parent_data.append(item)
+
+        return parent_data
+
+
     def _get_data_from_csv(
         self, csv_dir: str, filename: str
     ) -> Dict[str, ProductItem]:
@@ -458,12 +534,6 @@ def _get_data_from_csv(
             )
         }
         missing_paths: Set[str] = folder_paths - set(folder_ids_by_path.keys())
-        if missing_paths:
-            ending = "" if len(missing_paths) == 1 else "s"
-            joined_paths = "\n".join(sorted(missing_paths))
-            raise CreatorError(
-                f"Folder{ending} not found.\n{joined_paths}"
-            )
 
         task_names: Set[str] = {
             product_item.task_name
@@ -480,8 +550,25 @@ def _get_data_from_csv(
             task_entities_by_folder_id[folder_id].append(task_entity)
 
         missing_tasks: Set[str] = set()
+        if missing_paths and not self.folder_creation_config["enabled"]:
+            error_msg = (
+                "Folder creation is disabled but found missing folder(s): %r" %
+                ",".join(missing_paths)
+            )
+            raise CreatorError(error_msg)
+
         for product_item in product_items_by_name.values():
             folder_path = product_item.folder_path
+
+            if folder_path in missing_paths:
+                product_item.has_promised_context = True
+                product_item.task_type = None
+                product_item.parents = self._compute_parents_data(
+                    project_name,
+                    product_item
+                )
+                continue
+
             task_name = product_item.task_name
             folder_id = folder_ids_by_path[folder_path]
             task_entities = task_entities_by_folder_id[folder_id]
@@ -758,6 +845,24 @@ def _prepare_representations(
                 explicit_output_name
             )
 
+    def _get_task_type_from_task_name(self, task_name: str):
+        """ Retrieve task type from task name.
+
+        Args:
+            task_name (str): The task name.
+
+        Returns:
+            str. The task type computed from settings.
+        """
+        for task_setting in self.folder_creation_config["task_type_regexes"]:
+            if re.match(task_setting["regex"], task_name):
+                task_type = task_setting["task_type"]
+                break
+        else:
+            task_type = self.folder_creation_config["task_create_type"]
+
+        return task_type
+
     def _create_instances_from_csv_data(self, csv_dir: str, filename: str):
         """Create instances from csv data"""
         # from special function get all data from csv file and convert them
@@ -835,6 +940,31 @@ def _create_instances_from_csv_data(self, csv_dir: str, filename: str):
                 "prepared_data_for_repres": []
             }
 
+            if product_item.has_promised_context:
+                hierarchy, folder_name = folder_path.rsplit("/", 1)
+                families.append("shot")
+                instance_data.update(
+                    {
+                        "newHierarchyIntegration": True,
+                        "hierarchy": hierarchy,
+                        "parents": product_item.parents,
+                        "families": families,
+                        "heroTrack": True,
+                    }
+                )
+
+                folder_type = self._get_folder_type_from_regex_settings(folder_name)
+                instance_data["folder_type"] = folder_type
+
+                if product_item.task_name:
+                    task_type = self._get_task_type_from_task_name(
+                        product_item.task_name
+                    )
+                    tasks = instance_data.setdefault("tasks", {})
+                    tasks[product_item.task_name] = {
+                        "type": task_type
+                    }
+
             # create new instance
             new_instance: CreatedInstance = CreatedInstance(
                 product_item.product_type,
@@ -843,6 +973,10 @@ def _create_instances_from_csv_data(self, csv_dir: str, filename: str):
                 self
             )
             self._prepare_representations(product_item, new_instance)
+
+            if product_item.has_promised_context:
+                new_instance.transient_data["has_promised_context"] = True
+
             instances.append(new_instance)
 
         return instances

server/settings/creator_plugins.py

@@ -1,5 +1,10 @@
 from pydantic import validator
-from ayon_server.settings import BaseSettingsModel, SettingsField
+from ayon_server.settings import (
+    BaseSettingsModel,
+    SettingsField,
+    folder_types_enum,
+    task_types_enum,
+)
 from ayon_server.settings.validators import ensure_unique_names
 from ayon_server.exceptions import BadRequestException
 
@@ -128,6 +133,73 @@ def validate_unique_outputs(cls, value):
         return value
 
 
+class FolderTypeRegexItem(BaseSettingsModel):
+    _layout = "expanded"
+    regex: str = SettingsField("", title="Folder Regex")
+    folder_type: str = SettingsField(
+        "Folder",
+        title="Folder Type",
+        enum_resolver=folder_types_enum,
+        description=(
+            "Project's Anatomy folder type to create when regex matches."),
+    )
+
+
+class TaskTypeRegexItem(BaseSettingsModel):
+    _layout = "expanded"
+    regex: str = SettingsField("", title="Task Regex")
+    task_type: str = SettingsField(
+        "",
+        title="Task Type",
+        enum_resolver=task_types_enum,
+        description=(
+            "New task type to create when regex matches."),
+    )
+
+
+class FolderCreationConfigModel(BaseSettingsModel):
+    """Allow to create folder hierarchy when non-existing."""
+
+    enabled: bool = SettingsField(
+        title="Enabled folder creation",
+        default=False,
+    )
+
+    folder_type_regexes: list[FolderTypeRegexItem] = SettingsField(
+        default_factory=FolderTypeRegexItem,
+        description=(
+            "Using Regex expressions to create missing folders. \nThose can be used"
+            " to define which folder types are used for new folder creation"
+            " depending on their names."
+        )
+    )
+
+    folder_create_type: str = SettingsField(
+        "Folder",
+        title="Default Folder Type",
+        enum_resolver=folder_types_enum,
+        description=(
+            "Default folder type for new folder(s) creation."),
+    )
+
+    task_type_regexes: list[TaskTypeRegexItem] = SettingsField(
+        default_factory=TaskTypeRegexItem,
+        description=(
+            "Using Regex expressions to create missing tasks. \nThose can be used"
+            " to define which task types are used for new folder+task creation"
+            " depending on their names."
+        )
+    )
+
+    task_create_type: str = SettingsField(
+        "",
+        title="Default Task Type",
+        enum_resolver=task_types_enum,
+        description=(
+            "Default task type for new task(s) creation."),
+    )
+
+
 class IngestCSVPluginModel(BaseSettingsModel):
     """Allows to publish multiple video files in one go. <br />Name of matching
      asset is parsed from file names ('asset.mov', 'asset_v001.mov',
@@ -148,6 +220,11 @@ class IngestCSVPluginModel(BaseSettingsModel):
         default_factory=RepresentationConfigModel
     )
 
+    folder_creation_config: FolderCreationConfigModel = SettingsField(
+        title="Folder creation config",
+        default_factory=FolderCreationConfigModel
+    )
+
 
 class TrayPublisherCreatePluginsModel(BaseSettingsModel):
     BatchMovieCreator: BatchMovieCreatorPlugin = SettingsField(
@@ -336,6 +413,16 @@ class TrayPublisherCreatePluginsModel(BaseSettingsModel):
                     ]
                 }
             ]
+        },
+        "folder_creation_config": {
+            "enabled": False,
+            "folder_type_regexes": [
+                {"regex": "(sh.*)", "folder_type": "Shot"},
+                {"regex": "(seq.*)", "folder_type": "Sequence"}
+            ],
+            "folder_create_type": "Folder",
+            "task_type_regexes": [],
+            "task_create_type": "Generic",
         }
     }
 }