actions

@dataclass
class ActionForm:
    """Form for loader action.

    If an action needs to collect information from a user before or during of
        the action execution, it can return a response with a form. When the
        form is submitted, a new execution of the action is triggered.

    It is also possible to just show a label message without the submit
        button to make sure the user has seen the message.

    Attributes:
        title (str): Title of the form -> title of the window.
        fields (list[AbstractAttrDef]): Fields of the form.
        submit_label (Optional[str]): Label of the submit button. Is hidden
            if is set to None.
        submit_icon (Optional[dict[str, Any]]): Icon definition of the submit
            button.
        cancel_label (Optional[str]): Label of the cancel button. Is hidden
            if is set to None. User can still close the window tho.
        cancel_icon (Optional[dict[str, Any]]): Icon definition of the cancel
            button.

    """
    title: str
    fields: list[AbstractAttrDef]
    submit_label: Optional[str] = "Submit"
    submit_icon: Optional[dict[str, Any]] = None
    cancel_label: Optional[str] = "Cancel"
    cancel_icon: Optional[dict[str, Any]] = None

    def to_json_data(self) -> dict[str, Any]:
        fields = self.fields
        if fields is not None:
            fields = serialize_attr_defs(fields)
        return {
            "title": self.title,
            "fields": fields,
            "submit_label": self.submit_label,
            "submit_icon": self.submit_icon,
            "cancel_label": self.cancel_label,
            "cancel_icon": self.cancel_icon,
        }

    @classmethod
    def from_json_data(cls, data: dict[str, Any]) -> "ActionForm":
        fields = data["fields"]
        if fields is not None:
            data["fields"] = deserialize_attr_defs(fields)
        return cls(**data)

class InventoryAction:
    """A custom action for the scene inventory tool

    If registered the action will be visible in the Right Mouse Button menu
    under the submenu "Actions".

    """

    label = None
    icon = None
    color = None
    order = 0

    log = logging.getLogger("InventoryAction")
    log.propagate = True

    @staticmethod
    def is_compatible(container):
        """Override function in a custom class

        This method is specifically used to ensure the action can operate on
        the container.

        Args:
            container(dict): the data of a loaded asset, see host.ls()

        Returns:
            bool
        """
        return bool(container.get("objectName"))

    def process(self, containers):
        """Override function in a custom class

        This method will receive all containers even those which are
        incompatible. It is advised to create a small filter along the lines
        of this example:

        valid_containers = filter(self.is_compatible(c) for c in containers)

        The return value will need to be a True-ish value to trigger
        the data_changed signal in order to refresh the view.

        You can return a list of container names to trigger GUI to select
        treeview items.

        You can return a dict to carry extra GUI options. For example:
            {
                "objectNames": [container names...],
                "options": {"mode": "toggle",
                            "clear": False}
            }
        Currently workable GUI options are:
            - clear (bool): Clear current selection before selecting by action.
                            Default `True`.
            - mode (str): selection mode, use one of these:
                          "select", "deselect", "toggle". Default is "select".

        Args:
            containers (list): list of dictionaries

        Return:
            bool, list or dict

        """
        return True

    @classmethod
    def filepath_from_context(cls, context):
        return get_representation_path_from_context(context)

class LauncherAction(object):
    """A custom action available"""
    name = None
    label = None
    icon = None
    color = None
    order = 0

    log = logging.getLogger("LauncherAction")
    log.propagate = True

    def is_compatible(self, selection):
        """Return whether the class is compatible with the Session.

        Args:
            selection (LauncherActionSelection): Data with selection.

        """
        return True

    def process(self, selection, **kwargs):
        """Process the action.

        Args:
            selection (LauncherActionSelection): Data with selection.
            **kwargs: Additional arguments.

        """
        pass

