publish

class AYONPyblishPluginMixin:
    # TODO
    # executable_in_thread = False
    #
    # state_message = None
    # state_percent = None
    # _state_change_callbacks = []
    #
    # def set_state(self, percent=None, message=None):
    #     """Inner callback of plugin that would help to show in UI state.
    #
    #     Plugin have registered callbacks on state change which could trigger
    #     update message and percent in UI and repaint the change.
    #
    #     This part must be optional and should not be used to display errors
    #     or for logging.
    #
    #     Message should be short without details.
    #
    #     Args:
    #         percent(int): Percent of processing in range <1-100>.
    #         message(str): Message which will be shown to user (if in UI).
    #     """
    #     if percent is not None:
    #         self.state_percent = percent
    #
    #     if message:
    #         self.state_message = message
    #
    #     for callback in self._state_change_callbacks:
    #         callback(self)

    @classmethod
    def register_create_context_callbacks(
        cls, create_context: "CreateContext"
    ):
        """Register callbacks for create context.

        It is possible to register callbacks listening to changes happened
        in create context.

        Methods available on create context:
        - add_instances_added_callback
        - add_instances_removed_callback
        - add_value_changed_callback
        - add_pre_create_attr_defs_change_callback
        - add_create_attr_defs_change_callback
        - add_publish_attr_defs_change_callback

        Args:
            create_context (CreateContext): Create context.

        """
        pass

    @classmethod
    def get_attribute_defs(cls):
        """Publish attribute definitions.

        Attributes available for all families in plugin's `families` attribute.

        Returns:
            list[AbstractAttrDef]: Attribute definitions for plugin.

        """
        return []

    @classmethod
    def get_attr_defs_for_context(cls, create_context: "CreateContext"):
        """Publish attribute definitions for context.

        Attributes available for all families in plugin's `families` attribute.

        Args:
            create_context (CreateContext): Create context.

        Returns:
            list[AbstractAttrDef]: Attribute definitions for plugin.

        """
        if cls.__instanceEnabled__:
            return []
        return cls.get_attribute_defs()

    @classmethod
    def instance_matches_plugin_families(
        cls, instance: Optional["CreatedInstance"]
    ):
        """Check if instance matches families.

        Args:
            instance (Optional[CreatedInstance]): Instance to check. Or None
                for context.

        Returns:
            bool: True if instance matches plugin families.

        """
        if instance is None:
            return not cls.__instanceEnabled__

        if not cls.__instanceEnabled__:
            return False

        families = [instance.product_base_type]
        families.extend(instance.families)
        for _ in pyblish.logic.plugins_by_families([cls], families):
            return True
        return False

    @classmethod
    def get_attr_defs_for_instance(
        cls, create_context: "CreateContext", instance: "CreatedInstance"
    ):
        """Publish attribute definitions for an instance.

        Attributes available for all families in plugin's `families` attribute.

        Args:
            create_context (CreateContext): Create context.
            instance (CreatedInstance): Instance for which attributes are
                collected.

        Returns:
            list[AbstractAttrDef]: Attribute definitions for plugin.

        """
        if not cls.instance_matches_plugin_families(instance):
            return []
        return cls.get_attribute_defs()

    @classmethod
    def convert_attribute_values(
        cls, create_context: "CreateContext", instance: "CreatedInstance"
    ):
        """Convert attribute values for instance.

        Args:
            create_context (CreateContext): Create context.
            instance (CreatedInstance): Instance for which attributes are
                converted.

        """
        return

    @staticmethod
    def get_attr_values_from_data_for_plugin(plugin, data):
        """Get attribute values for attribute definitions from data.

        Args:
            plugin (Union[publish.api.Plugin, Type[publish.api.Plugin]]): The
                plugin for which attributes are extracted.
            data(dict): Data from instance or context.
        """

        if not inspect.isclass(plugin):
            plugin = plugin.__class__

        return (
            data
            .get("publish_attributes", {})
            .get(plugin.__name__, {})
        )

    def get_attr_values_from_data(self, data):
        """Get attribute values for attribute definitions from data.

        Args:
            data(dict): Data from instance or context.
        """

        return self.get_attr_values_from_data_for_plugin(self.__class__, data)

