# Copyright (c) Microsoft Corporation. All rights reserved. # Licensed under the MIT License. from abc import abstractmethod import copy from typing import Dict, List from botbuilder.core.turn_context import TurnContext from botbuilder.schema import InputHints, ActivityTypes from botbuilder.dialogs.choices import ( Choice, ChoiceFactory, ChoiceFactoryOptions, ListStyle, ) from botbuilder.schema import Activity from .prompt_options import PromptOptions from .prompt_validator_context import PromptValidatorContext from ..dialog_reason import DialogReason from ..dialog import Dialog from ..dialog_instance import DialogInstance from ..dialog_turn_result import DialogTurnResult from ..dialog_context import DialogContext class Prompt(Dialog): """ Defines the core behavior of prompt dialogs. Extends the :class:`Dialog` base class. .. remarks:: When the prompt ends, it returns an object that represents the value it was prompted for. Use :meth:`DialogSet.add()` or :meth:`ComponentDialog.add_dialog()` to add a prompt to a dialog set or component dialog, respectively. Use :meth:`DialogContext.prompt()` or :meth:`DialogContext.begin_dialog()` to start the prompt. If you start a prompt from a :class:`WaterfallStep` in a :class:`WaterfallDialog`, then the prompt result will be available in the next step of the waterfall. """ ATTEMPT_COUNT_KEY = "AttemptCount" persisted_options = "options" persisted_state = "state" def __init__(self, dialog_id: str, validator: object = None): """ Creates a new :class:`Prompt` instance. :param dialog_id: Unique Id of the prompt within its parent :class:`DialogSet` :class:`ComponentDialog` :type dialog_id: str :param validator: Optionally provide additional validation and re-prompting logic :type validator: Object """ super(Prompt, self).__init__(dialog_id) self._validator = validator async def begin_dialog( self, dialog_context: DialogContext, options: object = None ) -> DialogTurnResult: """ Starts a prompt dialog. Called when a prompt dialog is pushed onto the dialog stack and is being activated. :param dialog_context: The dialog context for the current turn of the conversation :type dialog_context: :class:`DialogContext` :param options: Optional, additional information to pass to the prompt being started :type options: Object :return: The dialog turn result :rtype: :class:`DialogTurnResult` .. note:: The result indicates whether the prompt is still active after the turn has been processed. """ if not dialog_context: raise TypeError("Prompt(): dc cannot be None.") if not isinstance(options, PromptOptions): raise TypeError("Prompt(): Prompt options are required for Prompt dialogs.") # Ensure prompts have input hint set if options.prompt is not None and not options.prompt.input_hint: options.prompt.input_hint = InputHints.expecting_input if options.retry_prompt is not None and not options.retry_prompt.input_hint: options.retry_prompt.input_hint = InputHints.expecting_input # Initialize prompt state state = dialog_context.active_dialog.state state[self.persisted_options] = options state[self.persisted_state] = {} # Send initial prompt await self.on_prompt( dialog_context.context, state[self.persisted_state], state[self.persisted_options], False, ) return Dialog.end_of_turn async def continue_dialog(self, dialog_context: DialogContext): """ Continues a dialog. :param dialog_context: The dialog context for the current turn of the conversation :type dialog_context: :class:`DialogContext` :return: The dialog turn result :rtype: :class:`DialogTurnResult` .. remarks:: Called when a prompt dialog is the active dialog and the user replied with a new activity. If the task is successful, the result indicates whether the dialog is still active after the turn has been processed by the dialog. The prompt generally continues to receive the user's replies until it accepts the user's reply as valid input for the prompt. """ if not dialog_context: raise TypeError("Prompt(): dc cannot be None.") # Don't do anything for non-message activities if dialog_context.context.activity.type != ActivityTypes.message: return Dialog.end_of_turn # Perform base recognition instance = dialog_context.active_dialog state = instance.state[self.persisted_state] options = instance.state[self.persisted_options] recognized = await self.on_recognize(dialog_context.context, state, options) # Validate the return value is_valid = False if self._validator is not None: prompt_context = PromptValidatorContext( dialog_context.context, recognized, state, options ) is_valid = await self._validator(prompt_context) if options is None: options = PromptOptions() options.number_of_attempts += 1 else: if recognized.succeeded: is_valid = True # Return recognized value or re-prompt if is_valid: return await dialog_context.end_dialog(recognized.value) if not dialog_context.context.responded: await self.on_prompt(dialog_context.context, state, options, True) return Dialog.end_of_turn async def resume_dialog( self, dialog_context: DialogContext, reason: DialogReason, result: object ) -> DialogTurnResult: """ Resumes a dialog. :param dialog_context: The dialog context for the current turn of the conversation. :type dialog_context: :class:`DialogContext` :param reason: An enum indicating why the dialog resumed. :type reason: :class:`DialogReason` :param result: Optional, value returned from the previous dialog on the stack. :type result: object :return: The dialog turn result :rtype: :class:`DialogTurnResult` .. remarks:: Called when a prompt dialog resumes being the active dialog on the dialog stack, such as when the previous active dialog on the stack completes. If the task is successful, the result indicates whether the dialog is still active after the turn has been processed by the dialog. Prompts are typically leaf nodes on the stack but the dev is free to push other dialogs on top of the stack which will result in the prompt receiving an unexpected call to :meth:resume_dialog() when the pushed on dialog ends. Simply re-prompt the user to avoid that the prompt ends prematurely. """ await self.reprompt_dialog(dialog_context.context, dialog_context.active_dialog) return Dialog.end_of_turn async def reprompt_dialog(self, context: TurnContext, instance: DialogInstance): """ Reprompts user for input. :param context: Context for the current turn of conversation with the user :type context: :class:`botbuilder.core.TurnContext` :param instance: The instance of the dialog on the stack :type instance: :class:`DialogInstance` :return: A task representing the asynchronous operation """ state = instance.state[self.persisted_state] options = instance.state[self.persisted_options] await self.on_prompt(context, state, options, False) @abstractmethod async def on_prompt( self, turn_context: TurnContext, state: Dict[str, object], options: PromptOptions, is_retry: bool, ): """ Prompts user for input. When overridden in a derived class, prompts the user for input. :param turn_context: Context for the current turn of conversation with the user :type turn_context: :class:`botbuilder.core.TurnContext` :param state: Contains state for the current instance of the prompt on the dialog stack :type state: :class:`Dict` :param options: A prompt options object constructed from:meth:`DialogContext.prompt()` :type options: :class:`PromptOptions` :param is_retry: Determines whether `prompt` or `retry_prompt` should be used :type is_retry: bool :return: A task representing the asynchronous operation. """ @abstractmethod async def on_recognize( self, turn_context: TurnContext, state: Dict[str, object], options: PromptOptions, ): """ Recognizes the user's input. :param turn_context: Context for the current turn of conversation with the user :type turn_context: :class:`botbuilder.core.TurnContext` :param state: Contains state for the current instance of the prompt on the dialog stack :type state: :class:`Dict` :param options: A prompt options object constructed from :meth:`DialogContext.prompt()` :type options: :class:`PromptOptions` :return: A task representing the asynchronous operation. .. note:: When overridden in a derived class, attempts to recognize the user's input. """ def append_choices( self, prompt: Activity, channel_id: str, choices: List[Choice], style: ListStyle, options: ChoiceFactoryOptions = None, ) -> Activity: """ Composes an output activity containing a set of choices. :param prompt: The prompt to append the user's choice to :type prompt: :param channel_id: Id of the channel the prompt is being sent to :type channel_id: str :param: choices: List of choices to append :type choices: :class:`List` :param: style: Configured style for the list of choices :type style: :class:`ListStyle` :param: options: Optional formatting options to use when presenting the choices :type style: :class:`ChoiceFactoryOptions` :return: A task representing the asynchronous operation .. remarks:: If the task is successful, the result contains the updated activity. When overridden in a derived class, appends choices to the activity when the user is prompted for input. This is an helper function to compose an output activity containing a set of choices. """ # Get base prompt text (if any) text = prompt.text if prompt is not None and prompt.text else "" # Create temporary msg # TODO: fix once ChoiceFactory complete def inline() -> Activity: return ChoiceFactory.inline(choices, text, None, options) def list_style() -> Activity: return ChoiceFactory.list_style(choices, text, None, options) def suggested_action() -> Activity: return ChoiceFactory.suggested_action(choices, text) def hero_card() -> Activity: return ChoiceFactory.hero_card(choices, text) def list_style_none() -> Activity: activity = Activity(type=ActivityTypes.message) activity.text = text return activity def default() -> Activity: return ChoiceFactory.for_channel(channel_id, choices, text, None, options) # Maps to values in ListStyle Enum switcher = { 0: list_style_none, 1: default, 2: inline, 3: list_style, 4: suggested_action, 5: hero_card, } msg = switcher.get(int(style.value), default)() # Update prompt with text, actions and attachments if prompt: # clone the prompt the set in the options (note ActivityEx has Properties so this is the safest mechanism) prompt = copy.copy(prompt) prompt.text = msg.text if ( msg.suggested_actions is not None and msg.suggested_actions.actions is not None and msg.suggested_actions.actions ): prompt.suggested_actions = msg.suggested_actions if msg.attachments: if prompt.attachments: prompt.attachments.extend(msg.attachments) else: prompt.attachments = msg.attachments return prompt # TODO: Update to InputHints.ExpectingInput; msg.input_hint = None return msg