chatbot_input_util library file

"""Multiline input utilities for chatbot-style console programs.

The module exposes :class:`MultilineInput`, a standalone stateful input
collector that returns complete multiline messages while still allowing
single-word slash commands to interrupt the draft. Interrupted drafts are
retained by the session so callers can run an application command and then
return the user to the message they were composing.

The module also exposes :func:`input_multiline`, a stateless, input-like helper.
It creates an internal session for one completed input operation, disables
mid-input application command interruption, and discards the session when the
operation finishes.

Supported runtime:

- Python 3.10 or newer. The module uses built-in generic annotations and PEP
  604 unions without postponed annotation evaluation.
- prompt_toolkit is optional. When installed, the supported floor is
  prompt_toolkit 3.0.0. The integration uses PromptSession, KeyBindings,
  multiline prompts, default text, application exit results, current buffer
  access, and document cursor properties that exist in the 3.0 API.
- keyboard is optional. It is used as a best-effort modified-Enter
  detector where the active frontend cannot expose that key combination
  directly; when unavailable or unsupported, the slash send marker remains the
  portable send path.

The input session handles its own editor command, `/back`. Applications using
this class do not need to implement that command; they can simply document it
if it is useful for users. On a terminal, /back reprints draft input, and may
return caret to end-of-line.
"""

__all__ = ["MultilineInput", "input_multiline"]