class AbstractCollectRender(
    pyblish.api.ContextPlugin, metaclass=AbstractMetaContextPlugin
):
    """Gather all publishable render layers from renderSetup."""

    order = pyblish.api.CollectorOrder + 0.01
    label = "Collect Render"
    sync_workfile_version = False

    def __init__(self, *args, **kwargs):
        """Constructor."""
        super(AbstractCollectRender, self).__init__(*args, **kwargs)
        self._file_path = None
        self._context = None

    def process(self, context):
        """Entry point to collector."""
        self._context = context
        for instance in context:
            # make sure workfile instance publishing is enabled
            try:
                if "workfile" in instance.data["families"]:
                    instance.data["publish"] = True
                # TODO merge renderFarm and render.farm
                if ("renderFarm" in instance.data["families"] or
                        "render.farm" in instance.data["families"]):
                    instance.data["remove"] = True
            except KeyError:
                # be tolerant if 'families' is missing.
                pass

        self._file_path = context.data["currentFile"].replace("\\", "/")

        render_instances = self.get_instances(context)
        for render_instance in render_instances:
            exp_files = self.get_expected_files(render_instance)
            assert exp_files, "no file names were generated, this is bug"

            # if we want to attach render to product, check if we have AOV's
            # in expectedFiles. If so, raise error as we cannot attach AOV
            # (considered to be product on its own) to another product
            if render_instance.attachTo:
                assert isinstance(exp_files, list), (
                    "attaching multiple AOVs or renderable cameras to "
                    "product is not supported"
                )

            frame_start_render = int(render_instance.frameStart)
            frame_end_render = int(render_instance.frameEnd)
            # TODO: Refactor hacky frame range workaround below
            if (render_instance.ignoreFrameHandleCheck or
                    int(context.data['frameStartHandle']) == frame_start_render
                    and int(context.data['frameEndHandle']) == frame_end_render):  # noqa: W503, E501
                # only for Harmony where frame range cannot be set by DB
                handle_start = context.data['handleStart']
                handle_end = context.data['handleEnd']
                frame_start = context.data['frameStart']
                frame_end = context.data['frameEnd']
                frame_start_handle = context.data['frameStartHandle']
                frame_end_handle = context.data['frameEndHandle']
            elif (hasattr(render_instance, "frameStartHandle")
                  and hasattr(render_instance, "frameEndHandle")):
                handle_start = int(render_instance.handleStart)
                handle_end = int(render_instance.handleEnd)
                frame_start = int(render_instance.frameStart)
                frame_end = int(render_instance.frameEnd)
                frame_start_handle = int(render_instance.frameStartHandle)
                frame_end_handle = int(render_instance.frameEndHandle)
            else:
                handle_start = 0
                handle_end = 0
                frame_start = frame_start_render
                frame_end = frame_end_render
                frame_start_handle = frame_start_render
                frame_end_handle = frame_end_render

            data = {
                "handleStart": handle_start,
                "handleEnd": handle_end,
                "frameStart": frame_start,
                "frameEnd": frame_end,
                "frameStartHandle": frame_start_handle,
                "frameEndHandle": frame_end_handle,
                "byFrameStep": int(render_instance.frameStep),

                "author": context.data["user"],
                # Add source to allow tracing back to the scene from
                # which was submitted originally
                "expectedFiles": exp_files,
            }
            if self.sync_workfile_version:
                data["version"] = context.data["version"]

            # add additional data
            data = self.add_additional_data(data)

            instance = render_instance.source_instance
            if instance is None:
                instance = context.create_instance(render_instance.name)

            render_instance_dict = attr.asdict(render_instance)
            instance.data.update(render_instance_dict)
            instance.data.update(data)

        self.post_collecting_action()

    @abstractmethod
    def get_instances(self, context):
        """Get all renderable instances and their data.

        Args:
            context (pyblish.api.Context): Context object.

        Returns:
            list of :class:`RenderInstance`: All collected renderable instances
                (like render layers, write nodes, etc.)

        """
        pass

    @abstractmethod
    def get_expected_files(self, render_instance):
        """Get list of expected files.

        Returns:
            list: expected files. This can be either simple list of files with
                their paths, or list of dictionaries, where key is name of AOV
                for example and value is list of files for that AOV.

        Example::

            ['/path/to/file.001.exr', '/path/to/file.002.exr']

            or as dictionary:

            [
                {
                    "beauty": ['/path/to/beauty.001.exr', ...],
                    "mask": ['/path/to/mask.001.exr']
                }
            ]

        """
        pass

    def add_additional_data(self, data):
        """Add additional data to collected instance.

        This can be overridden by host implementation to add custom
        additional data.

        """
        return data

    def post_collecting_action(self):
        """Execute some code after collection is done.

        This is useful for example for restoring current render layer.

        """
        pass