class LauncherActionSelection:
    """Object helper to pass selection to actions.

    Object support backwards compatibility for 'session' from OpenPype where
    environment variable keys were used to define selection.

    Args:
        project_name (str): Selected project name.
        folder_id (str): Selected folder id.
        task_id (str): Selected task id.
        folder_path (Optional[str]): Selected folder path.
        task_name (Optional[str]): Selected task name.
        project_entity (Optional[dict[str, Any]]): Project entity.
        folder_entity (Optional[dict[str, Any]]): Folder entity.
        task_entity (Optional[dict[str, Any]]): Task entity.

    """
    def __init__(
        self,
        project_name,
        folder_id,
        task_id,
        workfile_id,
        folder_path=None,
        task_name=None,
        project_entity=None,
        folder_entity=None,
        task_entity=None,
        workfile_entity=None,
        project_settings=None,
    ):
        self._project_name = project_name
        self._folder_id = folder_id
        self._task_id = task_id
        self._workfile_id = workfile_id

        self._folder_path = folder_path
        self._task_name = task_name

        self._project_entity = project_entity
        self._folder_entity = folder_entity
        self._task_entity = task_entity
        self._workfile_entity = workfile_entity

        self._project_settings = project_settings

    def __getitem__(self, key):
        warnings.warn(
            (
                "Using deprecated access to selection data. Please use"
                " attributes and methods"
                " defined by 'LauncherActionSelection'."
            ),
            category=DeprecationWarning
        )
        if key in {"AYON_PROJECT_NAME", "AVALON_PROJECT"}:
            return self.project_name
        if key in {"AYON_FOLDER_PATH", "AVALON_ASSET"}:
            return self.folder_path
        if key in {"AYON_TASK_NAME", "AVALON_TASK"}:
            return self.task_name
        raise KeyError(f"Key: {key} not found")

    def __iter__(self):
        for key in self.keys():
            yield key

    def __contains__(self, key):
        warnings.warn(
            (
                "Using deprecated access to selection data. Please use"
                " attributes and methods"
                " defined by 'LauncherActionSelection'."
            ),
            category=DeprecationWarning
        )
        # Fake missing keys check for backwards compatibility
        if key in {
            "AYON_PROJECT_NAME",
            "AVALON_PROJECT",
        }:
            return self._project_name is not None
        if key in {
            "AYON_FOLDER_PATH",
            "AVALON_ASSET",
        }:
            return self._folder_id is not None
        if key in {
            "AYON_TASK_NAME",
            "AVALON_TASK",
        }:
            return self._task_id is not None
        return False

    def get(self, key, default=None):
        """

        Deprecated:
            Added for backwards compatibility with older actions.

        """
        warnings.warn(
            (
                "Using deprecated access to selection data. Please use"
                " attributes and methods"
                " defined by 'LauncherActionSelection'."
            ),
            category=DeprecationWarning
        )
        try:
            return self[key]
        except KeyError:
            return default

    def items(self):
        """

        Deprecated:
            Added for backwards compatibility with older actions.

        """
        for key, value in (
            ("AYON_PROJECT_NAME", self.project_name),
            ("AYON_FOLDER_PATH", self.folder_path),
            ("AYON_TASK_NAME", self.task_name),
        ):
            if value is not None:
                yield (key, value)

    def keys(self):
        """

        Deprecated:
            Added for backwards compatibility with older actions.

        """
        for key, _ in self.items():
            yield key

    def values(self):
        """

        Deprecated:
            Added for backwards compatibility with older actions.

        """
        for _, value in self.items():
            yield value

    def get_project_name(self):
        """Selected project name.

        Returns:
            Union[str, None]: Selected project name.

        """
        return self._project_name

    def get_folder_id(self):
        """Selected folder id.

        Returns:
            Union[str, None]: Selected folder id.

        """
        return self._folder_id

    def get_folder_path(self):
        """Selected folder path.

        Returns:
            Union[str, None]: Selected folder path.

        """
        if self._folder_id is None:
            return None
        if self._folder_path is None:
            self._folder_path = self.folder_entity["path"]
        return self._folder_path

    def get_task_id(self):
        """Selected task id.

        Returns:
            Union[str, None]: Selected task id.

        """
        return self._task_id

    def get_task_name(self):
        """Selected task name.

        Returns:
            Union[str, None]: Selected task name.

        """
        if self._task_id is None:
            return None
        if self._task_name is None:
            self._task_name = self.task_entity["name"]
        return self._task_name

    def get_workfile_id(self):
        """Selected workfile id.

        Returns:
            Union[str, None]: Selected workfile id.

        """
        return self._workfile_id

    def get_project_entity(self):
        """Project entity for the selection.

        Returns:
            Union[dict[str, Any], None]: Project entity.

        """
        if self._project_name is None:
            return None
        if self._project_entity is None:
            self._project_entity = ayon_api.get_project(self._project_name)
        return self._project_entity

    def get_folder_entity(self):
        """Folder entity for the selection.

        Returns:
            Union[dict[str, Any], None]: Folder entity.

        """
        if self._project_name is None or self._folder_id is None:
            return None
        if self._folder_entity is None:
            self._folder_entity = ayon_api.get_folder_by_id(
                self._project_name, self._folder_id
            )
        return self._folder_entity

    def get_task_entity(self):
        """Task entity for the selection.

        Returns:
            Union[dict[str, Any], None]: Task entity.

        """
        if (
            self._project_name is None
            or self._task_id is None
        ):
            return None
        if self._task_entity is None:
            self._task_entity = ayon_api.get_task_by_id(
                self._project_name, self._task_id
            )
        return self._task_entity

    def get_workfile_entity(self):
        """Workfile entity for the selection.

        Returns:
            Union[dict[str, Any], None]: Workfile entity.

        """
        if (
            self._project_name is None
            or self._workfile_id is None
        ):
            return None
        if self._workfile_entity is None:
            self._workfile_entity = ayon_api.get_workfile_info_by_id(
                self._project_name, self._workfile_id
            )
        return self._workfile_entity

    def get_project_settings(self):
        """Project settings for the selection.

        Returns:
            dict[str, Any]: Project settings or studio settings if
                project is not selected.

        """
        if self._project_settings is None:
            if self._project_name is None:
                settings = get_studio_settings()
            else:
                settings = get_project_settings(self._project_name)
            self._project_settings = settings
        return self._project_settings

    @property
    def is_project_selected(self):
        """Return whether a project is selected.

        Returns:
            bool: Whether a project is selected.

        """
        return self._project_name is not None

    @property
    def is_folder_selected(self):
        """Return whether a folder is selected.

        Returns:
            bool: Whether a folder is selected.

        """
        return self._folder_id is not None

    @property
    def is_task_selected(self):
        """Return whether a task is selected.

        Returns:
            bool: Whether a task is selected.

        """
        return self._task_id is not None

    @property
    def is_workfile_selected(self):
        """Return whether a task is selected.

        Returns:
            bool: Whether a task is selected.

        """
        return self._workfile_id is not None

    project_name = property(get_project_name)
    folder_id = property(get_folder_id)
    task_id = property(get_task_id)
    workfile_id = property(get_workfile_id)
    folder_path = property(get_folder_path)
    task_name = property(get_task_name)

    project_entity = property(get_project_entity)
    folder_entity = property(get_folder_entity)
    task_entity = property(get_task_entity)
    workfile_entity = property(get_workfile_entity)