class MultilineInput:
    """docstring"""
    class _PromptToolkitFrontendUnavailable(Exception):
        """Internal signal to retry the prompt without prompt_toolkit."""

    SEND_MARKER = "/"
    BACK_COMMAND = "/back"
    DEFAULT_PROMPT = "-- your message? --"
    PROMPT_SEND_HINT = "{send_marker!r} alone on a line to send."
    PROMPT_MODIFIED_ENTER_HINT = (
        "{modified_enter} to send, or {send_marker!r} alone on a line."
    )
    PROMPT_ADDITIONAL_HELP: tuple[str, ...] = ("/help for commands.",)
    MODIFIED_ENTER_BRIDGE_SECONDS = 0.35

    def __init__(self, prompt_message: str | None = None) -> None:
        self.prompt_message = (
            self.DEFAULT_PROMPT if prompt_message is None else prompt_message
        )
        self.help_enabled = True
        self.sending_help = True
        self.command_interrupts_enabled = True
        self.transcript_echo_enabled = True
        self._draft = ""
        self._keyboard_loaded = False
        self._keyboard: object | None = None
        self._hotkey_support_cache: dict[str, bool] = {}
        self._prompt_toolkit_available_cache: bool | None = None
        self._prompt_toolkit_modified_enter_cache: tuple[str, str] | None = None
        self._prompt_toolkit_modified_enter_checked = False
        self._prompt_toolkit_runtime_disabled = False
        # Internal debugging handle:
        # force prompt_toolkit even in IDLE / non-TTY / unsupported TERM contexts.
        self._FORCE_PTK = False

    @staticmethod
    def is_slash_command(line: str) -> bool:
        """
        Return True when a console line is exactly one slash command token.
        """
        value = line.rstrip()
        return (
            value.startswith("/")
            and not value.startswith("//")
            and len(value) > 1
            and not any(ch.isspace() for ch in value)
        )

    @classmethod
    def _is_slash_command(cls, line: str) -> bool:
        return cls.is_slash_command(line)

    @classmethod
    def _is_back_command(cls, line: str) -> bool:
        return (
            cls._is_slash_command(line)
            and line.rstrip().lower() == cls.BACK_COMMAND
        )

    @classmethod
    def _is_send_marker(cls, line: str) -> bool:
        return line.rstrip() == cls.SEND_MARKER

    @staticmethod
    def _postprocess(raw: str) -> str:
        lines = raw.splitlines()
        for i, line in enumerate(lines):
            if line.startswith("//"):
                lines[i] = "/" + line[2:]
        return "\n".join(lines)

    def _finalize_message(self, raw: str, *, strip: bool) -> str:
        self._draft = ""
        out = self._postprocess(raw)
        return out.strip() if strip else out

    @staticmethod
    def _modified_enter_candidates() -> tuple[tuple[str, str, str], ...]:
        """
        Return hotkey name, modifier name, and display label candidates.
        """
        import sys

        if sys.platform == "darwin":
            return (
                ("command+enter", "command", "Command+Enter"),
                ("cmd+enter", "cmd", "Command+Enter"),
                ("ctrl+enter", "ctrl", "Ctrl+Enter"),
            )

        return (("ctrl+enter", "ctrl", "Ctrl+Enter"),)

    @staticmethod
    def _is_keyboard_backend_error(exc: Exception) -> bool:
        error_types: tuple[type[BaseException], ...] = (
            AttributeError,
            ImportError,
            KeyError,
            RuntimeError,
            OSError,
            ValueError,
        )

        try:
            from subprocess import CalledProcessError
        except ImportError:
            return isinstance(exc, error_types)

        return isinstance(exc, error_types + (CalledProcessError,))

    def _load_keyboard(self) -> object | None:
        if self._keyboard_loaded:
            return self._keyboard

        self._keyboard_loaded = True
        try:
            import keyboard  # type: ignore  # pylint: disable=C0415
        except ImportError as exc:
            if exc.name and exc.name.startswith("keyboard"):
                return None
            raise

        self._keyboard = keyboard
        return self._keyboard

    def _keyboard_hotkey_supported(self, hotkey_name: str) -> bool:
        cached = self._hotkey_support_cache.get(hotkey_name)
        if cached is not None:
            return cached

        keyboard = self._load_keyboard()
        if keyboard is None:
            self._hotkey_support_cache[hotkey_name] = False
            return False

        try:
            hotkey = keyboard.add_hotkey(  # type: ignore[attr-defined]
                hotkey_name,
                lambda: None,
                suppress=False,
            )
        except Exception as exc:
            if not self._is_keyboard_backend_error(exc):
                raise
            self._hotkey_support_cache[hotkey_name] = False
            return False

        try:
            keyboard.remove_hotkey(hotkey)  # type: ignore[attr-defined]
        except Exception as exc:
            if not self._is_keyboard_backend_error(exc):
                raise

        self._hotkey_support_cache[hotkey_name] = True
        return True

    def _supported_modified_enter_candidate(
        self,
    ) -> tuple[str, str, str] | None:
        for candidate in self._modified_enter_candidates():
            hotkey_name, _, _ = candidate
            if self._keyboard_hotkey_supported(hotkey_name):
                return candidate
        return None

    def _add_modified_enter_hotkey(
        self,
        callback: object,
    ) -> tuple[object | None, list[object], tuple[str, str, str] | None]:
        keyboard = self._load_keyboard()
        if keyboard is None:
            return None, [], None

        for candidate in self._modified_enter_candidates():
            hotkey_name, _, _ = candidate
            if self._hotkey_support_cache.get(hotkey_name) is False:
                continue

            try:
                hotkey = keyboard.add_hotkey(  # type: ignore[attr-defined]
                    hotkey_name,
                    callback,
                    suppress=False,
                )
            except Exception as exc:
                if not self._is_keyboard_backend_error(exc):
                    raise
                self._hotkey_support_cache[hotkey_name] = False
                continue

            self._hotkey_support_cache[hotkey_name] = True
            return keyboard, [hotkey], candidate

        return keyboard, [], None

    @staticmethod
    def _remove_keyboard_hotkeys(
        keyboard: object | None,
        hotkeys: list[object],
    ) -> None:
        if keyboard is None:
            return

        for hotkey in hotkeys:
            try:
                keyboard.remove_hotkey(hotkey)  # type: ignore[attr-defined]
            except Exception as exc:
                if not MultilineInput._is_keyboard_backend_error(exc):
                    raise

    @staticmethod
    def _modified_enter_pressed(
        keyboard: object | None,
        candidate: tuple[str, str, str] | None,
    ) -> bool:
        if keyboard is None or candidate is None:
            return False

        _, modifier_name, _ = candidate
        try:
            return bool(keyboard.is_pressed(modifier_name))  # type: ignore[attr-defined]
        except Exception as exc:
            if not MultilineInput._is_keyboard_backend_error(exc):
                raise
            return False

    def _modified_enter_prompt_label(self, *, prompt_toolkit_mode: bool) -> str | None:
        if prompt_toolkit_mode:
            ptk_candidate = self._prompt_toolkit_modified_enter_candidate()
            if ptk_candidate is not None:
                _, label = ptk_candidate
                return label

        candidate = self._supported_modified_enter_candidate()
        if candidate is None:
            return None

        _, _, label = candidate
        return label

    @staticmethod
    def _prompt_toolkit_modified_enter_candidates() -> tuple[tuple[str, str], ...]:
        """
        Return prompt_toolkit key names that may represent modified Enter.

        Plain Enter is Control-M in many terminals, so this deliberately avoids
        aliases such as "c-m" and "c-j"; binding those would make ordinary
        Enter send the message. A two-key Escape-then-Enter sequence is
        intentionally not a candidate because the send chord is Ctrl+Enter.
        """
        return (("c-enter", "Ctrl+Enter"),)

    def _prompt_toolkit_modified_enter_candidate(self) -> tuple[str, str] | None:
        if self._prompt_toolkit_modified_enter_checked:
            return self._prompt_toolkit_modified_enter_cache

        try:
            from prompt_toolkit.key_binding import KeyBindings  # pylint: disable=C0415
        except ImportError as exc:
            if exc.name and exc.name.startswith("prompt_toolkit"):
                self._prompt_toolkit_modified_enter_checked = True
                return None
            raise

        for key_name, label in self._prompt_toolkit_modified_enter_candidates():
            bindings = KeyBindings()
            try:
                bindings.add(key_name)(lambda event: None)
            except ValueError:
                continue

            self._prompt_toolkit_modified_enter_cache = (key_name, label)
            self._prompt_toolkit_modified_enter_checked = True
            return key_name, label

        self._prompt_toolkit_modified_enter_checked = True
        return None

    def _format_prompt_fragment(
        self,
        template: str,
        *,
        modified_enter: str | None,
    ) -> str:
        return template.format(
            send_marker=self.SEND_MARKER,
            back_command=self.BACK_COMMAND,
            modified_enter=modified_enter or "",
        ).strip()

    def _prompt_additional_help(self) -> tuple[str, ...]:
        additional_help = self.PROMPT_ADDITIONAL_HELP
        if isinstance(additional_help, str):
            return (additional_help,)
        return tuple(additional_help)

    def _store_draft_and_return_command(
        self,
        draft_lines: list[str],
        command_line: str,
        *,
        strip: bool,
    ) -> str:
        # Keep the draft in its raw input form. Escaped slash lines such as
        # "//help" must remain escaped until the user finally sends the message;
        # otherwise a restored literal slash line could be mistaken for a command.
        self._draft = "\n".join(draft_lines)
        return command_line.strip() if strip else command_line

    def _should_return_command(
        self,
        command_line: str,
        draft_lines: list[str],
    ) -> bool:
        if not self._is_slash_command(command_line):
            return False

        if self._is_back_command(command_line):
            return True

        return self.command_interrupts_enabled or not draft_lines

    def _split_latest_command(
        self,
        raw: str,
        *,
        strip: bool,
    ) -> str | None:
        """
        Return a latest-line command and preserve prior text as draft.
        """
        lines = raw.splitlines()
        if not lines:
            return None

        command_line = lines[-1]
        draft_lines = lines[:-1]
        if not self._should_return_command(command_line, draft_lines):
            return None

        return self._store_draft_and_return_command(
            draft_lines,
            command_line,
            strip=strip,
        )

    def _format_prompt(self, *, prompt_toolkit_mode: bool) -> str:
        modified_enter_label = (
            self._modified_enter_prompt_label(
                prompt_toolkit_mode=prompt_toolkit_mode,
            )
            if self.sending_help
            else None
        )
        prompt_parts: list[str] = []

        base_prompt = self.prompt_message.strip()
        if base_prompt:
            prompt_parts.append(base_prompt)

        if self.sending_help:
            if modified_enter_label is None:
                send_help = self._format_prompt_fragment(
                    self.PROMPT_SEND_HINT,
                    modified_enter=None,
                )
            else:
                send_help = self._format_prompt_fragment(
                    self.PROMPT_MODIFIED_ENTER_HINT,
                    modified_enter=modified_enter_label,
                )
            if send_help:
                prompt_parts.append(send_help)

        if self.help_enabled:
            for additional_help in self._prompt_additional_help():
                help_text = self._format_prompt_fragment(
                    additional_help,
                    modified_enter=modified_enter_label,
                )
                if help_text:
                    prompt_parts.append(help_text)

        return " ".join(prompt_parts)

    def _print_prompt(self, *, prompt_toolkit_mode: bool) -> None:
        prompt = self._format_prompt(prompt_toolkit_mode=prompt_toolkit_mode)
        if prompt:
            print(prompt)

        if not self._draft:
            return

        # prompt_toolkit renders the restored draft in the editable input
        # buffer. Plain input has no editable buffer, so echo the retained
        # draft directly after the normal prompt header.
        if not prompt_toolkit_mode:
            print(self._draft)

    def _print_prompt_toolkit_transcript(self, text: str) -> None:
        if not self.transcript_echo_enabled or not text:
            return

        import sys

        sys.stdout.write(text)
        if not text.endswith("\n"):
            sys.stdout.write("\n")
        sys.stdout.flush()

    @staticmethod
    def _terminal_supports_prompt_toolkit() -> bool:
        import os
        import sys

        try:
            stdin_tty = sys.stdin.isatty()
            stdout_tty = sys.stdout.isatty()
        except (AttributeError, OSError, ValueError):
            return False

        in_idle = "idlelib" in sys.modules or (
            hasattr(sys.stdin, "__class__")
            and getattr(sys.stdin.__class__, "__module__", "").startswith("idlelib")
        )

        if os.name == "nt":
            term_ok = True
        else:
            term_ok = os.environ.get("TERM") not in (None, "", "dumb")

        return stdin_tty and stdout_tty and term_ok and not in_idle

    def _prompt_toolkit_available(self) -> bool:
        cached = self._prompt_toolkit_available_cache
        if cached is not None:
            return cached

        try:
            from prompt_toolkit import PromptSession  # pylint: disable=C0415
            from prompt_toolkit.key_binding import KeyBindings  # pylint: disable=C0415
        except ImportError as exc:
            if exc.name and exc.name.startswith("prompt_toolkit"):
                self._prompt_toolkit_available_cache = False
                return False
            raise

        self._prompt_toolkit_available_cache = bool(PromptSession and KeyBindings)
        return self._prompt_toolkit_available_cache

    @staticmethod
    def _prompt_toolkit_frontend_error_types() -> tuple[type[BaseException], ...]:
        """Return prompt_toolkit frontend failures that merit plain fallback."""
        import sys

        error_types: tuple[type[BaseException], ...] = (OSError, RuntimeError)
        if sys.platform != "win32":
            return error_types

        try:
            from prompt_toolkit.output.win32 import (  # pylint: disable=C0415
                NoConsoleScreenBufferError,
            )
        except ImportError as exc:
            if exc.name and exc.name.startswith("prompt_toolkit"):
                return error_types
            raise

        return error_types + (NoConsoleScreenBufferError,)

    def _collect_with_plain_input(self, *, strip: bool) -> str | None:
        # Line-based collector that works in IDLE and any non-TTY stdin environment.
        # Best-effort modified-Enter detection via optional `keyboard`.
        send_requested = False

        def _mark_send() -> None:
            nonlocal send_requested
            send_requested = True

        keyboard, hotkeys, active_candidate = self._add_modified_enter_hotkey(
            _mark_send,
        )

        def _mod_enter_send_requested() -> bool:
            nonlocal send_requested
            if send_requested:
                send_requested = False
                return True
            return self._modified_enter_pressed(keyboard, active_candidate)

        try:
            lines: list[str] = self._draft.splitlines() if self._draft else []

            try:
                first = input()
            except (EOFError, KeyboardInterrupt):
                return None

            if self._is_send_marker(first):
                return self._finalize_message("\n".join(lines), strip=strip)

            if self._should_return_command(first, lines):
                return self._store_draft_and_return_command(
                    lines,
                    first,
                    strip=strip,
                )

            lines.append(first)
            if _mod_enter_send_requested():
                return self._finalize_message("\n".join(lines), strip=strip)

            while True:
                try:
                    line = input()
                except EOFError:
                    break
                except KeyboardInterrupt:
                    return None

                if self._is_send_marker(line):
                    break

                if self._should_return_command(line, lines):
                    return self._store_draft_and_return_command(
                        lines,
                        line,
                        strip=strip,
                    )

                lines.append(line)
                if _mod_enter_send_requested():
                    break

            return self._finalize_message("\n".join(lines), strip=strip)
        finally:
            self._remove_keyboard_hotkeys(keyboard, hotkeys)

    def _collect_with_prompt_toolkit(self, *, strip: bool) -> str | None:
        # Real terminal: prompt_toolkit provides editable restored drafts.
        try:
            from prompt_toolkit import PromptSession  # pylint: disable=C0415
            from prompt_toolkit.key_binding import KeyBindings  # pylint: disable=C0415
        except ImportError as exc:
            if exc.name and exc.name.startswith("prompt_toolkit"):
                return self._collect_with_plain_input(strip=strip)
            raise

        keyboard: object | None = None
        hotkeys: list[object] = []
        ptk_frontend_error_types = self._prompt_toolkit_frontend_error_types()

        try:
            session = PromptSession(
                erase_when_done=self.transcript_echo_enabled,
            )
            kb = KeyBindings()
            ptk_modified_enter = self._prompt_toolkit_modified_enter_candidate()
            send_requested_at: float | None = None

            def _send_current_buffer(event):
                event.app.exit(result=event.app.current_buffer.text)

            def _mark_send() -> None:
                nonlocal send_requested_at
                from time import monotonic

                send_requested_at = monotonic()

            if ptk_modified_enter is None:
                keyboard, hotkeys, _ = self._add_modified_enter_hotkey(_mark_send)

            def _keyboard_send_requested() -> bool:
                nonlocal send_requested_at
                if send_requested_at is None:
                    return False

                requested_at = send_requested_at
                send_requested_at = None

                from time import monotonic

                return (
                    monotonic() - requested_at
                    <= self.MODIFIED_ENTER_BRIDGE_SECONDS
                )

            @kb.add("c-d")
            def _ctrl_d(event):  # type: ignore[no-redef]
                _send_current_buffer(event)

            if ptk_modified_enter is not None:
                ptk_modified_enter_key, _ = ptk_modified_enter

                @kb.add(ptk_modified_enter_key)
                def _modified_enter(event):  # type: ignore[no-redef]
                    _send_current_buffer(event)

            @kb.add("enter")
            def _enter(event):  # type: ignore[no-redef]
                buf = event.app.current_buffer
                doc = buf.document
                text = buf.text

                if _keyboard_send_requested():
                    event.app.exit(result=text)
                    return

                lines = text.splitlines()
                row = doc.cursor_position_row
                col = doc.cursor_position_col
                if lines and row == (len(lines) - 1):
                    current_line = doc.current_line
                    if (
                        self._is_send_marker(current_line)
                        and col == len(current_line)
                    ):
                        event.app.exit(result="\n".join(lines[:-1]))
                        return
                    if (
                        self._should_return_command(
                            current_line,
                            lines[:-1],
                        )
                        and col == len(current_line)
                    ):
                        event.app.exit(result=text)
                        return

                buf.insert_text("\n")

            try:
                raw = session.prompt(
                    "",
                    multiline=True,
                    key_bindings=kb,
                    default=self._draft,
                )
            except (EOFError, KeyboardInterrupt):
                return None

            command = self._split_latest_command(raw, strip=strip)
            if command is not None:
                if not self._is_back_command(command):
                    self._print_prompt_toolkit_transcript(command)
                return command

            message = self._finalize_message(raw, strip=strip)
            self._print_prompt_toolkit_transcript(message)
            return message
        except ptk_frontend_error_types as exc:
            if self._FORCE_PTK:
                raise
            self._prompt_toolkit_runtime_disabled = True
            raise self._PromptToolkitFrontendUnavailable() from exc
        finally:
            self._remove_keyboard_hotkeys(keyboard, hotkeys)

    def get_input(self, strip: bool = True) -> str | None:
        while True:
            use_ptk = (
                (
                    self._FORCE_PTK
                    or (
                        not self._prompt_toolkit_runtime_disabled
                        and self._terminal_supports_prompt_toolkit()
                    )
                )
                and self._prompt_toolkit_available()
            )
            self._print_prompt(prompt_toolkit_mode=use_ptk)
            try:
                if use_ptk:
                    result = self._collect_with_prompt_toolkit(strip=strip)
                else:
                    result = self._collect_with_plain_input(strip=strip)
            except self._PromptToolkitFrontendUnavailable:
                continue

            if result is None:
                return None

            if self._is_back_command(result):
                continue

            return result


