addon

class AYONAddon(ABC):
    """Base class of AYON addon.

    Attributes:
        enabled (bool): Is addon enabled.

    Args:
        manager (AddonsManager): Manager object who discovered addon.
        settings (dict[str, Any]): AYON settings.

    """
    enabled: bool = True
    _id = None

    # Temporary variable for 'version' property
    _missing_version_warned = False

    def __init__(
        self, manager: AddonsManager, settings: dict[str, Any]
    ) -> None:
        self.manager = manager

        self.log = Logger.get_logger(self.name)

        self.initialize(settings)

    @property
    def id(self) -> str:
        """Random id of addon object.

        Returns:
            str: Object id.

        """
        if self._id is None:
            self._id = uuid4()
        return self._id

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

        Returns:
            str: Addon name.

        """
        pass

    @property
    def version(self) -> str:
        """Addon version.

        Todo:
            Should be abstract property (required). Introduced in
                ayon-core 0.3.3 .

        Returns:
            str: Addon version as semver compatible string.

        """
        if not self.__class__._missing_version_warned:
            self.__class__._missing_version_warned = True
            print(
                f"DEV WARNING: Addon '{self.name}' does not have"
                f" defined version."
            )
        return "0.0.0"

    def initialize(self, settings: dict[str, Any]) -> None:
        """Initialization of addon attributes.

        It is not recommended to override __init__ that's why specific method
        was implemented.

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

        """
        pass

    def connect_with_addons(self, enabled_addons: list[AYONAddon]) -> None:
        """Connect with other enabled addons.

        Args:
            enabled_addons (list[AYONAddon]): Addons that are enabled.

        """
        pass

    def ensure_is_process_ready(
        self, process_context: ProcessContext
    ) -> None:
        """Make sure addon is prepared for a process.

        This method is called when some action makes sure that addon has set
        necessary data. For example if user should be logged in
        and filled credentials in environment variables this method should
        ask user for credentials.

        Implementation of this method is optional.

        Note:
            The logic can be similar to logic in tray, but tray does not
                require to be logged in.

        Args:
            process_context (ProcessContext): Context of child
                process.

        """
        pass

    def get_global_environments(self) -> dict[str, str]:
        """Get global environments values of addon.

        Environment variables that can be get only from system settings.

        Returns:
            dict[str, str]: Environment variables.

        """
        return {}

    def on_host_install(
        self,
        host: HostBase,
        host_name: str,
        project_name: str,
    ) -> None:
        """Host was installed which gives option to handle in-host logic.

        It is a good option to register in-host event callbacks which are
        specific for the addon. The addon is kept in memory for rest of
        the process.

        Arguments may change in future. E.g. 'host_name' should be possible
        to receive from 'host' object.

        Args:
            host (HostBase): Access to installed/registered
                host object.
            host_name (str): Name of host.
            project_name (str): Project name which is main part of host
                context.

        """
        pass

    def cli(self, addon_click_group: click.Group) -> None:
        """Add commands to click group.

        The best practise is to create click group for whole addon which is
        used to separate commands.

        Example:
            class MyPlugin(AYONAddon):
                ...
                def cli(self, addon_click_group):
                    addon_click_group.add_command(cli_main)


            @click.group(<addon name>, help="<Any help shown in cmd>")
            def cli_main():
                pass

            @cli_main.command()
            def mycommand():
                print("my_command")

        Args:
            addon_click_group (click.Group): Group to which can be added
                commands.

        """
        pass

class AddonsManager:
    """Manager of addons that helps to load and prepare them to work.

    Args:
        settings (Optional[dict[str, Any]]): AYON studio settings.
        initialize (Optional[bool]): Initialize addons on init.
            True by default.

    """
    # Helper attributes for report
    _report_total_key = "Total"
    _log = None

    def __init__(
        self,
        settings: Optional[dict[str, Any]] = None,
        initialize: bool = True,
    ) -> None:
        self._settings = settings

        self._addons: list[AYONAddon] = []
        self._addons_by_id: dict[str, AYONAddon] = {}
        self._addons_by_name: dict[str, AYONAddon] = {}
        # For report of time consumption
        self._report: dict[str, dict[str, int]] = {}

        if initialize:
            self.initialize_addons()
            self.connect_addons()

    def __getitem__(self, addon_name: str) -> AYONAddon:
        return self._addons_by_name[addon_name]

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

    def get(
        self, addon_name: str, default: Optional[Any] = None
    ) -> Union[AYONAddon, Any]:
        """Access addon by name.

        Args:
            addon_name (str): Name of addon which should be returned.
            default (Any): Default output if addon is not available.

        Returns:
            Union[AYONAddon, Any]: Addon found by name or `default`.

        """
        return self._addons_by_name.get(addon_name, default)

    @property
    def addons(self) -> list[AYONAddon]:
        return list(self._addons)

    @property
    def addons_by_id(self) -> dict[str, AYONAddon]:
        return dict(self._addons_by_id)

    @property
    def addons_by_name(self) -> dict[str, AYONAddon]:
        return dict(self._addons_by_name)

    def get_enabled_addon(
        self, addon_name: str, default: Optional[Any] = None
    ) -> Union[AYONAddon, Any]:
        """Fast access to enabled addon.

        If addon is available but is not enabled default value is returned.

        Args:
            addon_name (str): Name of addon which should be returned.
            default (Any): Default output if addon is not available or is
                not enabled.

        Returns:
            Union[AYONAddon, Any]: Enabled addon found by name or None.

        """
        addon = self.get(addon_name)
        if addon is not None and addon.enabled:
            return addon
        return default

    def get_enabled_addons(self) -> list[AYONAddon]:
        """Enabled addons initialized by the manager.

        Returns:
            list[AYONAddon]: Initialized and enabled addons.

        """
        return [
            addon
            for addon in self._addons
            if addon.enabled
        ]

    def initialize_addons(self) -> None:
        """Import and initialize addons."""
        # Make sure modules are loaded
        load_addons()

        self.log.debug("*** AYON addons initialization.")

        # Prepare settings for addons
        settings = self._settings
        if settings is None:
            settings = get_studio_settings()

        report = {}
        time_start = time.time()
        prev_start_time = time_start

        addon_classes = []
        for module in _LoadCache.addon_modules:
            # Go through globals in `ayon_core.modules`
            for name in dir(module):
                modules_item = getattr(module, name, None)
                # Filter globals that are not classes which inherit from
                #   AYONAddon
                if (
                    not inspect.isclass(modules_item)
                    or modules_item is AYONAddon
                    or not issubclass(modules_item, AYONAddon)
                ):
                    continue

                # Check if class is abstract (Developing purpose)
                if inspect.isabstract(modules_item):
                    # Find abstract attributes by convention on `abc` module
                    not_implemented = []
                    for attr_name in dir(modules_item):
                        attr = getattr(modules_item, attr_name, None)
                        abs_method = getattr(
                            attr, "__isabstractmethod__", None
                        )
                        if attr and abs_method:
                            not_implemented.append(attr_name)

                    # Log missing implementations
                    self.log.warning((
                        "Skipping abstract Class: {}."
                        " Missing implementations: {}"
                    ).format(name, ", ".join(not_implemented)))
                    continue

                addon_classes.append(modules_item)

        for addon_cls in addon_classes:
            name = addon_cls.__name__
            try:
                addon = addon_cls(self, settings)
                # Store initialized object
                self._addons.append(addon)
                self._addons_by_id[addon.id] = addon
                self._addons_by_name[addon.name] = addon

                now = time.time()
                report[addon.__class__.__name__] = now - prev_start_time
                prev_start_time = now

            except Exception:
                self.log.warning(
                    "Initialization of addon '{}' failed.".format(name),
                    exc_info=True
                )

        for addon_name in sorted(self._addons_by_name.keys()):
            addon = self._addons_by_name[addon_name]
            enabled_str = "X" if addon.enabled else " "
            self.log.debug(
                f"[{enabled_str}] {addon.name} ({addon.version})"
            )

        if self._report is not None:
            report[self._report_total_key] = time.time() - time_start
            self._report["Initialization"] = report

    def connect_addons(self) -> None:
        """Trigger connection with other enabled addons.

        Addons should handle their interfaces in `connect_with_addons`.
        """
        report = {}
        time_start = time.time()
        prev_start_time = time_start
        enabled_addons = self.get_enabled_addons()
        self.log.debug(f"Has {len(enabled_addons)} enabled addons.")
        for addon in enabled_addons:
            try:
                addon.connect_with_addons(enabled_addons)

            except Exception:
                self.log.error(
                    "BUG: Module failed on connection with other modules.",
                    exc_info=True
                )

            now = time.time()
            report[addon.__class__.__name__] = now - prev_start_time
            prev_start_time = now

        if self._report is not None:
            report[self._report_total_key] = time.time() - time_start
            self._report["Connect modules"] = report

    def collect_global_environments(self) -> dict[str, str]:
        """Helper to collect global environment variabled from modules.

        Returns:
            dict: Global environment variables from enabled modules.

        Raises:
            AssertionError: Global environment variables must be unique for
                all modules.
        """
        module_envs = {}
        for module in self.get_enabled_addons():
            # Collect global module's global environments
            _envs = module.get_global_environments()
            for key, value in _envs.items():
                if key in module_envs:
                    # TODO better error message
                    raise AssertionError(
                        "Duplicated environment key {}".format(key)
                    )
                module_envs[key] = value
        return module_envs

    def collect_plugin_paths(self) -> dict[str, list[str]]:
        """Helper to collect all plugins from modules inherited IPluginPaths.

        Unknown keys are logged out.

        Deprecated:
            Use targeted methods 'collect_launcher_action_paths',
                'collect_create_plugin_paths', 'collect_load_plugin_paths',
                'collect_publish_plugin_paths' and
                 'collect_inventory_action_paths' to collect plugin paths.

        Returns:
            dict: Output is dictionary with keys "publish", "create", "load",
                "actions" and "inventory" each containing list of paths.

        """
        warnings.warn(
            "Used deprecated method 'collect_plugin_paths'. Please use"
            " targeted methods 'collect_launcher_action_paths',"
            " 'collect_create_plugin_paths', 'collect_load_plugin_paths'"
            " 'collect_publish_plugin_paths' and"
            " 'collect_inventory_action_paths'",
            DeprecationWarning,
            stacklevel=2
        )
        # Output structure
        output = {
            "publish": [],
            "create": [],
            "load": [],
            "actions": [],
            "inventory": []
        }
        unknown_keys_by_addon = {}
        for addon in self.get_enabled_addons():
            # Skip module that do not inherit from `IPluginPaths`
            if not isinstance(addon, IPluginPaths):
                continue
            plugin_paths = addon.get_plugin_paths()
            for key, value in plugin_paths.items():
                # Filter unknown keys
                if key not in output:
                    if addon.name not in unknown_keys_by_addon:
                        unknown_keys_by_addon[addon.name] = []
                    unknown_keys_by_addon[addon.name].append(key)
                    continue

                # Skip if value is empty
                if not value:
                    continue

                # Convert to list if value is not list
                if not isinstance(value, (list, tuple, set)):
                    value = [value]
                output[key].extend(value)

        # Report unknown keys (Developing purposes)
        if unknown_keys_by_addon:
            expected_keys = ", ".join([
                f'"{key}"' for key in output.keys()
            ])
            msg_template = "Addon: \"{}\" - got key {}"
            msg_items = []
            for addon_name, keys in unknown_keys_by_addon.items():
                joined_keys = ", ".join([
                    "\"{}\"".format(key) for key in keys
                ])
                msg_items.append(msg_template.format(addon_name, joined_keys))
            joined_items = " | ".join(msg_items)
            self.log.warning(
                f"Expected keys from `get_plugin_paths` are {expected_keys}."
                f" {joined_items}"
            )
        return output

    def _collect_plugin_paths(self, method_name: str, *args, **kwargs):
        output = []
        for addon in self.get_enabled_addons():
            # Skip addon that do not inherit from `IPluginPaths`
            if not isinstance(addon, IPluginPaths):
                continue

            paths = None
            method = getattr(addon, method_name)
            try:
                paths = method(*args, **kwargs)
            except Exception:
                self.log.warning(
                    "Failed to get plugin paths from addon"
                    f" '{addon.name}' using '{method_name}'.",
                    exc_info=True
                )

            if not paths:
                continue

            if isinstance(paths, str):
                paths = [paths]
                self.log.warning(
                    f"Addon '{addon.name}' returned invalid output type"
                    f" from '{method_name}'."
                    f" Got 'str' expected 'list[str]'."
                )
            output.extend(paths)
        return output

    def collect_launcher_action_paths(self) -> list[str]:
        """Helper to collect launcher action paths from addons.

        Returns:
            list: List of paths to launcher actions.

        """
        output = self._collect_plugin_paths(
            "get_launcher_action_paths"
        )
        # Add default core actions
        actions_dir = os.path.join(AYON_CORE_ROOT, "plugins", "actions")
        output.insert(0, actions_dir)
        return output

    def collect_create_plugin_paths(self, host_name: str) -> list[str]:
        """Helper to collect creator plugin paths from addons.

        Args:
            host_name (str): For which host are creators meant.

        Returns:
            list[str]: List of creator plugin paths.

        """
        return self._collect_plugin_paths(
            "get_create_plugin_paths",
            host_name
        )

    collect_creator_plugin_paths = collect_create_plugin_paths

    def collect_load_plugin_paths(self, host_name: str) -> list[str]:
        """Helper to collect load plugin paths from addons.

        Args:
            host_name (str): For which host are load plugins meant.

        Returns:
            list[str]: List of load plugin paths.

        """
        return self._collect_plugin_paths(
            "get_load_plugin_paths",
            host_name
        )

    def collect_publish_plugin_paths(self, host_name: str) -> list[str]:
        """Helper to collect load plugin paths from addons.

        Args:
            host_name (str): For which host are load plugins meant.

        Returns:
            list[str]: List of pyblish plugin paths.

        """
        return self._collect_plugin_paths(
            "get_publish_plugin_paths",
            host_name
        )

    def collect_inventory_action_paths(self, host_name: str) -> list[str]:
        """Helper to collect load plugin paths from addons.

        Args:
            host_name (str): For which host are load plugins meant.

        Returns:
            list: List of pyblish plugin paths.

        """
        return self._collect_plugin_paths(
            "get_inventory_action_paths",
            host_name
        )

    def get_host_addon(self, host_name: str) -> Optional[AYONAddon]:
        """Find host addon by host name.

        Args:
            host_name (str): Host name for which is found host addon.

        Returns:
            Optional[AYONAddon]: Found host addon by name or `None`.
        """

        for addon in self.get_enabled_addons():
            if (
                isinstance(addon, IHostAddon)
                and addon.host_name == host_name
            ):
                return addon
        return None

    def get_host_names(self) -> set[str]:
        """List of available host names based on host addons.

        Returns:
            set[str]: All available host names based on enabled addons
                inheriting 'IHostAddon'.

        """
        return {
            addon.host_name
            for addon in self.get_enabled_addons()
            if isinstance(addon, IHostAddon)
        }

    def print_report(self) -> None:
        """Print out report of time spent on addons initialization parts.

        Reporting is not automated must be implemented for each initialization
        part separately. Reports must be stored to `_report` attribute.
        Print is skipped if `_report` is empty.

        Attribute `_report` is dictionary where key is "label" describing
        the processed part and value is dictionary where key is addon's
        class name and value is time delta of it's processing.

        It is good idea to add total time delta on processed part under key
        which is defined in attribute `_report_total_key`. By default has value
        `"Total"` but use the attribute please.

        ```javascript
        {
            "Initialization": {
                "FtrackAddon": 0.003,
                ...
                "Total": 1.003,
            },
            ...
        }
        ```
        """

        if not self._report:
            return

        available_col_names = set()
        for addon_names in self._report.values():
            available_col_names |= set(addon_names.keys())

        # Prepare ordered dictionary for columns
        addons_info = [
            _AddonReportInfo.from_addon(addon, self._report)
            for addon in self.addons
            if addon.__class__.__name__ in available_col_names
        ]
        addons_info.sort(key=lambda x: x.name)

        addon_name_rows = [
            addon_info.name
            for addon_info in addons_info
        ]
        addon_version_rows = [
            addon_info.version
            for addon_info in addons_info
        ]

        # Add total key (as last addon)
        addon_name_rows.append(self._report_total_key)
        addon_version_rows.append(f"({len(addons_info)})")

        cols = collections.OrderedDict()
        # Add addon names to first columnt
        cols["Addon name"] = addon_name_rows
        cols["Version"] = addon_version_rows

        # Add columns from report
        total_by_addon = {
            row: 0
            for row in addon_name_rows
        }
        for label in self._report.keys():
            rows = []
            col_total = 0
            for addon_info in addons_info:
                value = addon_info.report_value_by_label.get(label)
                if value is None:
                    rows.append("N/A")
                    continue
                rows.append("{:.3f}".format(value))
                total_by_addon[addon_info.name] += value
                col_total += value
            total_by_addon[self._report_total_key] += col_total
            rows.append("{:.3f}".format(col_total))
            cols[label] = rows
        # Add to also total column that should sum the row
        cols[self._report_total_key] = [
            "{:.3f}".format(total_by_addon[addon_name])
            for addon_name in cols["Addon name"]
        ]

        # Prepare column widths and total row count
        # - column width is by
        col_widths = {}
        total_rows = None
        for key, values in cols.items():
            if total_rows is None:
                total_rows = 1 + len(values)
            max_width = len(key)
            for value in values:
                value_length = len(value)
                if value_length > max_width:
                    max_width = value_length
            col_widths[key] = max_width

        rows = []
        for _idx in range(total_rows):
            rows.append([])

        for key, values in cols.items():
            width = col_widths[key]
            idx = 0
            rows[idx].append(key.ljust(width))
            for value in values:
                idx += 1
                rows[idx].append(value.ljust(width))

        filler_parts = []
        for width in col_widths.values():
            filler_parts.append(width * "-")
        filler = "+".join(filler_parts)

        formatted_rows = [filler]
        last_row_idx = len(rows) - 1
        for idx, row in enumerate(rows):
            # Add filler before last row
            if idx == last_row_idx:
                formatted_rows.append(filler)

            formatted_rows.append("|".join(row))

            # Add filler after first row
            if idx == 0:
                formatted_rows.append(filler)

        # Join rows with newline char and add new line at the end
        output = "\n".join(formatted_rows) + "\n"
        print(output)

class IHostAddon(AYONInterface):
    """Addon which also contain a host implementation."""

    @property
    @abstractmethod
    def host_name(self) -> str:
        """Name of host which addon represents."""

    def get_workfile_extensions(self) -> list[str]:  # noqa: PLR6301
        """Define workfile extensions for host.

        Not all hosts support workfiles thus this is optional implementation.

        Returns:
            List[str]: Extensions used for workfiles with dot.

        """
        return []

class IPluginPaths(AYONInterface):
    """Addon wants to register plugin paths."""

    def get_plugin_paths(self) -> dict[str, list[str]]:
        """Return plugin paths for addon.

        This method was abstract (required) in the past, so raise the required
            'core' addon version when 'get_plugin_paths' is removed from
            addon.

        Deprecated:
            Please implement specific methods 'get_create_plugin_paths',
                'get_load_plugin_paths', 'get_inventory_action_paths' and
                'get_publish_plugin_paths' to return plugin paths.

        Returns:
            dict[str, list[str]]: Plugin paths for addon.

        """
        return {}

    def _get_plugin_paths_by_type(
        self, plugin_type: str
    ) -> list[str]:
        """Get plugin paths by type.

        Args:
            plugin_type (str): Type of plugin paths to get.

        Returns:
            list[str]: List of plugin paths.

        """
        paths = self.get_plugin_paths()
        if not paths or plugin_type not in paths:
            return []

        paths = paths[plugin_type]
        if not paths:
            return []

        if not isinstance(paths, (list, tuple, set)):
            paths = [paths]

        new_function_name = "get_launcher_action_paths"
        if plugin_type == "create":
            new_function_name = "get_create_plugin_paths"
        elif plugin_type == "load":
            new_function_name = "get_load_plugin_paths"
        elif plugin_type == "publish":
            new_function_name = "get_publish_plugin_paths"
        elif plugin_type == "inventory":
            new_function_name = "get_inventory_action_paths"

        warnings.warn(
            f"Addon '{self.name}' returns '{plugin_type}' paths using"
            " 'get_plugin_paths' method. Please implement"
            f" '{new_function_name}' instead.",
            DeprecationWarning,
            stacklevel=2
        )
        return paths

    def get_launcher_action_paths(self) -> list[str]:
        """Receive launcher actions paths.

        Give addons ability to add launcher actions paths.

        Returns:
            list[str]: List of launcher action paths.

        """
        return self._get_plugin_paths_by_type("actions")

    def get_create_plugin_paths(self, host_name: str) -> list[str]:
        """Receive create plugin paths.

        Give addons ability to add create plugin paths based on host name.

        Notes:
            Default implementation uses 'get_plugin_paths' and always return
                all create plugin paths.

        Args:
            host_name (str): For which host are the plugins meant.

        Returns:
            list[str]: List of create plugin paths.

        """
        return self._get_plugin_paths_by_type("create")

    def get_load_plugin_paths(self, host_name: str) -> list[str]:
        """Receive load plugin paths.

        Give addons ability to add load plugin paths based on host name.

        Notes:
            Default implementation uses 'get_plugin_paths' and always return
                all load plugin paths.

        Args:
            host_name (str): For which host are the plugins meant.

        Returns:
            list[str]: List of load plugin paths.

        """
        return self._get_plugin_paths_by_type("load")

    def get_publish_plugin_paths(self, host_name: str) -> list[str]:
        """Receive publish plugin paths.

        Give addons ability to add publish plugin paths based on host name.

        Notes:
           Default implementation uses 'get_plugin_paths' and always return
               all publish plugin paths.

        Args:
           host_name (str): For which host are the plugins meant.

        Returns:
            list[str]: List of publish plugin paths.

        """
        return self._get_plugin_paths_by_type("publish")

    def get_inventory_action_paths(self, host_name: str) -> list[str]:
        """Receive inventory action paths.

        Give addons ability to add inventory action plugin paths.

        Notes:
           Default implementation uses 'get_plugin_paths' and always return
               all publish plugin paths.

        Args:
           host_name (str): For which host are the plugins meant.

        Returns:
            list[str]: List of inventory action plugin paths.

        """
        return self._get_plugin_paths_by_type("inventory")