@dataclass
class LoaderActionItem:
    """Item of loader action.

    Action plugins return these items as possible actions to run for a given
        context.

    Because the action item can be related to a specific entity
        and not the whole selection, they also have to define the entity type
        and ids to be executed on.

    Attributes:
        label (str): Text shown in UI.
        order (int): Order of the action in UI.
        group_label (Optional[str]): Label of the group to which the action
            belongs.
        icon (Optional[dict[str, Any]): Icon definition.
        data (Optional[DataType]): Action item data.
        identifier (Optional[str]): Identifier of the plugin which
            created the action item. Is filled automatically. Is not changed
            if is filled -> can lead to different plugin.

    """
    label: str
    order: int = 0
    group_label: Optional[str] = None
    icon: Optional[dict[str, Any]] = None
    data: Optional[DataType] = None
    # Is filled automatically
    identifier: str = None

class LoaderActionPlugin(ABC):
    """Plugin for loader actions.

    Plugin is responsible for getting action items and executing actions.

    """
    _log: Optional[logging.Logger] = None
    enabled: bool = True

    def __init__(self, context: "LoaderActionsContext") -> None:
        self._context = context
        self.apply_settings(context.get_studio_settings())

    def apply_settings(self, studio_settings: dict[str, Any]) -> None:
        """Apply studio settings to the plugin.

        Args:
            studio_settings (dict[str, Any]): Studio settings.

        """
        pass

    @property
    def log(self) -> logging.Logger:
        if self._log is None:
            self._log = Logger.get_logger(self.__class__.__name__)
        return self._log

    @property
    def identifier(self) -> str:
        """Identifier of the plugin.

        Returns:
            str: Plugin identifier.

        """
        return self.__class__.__name__

    @property
    def host_name(self) -> Optional[str]:
        """Name of the current host."""
        return self._context.get_host_name()

    @abstractmethod
    def get_action_items(
        self, selection: LoaderActionSelection
    ) -> list[LoaderActionItem]:
        """Action items for the selection.

        Args:
            selection (LoaderActionSelection): Selection.

        Returns:
            list[LoaderActionItem]: Action items.

        """
        pass

    @abstractmethod
    def execute_action(
        self,
        selection: LoaderActionSelection,
        data: Optional[DataType],
        form_values: dict[str, Any],
    ) -> Optional[LoaderActionResult]:
        """Execute an action.

        Args:
            selection (LoaderActionSelection): Selection wrapper. Can be used
                to get entities or get context of original selection.
            data (Optional[DataType]): Additional action item data.
            form_values (dict[str, Any]): Attribute values.

        Returns:
            Optional[LoaderActionResult]: Result of the action execution.

        """
        pass

