host

class AbstractHost(ABC):
    """Abstract definition of host implementation."""
    @property
    @abstractmethod
    def log(self) -> logging.Logger:
        pass

    @property
    @abstractmethod
    def name(self) -> str:
        """Host name."""
        pass

    @abstractmethod
    def get_app_information(self) -> ApplicationInformation:
        """Information about the application where host is running.

        Returns:
            ApplicationInformation: Application information.

        """
        pass

    @abstractmethod
    def get_current_context(self) -> HostContextData:
        """Get the current context of the host.

        Current context is defined by project name, folder path and task name.

        Returns:
            HostContextData: The current context of the host.

        """
        pass

    @abstractmethod
    def set_current_context(
        self,
        folder_entity: dict[str, Any],
        task_entity: dict[str, Any],
        *,
        reason: ContextChangeReason = ContextChangeReason.undefined,
        project_entity: Optional[dict[str, Any]] = None,
        anatomy: Optional[Anatomy] = None,
    ) -> HostContextData:
        """Change context of the host.

        Args:
            folder_entity (dict[str, Any]): Folder entity.
            task_entity (dict[str, Any]): Task entity.
            reason (ContextChangeReason): Reason for change.
            project_entity (dict[str, Any]): Project entity.
            anatomy (Anatomy): Anatomy entity.

        """
        pass

    @abstractmethod
    def get_current_project_name(self) -> str:
        """Get the current project name.

        Returns:
            Optional[str]: The current project name.

        """
        pass

    @abstractmethod
    def get_current_folder_path(self) -> Optional[str]:
        """Get the current folder path.

        Returns:
            Optional[str]: The current folder path.

        """
        pass

    @abstractmethod
    def get_current_task_name(self) -> Optional[str]:
        """Get the current task name.

        Returns:
            Optional[str]: The current task name.

        """
        pass

    @abstractmethod
    def get_context_title(self) -> str:
        """Get the context title used in UIs."""
        pass

@dataclass
class ApplicationInformation:
    """Application information.

    Attributes:
        app_name (Optional[str]): Application name. e.g. Maya, NukeX, Nuke
        app_version (Optional[str]): Application version. e.g. 15.2.1

    """
    app_name: Optional[str] = None
    app_version: Optional[str] = None

class ContextChangeReason(StrEnum):
    """Reasons for context change in the host."""
    undefined = "undefined"
    workfile_open = "workfile.opened"
    workfile_save = "workfile.saved"

