addon

class ApplicationsAddon(AYONAddon, IPluginPaths, ITrayAction):
    name = "applications"
    version = __version__

    # Tray action attributes
    label = "Process Monitor"
    admin_action = True

    _icons_cache: dict[str, bytes | None] = {}

    def tray_init(self) -> None:
        """Initialize the tray action."""
        self._process_monitor_window: Optional[ProcessMonitorWindow] = None

    def on_action_trigger(self) -> None:
        """Action triggered when the tray icon is clicked."""
        from ayon_applications.ui.process_monitor import (
            ProcessMonitorWindow,
        )
        if self._process_monitor_window is None:
            self._process_monitor_window = ProcessMonitorWindow()

        self._process_monitor_window.show()
        self._process_monitor_window.raise_()
        self._process_monitor_window.activateWindow()

    def get_app_environments_for_context(
        self,
        project_name: str,
        folder_path: str,
        task_name: str,
        full_app_name: str,
        env_group: Optional[str] = None,
        launch_type: Optional[str] = None,
        env: Optional[dict[str, str]] = None,
    ) -> dict[str, str]:
        """Calculate environment variables for launch context.

        Args:
            project_name (str): Project name.
            folder_path (str): Folder path.
            task_name (str): Task name.
            full_app_name (str): Full application name.
            env_group (Optional[str]): Environment group.
            launch_type (Optional[str]): Launch type.
            env (Optional[dict[str, str]]): Environment variables to update.

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

        """
        from ayon_applications.utils import get_app_environments_for_context

        if not full_app_name:
            return {}

        return get_app_environments_for_context(
            project_name,
            folder_path,
            task_name,
            full_app_name,
            env_group=env_group,
            launch_type=launch_type,
            env=env,
            addons_manager=self.manager
        )

    def get_farm_publish_environment_variables(
        self,
        project_name: str,
        folder_path: str,
        task_name: str,
        full_app_name: Optional[str] = None,
        env_group: Optional[str] = None,
    ) -> dict[str, str]:
        """Calculate environment variables for farm publish.

        Args:
            project_name (str): Project name.
            folder_path (str): Folder path.
            task_name (str): Task name.
            env_group (Optional[str]): Environment group.
            full_app_name (Optional[str]): Full application name. Value from
                environment variable 'AYON_APP_NAME' is used if 'None' is
                passed.

        Returns:
            dict[str, str]: Environment variables for farm publish.

        """
        if full_app_name is None:
            full_app_name = os.getenv("AYON_APP_NAME")

        return self.get_app_environments_for_context(
            project_name,
            folder_path,
            task_name,
            full_app_name,
            env_group=env_group,
            launch_type=LaunchTypes.farm_publish
        )

    def get_applications_manager(
        self, settings: Optional[dict[str, Any]] = None
    ) -> "ApplicationManager":
        """Get applications manager.

        Args:
            settings (Optional[dict]): Studio/project settings.

        Returns:
            ApplicationManager: Applications manager.

        """
        return ApplicationManager(settings)

    def get_plugin_paths(self) -> dict[str, list[str]]:
        return {}

    def get_publish_plugin_paths(self, host_name: str) -> list[str]:
        return [
            os.path.join(APPLICATIONS_ADDON_ROOT, "plugins", "publish")
        ]

    def get_launch_hook_paths(self, app: "Application") -> list[str]:
        return [
            os.path.join(APPLICATIONS_ADDON_ROOT, "hooks")
        ]

    def get_loader_action_plugin_paths(self, host_name):
        return [
            os.path.join(APPLICATIONS_ADDON_ROOT, "plugins", "load_actions"),
        ]

    def get_app_icon_path(self, icon_filename: str) -> str:
        """DEPRECATED Get icon path.

        Args:
            icon_filename (str): Icon filename.

        Returns:
            Optional[str]: Icon path or None if not found.

        """
        return get_app_icon_path(icon_filename)

    @classmethod
    def get_custom_icons_info(cls) -> list[dict[str, str]]:
        """List custom icons available on the server.

        Returns:
            list[dict[str, str]]: List of custom icons.

        """
        endpoint = f"addons/{cls.name}/{cls.version}/customIcons"
        response = ayon_api.get(endpoint)
        response.raise_for_status()
        return response.data["icons"]

    @classmethod
    def upload_custom_icon(
        cls, path: str, filename: str | None = None
    ) -> None:
        """Upload custom icon to AYON server.

        Args:
            path (str): Path to icon file.
            filename (str | None): Icon filename which will be used
                to store the icon on the server. This value is then used in
                settings.

        """
        if filename is None:
            filename = os.path.basename(path)
        endpoint = f"addons/{cls.name}/{cls.version}/customIcons/{filename}"
        response = ayon_api.upload_file(
            endpoint, path
        )
        response.raise_for_status()

    @classmethod
    def delete_custom_icon(cls, filename: str) -> None:
        """Delete custom icon from AYON server.

        Args:
            filename (str): Icon filename which will be deleted
                from the server.

        """
        endpoint = f"addons/{cls.name}/{cls.version}/customIcons/{filename}"
        response = ayon_api.delete(endpoint)
        response.raise_for_status()

    @classmethod
    def get_app_icon_url(
        cls, icon: dict[str, Any] | str, server: bool = False
    ) -> str | None:
        """Get icon path.

        icon filename can be either a full URL (http/https/file/...)
        or a bare filename. Full URLs are used as is while bare filenames
        resolve to the addons icons folder.

        Method does not validate if icon filename exist on server.

        Args:
            icon (dict[str, Any] | str): Icon name.
            server (bool): Return url to AYON server.

        Returns:
            str | None: Icon path or None is server url is not
                available.

        """
        if not icon:
            return None

        if isinstance(icon, str):
            icon_filename = icon
        elif isinstance(icon, dict):
            # NOTE At this moment the url always leads to addon's icons
            #   endpoint and last part of path is filename
            url = icon.get("url")
            if not isinstance(url, str):
                return None
            icon_filename = os.path.basename(url)

        else:
            return None

        # check if its a full URL
        try:
            url = urllib.parse.urlparse(icon_filename)
            if url.scheme:
                return icon_filename
        except Exception:
            pass

        icon_name = os.path.basename(icon_filename)
        if server:
            base_url = ayon_api.get_base_url()
            return (
                f"{base_url}/api/addons/{cls.name}/{cls.version}"
                f"/icons/{icon_name}"
            )
        server_url = os.getenv("AYON_WEBSERVER_URL")
        if not server_url:
            return None
        return "/".join([
            server_url, "addons", cls.name, "icons", icon_name
        ])

    @classmethod
    def get_application_items(
        cls,
        project_name: str | None = None,
        task_id: str | None = None,
        *,
        variant: str | None = None,
        version: str | None = None,
    ) -> list[dict[str, Any]]:
        """Get application items.

        This is meant as api for other addons to get application items for
            a given context. Can also filter applications for a specific task.

        It does handle project bundles and settings variant automatically.

        Args:
            project_name (str | None): Project name.
            task_id (str | None): Task id for which applications are fitlered.
            variant (str | None): Settings variant. Current settings variant
                is used if not passed in.
            version (str | None): Specific version of applications addon
                to get items for. If None, it will use the version
                resolved for current context (variant and project).

        Example application dict (may vary based on applications
            addon version):
            {
                "host_name": str
                "full_name": str
                "full_label": str
                "group_label": str
                "variant_label": str
                "icon": dict[str, str] | None
                "show_grouped": bool
            }

        Returns:
            list[dict]: Application items.

        """
        if variant is None:
            variant = get_settings_variant()

        query_params = {"variant": variant}
        if version is not None:
            query_params["version"] = version

        query = urllib.parse.urlencode(query_params)
        context_path = ""
        if project_name:
            context_path = f"/{project_name}"
            if task_id:
                context_path = f"{context_path}/task/{task_id}"

        response = ayon_api.get(
            f"addons/{cls.name}/{cls.version}/"
            f"apps{context_path}?{query}"
        )
        app_items = response.data["applications"]

        # Fill icon urls with 'addon_url' and prepare icon definitions
        if not version:
            version = cls.version
        addon_url = f"/addons/{cls.name}/{version}"

        for app_item in app_items:
            icon = app_item["icon"]
            if not icon:
                continue
            try:
                url = icon["url"].format(addon_url=addon_url)
            except Exception:
                continue
            app_item["icon"] = {
                "type": "ayon_url",
                "url": url.lstrip("/"),
            }
        return app_items

    @classmethod
    def get_tool_items(
        cls,
        project_name: str | None = None,
        *,
        variant: str | None = None,
        version: str | None = None,
    ) -> list[dict[str, Any]]:
        """Get tool items.

        This is meant as api for other addons to get tools items for a given
            context.

        It does handle project bundles and settings variant automatically.

        Args:
            project_name (str | None): Project name.
            variant (str | None): Settings variant. Current settings variant
                is used if not passed in.
            version (str | None): Specific version of applications addon
                to get items for. If None, it will use the version
                resolved for current context (variant and project).

        Example tool dict (may vary based on applications addon version):
            {
                "full_name": str,
                "full_label": str,
                "group_label": str,
                "variant_label": str,
                "host_names": list[str],
                "app_variants": list[str],
            }

        Returns:
            list[dict]: Tool items.

        """
        if variant is None:
            variant = get_settings_variant()

        query_params = {"variant": variant}
        if version is not None:
            query_params["version"] = version

        query = urllib.parse.urlencode(query_params)

        context_path = ""
        if project_name:
            context_path = f"/{project_name}"

        response = ayon_api.get(
            f"addons/{cls.name}/{cls.version}/"
            f"tools{context_path}?{query}"
        )
        return response.data["applications"]

    def launch_application(
        self,
        app_name: str,
        project_name: str,
        folder_path: str,
        task_name: str,
        workfile_path: Optional[str] = None,
        use_last_workfile: Optional[bool] = None,
    ):
        """Launch application.

        Args:
            app_name (str): Full application name e.g. 'maya/2024'.
            project_name (str): Project name.
            folder_path (str): Folder path.
            task_name (str): Task name.
            workfile_path (Optional[str]): Workfile path to use.
            use_last_workfile (Optional[bool]): Explicitly tell to use or
                not use last workfile. Ignored if 'workfile_path' is passed.

        """
        ensure_addons_are_process_ready(
            addon_name=self.name,
            addon_version=self.version,
            project_name=project_name,
        )
        headless = is_headless_mode_enabled()

        data = {
            "project_name": project_name,
            "folder_path": folder_path,
            "task_name": task_name,
        }
        # Backwards compatibility 'workfile_path' was added
        #   before 'use_last_workfile'
        if isinstance(workfile_path, bool):
            use_last_workfile = workfile_path
            workfile_path = None
            warnings.warn(
                "Passed 'use_last_workfile' as positional argument."
                " Use explicit 'use_last_workfile' keyword argument instead.",
                DeprecationWarning,
                stacklevel=2,
            )

        if workfile_path:
            data["workfile_path"] = workfile_path
            # Backwards compatibility to be able to use 'workfile_path'
            #   argument with older ayon-core
            # use_last_workfile = False
            data["last_workfile_path"] = workfile_path
            data["start_last_workfile"] = True

        elif use_last_workfile is not None:
            data["start_last_workfile"] = use_last_workfile

        # TODO handle raise errors
        failed = True
        message = None
        detail = None
        try:
            app_manager = self.get_applications_manager()
            app_manager.launch(app_name, **data)
            failed = False

        except (
            ApplicationLaunchFailed,
            ApplicationExecutableNotFound,
            ApplicationNotFound,
        ) as exc:
            message = str(exc)
            self.log.warning(f"Application launch failed: {message}")

        except Exception as exc:
            message = "An unexpected error happened"
            detail = "".join(traceback.format_exception(*sys.exc_info()))
            self.log.warning(
                f"Application launch failed: {str(exc)}",
                exc_info=True
            )

        if not failed:
            return

        if not headless:
            self._show_launch_error_dialog(message, detail)
        sys.exit(1)

    def webserver_initialization(self, manager: "WebServerManager") -> None:
        """Initialize webserver.

        Add localhost handler for icons requests.

        This was added for ftrack which is showing icons

        Args:
            manager (WebServerManager): Webserver manager.

        """
        def _cache_icon(filename: str, data: bytes | None) -> None:
            self.__class__._icons_cache[filename] = data
            if len(self.__class__._icons_cache) > 256:
                self.__class__._icons_cache.pop(
                    next(iter(self.__class__._icons_cache))
                )

        async def _get_web_icon(request):
            from aiohttp import web, ClientSession

            filename: str = os.path.basename(request.match_info["filename"])
            # TODO find better way how to cache
            if filename not in self.__class__._icons_cache:
                base_url = ayon_api.get_base_url()
                url = (
                    f"{base_url}/api/addons/{self.name}/{self.version}"
                    f"/icons/{filename}"
                )
                data = None
                async with ClientSession() as session:
                    async with session.get(url) as resp:
                        if resp.status != 200:
                            data = await resp.read()

                _cache_icon(filename, data)

            body = self.__class__._icons_cache[filename]
            if body is None:
                raise web.HTTPNotFound()
            return web.Response(body=body)

        manager.add_addon_route(
            self.name,
            "/icons/{filename}",
            "GET",
            _get_web_icon,
        )

    # --- CLI ---
    def cli(self, addon_click_group) -> None:
        main_group = click_wrap.group(
            self._cli_main, name=self.name, help="Applications addon"
        )
        (
            main_group.command(
                self._cli_extract_environments,
                name="extractenvironments",
                help=(
                    "Extract environment variables for context into json file"
                )
            )
            .argument("output_json_path")
            .option("--project", help="Project name", default=None)
            .option("--folder", help="Folder path", default=None)
            .option("--task", help="Task name", default=None)
            .option("--app", help="Full application name", default=None)
            .option(
                "--envgroup",
                help="Environment group (e.g. \"farm\")",
                default=None
            )
        )
        (
            main_group.command(
                self._cli_launch_context_names,
                name="launch",
                help="Launch application"
            )
            .option("--app", required=True, help="Full application name")
            .option("--project", required=True, help="Project name")
            .option("--folder", required=True, help="Folder path")
            .option("--task", required=True, help="Task name")
            .option(
                "--workfile-path",
                required=False,
                help="Workfile path",
                default=None,
            )
            .option(
                "--use-last-workfile",
                help="Use last workfile",
                default=None,
            )
        )
        (
            main_group.command(
                self._cli_launch_with_task_id,
                name="launch-by-id",
                help="Launch application"
            )
            .option("--app", required=True, help="Full application name")
            .option("--project", required=True, help="Project name")
            .option("--task-id", required=True, help="Task id")
            .option(
                "--workfile-path",
                required=False,
                help="Workfile path",
                default=None,
            )
            .option(
                "--use-last-workfile",
                help="Use last workfile",
                default=None,
            )
        )
        (
            main_group.command(
                self._cli_launch_with_workfile_id,
                name="launch-by-workfile-id",
                help="Launch application using workfile id"
            )
            .option("--app", required=True, help="Full application name")
            .option("--project", required=True, help="Project name")
            .option("--workfile-id", required=True, help="Workfile id")
        )
        (
            main_group.command(
                self._cli_launch_with_debug_terminal,
                name="launch-debug-terminal",
                help="Launch with debug terminal"
            )
            .option("--project", required=True, help="Project name")
            .option("--task-id", required=True, help="Task id")
            .option(
                "--app",
                required=False,
                help="Full application name",
                default=None,
            )
        )
        # Convert main command to click object and add it to parent group
        addon_click_group.add_command(
            main_group.to_click_obj()
        )

    def _cli_main(self) -> None:
        pass

    def _cli_extract_environments(
        self,
        output_json_path: str,
        project: str,
        folder: str,
        task: str,
        app: str,
        envgroup: str,
    ) -> None:
        """Produces json file with environment based on project and app.

        Called by farm integration to propagate environment into farm jobs.

        Args:
            output_json_path (str): Output json file path.
            project (str): Project name.
            folder (str): Folder path.
            task (str): Task name.
            app (str): Full application name e.g. 'maya/2024'.
            envgroup (str): Environment group.

        """
        if all((project, folder, task, app)):
            env = self.get_farm_publish_environment_variables(
                project, folder, task, app, env_group=envgroup,
            )
        else:
            env = os.environ.copy()

        output_dir = os.path.dirname(output_json_path)
        os.makedirs(output_dir, exist_ok=True)

        with open(output_json_path, "w") as file_stream:
            json.dump(env, file_stream, indent=4)

    def _cli_launch_context_names(
        self,
        project: str,
        folder: str,
        task: str,
        app: str,
        workfile_path: Optional[str] = None,
        use_last_workfile: Optional["BoolArg"] = None,
    ) -> None:
        """Launch application.

        Args:
            project (str): Project name.
            folder (str): Folder path.
            task (str): Task name.
            app (str): Full application name e.g. 'maya/2024'.
            workfile_path (str | None): Workfile path to use.
            use_last_workfile (Literal["1", "0"] | None): Explicitly tell
                to use last workfile.

        """
        if workfile_path:
            use_last_workfile = False

        elif use_last_workfile is not None:
            use_last_workfile = env_value_to_bool(
                use_last_workfile, default=None
            )

        self.launch_application(
            app,
            project,
            folder,
            task,
            workfile_path=workfile_path,
            use_last_workfile=use_last_workfile,
        )

    def _cli_launch_with_task_id(
        self,
        project: str,
        task_id: str,
        app: str,
        workfile_path: Optional[str] = None,
        use_last_workfile: Optional["BoolArg"] = None,
    ) -> None:
        """Launch application using project name, task id and full app name.

        Args:
            project (str): Project name.
            task_id (str): Task id.
            app (str): Full application name e.g. 'maya/2024'.
            workfile_path (str | None): Workfile path to use.
            use_last_workfile (Literal["1", "0"] | None): Explicitly tell
                to use last workfile.

        """
        if workfile_path:
            use_last_workfile = False

        elif use_last_workfile is not None:
            use_last_workfile = env_value_to_bool(
                value=use_last_workfile, default=None
            )

        task_entity = ayon_api.get_task_by_id(
            project, task_id, fields={"name", "folderId"}
        )
        folder_entity = ayon_api.get_folder_by_id(
            project, task_entity["folderId"], fields={"path"}
        )
        self.launch_application(
            app,
            project,
            folder_entity["path"],
            task_entity["name"],
            workfile_path=workfile_path,
            use_last_workfile=use_last_workfile,
        )

    def _cli_launch_with_workfile_id(
        self,
        project: str,
        workfile_id: str,
        app: str,
    ) -> None:
        from ayon_core.pipeline import Anatomy

        workfile_entity = ayon_api.get_workfile_info_by_id(
            project, workfile_id
        )
        task_id = workfile_entity["taskId"]
        task_entity = ayon_api.get_task_by_id(
            project, task_id, fields={"name", "folderId"}
        )
        folder_entity = ayon_api.get_folder_by_id(
            project, task_entity["folderId"], fields={"path"}
        )
        anatomy = Anatomy(project)
        workfile_path = anatomy.fill_root(workfile_entity["path"])
        self.launch_application(
            app,
            project,
            folder_entity["path"],
            task_entity["name"],
            workfile_path=workfile_path,
        )

    def _cli_launch_with_debug_terminal(
        self,
        project: str,
        task_id: str,
        app: Optional[str],
    ) -> None:
        from .ui.debug_terminal_launch import run_with_debug_terminal

        run_with_debug_terminal(self, project, task_id, app)

    def _show_launch_error_dialog(self, message: str, detail: str) -> None:
        script_path = os.path.join(
            APPLICATIONS_ADDON_ROOT, "ui", "launch_failed_dialog.py"
        )
        with tempfile.NamedTemporaryFile("w", delete=False) as tmp:
            tmp_path = tmp.name
            json.dump(
                {"message": message, "detail": detail},
                tmp.file
            )

        try:
            run_ayon_launcher_process(
                "--skip-bootstrap",
                script_path,
                tmp_path,
                add_sys_paths=True,
                creationflags=0,
            )

        finally:
            os.remove(tmp_path)