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,66 @@ def _resolve_repre_path(
 
         return filepath
 
+    def _validate_parents(self, project_name: str, product_item: ProductItem) -> list:
+        """ Ensure parent exists for provided product_item.folder_path
+
+        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:
+                for folder_setting in self.folder_creation_config["folder_type_regexes"]:
+                    if re.match(folder_setting["regex"], name):
+                        folder_type = folder_setting["folder_type"]
+                        break
+                else:
+                    folder_type = "Folder"  # default
+
+            parent_data.append({
+                "folder_type": folder_type,
+                "entity_name": name
+            })
+
+        return parent_data
+
+
     def _get_data_from_csv(
         self, csv_dir: str, filename: str
     ) -> Dict[str, ProductItem]:
@@ -458,12 +521,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 +537,31 @@ 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
+                try:
+                    product_item.parents = self._validate_parents(
+                        project_name,
+                        product_item
+                    )
+                except ValueError:
+                    raise CreatorError(
+                        f"Parent context must exists for new shots: {folder_path}"
+                    )
+
+                continue
+
             task_name = product_item.task_name
             folder_id = folder_ids_by_path[folder_path]
             task_entities = task_entities_by_folder_id[folder_id]
@@ -835,6 +915,26 @@ def _create_instances_from_csv_data(self, csv_dir: str, filename: str):
                 "prepared_data_for_repres": []
             }
 
+            if product_item.has_promised_context:
+                hierarchy, _ = folder_path.rsplit("/", 1)
+                families.append("csv_ingest_shot")
+                instance_data.update(
+                    {
+                        "newHierarchyIntegration": True,
+                        "hierarchy": hierarchy,
+                        "parents": product_item.parents
+                    }
+                )
+                # TODO create new task from provided task name
+                #if product_item.task_name:
+                #    tasks = instance_data.setdefault("tasks", [])
+                #    tasks.append(
+                #        {
+                #            "name": product_item.task_name,
+                #            "type": "Generic"
+                #        }
+                #    )
+
             # create new instance
             new_instance: CreatedInstance = CreatedInstance(
                 product_item.product_type,
@@ -843,6 +943,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

@@ -128,6 +128,30 @@ def validate_unique_outputs(cls, value):
         return value
 
 
+class FolderTypeRegexItem(BaseSettingsModel):
+    _layout = "expanded"
+    regex: str = SettingsField("", title="Folder Regex")
+    folder_type: str = SettingsField("", title="Folder Type")
+
+
+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."
+        )
+    )
+
+
 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 +172,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 +365,13 @@ class TrayPublisherCreatePluginsModel(BaseSettingsModel):
                     ]
                 }
             ]
-        }
+        },
+        "folder_creation_config": {
+            "enabled": False,
+            "folder_type_regexes": [
+                {"regex": "(sh.*)", "folder_type": "Shot"},
+                {"regex": "(seq.*)", "folder_type": "Sequence"}
+            ],
+        }        
     }
 }