class HostBase(AbstractHost):
    """Base of host implementation class.

    Host is pipeline implementation of DCC application. This class should help
    to identify what must/should/can be implemented for specific functionality.

    Compared to 'avalon' concept:
    What was before considered as functions in host implementation folder. The
    host implementation should primarily care about adding ability of creation
    (mark products to be published) and optionally about referencing published
    representations as containers.

    Host may need extend some functionality like working with workfiles
    or loading. Not all host implementations may allow that for those purposes
    can be logic extended with implementing functions for the purpose. There
    are prepared interfaces to be able identify what must be implemented to
    be able use that functionality.
    - current statement is that it is not required to inherit from interfaces
        but all of the methods are validated (only their existence!)

    # Installation of host before (avalon concept):
    ```python
    from ayon_core.pipeline import install_host
    import ayon_core.hosts.maya.api as host

    install_host(host)
    ```

    # Installation of host now:
    ```python
    from ayon_core.pipeline import install_host
    from ayon_core.hosts.maya.api import MayaHost

    host = MayaHost()
    install_host(host)
    ```

    Todo:
        - move content of 'install_host' as method of this class
            - register host object
            - install global plugin paths
        - store registered plugin paths to this object
        - handle current context (project, asset, task)
            - this must be done in many separated steps
        - have it's object of host tools instead of using globals

    This implementation will probably change over time when more
        functionality and responsibility will be added.
    """

    _log = None

    def __init__(self):
        """Initialization of host.

        Register DCC callbacks, host specific plugin paths, targets etc.
        (Part of what 'install' did in 'avalon' concept.)

        Note:
            At this moment global "installation" must happen before host
            installation. Because of this current limitation it is recommended
            to implement 'install' method which is triggered after global
            'install'.
        """

        pass

    def get_app_information(self) -> ApplicationInformation:
        """Running application information.

        Host integration should override this method and return correct
            information.

        Returns:
            ApplicationInformation: Application information.

        """
        return ApplicationInformation()

    def install(self):
        """Install host specific functionality.

        This is where should be added menu with tools, registered callbacks
        and other host integration initialization.

        It is called automatically when 'ayon_core.pipeline.install_host' is
        triggered.

        """
        pass

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

    def get_current_project_name(self) -> str:
        """
        Returns:
            str: Current project name.

        """
        return os.environ["AYON_PROJECT_NAME"]

    def get_current_folder_path(self) -> Optional[str]:
        """
        Returns:
            Optional[str]: Current asset name.

        """
        return os.environ.get("AYON_FOLDER_PATH")

    def get_current_task_name(self) -> Optional[str]:
        """
        Returns:
            Optional[str]: Current task name.

        """
        return os.environ.get("AYON_TASK_NAME")

    def get_current_context(self) -> HostContextData:
        """Get current context information.

        This method should be used to get current context of host. Usage of
        this method can be crucial for host implementations in DCCs where
        can be opened multiple workfiles at one moment and change of context
        can't be caught properly.

        Returns:
            HostContextData: Current context with 'project_name',
                'folder_path' and 'task_name'.

        """
        return {
            "project_name": self.get_current_project_name(),
            "folder_path": self.get_current_folder_path(),
            "task_name": self.get_current_task_name()
        }

    def set_current_context(
        self,
        folder_entity: dict[str, Any],
        task_entity: dict[str, Any],
        *,
        reason: ContextChangeReason = ContextChangeReason.undefined,
        project_entity: Optional[dict[str, Any]] = None,
        anatomy: Optional[Anatomy] = None,
    ) -> HostContextData:
        """Set current context information.

        This method should be used to set current context of host. Usage of
        this method can be crucial for host implementations in DCCs where
        can be opened multiple workfiles at one moment and change of context
        can't be caught properly.

        Notes:
            This method should not care about change of workdir and expect any
                of the arguments.

        Args:
            folder_entity (Optional[dict[str, Any]]): Folder entity.
            task_entity (Optional[dict[str, Any]]): Task entity.
            reason (ContextChangeReason): Reason for context change.
            project_entity (Optional[dict[str, Any]]): Project entity data.
            anatomy (Optional[Anatomy]): Anatomy instance for the project.

        Returns:
            dict[str, Optional[str]]: Context information with project name,
                folder path and task name.

        """
        from ayon_core.pipeline import Anatomy

        folder_path = folder_entity["path"]
        task_name = task_entity["name"]

        context = self.get_current_context()
        # Don't do anything if context did not change
        if (
            context["folder_path"] == folder_path
            and context["task_name"] == task_name
        ):
            return context

        project_name = self.get_current_project_name()
        if project_entity is None:
            project_entity = ayon_api.get_project(project_name)

        if anatomy is None:
            anatomy = Anatomy(project_name, project_entity=project_entity)

        context_change_data = ContextChangeData(
            project_entity,
            folder_entity,
            task_entity,
            reason,
            anatomy,
        )
        self._before_context_change(context_change_data)
        self._set_current_context(context_change_data)
        self._after_context_change(context_change_data)

        return self._emit_context_change_event(
            project_name,
            folder_path,
            task_name,
        )

    def get_context_title(self):
        """Context title shown for UI purposes.

        Should return current context title if possible.

        Note:
            This method is used only for UI purposes so it is possible to
                return some logical title for contextless cases.
            Is not meant for "Context menu" label.

        Returns:
            str: Context title.
            None: Default title is used based on UI implementation.
        """

        # Use current context to fill the context title
        current_context = self.get_current_context()
        project_name = current_context["project_name"]
        folder_path = current_context["folder_path"]
        task_name = current_context["task_name"]
        items = []
        if project_name:
            items.append(project_name)
            if folder_path:
                items.append(folder_path.lstrip("/"))
                if task_name:
                    items.append(task_name)
        if items:
            return "/".join(items)
        return None

    @contextlib.contextmanager
    def maintained_selection(self):
        """Some functionlity will happen but selection should stay same.

        This is DCC specific. Some may not allow to implement this ability
        that is reason why default implementation is empty context manager.

        Yields:
            None: Yield when is ready to restore selected at the end.
        """

        try:
            yield
        finally:
            pass

    def _emit_context_change_event(
        self,
        project_name: str,
        folder_path: Optional[str],
        task_name: Optional[str],
    ) -> HostContextData:
        """Emit context change event.

        Args:
            project_name (str): Name of the project.
            folder_path (Optional[str]): Path of the folder.
            task_name (Optional[str]): Name of the task.

        Returns:
            HostContextData: Data send to context change event.

        """
        data: HostContextData = {
            "project_name": project_name,
            "folder_path": folder_path,
            "task_name": task_name,
        }
        emit_event("taskChanged", data)
        return data

    def _set_current_context(
        self, context_change_data: ContextChangeData
    ) -> None:
        """Method that changes the context in host.

        Can be overriden for hosts that do need different handling of context
            than using environment variables.

        Args:
            context_change_data (ContextChangeData): Context change related
                data.

        """
        project_name = self.get_current_project_name()
        folder_path = None
        task_name = None
        if context_change_data.folder_entity:
            folder_path = context_change_data.folder_entity["path"]
            if context_change_data.task_entity:
                task_name = context_change_data.task_entity["name"]

        envs = {
            "AYON_PROJECT_NAME": project_name,
            "AYON_FOLDER_PATH": folder_path,
            "AYON_TASK_NAME": task_name,
        }

        # Update the Session and environments. Pop from environments all
        #   keys with value set to None.
        for key, value in envs.items():
            if value is None:
                os.environ.pop(key, None)
            else:
                os.environ[key] = value

    def _before_context_change(self, context_change_data: ContextChangeData):
        """Before context is changed.

        This method is called before the context is changed in the host.

        Can be overridden to implement host specific logic.

        Args:
            context_change_data (ContextChangeData): Object with information
                about context change.

        """
        pass

    def _after_context_change(self, context_change_data: ContextChangeData):
        """After context is changed.

        This method is called after the context is changed in the host.

        Can be overridden to implement host specific logic.

        Args:
            context_change_data (ContextChangeData): Object with information
                about context change.

        """
        pass