@dataclass
class LoaderActionResult:
    """Result of loader action execution.

    Attributes:
        message (Optional[str]): Message to show in UI.
        success (bool): If the action was successful. Affects color of
            the message.
        form (Optional[ActionForm]): Form to show in UI.
        form_values (Optional[dict[str, Any]]): Values for the form. Can be
            used if the same form is re-shown e.g. because a user forgot to
            fill a required field.

    """
    message: Optional[str] = None
    success: bool = True
    form: Optional[ActionForm] = None
    form_values: Optional[dict[str, Any]] = None

    def to_json_data(self) -> dict[str, Any]:
        form = self.form
        if form is not None:
            form = form.to_json_data()
        return {
            "message": self.message,
            "success": self.success,
            "form": form,
            "form_values": self.form_values,
        }

    @classmethod
    def from_json_data(cls, data: dict[str, Any]) -> "LoaderActionResult":
        form = data["form"]
        if form is not None:
            data["form"] = ActionForm.from_json_data(form)
        return LoaderActionResult(**data)

class LoaderActionSelection:
    """Selection of entities for loader actions.

    Selection tells action plugins what exactly is selected in the tool and
        which ids.

    Contains entity cache which can be used to get entities by their ids. Or
        to get project settings and anatomy.

    """
    def __init__(
        self,
        project_name: str,
        selected_ids: set[str],
        selected_type: LoaderSelectedType,
        *,
        project_anatomy: Optional[Anatomy] = None,
        project_settings: Optional[dict[str, Any]] = None,
        entities_cache: Optional[SelectionEntitiesCache] = None,
    ):
        self._project_name = project_name
        self._selected_ids = selected_ids
        self._selected_type = selected_type

        self._project_anatomy = project_anatomy
        self._project_settings = project_settings

        if entities_cache is None:
            entities_cache = SelectionEntitiesCache(project_name)
        self._entities_cache = entities_cache

    def get_entities_cache(self) -> SelectionEntitiesCache:
        return self._entities_cache

    def get_project_name(self) -> str:
        return self._project_name

    def get_selected_ids(self) -> set[str]:
        return set(self._selected_ids)

    def get_selected_type(self) -> str:
        return self._selected_type

    def get_project_settings(self) -> dict[str, Any]:
        if self._project_settings is None:
            self._project_settings = get_project_settings(self._project_name)
        return copy.deepcopy(self._project_settings)

    def get_project_anatomy(self) -> Anatomy:
        if self._project_anatomy is None:
            self._project_anatomy = Anatomy(
                self._project_name,
                project_entity=self.get_entities_cache().get_project(),
            )
        return self._project_anatomy

    project_name = property(get_project_name)
    selected_ids = property(get_selected_ids)
    selected_type = property(get_selected_type)
    project_settings = property(get_project_settings)
    project_anatomy = property(get_project_anatomy)
    entities = property(get_entities_cache)

    # --- Helper methods ---
    def versions_selected(self) -> bool:
        """Selected entity type is version.

        Returns:
            bool: True if selected entity type is version.

        """
        return self._selected_type == LoaderSelectedType.version

    def representations_selected(self) -> bool:
        """Selected entity type is representation.

        Returns:
            bool: True if selected entity type is representation.

        """
        return self._selected_type == LoaderSelectedType.representation

    def get_selected_version_entities(self) -> list[dict[str, Any]]:
        """Retrieve selected version entities.

        An empty list is returned if 'version' is not the selected
            entity type.

        Returns:
            list[dict[str, Any]]: List of selected version entities.

        """
        if self.versions_selected():
            return self.entities.get_versions(self.selected_ids)
        return []

    def get_selected_representation_entities(self) -> list[dict[str, Any]]:
        """Retrieve selected representation entities.

        An empty list is returned if 'representation' is not the selected
            entity type.

        Returns:
            list[dict[str, Any]]: List of selected representation entities.

        """
        if self.representations_selected():
            return self.entities.get_representations(self.selected_ids)
        return []

