create_csv_ingest

class IngestCSV(TrayPublishCreator):
    """CSV ingest creator class"""

    icon = "fa.file"

    label = "CSV Ingest"
    product_base_type = "csv_ingest_file"
    product_type = product_base_type
    identifier = "io.ayon.creators.traypublisher.csv_ingest"

    default_variants = ["Main"]

    description = "Ingest products' data from CSV file"
    detailed_description = """
Ingest products' data from CSV file following column and representation
configuration in project settings.
"""
    settings_category = "traypublisher"

    # Position in the list of creators.
    order = 10

    presets = []

    def get_instance_attr_defs(self):
        return [
            BoolDef(
                "add_review_family",
                default=True,
                label="Review"
            )
        ]

    def collect_instances(self):
        super().collect_instances()
        for instance in self.create_context.instances:
            if instance.creator_identifier == self.identifier:
                instance.transient_data["has_promised_context"] = True

            if instance.product_base_type == "csv_ingest_file":
                instance.set_mandatory(True)

    def get_pre_create_attr_defs(self):
        """Creating pre-create attributes at creator plugin.

        Returns:
            list: list of attribute object instances
        """
        # Use same attributes as for instance attributes
        preset_items = []
        for preset in self.presets:
            preset_items.append(
                {"value": preset["name"], "label": preset["name"]})

        if not preset_items:
            preset_items.append(
                {"value": None, "label": "< Missing preset >"})

        return [
            EnumDef(
                "preset",
                items=preset_items,
                label="Preset",
            ),
            FileDef(
                "csv_filepath_data",
                folders=False,
                extensions=[".csv"],
                allow_sequences=False,
                single_item=True,
                label="CSV File",
            ),
        ]

    def create(
        self,
        product_name: str,
        instance_data: dict[str, Any],
        pre_create_data: dict[str, Any]
    ):
        """Create product from each row found in the CSV.

        Args:
            product_name (str): The subset name.
            instance_data (dict): The instance data.
            pre_create_data (dict):
        """
        selected_preset = pre_create_data["preset"]
        preset_data = next(
            (
                preset for preset in self.presets
                if preset["name"] == selected_preset
            ),
            None,
        )

        if not preset_data:
            raise CreatorError(
                f"Invalid preset '{selected_preset}'"
            )

        csv_filepath_data = pre_create_data.get("csv_filepath_data", {})

        csv_dir = csv_filepath_data.get("directory", "")
        if not os.path.exists(csv_dir):
            raise CreatorError(
                f"Directory '{csv_dir}' does not exist."
            )
        filename = csv_filepath_data.get("filenames", [])
        self._process_csv_file(
            preset_data, product_name, instance_data, csv_dir, filename[0]
        )

    def _pass_data_to_csv_instance(
        self,
        instance_data: dict[str, Any],
        staging_dir: str,
        filename: str
    ):
        """Pass CSV representation file to instance data"""

        representation = {
            "name": "csv",
            "ext": "csv",
            "files": filename,
            "stagingDir": staging_dir,
            "stagingDir_persistent": True,
        }

        instance_data.update({
            "label": f"CSV: {filename}",
            "representations": [representation],
            "stagingDir": staging_dir,
            "stagingDir_persistent": True,
        })

    def _process_csv_file(
        self,
        preset_data: dict[str, Any],
        product_name: str,
        instance_data: dict[str, Any],
        csv_dir: str,
        filename: str
    ):
        """Process CSV file.

        Args:
            preset_data (dict[str, Any]): The selected preset data.
            product_name (str): The subset name.
            instance_data (dict): The instance data.
            csv_dir (str): The csv directory.
            filename (str): The filename.

        """
        # create new instance from the csv file via self function
        self._pass_data_to_csv_instance(
            instance_data,
            csv_dir,
            filename
        )

        csv_instance = CreatedInstance(
            product_base_type=self.product_base_type,
            product_type=self.product_base_type,
            product_name=product_name,
            data=instance_data,
            creator=self
        )

        csv_instance["csvFileData"] = {
            "filename": filename,
            "staging_dir": csv_dir,
        }
        csv_instance.set_mandatory(True)

        # create instances from csv data via self function
        instances, report_data = self._create_instances_from_csv_data(
            preset_data, csv_dir, filename)

        if report_data:
            csv_instance["csvReportData"] = report_data

        for instance in instances:
            instance.data["csv_parent_instance"] = csv_instance.id
            self._store_new_instance(instance)
        self._store_new_instance(csv_instance)

    def _resolve_repre_path(
        self, csv_dir: str, filepath: Union[str, None]
    ) -> Union[str, None]:
        if not filepath:
            return filepath

        # Validate only existence of file directory as filename
        #   may contain frame specific char (e.g. '%04d' or '####').
        filedir, filename = os.path.split(filepath)
        if not filedir or filedir == ".":
            # If filedir is empty or "." then use same directory as
            #   csv path
            filepath = os.path.join(csv_dir, filepath)

        elif not os.path.exists(filedir):
            # If filepath does not exist, first try to find it in the
            #   same directory as the csv file is, but keep original
            #   value otherwise.
            new_filedir = os.path.join(csv_dir, filedir)
            if os.path.exists(new_filedir):
                filepath = os.path.join(new_filedir, filename)

        return filepath

    def _get_folder_type_from_regex_settings(
        self,
        folder_name: str,
        folder_creation_config: dict
    ) -> str:
        """ Get the folder type that matches the regex settings.

        Args:
            folder_name (str): The folder name.
            folder_creation_config (dict): The folder creation configuration.

        Returns:
            str. The folder type to use.
        """
        for folder_setting in folder_creation_config["folder_type_regexes"]:
            if re.match(folder_setting["regex"], folder_name):
                folder_type = folder_setting["folder_type"]
                return folder_type

        return folder_creation_config["folder_create_type"]

    def _compute_parents_data(
        self,
        project_name: str,
        product_item: ProductItem,
        preset_data: dict[str, Any]
    ) -> 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.
            preset_data (dict): The preset data with folder creation settings.

        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, preset_data["folder_creation_config"])

            item = {
                "entity_name": name,
                "folder_type": folder_type,
            }
            parent_data.append(item)

        return parent_data

    def _resolve_row_folder(
        self,
        row: dict[str, Any],
        folders_by_name: dict[str, list[dict[str, Any]]],
        prevalidation: dict[str, Any],
        row_index: int = 0,
    ) -> tuple[Optional[dict[str, Any]], Optional[tuple[str, str]]]:
        """Resolve folder path from folder name when path is missing.

        If the row already has a non-empty ``Folder Path`` the row is
        returned unchanged.  Otherwise the project folders index is
        searched by ``Folder Name`` and – when exactly one match is found –
        the resolved path is injected together with any temporal attributes
        (Frame Start, Frame End, Handle Start, Handle End, FPS) that are
        absent from the row but present on the matched folder entity.

        When the relevant validator in ``prevalidation`` is configured with
        ``mode: \"ignore\"``, a failing row is *not* raised as a
        ``CreatorError``. Instead the method returns
        ``(None, (category, message))`` so the caller can skip the row and
        collect the report entry.
        method returns ``(None, (category, message))`` so the caller can
        skip the row and collect the report entry.

        Args:
            row (dict): A single CSV row (will be copied, not mutated).
            folders_by_name (dict): Project folders indexed by folder name.
                Values are lists of folder entity dicts
                (``id``, ``path``, ``name``, ``attrib``).
            prevalidation (dict): The ``prevalidation``
                for specific validator with output either ``error``
                or ``ignore``.
            row_index (int): 1-based row number in the CSV file (header = 1,
                first data row = 2).  Used in error messages for easier
                identification of the failing row.

        Returns:
            tuple: ``(resolved_row, None)`` on success, or
                ``(None, (category, message))`` when the row should be
                skipped and reported.

        Raises:
            CreatorError: When a folder resolution error occurs and the
                precreate validation is not configured to ignore it.
        """
        row = dict(row)  # work on a copy so the original is never mutated

        folder_path = (row.get("Folder Path") or "").strip()
        if folder_path:
            # Folder Path already provided. Nothing to resolve.
            return row, None

        file_path = (row.get("File Path") or "").strip()
        row_ctx = (
            f"Row {row_index} (File Path: '{file_path}'): \n"
            if row_index else ""
        )

        folder_name = (row.get("Folder Name") or "").strip()
        if not folder_name:
            error_msg = (
                f"{row_ctx}Both 'Folder Path' and 'Folder Name' are empty. \n"
                "Provide either a 'Folder Path' or a 'Folder Name' "
                "for this row."
            )
            if prevalidation["folder_not_exists"]["mode"] == "ignore":
                return None, ("Folder Does Not Exist", error_msg)
            raise CreatorError(error_msg)

        matching = folders_by_name.get(folder_name, [])
        if not matching:
            error_msg = (
                f"\n{row_ctx}No existing folder found "
                f"with name '{folder_name}'."
                " \nVerify the 'Folder Name' value is correct in the project,"
                " or use 'Folder Path' to specify the folder directly."
            )
            if prevalidation["folder_not_exists"]["mode"] == "ignore":
                return None, ("Folder Does Not Exist", error_msg)
            raise CreatorError(error_msg)

        if len(matching) > 1:
            paths = ", ".join(f["path"] for f in matching)
            error_msg = (
                f"{row_ctx}Multiple folders share the name '{folder_name}': \n"
                f"{paths}. "
                "Use 'Folder Path' to uniquely identify which folder to use."
            )
            if prevalidation["folder_name_duplicity"]["mode"] == "ignore":
                return None, ("Folder Name Duplicity", error_msg)
            raise CreatorError(error_msg)

        matched = matching[0]
        row["Folder Path"] = matched["path"]
        log.debug(
            "Resolved folder '%s' -> '%s' from project folders.",
            folder_name,
            matched["path"],
        )

        # Fill in temporal columns from folder entity attrs when absent.
        attrib = matched.get("attrib") or {}
        attr_column_map = (
            ("Frame Start", "frameStart"),
            ("Frame End", "frameEnd"),
            ("Handle Start", "handleStart"),
            ("Handle End", "handleEnd"),
            ("FPS", "fps"),
        )
        for csv_col, attr_key in attr_column_map:
            if (row.get(csv_col) or "").strip():
                continue  # CSV already has a value
            attr_val = attrib.get(attr_key)
            if attr_val is not None:
                row[csv_col] = str(attr_val)
                log.debug(
                    "Filled '%s' = %s from folder '%s' attrib.",
                    csv_col,
                    attr_val,
                    matched["path"],
                )

        return row, None

    def _get_data_from_csv(
        self,
        preset_data: dict[str, Any],
        csv_dir: str,
        filename: str,
    ) -> tuple[dict[str, ProductItem], dict[str, list[str]]]:
        """Parse the CSV file and build product items.

        Makes two targeted API calls:
        1. If any rows have an empty ``Folder Path`` but a non-empty
           ``Folder Name``, queries only those folder names to resolve
           paths and backfill temporal attributes from folder entities.
        2. After all rows are resolved, queries the final folder paths
           for ID lookup and validation.

        When the relevant validator in ``prevalidation`` is configured with
        ``mode: \"ignore\"``, rows that fail folder resolution are skipped
        instead of raising a ``CreatorError``.
        The collected report messages are returned as the second element
        of the tuple so the caller can store them on the CSV product
        instance.

        Args:
            preset_data (dict): The selected CSV ingest preset.
            csv_dir (str): Directory containing the CSV file.
            filename (str): CSV filename.

        Returns:
            tuple: ``(product_items_by_name, report_data)`` where
                *report_data* is a ``dict[str, list[str]]`` keyed
                by report category, containing any skipped-row messages.
        """
        project_name = self.create_context.get_current_project_name()
        csv_path = os.path.join(csv_dir, filename)

        # preset's variables
        columns_config = preset_data["columns_config"]
        representations_config = preset_data["representations_config"]
        folder_creation_config = preset_data["folder_creation_config"]
        prevalidation = preset_data["prevalidation"]

        # Make sure csv file contains all required columns.
        required_columns = [
            column["name"]
            for column in columns_config["columns"]
            if column["required_column"]
        ]

        # read csv file
        with open(csv_path, "r", encoding="utf-8-sig") as csv_file:
            csv_content = csv_file.read()

        # read csv file with DictReader
        csv_reader = csv.DictReader(
            StringIO(csv_content),
            delimiter=columns_config["csv_delimiter"]
        )

        # Fix fieldnames – strip accidental leading/trailing whitespace.
        all_columns = [
            " ".join(column.rsplit())
            for column in csv_reader.fieldnames
        ]
        csv_reader.fieldnames = all_columns

        # check if csv file contains all required columns
        if any(column not in all_columns for column in required_columns):
            raise CreatorError(
                f"Missing required columns: {required_columns}\n"
                f"All columns: {all_columns}"
            )

        # Read all rows upfront so we can make targeted API calls before
        # building product items.
        raw_rows: list[dict[str, Any]] = [dict(row) for row in csv_reader]

        # Collect only the names that actually need path resolution (i.e.
        # rows where Folder Path is absent/empty but Folder Name is set).
        folder_names_to_query: set[str] = set()
        for row in raw_rows:
            folder_path = (row.get("Folder Path") or "").strip()
            folder_name = (row.get("Folder Name") or "").strip()
            if not folder_path and folder_name:
                folder_names_to_query.add(folder_name)

        folders_by_name: dict[str, list[Any]] = collections.defaultdict(list)
        if folder_names_to_query:
            for folder in ayon_api.get_folders(
                project_name,
                folder_names=folder_names_to_query,
                fields={"id", "path", "name", "attrib", "folderType"},
            ):
                folders_by_name[folder["name"]].append(folder)

        # Resolve rows and build product items.  Rows that fail folder
        # resolution or frame range validation under "ignore_and_report" mode
        # are skipped; their messages accumulate in report_data.
        product_items_by_name: dict[str, ProductItem] = {}
        report_data: dict[str, list[str]] = {}
        # CSV header is row 1, so data rows start at 2.
        for row_index, row in enumerate(raw_rows, start=2):
            # Resolve Folder Path from Folder Name when path is absent and
            # inject temporal attrs from the matched folder entity.
            resolved_row, report_entry = self._resolve_row_folder(
                row, folders_by_name, prevalidation, row_index
            )
            if resolved_row is None and report_entry is not None:
                # Row failed folder resolution; record and skip.
                category, message = report_entry
                report_data.setdefault(category, []).append(message)
                log.warning(
                    "Skipping row due to precreate validation (%s): %s",
                    category,
                    message,
                )
                continue

            # Validate that frame range values are present after folder
            # resolution (folder attribs may have been back-filled above).
            frame_range_cols = [
                "Frame Start", "Frame End",
                "Handle Start", "Handle End", "FPS",
            ]
            missing_frame_cols = [
                col for col in frame_range_cols
                if not (resolved_row.get(col) or "").strip()
            ]
            if missing_frame_cols:
                file_path = (resolved_row.get("File Path") or "").strip()
                error_msg = (
                    f"Row {row_index} (File Path: '{file_path}'): "
                    f"Missing frame range values for column(s): \n\t"
                    f"{missing_frame_cols}. \n"
                    "Provide the values in the CSV or ensure the matched "
                    "folder has these attributes set."
                )
                if prevalidation["missing_frame_range_values"]["mode"] == "ignore":  # noqa
                    report_data.setdefault(
                        "Missing Frame Range Values", []
                    ).append(error_msg)
                    log.warning(
                        "Skipping row due to missing frame range values: %s",
                        error_msg,
                    )
                    continue
                raise CreatorError(error_msg)

            product_item_: ProductItem = ProductItem.from_csv_row(
                columns_config, resolved_row
            )
            unique_name = product_item_.unique_name
            row_passing_data = _collect_passing_data_columns(
                columns_config, resolved_row
            )
            row_repre_passing_data = [
                item
                for item in row_passing_data
                if item["data_type"] == "representation_data"
            ]
            row_product_passing_data = [
                item
                for item in row_passing_data
                if item["data_type"] != "representation_data"
            ]

            if unique_name not in product_items_by_name:
                product_item_.passing_data = row_product_passing_data
                product_items_by_name[unique_name] = product_item_
            else:
                existing_product_item = product_items_by_name[unique_name]
                existing_product_item.passing_data = _merge_passing_data_values(
                    existing_product_item.passing_data,
                    row_product_passing_data,
                    unique_name,
                    row_index,
                )

            product_item: ProductItem = product_items_by_name[unique_name]
            product_item.add_repre_item(
                RepreItem.from_csv_row(
                    columns_config,
                    representations_config,
                    resolved_row,
                    passing_data=row_repre_passing_data,
                )
            )

        # Query only the resolved paths to build the ID map and detect
        # missing folders.
        folder_paths: set[str] = {
            product_item.folder_path
            for product_item in product_items_by_name.values()
        }
        folder_ids_by_path: dict[str, str] = {
            folder_entity["path"]: folder_entity["id"]
            for folder_entity in ayon_api.get_folders(
                project_name,
                folder_paths=folder_paths,
                fields={"id", "path"},
            )
        }
        missing_paths: set[str] = folder_paths - set(folder_ids_by_path.keys())

        task_names: set[str] = {
            product_item.task_name
            for product_item in product_items_by_name.values()
        }
        task_entities_by_folder_id = collections.defaultdict(list)
        for task_entity in ayon_api.get_tasks(
            project_name,
            folder_ids=set(folder_ids_by_path.values()),
            task_names=task_names,
            fields={"folderId", "name", "taskType"}
        ):
            folder_id = task_entity["folderId"]
            task_entities_by_folder_id[folder_id].append(task_entity)

        missing_tasks: set[str] = set()
        if missing_paths and not 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,
                    preset_data,
                )
                continue

            task_name = product_item.task_name
            folder_id = folder_ids_by_path[folder_path]
            task_entities = task_entities_by_folder_id[folder_id]
            task_entity = next(
                (
                    task_entity
                    for task_entity in task_entities
                    if task_entity["name"] == task_name
                ),
                None
            )
            if task_entity is None:
                missing_tasks.add(f"{folder_path}/{task_name}")
                product_item.has_promised_context = True
                product_item.task_type = None
                product_item.parents = self._compute_parents_data(
                    project_name,
                    product_item,
                    preset_data,
                )
            else:
                product_item.task_type = task_entity["taskType"]

        if not folder_creation_config["enabled"] and missing_tasks:
            ending = "" if len(missing_tasks) == 1 else "s"
            joined_paths = "\n".join(sorted(missing_tasks))
            raise CreatorError(
                f"Task{ending} not found.\n{joined_paths}"
            )

        for product_item in product_items_by_name.values():
            repre_paths: set[str] = set()
            duplicated_paths: set[str] = set()
            for repre_item in product_item.repre_items:
                # Resolve relative paths in csv file
                repre_item.filepath = self._resolve_repre_path(
                    csv_dir, repre_item.filepath
                )
                repre_item.thumbnail_path = self._resolve_repre_path(
                    csv_dir, repre_item.thumbnail_path
                )

                filepath = repre_item.filepath
                if filepath in repre_paths:
                    duplicated_paths.add(filepath)
                repre_paths.add(filepath)

            if duplicated_paths:
                ending = "" if len(duplicated_paths) == 1 else "s"
                joined_names = "\n".join(sorted(duplicated_paths))
                raise CreatorError(
                    f"Duplicate filename{ending} in csv file.\n{joined_names}"
                )

        return product_items_by_name, report_data

    def _add_thumbnail_repre(
        self,
        thumbnails: set[str],
        instance: CreatedInstance,
        repre_item: RepreItem,
        multiple_thumbnails: bool = False,
    ) -> Union[str, None]:
        """Add thumbnail to instance.

        Add thumbnail as representation and set 'thumbnailPath' if is not set
            yet.

        Args:
            thumbnails (set[str]): set of all thumbnail paths that should
                create representation.
            instance (CreatedInstance): Instance from create plugin.
            repre_item (RepreItem): Representation item.
            multiple_thumbnails (bool): There are multiple representations
                with thumbnail.

        Returns:
            Uniom[str, None]: Explicit output name for thumbnail
                representation.

        """
        if not thumbnails:
            return None

        thumbnail_path = repre_item.thumbnail_path
        if not thumbnail_path or thumbnail_path not in thumbnails:
            return None

        thumbnails.remove(thumbnail_path)

        thumb_dir, thumb_file = os.path.split(thumbnail_path)
        thumb_basename, thumb_ext = os.path.splitext(thumb_file)

        # NOTE 'explicit_output_name' and custom repre name was set only
        #   when 'multiple_thumbnails' is True and 'review' tag is present.
        # That was changed to set 'explicit_output_name' is set when
        #   'multiple_thumbnails' is True.
        # is_reviewable = "review" in repre_item.tags

        repre_name = "thumbnail"
        explicit_output_name = None
        if multiple_thumbnails:
            repre_name = f"thumbnail_{thumb_basename}"
            explicit_output_name = repre_item.name

        thumbnail_repre_data = {
            "name": repre_name,
            "ext": thumb_ext.lstrip("."),
            "files": thumb_file,
            "stagingDir": thumb_dir,
            "stagingDir_persistent": True,
            "tags": ["thumbnail", "delete"],
        }
        if explicit_output_name:
            thumbnail_repre_data["outputName"] = explicit_output_name

        instance["prepared_data_for_repres"].append({
            "type": "thumbnail",
            "colorspace": None,
            "representation": thumbnail_repre_data,
        })
        # also add thumbnailPath for ayon to integrate
        if not instance.get("thumbnailPath"):
            instance["thumbnailPath"] = thumbnail_path

        return explicit_output_name

    def _add_representation(
        self,
        preset_data: dict[str, Any],
        instance: CreatedInstance,
        repre_item: RepreItem,
        explicit_output_name: Optional[str] = None
    ) -> None:
        """Get representation data

        Args:
            preset_data (dict[str, Any]): Preset data.
            instance (CreatedInstance): Created instance.
            repre_item (RepreItem): Representation item based on csv row.
            explicit_output_name (Optional[str]): Explicit output name.
                For grouping purposes with reviewable components.

        Exception:
            CreatorError: If representation not found.
        """
        representations_config = preset_data["representations_config"]

        # get extension of file
        basename: str = os.path.basename(repre_item.filepath)
        extension: str = os.path.splitext(basename)[-1].lower()

        # validate filepath is having correct extension based on output
        repre_config_data: Union[dict[str, Any], None] = None
        for repre in representations_config["representations"]:
            if repre["name"] == repre_item.name:
                repre_config_data = repre
                break

        if not repre_config_data:
            raise CreatorError(
                f"Representation '{repre_item.name}' not found "
                "in config representation data."
            )

        validate_extensions: list[str] = repre_config_data["extensions"]
        if extension not in validate_extensions:
            raise CreatorError(
                f"File extension '{extension}' not valid for "
                f"output '{validate_extensions}'."
            )

        is_sequence: bool = extension in IMAGE_EXTENSIONS
        # convert ### string in file name to %03d
        # this is for correct frame range validation
        # example: file.###.exr -> file.%03d.exr
        file_head = basename.split(".")[0]
        if "#" in basename:
            padding = len(basename.split("#")) - 1
            seq_padding = f"%0{padding}d"
            basename = basename.replace("#" * padding, seq_padding)
            file_head = basename.split(seq_padding)[0]
            is_sequence = True
        elif "%" in basename:
            pattern = re.compile(r"%\d+d|%d")
            padding = pattern.findall(basename)
            if not padding:
                raise CreatorError(
                    f"File sequence padding not found in '{basename}'."
                )
            file_head = basename.split("%")[0]
            is_sequence = True
        else:
            # in case it is still image
            is_sequence = False

        # make absolute path to file
        dirname: str = os.path.dirname(repre_item.filepath)

        # check if dirname exists
        if not os.path.isdir(dirname):
            raise CreatorError(
                f"Directory '{dirname}' does not exist."
            )

        frame_start: Union[int, None] = None
        frame_end: Union[int, None] = None
        files: Union[str, list[str]] = basename
        if is_sequence:
            # get only filtered files form dirname
            files_from_dir = [
                filename
                for filename in os.listdir(dirname)
                if filename.startswith(file_head)
            ]
            # collect all data from dirname
            cols, _ = clique.assemble(files_from_dir)
            if not cols:
                raise CreatorError(
                    f"No collections found in directory '{dirname}'."
                )

            col = cols[0]
            files = list(col)
            frame_start = min(col.indexes)
            frame_end = max(col.indexes)

        tags: list[str] = deepcopy(repre_item.tags)
        # if slate in repre_data is True then remove one frame from start
        if repre_item.slate_exists:
            tags.append("has_slate")

        # get representation data
        representation_data: dict[str, Any] = {
            "name": repre_item.name,
            "ext": extension[1:],
            "files": files,
            "stagingDir": dirname,
            "stagingDir_persistent": True,
            "tags": tags,
        }
        if extension in VIDEO_EXTENSIONS:
            representation_data.update({
                "fps": repre_item.fps,
                "outputName": repre_item.name,
            })

        if explicit_output_name:
            representation_data["outputName"] = explicit_output_name

        repre_passing_data = {
            item["name"]: item["value"]
            for item in repre_item.passing_data
            if item["data_type"] == "representation_data"
        }
        if repre_passing_data:
            representation_data["data"] = repre_passing_data

        if frame_start:
            representation_data["frameStart"] = frame_start
        if frame_end:
            representation_data["frameEnd"] = frame_end

        instance["prepared_data_for_repres"].append({
            "type": "media",
            "colorspace": repre_item.colorspace,
            "representation": representation_data,
        })

    def _prepare_representations(
        self,
        preset_data: dict[str, Any],
        product_item: ProductItem,
        instance: CreatedInstance
    ) -> None:
        """Prepare representations for the given product item.

        Args:
            preset_data (dict[str, Any]): The preset data.
            product_item (ProductItem): The product item.
            instance (CreatedInstance): The created instance.
        """
        # Collect thumbnail paths from all representation items
        #   to check if multiple thumbnails are present.
        # Once representation is created for certain thumbnail it is removed
        #   from the set.
        thumbnails: set[str] = {
            repre_item.thumbnail_path
            for repre_item in product_item.repre_items
            if repre_item.thumbnail_path
        }
        multiple_thumbnails: bool = len(thumbnails) > 1

        for repre_item in product_item.repre_items:
            explicit_output_name = self._add_thumbnail_repre(
                thumbnails,
                instance,
                repre_item,
                multiple_thumbnails,
            )

            # get representation data
            self._add_representation(
                preset_data,
                instance,
                repre_item,
                explicit_output_name
            )

    def _get_task_type_from_task_name(
        self,
        preset_data: dict[str, Any],
        task_name: str
    ) -> str:
        """Retrieve task type from task name.

        Args:
            preset_data (dict[str, Any]): The preset data.
            task_name (str): The task name.

        Returns:
            str. The task type computed from settings.
        """
        folder_creation_config = preset_data["folder_creation_config"]

        for task_setting in folder_creation_config["task_type_regexes"]:
            if re.match(task_setting["regex"], task_name):
                task_type = task_setting["task_type"]
                break
        else:
            task_type = folder_creation_config["task_create_type"]

        return task_type

    def _create_instances_from_csv_data(
        self,
        preset_data: dict[str, Any],
        csv_dir: str,
        filename: str
    ) -> tuple[list[CreatedInstance], dict[str, list[str]]]:
        """Create instances from csv data.

        Args:
            preset_data (dict[str, Any]): The preset data.
            csv_dir (str): The directory of the CSV file.
            filename (str): The name of the CSV file.

        Returns:
            tuple: ``(instances, report_data)`` where
                *report_data* holds any folder-resolution errors
                that were configured to be reported rather than raised.
        """
        # from special function get all data from csv file and convert them
        # to new instances
        product_items_by_name, report_data = (
            self._get_data_from_csv(preset_data, csv_dir, filename)
        )

        instances = []
        project_name: str = self.create_context.get_current_project_name()
        # Pre-fetch all existing entities to find matching
        folder_paths = {
            product_item.folder_path
            for product_item in product_items_by_name.values()
        }
        folder_entities_by_path: dict[str, dict[str, str]] = {
            folder_entity["path"]: folder_entity
            for folder_entity in ayon_api.get_folders(
                project_name,
                folder_paths=folder_paths,
            )
        }
        folder_paths_by_id = {
            f["id"]: f["path"]
            for f in folder_entities_by_path.values()
        }
        task_entities = list(ayon_api.get_tasks(
            project_name,
            folder_ids=folder_paths_by_id,
            fields={"name", "taskType", "folderId"},
        ))
        task_entities_by_folder_path = collections.defaultdict(list)
        for task_entity in task_entities:
            folder_id = task_entity["folderId"]
            folder_path = folder_paths_by_id[folder_id]
            task_entities_by_folder_path[folder_path].append(task_entity)

        for product_item in product_items_by_name.values():
            folder_path: str = product_item.folder_path
            hierarchy, folder_name = folder_path.rsplit("/", 1)

            folder_entity = folder_entities_by_path.get(folder_path)
            if folder_entity is not None:
                folder_type = folder_entity["folderType"]
            else:
                # TODO: find out how to define default folder type
                # - was hardcoded in pyblish plugin 'CollectShotInstances'
                folder_type: str = "Shot"
                if product_item.has_promised_context:
                    folder_type = self._get_folder_type_from_regex_settings(
                        folder_name, preset_data["folder_creation_config"]
                    )
                # Fake folder entity, this might break if more data
                #   is used in 'get_product_name'.
                folder_entity = {
                    "name": folder_name,
                    "label": folder_name,
                    "path": folder_path,
                    "folderType": folder_type,
                }

            instance_tasks = None
            task_name = None
            task_entity = None
            task_type = None
            if product_item.task_name:
                task_name_low = product_item.task_name.lower()
                for f_task_entity in task_entities_by_folder_path[folder_path]:
                    if f_task_entity["name"].lower() == task_name_low:
                        task_entity = f_task_entity
                        break

                if task_entity is not None:
                    task_name = task_entity["name"]
                    task_type = task_entity["taskType"]
                else:
                    task_name = product_item.task_name
                    task_type = self._get_task_type_from_task_name(
                        preset_data, task_name)
                    # Fake task entity, this might break if more data
                    #   is used in 'get_product_name'.
                    task_entity = {
                        "name": task_name,
                        "label": task_name,
                        "taskType": task_type,
                    }

                if product_item.has_promised_context:
                    instance_tasks = {task_name: {"type": task_type}}

            version: int = product_item.version
            product_name: str = get_product_name(
                project_name=project_name,
                folder_entity=folder_entity,
                task_entity=task_entity,
                product_base_type=product_item.product_base_type,
                product_type=product_item.product_type,
                host_name=self.host_name,
                variant=product_item.variant,
                project_settings=(
                    self.create_context.get_current_project_settings()
                ),
                project_entity=(
                    self.create_context.get_current_project_entity()
                ),
            )

            version_label: str = "[next]"
            if version is not None:
                version_label = f"{version:>03}"
            label: str = f"{folder_path}_{product_name}_v{version_label}"

            repre_items: list[RepreItem] = product_item.repre_items
            first_repre_item: RepreItem = repre_items[0]
            version_comment: Union[str, None] = next(
                (
                    repre_item.comment
                    for repre_item in repre_items
                    if repre_item.comment
                ),
                None
            )

            slate_exists: bool = any(
                repre_item.slate_exists
                for repre_item in repre_items
            )

            is_reviewable: bool = any(
                True
                for repre_item in repre_items
                if "review" in repre_item.tags
            )

            families: list[str] = ["csv_ingest"]
            if slate_exists:
                # adding slate to families mainly for loaders to be able
                # to filter out slates
                families.append("slate")

            if is_reviewable:
                # review family needs to be added for ExtractReview plugin
                families.append("review")

            instance_data = {
                "name": product_item.instance_name,
                "label": label,
                "folderPath": folder_path,
                "csv_preset_name": preset_data["name"],
                "task": task_name,
                "folder_type": folder_type,
                "families": families,
                "variant": product_item.variant,
                "source": Path(csv_dir, filename).as_posix(),
                "frameStart": first_repre_item.frame_start,
                "frameEnd": first_repre_item.frame_end,
                "handleStart": first_repre_item.handle_start,
                "handleEnd": first_repre_item.handle_end,
                "fps": first_repre_item.fps,
                "version": version,
                "comment": version_comment,
                "prepared_data_for_repres": [],
            }

            if instance_tasks:
                instance_data["tasks"] = instance_tasks

            if product_item.passing_data:
                instance_data["csv_passing_data"] = product_item.passing_data

            if product_item.has_promised_context:
                families.append("shot")
                instance_data.update(
                    {
                        "newHierarchyIntegration": True,
                        "hierarchy": hierarchy,
                        "parents": product_item.parents,
                        "families": families,
                        "heroTrack": True,
                    }
                )

                if product_item.pixel_aspect:
                    instance_data["pixelAspect"] = product_item.pixel_aspect

                if product_item.width and product_item.height:
                    instance_data.update(
                        {
                            "resolutionWidth": product_item.width,
                            "resolutionHeight": product_item.height,
                        }
                    )
                elif product_item.width or product_item.height:
                    log.warning(
                        (
                            "Ignoring incomplete provided resolution"
                            " %rx%r for shot %s."
                        ),
                        product_item.width,
                        product_item.height,
                        folder_name
                    )

            # create new instance
            new_instance: CreatedInstance = CreatedInstance(
                product_base_type=product_item.product_base_type,
                product_type=product_item.product_type,
                product_name=product_name,
                data=instance_data,
                creator=self
            )
            self._prepare_representations(
                preset_data, product_item, new_instance)

            if product_item.has_promised_context:
                new_instance.transient_data["has_promised_context"] = True

            instances.append(new_instance)

        return instances, report_data