class HostDirmap(ABC):
    """Abstract class for running dirmap on a workfile in a host.

    Dirmap is used to translate paths inside of host workfile from one
    OS to another. (Eg. arstist created workfile on Win, different artists
    opens same file on Linux.)

    Expects methods to be implemented inside of host:
        on_dirmap_enabled: run host code for enabling dirmap
        do_dirmap: run host code to do actual remapping
    """

    def __init__(
        self,
        host_name,
        project_name,
        project_settings=None,
        sitesync_addon=None
    ):
        self.host_name = host_name
        self.project_name = project_name
        self._project_settings = project_settings
        self._sitesync_addon = sitesync_addon
        # to limit reinit of Modules
        self._sitesync_addon_discovered = sitesync_addon is not None
        self._log = None

    @property
    def sitesync_addon(self):
        if not self._sitesync_addon_discovered:
            self._sitesync_addon_discovered = True
            manager = AddonsManager()
            self._sitesync_addon = manager.get("sitesync")
        return self._sitesync_addon

    @property
    def project_settings(self):
        if self._project_settings is None:
            self._project_settings = get_project_settings(self.project_name)
        return self._project_settings

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

    @abstractmethod
    def on_enable_dirmap(self):
        """Run host dependent operation for enabling dirmap if necessary."""
        pass

    @abstractmethod
    def dirmap_routine(self, source_path, destination_path):
        """Run host dependent remapping from source_path to destination_path"""
        pass

    def process_dirmap(self, mapping=None):
        # type: (dict) -> None
        """Go through all paths in Settings and set them using `dirmap`.

            If artists has Site Sync enabled, take dirmap mapping directly from
            Local Settings when artist is syncing workfile locally.

        """

        if not mapping:
            mapping = self.get_mappings()
        if not mapping:
            return

        self.on_enable_dirmap()

        for k, sp in enumerate(mapping["source_path"]):
            dst = mapping["destination_path"][k]
            try:
                # add trailing slash if missing
                sp = os.path.join(sp, '')
                dst = os.path.join(dst, '')
                print("{} -> {}".format(sp, dst))
                self.dirmap_routine(sp, dst)
            except IndexError:
                # missing corresponding destination path
                self.log.error((
                    "invalid dirmap mapping, missing corresponding"
                    " destination directory."
                ))
                break
            except RuntimeError:
                self.log.error(
                    "invalid path {} -> {}, mapping not registered".format(
                        sp, dst
                    )
                )
                continue

    def get_mappings(self):
        """Get translation from source_path to destination_path.

            It checks if Site Sync is enabled and user chose to use local
            site, in that case configuration in Local Settings takes precedence
        """
        mapping_sett = self.project_settings[self.host_name].get("dirmap", {})
        local_mapping = self._get_local_sync_dirmap()
        mapping_enabled = mapping_sett.get("enabled") or bool(local_mapping)
        if not mapping_enabled:
            return {}

        mapping = (
            local_mapping
            or mapping_sett["paths"]
            or {}
        )

        if (
            not mapping
            or not mapping.get("destination_path")
            or not mapping.get("source_path")
        ):
            return {}
        self.log.info("Processing directory mapping ...")
        self.log.info("mapping:: {}".format(mapping))
        return mapping

    def _get_local_sync_dirmap(self):
        """
            Returns dirmap if synch to local project is enabled.

            Only valid mapping is from roots of remote site to local site set
            in Local Settings.

            Returns:
                dict : { "source_path": [XXX], "destination_path": [YYYY]}
        """
        project_name = self.project_name

        sitesync_addon = self.sitesync_addon
        mapping = {}
        if (
            sitesync_addon is None
            or not sitesync_addon.enabled
            or not sitesync_addon.is_project_enabled(project_name, True)
        ):
            return mapping

        active_site = sitesync_addon.get_local_normalized_site(
            sitesync_addon.get_active_site(project_name))
        remote_site = sitesync_addon.get_local_normalized_site(
            sitesync_addon.get_remote_site(project_name))
        self.log.debug(
            "active {} - remote {}".format(active_site, remote_site)
        )

        if active_site == "local" and active_site != remote_site:
            sync_settings = sitesync_addon.get_sync_project_setting(
                project_name,
                exclude_locals=False,
                cached=False)

            # overrides for roots set in `Site Settings`
            active_roots_overrides = self._get_site_root_overrides(
                sitesync_addon, project_name, active_site)

            remote_roots_overrides = self._get_site_root_overrides(
                sitesync_addon, project_name, remote_site)

            current_platform = platform.system().lower()
            remote_provider = sitesync_addon.get_provider_for_site(
                project_name, remote_site
            )
            # dirmap has sense only with regular disk provider, in the workfile
            # won't be root on cloud or sftp provider so fallback to studio
            if remote_provider != "local_drive":
                remote_site = "studio"
            for root_name, active_site_dir in active_roots_overrides.items():
                remote_site_dir = (
                    remote_roots_overrides.get(root_name)
                    or sync_settings["sites"][remote_site]["root"][root_name]
                )

                if isinstance(remote_site_dir, dict):
                    remote_site_dir = remote_site_dir.get(current_platform)

                if not remote_site_dir:
                    continue

                if os.path.isdir(active_site_dir):
                    if "destination_path" not in mapping:
                        mapping["destination_path"] = []
                    mapping["destination_path"].append(active_site_dir)

                    if "source_path" not in mapping:
                        mapping["source_path"] = []
                    mapping["source_path"].append(remote_site_dir)

            self.log.debug("local sync mapping:: {}".format(mapping))
        return mapping

    def _get_site_root_overrides(
        self, sitesync_addon, project_name, site_name
    ):
        """Safely handle root overrides.

        SiteSync raises ValueError for non local or studio sites.
        """
        # TODO: could be removed when `get_site_root_overrides` is not raising
        #     an Error but just returns {}
        try:
            site_roots_overrides = sitesync_addon.get_site_root_overrides(
                project_name, site_name)
        except ValueError:
            site_roots_overrides = {}
        self.log.debug("{} roots overrides {}".format(
            site_name, site_roots_overrides))

        return site_roots_overrides

class ILoadHost(AbstractHost):
    """Implementation requirements to be able use reference of representations.

    The load plugins can do referencing even without implementation of methods
    here, but switch and removement of containers would not be possible.

    Questions:
        - Is list container dependency of host or load plugins?
        - Should this be directly in HostBase?
            - how to find out if referencing is available?
            - do we need to know that?
    """

    @staticmethod
    def get_missing_load_methods(host):
        """Look for missing methods on "old type" host implementation.

        Method is used for validation of implemented functions related to
        loading. Checks only existence of methods.

        Args:
            Union[ModuleType, AbstractHost]: Object of host where to look for
                required methods.

        Returns:
            list[str]: Missing method implementations for loading workflow.
        """

        if isinstance(host, ILoadHost):
            return []

        required = ["ls"]
        missing = []
        for name in required:
            if not hasattr(host, name):
                missing.append(name)
        return missing

    @staticmethod
    def validate_load_methods(host):
        """Validate implemented methods of "old type" host for load workflow.

        Args:
            Union[ModuleType, AbstractHost]: Object of host to validate.

        Raises:
            MissingMethodsError: If there are missing methods on host
                implementation.
        """
        missing = ILoadHost.get_missing_load_methods(host)
        if missing:
            raise MissingMethodsError(host, missing)

    @abstractmethod
    def get_containers(self):
        """Retrieve referenced containers from scene.

        This can be implemented in hosts where referencing can be used.

        Todo:
            Rename function to something more self explanatory.
                Suggestion: 'get_containers'

        Returns:
            list[dict]: Information about loaded containers.
        """

        pass

    # --- Deprecated method names ---
    def ls(self):
        """Deprecated variant of 'get_containers'.

        Todo:
            Remove when all usages are replaced.
        """

        return self.get_containers()