class ColormanagedPyblishPluginMixin(object):
    """Mixin for colormanaged plugins.

    This class is used to set colorspace data to a publishing
    representation. It contains a static method,
    get_colorspace_settings, which returns config and
    file rules data for the host context.
    It also contains a method, set_representation_colorspace,
    which sets colorspace data to the representation.
    The allowed file extensions are listed in the allowed_ext variable.
    The method first checks if the file extension is in
    the list of allowed extensions. If it is, it then gets the
    colorspace settings from the host context and gets a
    matching colorspace from rules. Finally, it infuses this
    data into the representation.
    """

    def get_colorspace_settings(self, context):
        """Returns solved settings for the host context.

        Args:
            context (publish.Context): publishing context

        Returns:
            tuple | bool: config, file rules or None
        """
        return get_colorspace_settings_from_publish_context(context.data)

    def set_representation_colorspace(
        self, representation, context,
        colorspace=None,
    ):
        """Sets colorspace data to representation.

        Args:
            representation (dict): publishing representation
            context (publish.Context): publishing context
            colorspace (str, optional): colorspace name. Defaults to None.

        Example:
            ```
            {
                # for other publish plugins and loaders
                "colorspace": "linear",
                "config": {
                    # for future references in case need
                    "path": "/abs/path/to/config.ocio",
                    # for other plugins within remote publish cases
                    "template": "{project[root]}/path/to/config.ocio"
                }
            }
            ```

        """

        # using cached settings if available
        set_colorspace_data_to_representation(
            representation, context.data,
            colorspace,
            log=self.log
        )

class ExpectedFiles(ABC):
    """Class grouping functionality for all supported renderers.

    Attributes:
        multipart (bool): Flag if multipart exrs are used.

    """

    multipart = False

    @abstractmethod
    def get(self, render_instance):
        """Get expected files for given renderer and render layer.

        This method should return dictionary of all files we are expecting
        to be rendered from the host. Usually `render_instance` corresponds
        to *render layer*. Result can be either flat list with the file
        paths or it can be list of dictionaries. Each key corresponds to
        for example AOV name or channel, etc.

        Example::

            ['/path/to/file.001.exr', '/path/to/file.002.exr']

            or as dictionary:

            [
                {
                    "beauty": ['/path/to/beauty.001.exr', ...],
                    "mask": ['/path/to/mask.001.exr']
                }
            ]


        Args:
            render_instance (:class:`RenderInstance`): Data passed from
                collector to determine files. This should be instance of
                :class:`abstract_collect_render.RenderInstance`

        Returns:
            list: Full paths to expected rendered files.
            list of dict: Path to expected rendered files categorized by
                AOVs, etc.

        """
        raise NotImplementedError()

class Extractor(pyblish.api.InstancePlugin):
    """Extractor base class.

    The extractor base class implements a "staging_dir" function used to
    generate a temporary directory for an instance to extract to.

    This temporary directory is generated through `tempfile.mkdtemp()`

    """

    order = 2.0

    def staging_dir(self, instance):
        """Provide a temporary directory in which to store extracted files

        Upon calling this method the staging directory is stored inside
        the instance.data['stagingDir']
        """

        return get_instance_staging_dir(instance)

class IntegrationTemplateItem:
    """Represents single template item.

    Template path, template data that was used in the template.

    Attributes:
        anatomy (Anatomy): Anatomy object.
        template_data (dict[str, Any]): Template data.
        template_object (AnatomyTemplateItem): Template object
    """
    anatomy: Anatomy
    template_data: dict[str, Any]
    template_object: "AnatomyTemplateItem"

    def __init__(self,
        anatomy: "Anatomy",
        template_data: dict[str, Any],
        template_object: "AnatomyTemplateItem",
    ) -> None:
        """Initialize TemplateItem.

        Args:
            anatomy (Anatomy): Anatomy object.
            template_data (dict[str, Any]): Template data.
            template_object (AnatomyTemplateItem): Template object.

        """
        self.anatomy = anatomy
        self.template_data = template_data
        self.template_object = template_object

class KnownPublishError(Exception):
    """Publishing crashed because of known error.

    Artist can't affect source of the error.

    Deprecated:
        Please use `PublishError` instead. Marked as deprecated 24/09/02.

    """
    pass

