"""A terminals extension app."""
import os
import shlex
import sys
from shutil import which

from jupyter_core.utils import ensure_async
from jupyter_server.extension.application import ExtensionApp
from jupyter_server.transutils import trans
from traitlets import Type

from . import api_handlers, handlers
from .terminalmanager import TerminalManager


class TerminalsExtensionApp(ExtensionApp):
    """A terminals extension app."""

    name = "jupyter_server_terminals"

    terminal_manager_class = Type(
        default_value=TerminalManager, help="The terminal manager class to use."
    ).tag(config=True)

    # Since use of terminals is also a function of whether the terminado package is
    # available, this variable holds the "final indication" of whether terminal functionality
    # should be considered (particularly during shutdown/cleanup).  It is enabled only
    # once both the terminals "service" can be initialized and terminals_enabled is True.
    # Note: this variable is slightly different from 'terminals_available' in the web settings
    # in that this variable *could* remain false if terminado is available, yet the terminal
    # service's initialization still fails.  As a result, this variable holds the truth.
    terminals_available = False

    def initialize_settings(self):
        """Initialize settings."""
        self.initialize_configurables()
        self.settings.update(
            {"terminals_available": True, "terminal_manager": self.terminal_manager}
        )

    def initialize_configurables(self):
        """Initialize configurables."""
        if os.name == "nt":
            default_shell = "powershell.exe"
        else:
            default_shell = which("sh")  # type:ignore[assignment]
        shell_override = self.serverapp.terminado_settings.get("shell_command")
        if isinstance(shell_override, str):
            shell_override = shlex.split(shell_override)
        shell = (
            [os.environ.get("SHELL") or default_shell] if shell_override is None else shell_override
        )
        # When the notebook server is not running in a terminal (e.g. when
        # it's launched by a JupyterHub spawner), it's likely that the user
        # environment hasn't been fully set up. In that case, run a login
        # shell to automatically source /etc/profile and the like, unless
        # the user has specifically set a preferred shell command.
        if os.name != "nt" and shell_override is None and not sys.stdout.isatty():
            shell.append("-l")

        self.terminal_manager = self.terminal_manager_class(
            shell_command=shell,
            extra_env={
                "JUPYTER_SERVER_ROOT": self.serverapp.root_dir,
                "JUPYTER_SERVER_URL": self.serverapp.connection_url,
            },
            parent=self.serverapp,
        )
        self.terminal_manager.log = self.serverapp.log

    def initialize_handlers(self):
        """Initialize handlers."""
        self.handlers.append(
            (
                r"/terminals/websocket/(\w+)",
                handlers.TermSocket,
                {"term_manager": self.terminal_manager},
            )
        )
        self.handlers.extend(api_handlers.default_handlers)
        self.serverapp.web_app.settings["terminal_manager"] = self.terminal_manager
        self.serverapp.web_app.settings["terminals_available"] = self.settings[
            "terminals_available"
        ]

    def current_activity(self):
        """Get current activity info."""
        if self.terminals_available:
            terminals = self.terminal_manager.terminals
            if terminals:
                return terminals

    async def cleanup_terminals(self):
        """Shutdown all terminals.

        The terminals will shutdown themselves when this process no longer exists,
        but explicit shutdown allows the TerminalManager to cleanup.
        """
        if not self.terminals_available:
            return

        terminal_manager = self.terminal_manager
        n_terminals = len(terminal_manager.list())
        terminal_msg = trans.ngettext(
            "Shutting down %d terminal", "Shutting down %d terminals", n_terminals
        )
        self.log.info(terminal_msg % n_terminals)
        await ensure_async(terminal_manager.terminate_all())

    async def stop_extension(self):
        """Stop the extension."""
        await self.cleanup_terminals()