def input_multiline(
    prompt: object = None,
    /,
    *,
    help: bool = True,  # pylint: disable=redefined-builtin
    sending_help: bool = True,
) -> str | None:
    """
    Return one complete multiline input value using an input-like API.

    The helper is stateless across calls. It creates a fresh
    MultilineInput, applies the prompt and guidance settings, disables
    mid-input application slash-command interruption, and discards the session
    after the input operation completes. A single-word slash command entered as
    the first line is still returned immediately. The built-in /back editor
    command remains available inside the one input operation.
    """
    session = MultilineInput(
        prompt_message=None if prompt is None else str(prompt),
    )
    session.help_enabled = help
    session.sending_help = sending_help
    session.command_interrupts_enabled = False
    return session.get_input(strip=False)


def multi_input_demo() -> None:
    """
    Run a small receiver loop demonstrating MultilineInput.
    """
    message_input = MultilineInput(
        prompt_message=MultilineInput.DEFAULT_PROMPT,
    )
    message_input.help_enabled = True
    message_input.sending_help = True
    message_input.command_interrupts_enabled = True

    def print_help() -> None:
        print("Demo commands:")
        print("  /help  Show this help text.")
        print("  /exit  Stop the demo.")
        print("  /quit  Stop the demo.")
        print("  /back  Restore the current draft in the input editor.")
        print("Any other single-word slash command is reported as received.")

    def handle_command(command: str) -> bool:
        """
        Return True when the receiver loop should continue.
        """
        print(f"Received command {command}")
        if command.lower() == "/help":
            print_help()
            return True
        if command.lower() in ["/exit", "/quit"]:
            print("Exiting...")
            return False
        return True

    print("MultilineInput() demo")

    while True:
        value = message_input.get_input()
        if value is None:
            print("Input closed.")
            return

        command = value.strip()
        if MultilineInput.is_slash_command(command):
            if not handle_command(command):
                return
            continue

        line_count = len(value.splitlines()) if value else 0
        print(f"Received message, {line_count} lines.")


if __name__ == "__main__":
    multi_input_demo()