class OptionalPyblishPluginMixin(AYONPyblishPluginMixin):
    """Prepare mixin for optional plugins.

    Defined active attribute definition prepared for published and
    prepares method which will check if is active or not.

    ```
    class ValidateScene(
        pyblish.api.InstancePlugin, OptionalPyblishPluginMixin
    ):
        def process(self, instance):
            # Skip the instance if is not active by data on the instance
            if not self.is_active(instance.data):
                return
    ```
    """

    # Allow exposing tooltip from class with `optional_tooltip` attribute
    optional_tooltip: Optional[str] = None

    @classmethod
    def get_attribute_defs(cls):
        """Attribute definitions based on plugin's optional attribute."""

        # Empty list if plugin is not optional
        if not getattr(cls, "optional", None):
            return []

        # Get active value from class as default value
        active = getattr(cls, "active", True)
        # Return boolean stored under 'active' key with label of the class name
        label = cls.label or cls.__name__

        return [
            BoolDef(
                "active",
                default=active,
                label=label,
                tooltip=cls.optional_tooltip,
            )
        ]

    def is_active(self, data):
        """Check if plugins is active for instance/context based on their data.

        Args:
            data(dict): Data from instance or context.
        """
        # Skip if is not optional and return True
        if not getattr(self, "optional", None):
            return True
        attr_values = self.get_attr_values_from_data(data)
        active = attr_values.get("active")
        if active is None:
            active = getattr(self, "active", True)
        return active

@dataclass
class PublishActionResult:
    """Result of triggered action.

    Attributes:
        success (bool): Whether action was successful.
        action (ActionType): Triggered action.
        exception (Exception | None): Exception that happened if success is
            'False'.

    """
    success: bool
    action: ActionType
    exception: Exception | None = None

class PublishError(Exception):
    """Publishing crashed because of known error.

    Message will be shown in UI for artist.

    Args:
        message (str): Message of error. Short explanation an issue.
        title (Optional[str]): Title showed in UI.
        description (Optional[str]): Detailed description of an error.
            It is possible to use Markdown syntax.

    """
    def __init__(self, message, title=None, description=None, detail=None):
        self.message = message
        self.title = title
        self.description = description or message
        self.detail = detail
        super().__init__(message)

