console.py

# this builtin is needed so we can overwrite in test
import asyncio
import json
import logging
import os

from typing import (
    Any,
    AsyncGenerator,
    Dict,
    List,
    Optional,
    Text,
    overload,
)

import aiohttp
import questionary
from aiohttp import ClientTimeout
from prompt_toolkit.styles import Style

import rasa.shared.utils.cli
import rasa.shared.utils.io
from rasa.cli import utils as cli_utils
from rasa.core import utils
from rasa.core.channels.rest import RestInput
from rasa.core.constants import DEFAULT_SERVER_URL, DEFAULT_STREAM_READING_TIMEOUT
from rasa.shared.constants import INTENT_MESSAGE_PREFIX
from rasa.shared.utils.io import DEFAULT_ENCODING

logger = logging.getLogger(__name__)

STREAM_READING_TIMEOUT_ENV = "RASA_SHELL_STREAM_READING_TIMEOUT_IN_SECONDS"


def print_buttons(
    message: Dict[Text, Any],
    is_latest_message: bool = False,
    color: Text = rasa.shared.utils.io.bcolors.OKBLUE,
) -> Optional[questionary.Question]:
    """Create CLI buttons from message data."""
    if is_latest_message:
        choices = cli_utils.button_choices_from_message_data(
            message, allow_free_text_input=True
        )
        question = questionary.select(
            message.get("text"),
            choices,
            style=Style([("qmark", "#6d91d3"), ("", "#6d91d3"), ("answer", "#b373d6")]),
        )
        return question
    else:
        rasa.shared.utils.cli.print_color("Buttons:", color=color)
        for idx, button in enumerate(message.get("buttons")):
            rasa.shared.utils.cli.print_color(
                cli_utils.button_to_string(button, idx), color=color
            )
        return None


def _print_bot_output(
    message: Dict[Text, Any],
    is_latest_message: bool = False,
    color: Text = rasa.shared.utils.io.bcolors.OKBLUE,
) -> Optional[questionary.Question]:
    if "buttons" in message:
        question = print_buttons(message, is_latest_message, color)
        if question:
            return question

    if "text" in message:
        rasa.shared.utils.cli.print_color(message["text"], color=color)

    if "image" in message:
        rasa.shared.utils.cli.print_color("Image: " + message["image"], color=color)

    if "attachment" in message:
        rasa.shared.utils.cli.print_color(
            "Attachment: " + message["attachment"], color=color
        )

    if "elements" in message:
        rasa.shared.utils.cli.print_color("Elements:", color=color)
        for idx, element in enumerate(message["elements"]):
            rasa.shared.utils.cli.print_color(
                cli_utils.element_to_string(element, idx), color=color
            )

    if "quick_replies" in message:
        rasa.shared.utils.cli.print_color("Quick Replies:", color=color)
        for idx, element in enumerate(message["quick_replies"]):
            rasa.shared.utils.cli.print_color(
                cli_utils.button_to_string(element, idx), color=color
            )

    if "custom" in message:
        rasa.shared.utils.cli.print_color("Custom json:", color=color)
        rasa.shared.utils.cli.print_color(
            rasa.shared.utils.io.json_to_string(message["custom"]), color=color
        )

    return None


@overload
async def _get_user_input(previous_response: None) -> Text:
    ...


@overload
async def _get_user_input(previous_response: Dict[str, Any]) -> Optional[Text]:
    ...


async def _get_user_input(
    previous_response: Optional[Dict[str, Any]]
) -> Optional[Text]:
    button_response = None
    if previous_response is not None:
        button_response = _print_bot_output(previous_response, is_latest_message=True)

    if button_response is not None:
        response = await cli_utils.payload_from_button_question(button_response)
        if response == cli_utils.FREE_TEXT_INPUT_PROMPT:
            # Re-prompt user with a free text input
            response = await _get_user_input(None)
    else:
        question = questionary.text(
            "",
            qmark="Your input ->",
            style=Style([("qmark", "#b373d6"), ("", "#b373d6")]),
        )
        response = await question.ask_async()
    return response.strip() if response is not None else None


async def send_message_receive_block(
    server_url: Text, auth_token: Text, sender_id: Text, message: Text
) -> List[Dict[Text, Any]]:
    """Posts message and returns response."""
    payload = {"sender": sender_id, "message": message}

    url = f"{server_url}/webhooks/rest/webhook?token={auth_token}"
    async with aiohttp.ClientSession() as session:
        async with session.post(url, json=payload, raise_for_status=True) as resp:
            return await resp.json()


async def _send_message_receive_stream(
    server_url: Text,
    auth_token: Text,
    sender_id: Text,
    message: Text,
    request_timeout: Optional[int] = None,
) -> AsyncGenerator[Dict[Text, Any], None]:
    payload = {"sender": sender_id, "message": message}

    url = f"{server_url}/webhooks/rest/webhook?stream=true&token={auth_token}"

    # Define timeout to not keep reading in case the server crashed in between
    timeout = _get_stream_reading_timeout(request_timeout)

    async with aiohttp.ClientSession(timeout=timeout) as session:
        async with session.post(url, json=payload, raise_for_status=True) as resp:

            async for line in resp.content:
                if line:
                    yield json.loads(line.decode(DEFAULT_ENCODING))


def _get_stream_reading_timeout(request_timeout: Optional[int] = None) -> ClientTimeout:
    """Define the ClientTimeout with fallbacks.

    First use the `request_timeout` function parameter if available, this comes from the
    `--request-timeout` command line argument.
    If that fails fallback `STREAM_READING_TIMEOUT_ENV`, a commandline argument.
    Lastly fallback to `DEFAULT_STREAM_READING_TIMEOUT` from rasa.core.constants
    """
    timeout_str = (
        request_timeout
        if request_timeout is not None
        else os.environ.get(STREAM_READING_TIMEOUT_ENV, DEFAULT_STREAM_READING_TIMEOUT)
    )

    return ClientTimeout(int(timeout_str))


async def record_messages(
    sender_id: Text,
    server_url: Text = DEFAULT_SERVER_URL,
    auth_token: Text = "",
    max_message_limit: Optional[int] = None,
    use_response_stream: bool = True,
    request_timeout: Optional[int] = None,
) -> int:
    """Read messages from the command line and print bot responses."""
    exit_text = INTENT_MESSAGE_PREFIX + "stop"

    rasa.shared.utils.cli.print_success(
        "Bot loaded. Type a message and press enter "
        "(use '{}' to exit): ".format(exit_text)
    )

    num_messages = 0
    previous_response = None
    await asyncio.sleep(0.5)  # Wait for server to start
    while not utils.is_limit_reached(num_messages, max_message_limit):
        text = await _get_user_input(previous_response)

        if text == exit_text or text is None:
            break

        if use_response_stream:
            bot_responses_stream = _send_message_receive_stream(
                server_url, auth_token, sender_id, text, request_timeout=request_timeout
            )
            previous_response = None
            async for response in bot_responses_stream:
                if previous_response is not None:
                    _print_bot_output(previous_response)
                previous_response = response
        else:
            bot_responses = await send_message_receive_block(
                server_url, auth_token, sender_id, text
            )
            previous_response = None
            for response in bot_responses:
                if previous_response is not None:
                    _print_bot_output(previous_response)
                previous_response = response

        num_messages += 1
        await asyncio.sleep(0)  # Yield event loop for others coroutines
    return num_messages


class CmdlineInput(RestInput):
    @classmethod
    def name(cls) -> Text:
        return "cmdline"

    def url_prefix(self) -> Text:
        return RestInput.name()