class INewPublisher(IPublishHost):
    """Legacy interface replaced by 'IPublishHost'.

    Deprecated:
        'INewPublisher' is replaced by 'IPublishHost' please change your
        imports.
        There is no "reasonable" way hot mark these classes as deprecated
        to show warning of wrong import. Deprecated since 3.14.* will be
        removed in 3.15.*
    """

    pass

class IPublishHost(AbstractHost):
    """Functions related to new creation system in new publisher.

    New publisher is not storing information only about each created instance
    but also some global data. At this moment are data related only to context
    publish plugins but that can extend in future.
    """

    @staticmethod
    def get_missing_publish_methods(host):
        """Look for missing methods on "old type" host implementation.

        Method is used for validation of implemented functions related to
        new publish creation. Checks only existence of methods.

        Args:
            Union[ModuleType, AbstractHost]: Host module where to look for
                required methods.

        Returns:
            list[str]: Missing method implementations for new publisher
                workflow.
        """

        if isinstance(host, IPublishHost):
            return []

        required = [
            "get_context_data",
            "update_context_data",
            "get_context_title",
            "get_current_context",
        ]
        missing = []
        for name in required:
            if not hasattr(host, name):
                missing.append(name)
        return missing

    @staticmethod
    def validate_publish_methods(host):
        """Validate implemented methods of "old type" host.

        Args:
            Union[ModuleType, AbstractHost]: Host module to validate.

        Raises:
            MissingMethodsError: If there are missing methods on host
                implementation.
        """
        missing = IPublishHost.get_missing_publish_methods(host)
        if missing:
            raise MissingMethodsError(host, missing)

    @abstractmethod
    def get_context_data(self):
        """Get global data related to creation-publishing from workfile.

        These data are not related to any created instance but to whole
        publishing context. Not saving/returning them will cause that each
        reset of publishing resets all values to default ones.

        Context data can contain information about enabled/disabled publish
        plugins or other values that can be filled by artist.

        Returns:
            dict: Context data stored using 'update_context_data'.
        """

        pass

    @abstractmethod
    def update_context_data(self, data, changes):
        """Store global context data to workfile.

        Called when some values in context data has changed.

        Without storing the values in a way that 'get_context_data' would
        return them will each reset of publishing cause loose of filled values
        by artist. Best practice is to store values into workfile, if possible.

        Args:
            data (dict): New data as are.
            changes (dict): Only data that has been changed. Each value has
                tuple with '(<old>, <new>)' value.
        """

        pass