class PublishLogic:
    """Wrapper for publising logic.

    This class is used to manage publishing logic. Can be used to run
        publishing and then create publish report.

    Args:
        reset (bool): Reset publish logic with current context
            on initialization. If False, an explicit reset has to be called
            before publishing.

    """
    def __init__(self, *, reset=True) -> None:
        self._log_handler: MessageHandler = MessageHandler()

        self._log_to_console: bool = True

        self._publish_state: PublishState = PublishState()

        self._publish_plugins: list[PluginType] = []
        self._publish_plugins_by_id: dict[str, PluginType] = {}
        self._publish_plugins_by_name: dict[str, PluginType] = {}

        # pyblish.api.Context
        self._pyblish_context: pyblish.api.Context = pyblish.api.Context()
        # Pyblish report
        self._publish_report: PublishReportMaker = PublishReportMaker()

        # Plugin iterator
        self._publish_iterator: _PublishIterator = _PublishIterator()

        if reset:
            self.reset_with_current_context()

    def reset_with_current_context(self) -> None:
        """Reset publish logic with current context."""
        from ayon_core.pipeline.create import CreateContext
        from ayon_core.pipeline.context_tools import (
            registered_host,
            get_current_project_name,
        )

        host = registered_host()
        create_context = None
        if host is not None:
            create_context = CreateContext(host)

        if create_context is not None:
            self.reset_from_create_context(create_context)
            return

        project_name = get_current_project_name()
        self.reset(project_name)

    def reset(
        self,
        project_name: str,
        context: pyblish.api.Context | None = None,
        plugins: list[PluginType] | None = None,
        targets: list[str] | None = None,
        create_context: CreateContext | None = None,
        publish_discover_result: DiscoverResult | None = None,
        project_settings: dict[str, Any] | None = None,
    ) -> None:
        """Reset publish logic with given parameters.

        Args:
            project_name (str): Name of the project.
            context (pyblish.api.Context | None): Pyblish context to use.
                If None, a new context will be created.
            plugins (list[PluginType] | None): List of publish plugins to use.
                If None, plugins will be discovered.
            targets (list[str] | None): List of publish targets to use.
                If None, all registered targets will be used.
            create_context (CreateContext | None): Create context to use.
                If None, a new create context will be created if possible.
            publish_discover_result (DiscoverResult | None): Discover result
                for publish plugins. If None, plugins will be discovered.

        """
        if plugins is not None and publish_discover_result is None:
            publish_discover_result = DiscoverResult(pyblish.api.Plugin)
            publish_discover_result.plugins = plugins

        self._reset(
            project_name,
            publish_context=context,
            publish_targets=targets,
            create_context=create_context,
            publish_discover_result=publish_discover_result,
            project_settings=project_settings,
        )

    def reset_from_create_context(
        self,
        create_context: CreateContext,
        context: pyblish.api.Context | None = None,
        targets: list[str] | None = None,
    ) -> None:
        """Reset publish logic with given create context.

        Args:
            create_context (CreateContext): Create context to use.
            context (pyblish.api.Context | None): Pyblish context to use.
                If None, a new context will be created.
            targets (list[str] | None): List of publish targets to use.
                If None, all registered targets will be used.

        """
        self._reset(
            create_context.project_name,
            publish_context=context,
            publish_targets=targets,
            publish_discover_result=create_context.publish_discover_result,
            creator_discover_result=create_context.creator_discover_result,
            convertor_discover_result=create_context.convertor_discover_result,
            create_context=create_context,
        )

    @staticmethod
    def get_publish_plugin_id(plugin: PluginType) -> str:
        """Get id of plugin based on plugin object.

        It's used for validation errors report.

        Args:
            plugin (PluginType): Publish plugin for which id should
                be returned.

        Returns:
            str: Plugin id.

        """
        return plugin.id

    def get_pyblish_context(self) -> pyblish.api.Context:
        """Get pyblish context used for publishing."""
        return self._pyblish_context

    pyblish_context: pyblish.api.Context = property(get_pyblish_context)

    def get_publish_plugin_by_id(
        self, plugin_id: str
    ) -> PluginType | None:
        return self._publish_plugins_by_id.get(plugin_id)

    def get_publish_plugin_by_name(
        self, plugin_name: str
    ) -> PluginType | None:
        return self._publish_plugins_by_name.get(plugin_name)

    def set_log_to_console(self, value: bool) -> None:
        """Set whether to log to console."""
        self._log_to_console = value

    def iter_publish_plugins(self) -> Iterable[PluginType]:
        """Iterate over publish plugins.

        Give developers access to publish plugins that are used in current
            publishing process. Can be used to change their attributes, e.g.
            to disable them.

        Returns:
            Iterable[PluginType]: Iterable with publish plugins.

        """
        yield from self._publish_plugins

    def publish(self) -> None:
        for info in self.publish_iter():
            info()

    def publish_iter(self) -> Generator[PublishIterTask, None, None]:
        """Publish iteration generator.

        Yields:
            PublishIterTask: Information about next processing function in
                publishing. Can be called to process the next function.

        """
        self._publish_state.is_running = True
        self._publish_state.started = True
        for task in self._publish_iterator:
            yield task

    def get_stop_after_validation(self) -> bool:
        """Check if publishing will stop after validation.

        Returns:
            bool: True if publishing will stop after validation,
                False otherwise.

        """
        return self._publish_state.stop_after_validation

    def set_stop_after_validation(self, value: bool) -> None:
        """Publishing will stop after validation.

        Args:
            value (bool): If True, publishing will stop after validation.

        """
        self._publish_state.stop_after_validation = value

    def set_strict_validation_error_handling(self, value: bool) -> None:
        """Set strict validation error handling.

        Args:
            value (bool): If True, publishing will stop after validation.

        """
        self._publish_state.strict_validation_error_handling = value

    def stop_publish(self) -> None:
        """Mark publishing to stop on next iteration.

        This is useful for UI purposes.
        """
        self._publish_state.is_running = False

    def is_running(self) -> bool:
        """Check if publishing is currently meant to be running.

        It is possible that publishing is currently in process of stopping,
            but this method will return False.

        Returns:
            bool: True if publishing is currently meant to be running,
                False otherwise.

        """
        return self._publish_state.is_running

    def has_started(self) -> bool:
        """Publishing has started."""
        return self._publish_state.started

    def has_finished(self) -> bool:
        """Publishing has successfully finished.

        It is not possible that 'has_finished' is True and 'has_failed'
            is True at the same time.

        Returns:
            bool: True if publishing has successfully finished,
                False otherwise.

        """
        return self._publish_state.finished

    def has_validated(self) -> bool:
        """Validation order passed.

        Returns:
            bool: True if validation order passed, False otherwise.

        """
        return self._publish_state.validated

    def publish_can_continue(self) -> bool:
        """Publishing can continue.

        Returns:
            bool: True if publishing still can continue, False otherwise.

        """
        return self._publish_state.can_continue()

    def get_progress(self) -> int:
        """Get current progress of publishing.

        Returns:
            int: Current progress of publishing. Value is between 0 and
                max progress. 0 means no plugin was processed yet.

        """
        return self._publish_state.progress

    def get_max_progress(self) -> int:
        """Get max progress of publishing."""
        return self._publish_state.max_progress

    def has_validation_errors(self) -> bool:
        """Check if publishing has validation errors."""
        return bool(self._publish_state.validation_errors)

    def has_failed(self) -> bool:
        """Check if publishing has failed.

        It is possible that publishing even didn't start, or still can
            continue even if publishing has crashed. Don't use this method
            to check if publishing should be running.

        Returns:
            bool: True if publishing has crashed, False otherwise.

        """
        return self._publish_state.fail_reason != PublishFailReason.No

    def get_fail_reason(self) -> PublishFailReason:
        """Get the reason for publishing failure.

        Returns:
            PublishFailReason: The reason for publishing failure.

        """
        return self._publish_state.fail_reason

    def get_publish_report(self) -> PublishReport:
        """Get report for the current state of publishing."""
        self._publish_report.update_publish_instances(self._pyblish_context)
        return self._publish_report.get_report()

    def get_publish_report_data(self) -> dict[str, Any]:
        """Extract publish report data for the current state of publishing.

        This method should be used if report data are being stored to a file.
            It does mark the current plugin as 'passed' for UI report viewer.

        Returns:
            dict[str, Any]: Publish report data.

        """
        report: PublishReport = self.get_publish_report()
        return report.to_data(self._publish_report.current_plugin_id)

    def store_publish_report(self, filepath: str) -> None:
        """Store publish report to a file."""
        report: PublishReport = self.get_publish_report()
        report.store_to_file(
            filepath,
            self._publish_report.current_plugin_id
        )

    def get_error_info(self) -> PublishErrorInfo | None:
        """Get information about publish error.

        Use to get error if fail reason is 'Error'.

        Returns:
            PublishErrorInfo | None: Information about publish error or None
                if no error happened.

        """
        return self._publish_state.error_info

    def get_validation_errors_info(self) -> list[PublishErrorInfo]:
        """Get information about validation errors.

        Use to get errors if fail reason is 'ValidationErrors'.

        Returns:
            list[PublishErrorInfo]: Validation errors info.

        """
        return self._publish_state.validation_errors

    def iter_all_error_info(self) -> Iterable[PublishErrorInfo]:
        """Iterate over all error information."""
        if self._publish_state.error_info is not None:
            yield self._publish_state.error_info
        yield from self._publish_state.validation_errors

    def set_comment(self, comment: str) -> None:
        # Ignore change of comment when publishing started
        if self._publish_state.started:
            raise ValueError(
                "Cannot change comment when publishing started."
            )
        self._pyblish_context.data["comment"] = comment
        self._publish_state.comment_is_set = True

    @staticmethod
    def get_publish_plugin_actions(
        plugin: PluginType
    ) -> list[ActionType]:
        actions = getattr(plugin, "actions", None)
        if isinstance(actions, list):
            return actions
        return []

    def run_action(
        self,
        plugin: PluginType | str,
        action: ActionType | str,
    ) -> PublishActionResult:
        if isinstance(plugin, str):
            _plugin = self._publish_plugins_by_id.get(plugin)
            if _plugin is None:
                _plugin = self._publish_plugins_by_name[plugin]
            plugin = _plugin

        if isinstance(action, str):
            action_id = action
            action = None
            for plugin_action in self.get_publish_plugin_actions(plugin):
                if plugin_action.id == action_id:
                    action = plugin_action
                    break

            if action is None:
                raise ValueError(
                    f"Action '{action_id}' not found in plugin '{plugin}'"
                )

        result = pyblish.plugin.process(
            plugin, self._pyblish_context, None, action.id
        )
        self._publish_report.add_action_result(action, result)

        exception = result.get("error")
        if not exception:
            return PublishActionResult(True, action)
        return PublishActionResult(False, action, exception)

    def _reset(
        self,
        project_name: str,
        publish_context: pyblish.api.Context | None = None,
        publish_targets: list[str] | None = None,
        # Hosts filtering is embedded into pyblish register and discover.
        # - hard to change how it works, can be changed only for paths
        #   discovery
        # publish_hosts: list[str] | None = None,
        publish_discover_result: DiscoverResult | None = None,
        creator_discover_result: DiscoverResult | None = None,
        convertor_discover_result: DiscoverResult | None = None,
        create_context: CreateContext | None = None,
        project_settings: dict[str, Any] | None = None,
    ) -> None:
        # Publish context preparation
        if publish_context is None:
            publish_context = pyblish.api.Context()

        if publish_context.data.get("comment") is None:
            publish_context.data["comment"] = ""

        # Plugins preparation
        if create_context is not None:
            publish_context.data["create_context"] = create_context

            # Use discover results from create context
            if publish_discover_result is None:
                publish_discover_result = (
                    create_context.publish_discover_result
                )
            if creator_discover_result is None:
                creator_discover_result = (
                    create_context.creator_discover_result
                )
            if convertor_discover_result is None:
                convertor_discover_result = (
                    create_context.convertor_discover_result
                )

            if project_settings is None:
                project_settings = (
                    create_context.get_current_project_settings()
                )

        if publish_discover_result is None:
            publish_discover_result: DiscoverResult = (
                publish_plugins_discover()
            )

        if creator_discover_result is None:
            creator_discover_result = DiscoverResult(None)

        if convertor_discover_result is None:
            convertor_discover_result = DiscoverResult(None)

        if project_settings is None:
            project_settings = get_project_settings(project_name)

        discovered_publish_plugins = publish_discover_result.plugins
        crashed_file_paths = publish_discover_result.crashed_file_paths

        if publish_targets is None:
            publish_targets = pyblish.logic.registered_targets()

        publish_targets = list(publish_targets)
        if "default" not in publish_targets:
            publish_targets.append("default")

        plugins_by_targets = pyblish.logic.plugins_by_targets(
            discovered_publish_plugins, publish_targets
        )

        blocking_crashed_paths = filter_crashed_publish_paths(
            project_name,
            set(crashed_file_paths),
            project_settings=project_settings,
        )

        self._publish_report.reset_with_discover_results(
            creator_discover_result,
            convertor_discover_result,
            publish_discover_result,
            blocking_crashed_paths,
        )

        plugins_by_targets_set = set(plugins_by_targets)
        for plugin in discovered_publish_plugins:
            if plugin not in plugins_by_targets_set:
                self._publish_report.set_plugin_skipped(
                    self.get_publish_plugin_id(plugin)
                )

        self._pyblish_context = publish_context
        self._publish_plugins = plugins_by_targets
        self._publish_plugins_by_id = {
            self.get_publish_plugin_id(plugin): plugin
            for plugin in plugins_by_targets
        }
        self._publish_plugins_by_name = {
            plugin.__name__: plugin
            for plugin in plugins_by_targets
        }
        strict_validation_error_handling = (
            self._publish_state.strict_validation_error_handling
        )
        self._publish_state = PublishState()
        self._publish_state.strict_validation_error_handling = (
            strict_validation_error_handling
        )
        self._publish_state.max_progress = len(plugins_by_targets)
        if blocking_crashed_paths:
            self._publish_state.fail_reason = PublishFailReason.BlockingPaths

        self._publish_iterator.update_iter(self._inner_publish_iter())

    def _inner_publish_iter(self) -> PublishIterGen:
        """Main logic center of publishing.

        Iterator returns `PublishIterTask` objects with callbacks that should
        be processed in main thread (threaded in future?). Cares about
        changing states of currently processed publish plugin and instance.
        Also change state of processed orders like validation order
        has passed etc.

        Also stops publishing, if should stop on validation.

        Yields:
            PublishIterTask: Information about next processing function in
                publishing.
            None: If publishing should stop, yields None to indicate
                that publishing should stop.

        Raises:
            ValueError: If a previous item did not execute and the next one
                was requested.

        """
        for idx, plugin in enumerate(self._publish_plugins):
            if self._publish_state.should_stop():
                self.stop_publish()

            while not self.is_running():
                yield None

            plugin_id = self.get_publish_plugin_id(plugin)
            self._publish_state.progress = idx

            # Check if plugin is over validation order and stop if needed
            if (
                not self._publish_state.validated
                and plugin.order >= VALIDATION_ORDER
            ):
                self._publish_state.validated = True
                if (
                    self._publish_state.stop_after_validation
                    or self.has_validation_errors()
                ):
                    self.stop_publish()
                    yield None

            # Add plugin to publish report
            self._publish_report.add_plugin_iter(
                plugin_id, self._pyblish_context
            )

            # Check if plugin should be skipped because is not enabled
            #   or active.
            if self._should_skip_plugin(plugin):
                self._publish_report.set_plugin_skipped(plugin_id)
                continue

            # Plugin is instance plugin
            if plugin.__instanceEnabled__:
                instances = pyblish.logic.instances_by_plugin(
                    self._pyblish_context, plugin
                )
                if not instances:
                    self._publish_report.set_plugin_skipped(plugin_id)
                    continue

                for instance in instances:
                    while not self.is_running():
                        yield None

                    if instance.data.get("publish") is False:
                        continue

                    item_label = (
                        instance.data.get("label")
                        or instance.data["name"]
                    )
                    task = PublishIterTask(
                        self,
                        plugin,
                        instance,
                        item_label,
                    )
                    yield task

                    if not task.executed:
                        raise ValueError("Previous task did not execute")

            else:
                families = collect_families_from_instances(
                    self._pyblish_context, only_active=True
                )
                plugins = pyblish.logic.plugins_by_families(
                    [plugin], families
                )
                if not plugins:
                    self._publish_report.set_plugin_skipped(plugin_id)
                    continue

                item_label = (
                    self._pyblish_context.data.get("label")
                    or self._pyblish_context.data.get("name")
                    or "Context"
                )
                task = PublishIterTask(
                    self,
                    plugin,
                    None,
                    item_label,
                )
                yield task
                if not task.executed:
                    raise ValueError("Previous task did not execute")

            self._publish_report.set_plugin_passed(plugin_id)

        # Cleanup of publishing process
        self._publish_state.finished = True
        self._publish_state.progress = self._publish_state.max_progress

        self.stop_publish()
        while True:
            yield None

    @contextmanager
    def _log_manager(self, plugin: PluginType):
        root = logging.getLogger()
        ayon_root = Logger.get_root_logger()
        plugin_log_has_handler = False
        orig_propagate = plugin.log.propagate
        if not self._log_to_console:
            plugin.log.propagate = False

        if not plugin.log.propagate:
            plugin_log_has_handler = True
            plugin.log.addHandler(self._log_handler)
        root.addHandler(self._log_handler)
        ayon_root.addHandler(self._log_handler)

        try:
            yield self._log_handler

        finally:
            if plugin_log_has_handler:
                plugin.log.removeHandler(self._log_handler)
            plugin.log.propagate = orig_propagate
            root.removeHandler(self._log_handler)
            ayon_root.removeHandler(self._log_handler)
            self._log_handler.clear_records()

    def _process_plugin(
        self,
        plugin: PluginType,
        instance: pyblish.api.Instance | None
    ) -> None:
        plugin_id = self.get_publish_plugin_id(plugin)
        with self._log_manager(plugin) as log_handler:
            result = pyblish.plugin.process(
                plugin, self._pyblish_context, instance
            )
            if log_handler is not None:
                records = log_handler.get_records()
                exception = result.get("error")
                if exception is not None and records:
                    last_record = records[-1]
                    if (
                        last_record.name == "pyblish.plugin"
                        and last_record.levelno == logging.ERROR
                    ):
                        # Remove last record made by pyblish
                        # - `log.exception(formatted_traceback)`
                        records.pop(-1)
                result["records"] = records

        exception = result.get("error")
        if exception:
            error_info = PublishErrorInfo.from_result(
                plugin, instance, exception
            )
            is_validation_error = False
            if (
                isinstance(exception, PublishValidationError)
                and not self._publish_state.validated
            ):
                is_validation_error = True
                self._publish_state.validation_errors.append(error_info)
                self._publish_state.fail_reason = (
                    PublishFailReason.ValidationErrors
                )

            else:
                if (
                    isinstance(exception, PublishError)
                    and not exception.title
                ):
                    exception.title = plugin.label or plugin.__name__

                self._publish_state.error_info = error_info
                self._publish_state.fail_reason = PublishFailReason.Error

            # Store additional metadata to result for report maker
            result["is_validation_error"] = is_validation_error

        self._publish_report.add_result(plugin_id, result)

    def _should_skip_plugin(self, plugin: PluginType) -> bool:
        """Decide if publish plugin should be skipped.

        The pyblish base does define 'active' on plugin which is used to mark
            plugin as active or not. But if plugin is also 'optional' it is
            possible to change the 'active' value.

        AYON does use 'optional' to mark plugin as optional. But the
            optional logic does not change 'active', instead it is handled by
            'process' logic. So AYON's 'OptionalPyblishPluginMixin' will
            always be 'active'.

        With that AYON plugins have to somehow mark plugin as disabled, that
            is why 'enabled' can be set on plugins. If plugin is disabled
            it is skipped no matter what.

        Args:
            plugin (PluginType): Plugin which should be checked
                if it should be skipped.

        Returns:
            bool: Skip the plugin.

        """
        if getattr(plugin, "enabled", True) is False:
            return True

        if plugin.active:
            return False

        # Custom handling of AYON's optional plugins
        if (
            plugin.optional
            and OptionalPyblishPluginMixin in inspect.getmro(plugin)
        ):
            return False
        return True