class LoaderActionsContext:
    """Wrapper for loader actions and their logic.

    Takes care about the public api of loader actions and internal logic like
        discovery and initialization of plugins.

    """
    def __init__(
        self,
        studio_settings: Optional[dict[str, Any]] = None,
        addons_manager: Optional[AddonsManager] = None,
        host: Optional[AbstractHost] = _PLACEHOLDER,
    ) -> None:
        self._log = Logger.get_logger(self.__class__.__name__)

        self._addons_manager = addons_manager
        self._host = host

        # Attributes that are re-cached on reset
        self._studio_settings = studio_settings
        self._plugins = None

    def reset(
        self, studio_settings: Optional[dict[str, Any]] = None
    ) -> None:
        """Reset context cache.

        Reset plugins and studio settings to reload them.

        Notes:
             Does not reset the cache of AddonsManger because there should not
                be a reason to do so.

        """
        self._studio_settings = studio_settings
        self._plugins = None

    def get_addons_manager(self) -> AddonsManager:
        if self._addons_manager is None:
            self._addons_manager = AddonsManager(
                settings=self.get_studio_settings()
            )
        return self._addons_manager

    def get_host(self) -> Optional[AbstractHost]:
        """Get current host integration.

        Returns:
            Optional[AbstractHost]: Host integration. Can be None if host
                integration is not registered -> probably not used in the
                host integration process.

        """
        if self._host is _PLACEHOLDER:
            from ayon_core.pipeline import registered_host

            self._host = registered_host()
        return self._host

    def get_host_name(self) -> Optional[str]:
        host = self.get_host()
        if host is None:
            return None
        return host.name

    def get_studio_settings(self) -> dict[str, Any]:
        if self._studio_settings is None:
            self._studio_settings = get_studio_settings()
        return copy.deepcopy(self._studio_settings)

    def get_action_items(
        self, selection: LoaderActionSelection
    ) -> list[LoaderActionItem]:
        """Collect action items from all plugins for given selection.

        Args:
            selection (LoaderActionSelection): Selection wrapper.

        """
        output = []
        for plugin_id, plugin in self._get_plugins().items():
            try:
                for action_item in plugin.get_action_items(selection):
                    if action_item.identifier is None:
                        action_item.identifier = plugin_id
                    output.append(action_item)

            except Exception:
                self._log.warning(
                    "Failed to get action items for"
                    f" plugin '{plugin.identifier}'",
                    exc_info=True,
                )
        return output

    def execute_action(
        self,
        identifier: str,
        selection: LoaderActionSelection,
        data: Optional[DataType],
        form_values: dict[str, Any],
    ) -> Optional[LoaderActionResult]:
        """Trigger action execution.

        Args:
            identifier (str): Identifier of the plugin.
            selection (LoaderActionSelection): Selection wrapper. Can be used
                to get what is selected in UI and to get access to entity
                cache.
            data (Optional[DataType]): Additional action item data.
            form_values (dict[str, Any]): Form values related to action.
                Usually filled if action returned response with form.

        """
        plugins_by_id = self._get_plugins()
        plugin = plugins_by_id[identifier]
        return plugin.execute_action(
            selection,
            data,
            form_values,
        )

    def _get_plugins(self) -> dict[str, LoaderActionPlugin]:
        if self._plugins is None:
            host_name = self.get_host_name()
            addons_manager = self.get_addons_manager()
            all_paths = [
                os.path.join(AYON_CORE_ROOT, "plugins", "loader")
            ]
            for addon in addons_manager.addons:
                if not isinstance(addon, IPluginPaths):
                    continue

                try:
                    if is_func_signature_supported(
                        addon.get_loader_action_plugin_paths,
                        host_name
                    ):
                        paths = addon.get_loader_action_plugin_paths(
                            host_name
                        )
                    else:
                        paths = addon.get_loader_action_plugin_paths()
                except Exception:
                    self._log.warning(
                        "Failed to get plugin paths for addon",
                        exc_info=True
                    )
                    continue

                if paths:
                    all_paths.extend(paths)

            result = discover_plugins(LoaderActionPlugin, all_paths)
            result.log_report()
            plugins = {}
            for cls in result.plugins:
                try:
                    plugin = cls(self)
                    if not plugin.enabled:
                        continue

                    plugin_id = plugin.identifier
                    if plugin_id not in plugins:
                        plugins[plugin_id] = plugin
                        continue

                    self._log.warning(
                        f"Duplicated plugins identifier found '{plugin_id}'."
                    )

                except Exception:
                    self._log.warning(
                        f"Failed to initialize plugin '{cls.__name__}'",
                        exc_info=True,
                    )
            self._plugins = plugins
        return self._plugins