class IWorkfileHost(AbstractHost):
    """Implementation requirements to be able to use workfiles utils and tool.

    Some of the methods are pre-implemented as they generally do the same in
        all host integrations.

    """
    @abstractmethod
    def save_workfile(self, dst_path: Optional[str] = None) -> None:
        """Save the currently opened scene.

        Args:
            dst_path (str): Where the current scene should be saved. Or use
                the current path if 'None' is passed.

        """
        pass

    @abstractmethod
    def open_workfile(self, filepath: str) -> None:
        """Open passed filepath in the host.

        Args:
            filepath (str): Path to workfile.

        """
        pass

    @abstractmethod
    def get_current_workfile(self) -> Optional[str]:
        """Retrieve a path to current opened file.

        Returns:
            Optional[str]: Path to the file which is currently opened. None if
                nothing is opened or the current workfile is unsaved.

        """
        return None

    def workfile_has_unsaved_changes(self) -> Optional[bool]:
        """Currently opened scene is saved.

        Not all hosts can know if the current scene is saved because the API
            of DCC does not support it.

        Returns:
            Optional[bool]: True if scene is saved and False if has unsaved
                modifications. None if can't tell if workfiles has
                modifications.

        """
        return None

    def get_workfile_extensions(self) -> list[str]:
        """Extensions that can be used to save the workfile to.

        Notes:
            Method may not be used if 'list_workfiles' and
                'list_published_workfiles' are re-implemented with different
                logic.

        Returns:
            list[str]: List of extensions that can be used for saving.

        """
        return []

    def save_workfile_with_context(
        self,
        filepath: str,
        folder_entity: dict[str, Any],
        task_entity: dict[str, Any],
        *,
        version: Optional[int] = None,
        comment: Optional[str] = None,
        description: Optional[str] = None,
        prepared_data: Optional[SaveWorkfileOptionalData] = None,
    ) -> None:
        """Save the current workfile with context.

        Arguments 'rootless_path', 'workfile_entities', 'project_entity'
            and 'anatomy' can be filled to enhance efficiency if you already
            have access to the values.

        Argument 'project_settings' is used to calculate 'rootless_path'
            if it is not provided.

        Notes:
            Should this method care about context change?

        Args:
            filepath (str): Where the current scene should be saved.
            folder_entity (dict[str, Any]): Folder entity.
            task_entity (dict[str, Any]): Task entity.
            version (Optional[int]): Version of the workfile. Information
                for workfile entity. Recommended to fill.
            comment (Optional[str]): Comment for the workfile.
                Usually used in the filename template.
            description (Optional[str]): Artist note for the workfile entity.
            prepared_data (Optional[SaveWorkfileOptionalData]): Prepared data
                for speed enhancements.

        """
        project_name = self.get_current_project_name()
        save_workfile_context = get_save_workfile_context(
            project_name,
            filepath,
            folder_entity,
            task_entity,
            host_name=self.name,
            prepared_data=prepared_data,
        )

        self._before_workfile_save(save_workfile_context)
        event_data = self._get_workfile_event_data(
            project_name,
            folder_entity,
            task_entity,
            filepath,
        )
        self._emit_workfile_save_event(event_data, after_save=False)

        workdir = os.path.dirname(filepath)
        if not os.path.exists(workdir):
            os.makedirs(workdir, exist_ok=True)

        # Set 'AYON_WORKDIR' environment variable
        os.environ["AYON_WORKDIR"] = workdir

        self.set_current_context(
            folder_entity,
            task_entity,
            reason=ContextChangeReason.workfile_save,
            project_entity=save_workfile_context.project_entity,
            anatomy=save_workfile_context.anatomy,
        )

        self.save_workfile(filepath)

        self._save_workfile_entity(
            save_workfile_context,
            version,
            comment,
            description,
        )
        self._after_workfile_save(save_workfile_context)
        self._emit_workfile_save_event(event_data)

    def open_workfile_with_context(
        self,
        filepath: str,
        folder_entity: dict[str, Any],
        task_entity: dict[str, Any],
        *,
        prepared_data: Optional[OpenWorkfileOptionalData] = None,
    ) -> None:
        """Open passed filepath in the host with context.

        This function should be used to open workfile in different context.

        Notes:
            Should this method care about context change?

        Args:
            filepath (str): Path to workfile.
            folder_entity (dict[str, Any]): Folder id.
            task_entity (dict[str, Any]): Task id.
            prepared_data (Optional[WorkfileOptionalData]): Prepared data
                for speed enhancements.

        """
        context = self.get_current_context()
        project_name = context["project_name"]

        open_workfile_context = get_open_workfile_context(
            project_name,
            filepath,
            folder_entity,
            task_entity,
            prepared_data=prepared_data,
        )

        workdir = os.path.dirname(filepath)
        # Set 'AYON_WORKDIR' environment variable
        os.environ["AYON_WORKDIR"] = workdir

        event_data = self._get_workfile_event_data(
            project_name, folder_entity, task_entity, filepath
        )
        self._before_workfile_open(open_workfile_context)
        self._emit_workfile_open_event(event_data, after_open=False)

        self.set_current_context(
            folder_entity,
            task_entity,
            reason=ContextChangeReason.workfile_open,
            project_entity=open_workfile_context.project_entity,
            anatomy=open_workfile_context.anatomy,
        )

        self.open_workfile(filepath)

        self._after_workfile_open(open_workfile_context)
        self._emit_workfile_open_event(event_data)

    def list_workfiles(
        self,
        project_name: str,
        folder_entity: dict[str, Any],
        task_entity: dict[str, Any],
        *,
        prepared_data: Optional[ListWorkfilesOptionalData] = None,
    ) -> list[WorkfileInfo]:
        """List workfiles in the given task.

        The method should also return workfiles that are not available on
            disk, but are in the AYON database.

        Notes:
        - Better method name?
        - This method is pre-implemented as the logic can be shared across
            95% of host integrations. Ad-hoc implementation to give host
            integration workfile api functionality.

        Args:
            project_name (str): Project name.
            folder_entity (dict[str, Any]): Folder entity.
            task_entity (dict[str, Any]): Task entity.
            prepared_data (Optional[ListWorkfilesOptionalData]): Prepared
                data for speed enhancements.

        Returns:
            list[WorkfileInfo]: List of workfiles.

        """
        from ayon_core.pipeline.template_data import get_template_data
        from ayon_core.pipeline.workfile.path_resolving import (
            get_workdir_with_workdir_data,
            WorkfileDataParser,
        )

        extensions = self.get_workfile_extensions()
        if not extensions:
            return []

        list_workfiles_context = get_list_workfiles_context(
            project_name,
            folder_entity,
            task_entity,
            host_name=self.name,
            prepared_data=prepared_data,
        )

        workfile_entities_by_path = {}
        for workfile_entity in list_workfiles_context.workfile_entities:
            rootless_path = workfile_entity["path"]
            path = os.path.normpath(
                list_workfiles_context.anatomy.fill_root(rootless_path)
            )
            workfile_entities_by_path[path] = workfile_entity

        workdir_data = get_template_data(
            list_workfiles_context.project_entity,
            folder_entity,
            task_entity,
            host_name=self.name,
        )
        workdir = get_workdir_with_workdir_data(
            workdir_data,
            project_name,
            anatomy=list_workfiles_context.anatomy,
            template_key=list_workfiles_context.template_key,
            project_settings=list_workfiles_context.project_settings,
        )

        file_template = list_workfiles_context.anatomy.get_template_item(
            "work", list_workfiles_context.template_key, "file"
        )
        rootless_workdir = workdir.rootless
        if platform.system().lower() == "windows":
            rootless_workdir = rootless_workdir.replace("\\", "/")

        filenames = []
        if os.path.exists(workdir):
            filenames = list(os.listdir(workdir))

        data_parser = WorkfileDataParser(file_template, workdir_data)
        items = []
        for filename in filenames:
            # TODO add 'default' support for folders
            ext = os.path.splitext(filename)[1].lower()
            if ext not in extensions:
                continue

            filepath = os.path.join(workdir, filename)

            rootless_path = f"{rootless_workdir}/{filename}"
            workfile_entity = workfile_entities_by_path.pop(
                filepath, None
            )
            version = comment = None
            if workfile_entity is not None:
                _data = workfile_entity["data"]
                version = _data.get("version")
                comment = _data.get("comment")

            if version is None:
                parsed_data = data_parser.parse_data(filename)
                version = parsed_data.version
                comment = parsed_data.comment

            item = WorkfileInfo.new(
                filepath,
                rootless_path,
                version=version,
                comment=comment,
                available=True,
                workfile_entity=workfile_entity,
            )
            items.append(item)

        for filepath, workfile_entity in workfile_entities_by_path.items():
            # Workfile entity is not in the filesystem
            #   but it is in the database
            rootless_path = workfile_entity["path"]
            ext = os.path.splitext(rootless_path)[1].lower()
            if ext not in extensions:
                continue

            _data = workfile_entity["data"]
            version = _data.get("version")
            comment = _data.get("comment")
            if version is None:
                filename = os.path.basename(rootless_path)
                parsed_data = data_parser.parse_data(filename)
                version = parsed_data.version
                comment = parsed_data.comment

            available = os.path.exists(filepath)
            items.append(WorkfileInfo.new(
                filepath,
                rootless_path,
                version=version,
                comment=comment,
                available=available,
                workfile_entity=workfile_entity,
            ))

        return items

    def list_published_workfiles(
        self,
        project_name: str,
        folder_id: str,
        *,
        prepared_data: Optional[ListPublishedWorkfilesOptionalData] = None,
    ) -> list[PublishedWorkfileInfo]:
        """List published workfiles for the given folder.

        The default implementation looks for products with the 'workfile'
            product type.

        Pre-fetched entities have mandatory fields to be fetched:
            - Version: 'id', 'author', 'taskId'
            - Representation: 'id', 'versionId', 'files'

        Args:
            project_name (str): Project name.
            folder_id (str): Folder id.
            prepared_data (Optional[ListPublishedWorkfilesOptionalData]):
                Prepared data for speed enhancements.

        Returns:
            list[PublishedWorkfileInfo]: Published workfile information for
                the given context.

        """
        list_workfiles_context = get_list_published_workfiles_context(
            project_name,
            folder_id,
            prepared_data=prepared_data,
        )
        if not list_workfiles_context.repre_entities:
            return []

        versions_by_id = {
            version_entity["id"]: version_entity
            for version_entity in list_workfiles_context.version_entities
        }
        extensions = {
            ext.lstrip(".")
            for ext in self.get_workfile_extensions()
        }
        items = []
        for repre_entity in list_workfiles_context.repre_entities:
            version_id = repre_entity["versionId"]
            version_entity = versions_by_id[version_id]
            task_id = version_entity["taskId"]

            # Filter by extension
            workfile_path = None
            for repre_file in repre_entity["files"]:
                ext = (
                    os.path.splitext(repre_file["name"])[1]
                    .lower()
                    .lstrip(".")
                )
                if ext in extensions:
                    workfile_path = repre_file["path"]
                    break

            if not workfile_path:
                continue

            try:
                workfile_path = workfile_path.format(
                    root=list_workfiles_context.anatomy.roots
                )
            except Exception:
                self.log.warning(
                    "Failed to format workfile path.", exc_info=True
                )

            is_available = False
            file_size = file_modified = file_created = None
            if workfile_path and os.path.exists(workfile_path):
                filestat = os.stat(workfile_path)
                is_available = True
                file_size = filestat.st_size
                file_created = filestat.st_ctime
                file_modified = filestat.st_mtime

            workfile_item = PublishedWorkfileInfo.new(
                project_name,
                folder_id,
                task_id,
                repre_entity,
                filepath=workfile_path,
                author=version_entity["author"],
                available=is_available,
                file_size=file_size,
                file_created=file_created,
                file_modified=file_modified,
            )
            items.append(workfile_item)

        return items

    def copy_workfile(
        self,
        src_path: str,
        dst_path: str,
        folder_entity: dict[str, Any],
        task_entity: dict[str, Any],
        *,
        version: Optional[int] = None,
        comment: Optional[str] = None,
        description: Optional[str] = None,
        open_workfile: bool = True,
        prepared_data: Optional[CopyWorkfileOptionalData] = None,
    ) -> None:
        """Save workfile path with target folder and task context.

        It is expected that workfile is saved to the current project, but
            can be copied from the other project.

        Arguments 'rootless_path', 'workfile_entities', 'project_entity'
            and 'anatomy' can be filled to enhance efficiency if you already
            have access to the values.

        Argument 'project_settings' is used to calculate 'rootless_path'
            if it is not provided.

        Args:
            src_path (str): Path to the source scene.
            dst_path (str): Where the scene should be saved.
            folder_entity (dict[str, Any]): Folder entity.
            task_entity (dict[str, Any]): Task entity.
            version (Optional[int]): Version of the workfile. Information
                for workfile entity. Recommended to fill.
            comment (Optional[str]): Comment for the workfile.
            description (Optional[str]): Artist note for the workfile entity.
            open_workfile (bool): Open workfile when copied.
            prepared_data (Optional[CopyWorkfileOptionalData]): Prepared data
                for speed enhancements.

        """
        project_name = self.get_current_project_name()
        copy_workfile_context: CopyWorkfileContext = get_copy_workfile_context(
            project_name,
            src_path,
            dst_path,
            folder_entity,
            task_entity,
            version=version,
            comment=comment,
            description=description,
            open_workfile=open_workfile,
            host_name=self.name,
            prepared_data=prepared_data,
        )
        self._copy_workfile(
            copy_workfile_context,
            version=version,
            comment=comment,
            description=description,
            open_workfile=open_workfile,
        )

    def copy_workfile_representation(
        self,
        src_project_name: str,
        src_representation_entity: dict[str, Any],
        dst_path: str,
        folder_entity: dict[str, Any],
        task_entity: dict[str, Any],
        *,
        version: Optional[int] = None,
        comment: Optional[str] = None,
        description: Optional[str] = None,
        open_workfile: bool = True,
        prepared_data: Optional[CopyPublishedWorkfileOptionalData] = None,
    ) -> None:
        """Copy workfile representation.

        Use representation as a source for the workfile.

        Arguments 'rootless_path', 'workfile_entities', 'project_entity'
            and 'anatomy' can be filled to enhance efficiency if you already
            have access to the values.

        Argument 'project_settings' is used to calculate 'rootless_path'
            if it is not provided.

        Args:
            src_project_name (str): Project name.
            src_representation_entity (dict[str, Any]): Representation
                entity.
            dst_path (str): Where the scene should be saved.
            folder_entity (dict[str, Any): Folder entity.
            task_entity (dict[str, Any]): Task entity.
            version (Optional[int]): Version of the workfile. Information
                for workfile entity. Recommended to fill.
            comment (Optional[str]): Comment for the workfile.
            description (Optional[str]): Artist note for the workfile entity.
            open_workfile (bool): Open workfile when copied.
            prepared_data (Optional[CopyPublishedWorkfileOptionalData]):
                Prepared data for speed enhancements.

        """
        project_name = self.get_current_project_name()
        copy_repre_workfile_context: CopyPublishedWorkfileContext = (
            get_copy_repre_workfile_context(
                project_name,
                src_project_name,
                src_representation_entity,
                dst_path,
                folder_entity,
                task_entity,
                version=version,
                comment=comment,
                description=description,
                open_workfile=open_workfile,
                host_name=self.name,
                prepared_data=prepared_data,
            )
        )
        self._copy_workfile(
            copy_repre_workfile_context,
            version=version,
            comment=comment,
            description=description,
            open_workfile=open_workfile,
        )

    # --- Deprecated method names ---
    @deprecated("Use 'get_workfile_extensions' instead")
    def file_extensions(self):
        """Deprecated variant of 'get_workfile_extensions'.

        Todo:
            Remove when all usages are replaced.

        """
        return self.get_workfile_extensions()

    @deprecated("Use 'save_workfile' instead")
    def save_file(self, dst_path=None):
        """Deprecated variant of 'save_workfile'.

        Todo:
            Remove when all usages are replaced

        """
        self.save_workfile(dst_path)

    @deprecated("Use 'open_workfile' instead")
    def open_file(self, filepath):
        """Deprecated variant of 'open_workfile'.

        Todo:
            Remove when all usages are replaced.

        """
        return self.open_workfile(filepath)

    @deprecated("Use 'get_current_workfile' instead")
    def current_file(self):
        """Deprecated variant of 'get_current_workfile'.

        Todo:
            Remove when all usages are replaced.

        """
        return self.get_current_workfile()

    @deprecated("Use 'workfile_has_unsaved_changes' instead")
    def has_unsaved_changes(self):
        """Deprecated variant of 'workfile_has_unsaved_changes'.

        Todo:
            Remove when all usages are replaced.

        """
        return self.workfile_has_unsaved_changes()

    def _copy_workfile(
        self,
        copy_workfile_context: CopyWorkfileContext,
        *,
        version: Optional[int],
        comment: Optional[str],
        description: Optional[str],
        open_workfile: bool,
    ) -> None:
        """Save workfile path with target folder and task context.

        It is expected that workfile is saved to the current project, but
            can be copied from the other project.

        Arguments 'rootless_path', 'workfile_entities', 'project_entity'
            and 'anatomy' can be filled to enhance efficiency if you already
            have access to the values.

        Argument 'project_settings' is used to calculate 'rootless_path'
            if it is not provided.

        Args:
            copy_workfile_context (CopyWorkfileContext): Prepared data
                for speed enhancements.
            version (Optional[int]): Version of the workfile. Information
                for workfile entity. Recommended to fill.
            comment (Optional[str]): Comment for the workfile.
            description (Optional[str]): Artist note for the workfile entity.
            open_workfile (bool): Open workfile when copied.

        """
        self._before_workfile_copy(copy_workfile_context)
        event_data = self._get_workfile_event_data(
            copy_workfile_context.project_name,
            copy_workfile_context.folder_entity,
            copy_workfile_context.task_entity,
            copy_workfile_context.dst_path,
        )
        self._emit_workfile_save_event(event_data, after_save=False)

        dst_dir = os.path.dirname(copy_workfile_context.dst_path)
        if not os.path.exists(dst_dir):
            os.makedirs(dst_dir, exist_ok=True)
        shutil.copy(
            copy_workfile_context.src_path,
            copy_workfile_context.dst_path
        )

        self._save_workfile_entity(
            copy_workfile_context,
            version,
            comment,
            description,
        )
        self._after_workfile_copy(copy_workfile_context)
        self._emit_workfile_save_event(event_data)

        if not open_workfile:
            return

        self.open_workfile_with_context(
            copy_workfile_context.dst_path,
            copy_workfile_context.folder_entity,
            copy_workfile_context.task_entity,
        )

    def _save_workfile_entity(
        self,
        save_workfile_context: SaveWorkfileContext,
        version: Optional[int],
        comment: Optional[str],
        description: Optional[str],
    ) -> Optional[dict[str, Any]]:
        """Create of update workfile entity to AYON based on provided data.

        Args:
            save_workfile_context (SaveWorkfileContext): Save workfile
                context with all prepared data.
            version (Optional[int]): Version of the workfile.
            comment (Optional[str]): Comment for the workfile.
            description (Optional[str]): Artist note for the workfile entity.

        Returns:
            Optional[dict[str, Any]]: Workfile entity.

        """
        from ayon_core.pipeline.workfile.utils import (
            save_workfile_info
        )

        project_name = self.get_current_project_name()
        if not description:
            description = None

        if not comment:
            comment = None

        rootless_path = save_workfile_context.rootless_path
        # It is not possible to create workfile infor without rootless path
        workfile_info = None
        if not rootless_path:
            return workfile_info

        if platform.system().lower() == "windows":
            rootless_path = rootless_path.replace("\\", "/")

        # Get application information
        app_info = self.get_app_information()
        data = {}
        if app_info.app_name:
            data["app_name"] = app_info.app_name
        if app_info.app_version:
            data["app_version"] = app_info.app_version

        # Use app group and app variant from applications addon (if available)
        app_addon_name = os.environ.get("AYON_APP_NAME")
        if not app_addon_name:
            app_addon_name = None

        app_addon_tools_s = os.environ.get("AYON_APP_TOOLS")
        app_addon_tools = []
        if app_addon_tools_s:
            app_addon_tools = app_addon_tools_s.split(";")

        data["ayon_app_name"] = app_addon_name
        data["ayon_app_tools"] = app_addon_tools

        workfile_info = save_workfile_info(
            project_name,
            save_workfile_context.task_entity["id"],
            rootless_path,
            self.name,
            version,
            comment,
            description,
            data=data,
            workfile_entities=save_workfile_context.workfile_entities,
        )
        return workfile_info

    def _create_extra_folders(
        self,
        folder_entity: dict[str, Any],
        task_entity: dict[str, Any],
        workdir: str,
    ) -> None:
        """Create extra folders in the workdir.

        This method should be called when workfile is saved or copied.

        Args:
            folder_entity (dict[str, Any]): Folder entity.
            task_entity (dict[str, Any]): Task entity.
            workdir (str): Workdir where workfile/s will be stored.

        """
        from ayon_core.pipeline.workfile.path_resolving import (
            create_workdir_extra_folders
        )

        project_name = self.get_current_project_name()

        # Create extra folders
        create_workdir_extra_folders(
            workdir,
            self.name,
            task_entity["taskType"],
            task_entity["name"],
            project_name
        )

    def _get_workfile_event_data(
        self,
        project_name: str,
        folder_entity: dict[str, Any],
        task_entity: dict[str, Any],
        filepath: str,
    ) -> dict[str, Optional[str]]:
        """Prepare workfile event data.

        Args:
            project_name (str): Name of the project where workfile lives.
            folder_entity (dict[str, Any]): Folder entity.
            task_entity (dict[str, Any]): Task entity.
            filepath (str): Path to the workfile.

        Returns:
            dict[str, Optional[str]]: Data for workfile event.

        """
        workdir, filename = os.path.split(filepath)
        return {
            "project_name": project_name,
            "folder_id": folder_entity["id"],
            "folder_path": folder_entity["path"],
            "task_id": task_entity["id"],
            "task_name": task_entity["name"],
            "host_name": self.name,
            "filepath": filepath,
            "filename": filename,
            "workdir_path": workdir,
        }

    def _before_workfile_open(
        self, open_workfile_context: OpenWorkfileContext
    ) -> None:
        """Before workfile is opened.

        This method is called before the workfile is opened in the host.

        Can be overridden to implement host specific logic.

        Args:
            open_workfile_context (OpenWorkfileContext): Context and path of
                workfile to open.

        """
        pass

    def _after_workfile_open(
        self, open_workfile_context: OpenWorkfileContext
    ) -> None:
        """After workfile is opened.

        This method is called after the workfile is opened in the host.

        Can be overridden to implement host specific logic.

        Args:
            open_workfile_context (OpenWorkfileContext): Context and path of
                opened workfile.

        """
        pass

    def _before_workfile_save(
        self, save_workfile_context: SaveWorkfileContext
    ) -> None:
        """Before workfile is saved.

        This method is called before the workfile is saved in the host.

        Can be overridden to implement host specific logic.

        Args:
            save_workfile_context (SaveWorkfileContext): Workfile path with
                target folder and task context.

        """
        pass

    def _after_workfile_save(
        self, save_workfile_context: SaveWorkfileContext
    ) -> None:
        """After workfile is saved.

        This method is called after the workfile is saved in the host.

        Can be overridden to implement host specific logic.

        Args:
            save_workfile_context (SaveWorkfileContext): Workfile path with
                target folder and task context.

        """
        workdir = os.path.dirname(save_workfile_context.dst_path)
        self._create_extra_folders(
            save_workfile_context.folder_entity,
            save_workfile_context.task_entity,
            workdir
        )

    def _before_workfile_copy(
        self, copy_workfile_context: CopyWorkfileContext
    ) -> None:
        """Before workfile is copied.

        This method is called before the workfile is copied by host
            integration.

        Can be overridden to implement host specific logic.

        Args:
            copy_workfile_context (CopyWorkfileContext): Source and destination
                path with context before workfile is copied.

        """
        pass

    def _after_workfile_copy(
        self, copy_workfile_context: CopyWorkfileContext
    ) -> None:
        """After workfile is copied.

        This method is called after the workfile is copied by host
            integration.

        Can be overridden to implement host specific logic.

        Args:
            copy_workfile_context (CopyWorkfileContext): Source and destination
                path with context after workfile is copied.

        """
        workdir = os.path.dirname(copy_workfile_context.dst_path)
        self._create_extra_folders(
            copy_workfile_context.folder_entity,
            copy_workfile_context.task_entity,
            workdir,
        )

    def _emit_workfile_open_event(
        self,
        event_data: dict[str, Optional[str]],
        after_open: bool = True,
    ) -> None:
        """Emit workfile save event.

        Emit event before and after workfile is opened.

        This method is not meant to be overridden.

        Other addons can listen to this event and do additional steps.

        Args:
            event_data (dict[str, Optional[str]]): Prepare event data.
            after_open (bool): Emit event after workfile is opened.

        """
        topics = []
        topic_end = "before"
        if after_open:
            topics.append("workfile.opened")
            topic_end = "after"

        # Keep backwards compatible event topic
        topics.append(f"workfile.open.{topic_end}")

        for topic in topics:
            emit_event(topic, event_data)

    def _emit_workfile_save_event(
        self,
        event_data: dict[str, Optional[str]],
        after_save: bool = True,
    ) -> None:
        """Emit workfile save event.

        Emit event before and after workfile is saved or copied.

        This method is not meant to be overridden.

        Other addons can listen to this event and do additional steps.

        Args:
            event_data (dict[str, Optional[str]]): Prepare event data.
            after_save (bool): Emit event after workfile is saved.

        """
        topics = []
        topic_end = "before"
        if after_save:
            topics.append("workfile.saved")
            topic_end = "after"

        # Keep backwards compatible event topic
        topics.append(f"workfile.save.{topic_end}")

        for topic in topics:
            emit_event(topic, event_data)