class LoaderSelectedType(StrEnum):
    """Selected entity type."""
    # folder = "folder"
    # task = "task"
    version = "version"
    representation = "representation"

class LoaderSimpleActionPlugin(LoaderActionPlugin):
    """Simple action plugin.

    This action will show exactly one action item defined by attributes
        on the class.

    Attributes:
        label: Label of the action item.
        order: Order of the action item.
        group_label: Label of the group to which the action belongs.
        icon: Icon definition shown next to label.

    """

    label: Optional[str] = None
    order: int = 0
    group_label: Optional[str] = None
    icon: Optional[dict[str, Any]] = None

    @abstractmethod
    def is_compatible(self, selection: LoaderActionSelection) -> bool:
        """Check if plugin is compatible with selection.

        Args:
            selection (LoaderActionSelection): Selection information.

        Returns:
            bool: True if plugin is compatible with selection.

        """
        pass

    @abstractmethod
    def execute_simple_action(
        self,
        selection: LoaderActionSelection,
        form_values: dict[str, Any],
    ) -> Optional[LoaderActionResult]:
        """Process action based on selection.

        Args:
            selection (LoaderActionSelection): Selection information.
            form_values (dict[str, Any]): Values from a form if there are any.

        Returns:
            Optional[LoaderActionResult]: Result of the action.

        """
        pass

    def get_action_items(
        self, selection: LoaderActionSelection
    ) -> list[LoaderActionItem]:
        if self.is_compatible(selection):
            label = self.label or self.__class__.__name__
            return [
                LoaderActionItem(
                    label=label,
                    order=self.order,
                    group_label=self.group_label,
                    icon=self.icon,
                )
            ]
        return []

    def execute_action(
        self,
        selection: LoaderActionSelection,
        data: Optional[DataType],
        form_values: dict[str, Any],
    ) -> Optional[LoaderActionResult]:
        return self.execute_simple_action(selection, form_values)