I cloned it and added some.
endpoints.yml
version: "2.0"
# This file contains the different endpoints your bot can use.
# Server where the models are pulled from.
# https://rasa.com/docs/rasa/user-guide/running-the-server/#fetching-models-from-a-server/
#models:
# url: http://my-server.com/models/default_core@latest
# wait_time_between_pulls: 10 # [optional](default: 100)
# Server which runs your custom actions.
# https://rasa.com/docs/rasa/core/actions/#custom-actions/
action_endpoint:
url: "http://localhost:5056/webhook/"
# action_endpoint:
# url: "http://app:5055/webhook"
# Tracker store which is used to store the conversations.
# By default the conversations are stored in memory.
# https://rasa.com/docs/rasa/api/tracker-stores/
#tracker_store:
# type: redis
# url: <host of the redis instance, e.g. localhost>
# port: <port of your redis instance, usually 6379>
# db: <number of your database within redis, e.g. 0>
# password: <password used for authentication>
# use_ssl: <whether or not the communication is encrypted, default false>
#tracker_store:
# type: mongod
# url: <url to your mongo instance, e.g. mongodb://localhost:27017>
# db: <name of the db within your mongo instance, e.g. rasa>
# username: <username used for authentication>
# password: <password used for authentication>
# Event broker which all conversation events should be streamed to.
# https://rasa.com/docs/rasa/api/event-brokers/
#event_broker:
# url: localhost
# username: username
# password: password
# queue: queue
stories.yml
version: "2.0"
stories:
- story: say ok when user affirms after utter_help
steps:
- action: utter_help
- intent: affirm
- action: utter_ok
- story: say ok when user affirms after action_show_balance
steps:
- action: action_show_balance
- intent: affirm
- action: utter_ok
- story: say ok when user affirms after action_show_recipients
steps:
- action: action_show_recipients
- intent: affirm
- action: utter_ok
- story: say ok when user affirms after action_show_transfer_charge
steps:
- action: action_show_transfer_charge
- intent: affirm
- action: utter_ok
- story: story_ask name and remember it
steps:
- intent: request_names
- action: name_form
- active_loop: name_form
- slot_was_set:
- requested_slot: first_name
- slot_was_set:
- first_name: vincent
- slot_was_set:
- requested_slot: last_name
- slot_was_set:
- last_name: vincent-mcvincent
- slot_was_set:
- requested_slot: null
- active_loop: null
- action: utter_submit
- action: utter_slots_values
rules.yml
version: "2.0"
rules:
- rule: Confirm that the bot is not a human
steps:
- intent: check_human
- action: utter_bot
- rule: Ask the user to rephrase whenever they send a message with low NLU confidence
steps:
- intent: nlu_fallback
- action: utter_default
- rule: answer out of scope
steps:
- intent: out_of_scope
- action: utter_out_of_scope
- rule: greet
steps:
- intent: greet
- action: utter_greet
- action: utter_help
- rule: say goodbye
steps:
- intent: goodbye
- action: utter_goodbye
- rule: help
steps:
- intent: help
- action: utter_help
- rule: thankyou
steps:
- intent: thankyou
- action: utter_noworries
- rule: is there a transfer charge
steps:
- intent: ask_transfer_charge
- action: action_show_transfer_charge
- rule: Show list of known recipients
steps:
- intent: check_recipients
- action: action_show_recipients
- rule: Show balance (bank account or credit card, based on account_type)
steps:
- intent: check_balance
- action: action_show_balance
- rule: Activate cc_payment_form when no other form is active
condition:
# this condition allows stories to handle form switching
- active_loop: null
steps:
- intent: pay_cc
- action: cc_payment_form
- active_loop: cc_payment_form
- rule: Activate transfer_money_form when no other form is active
condition:
# this condition allows stories to handle form switching
- active_loop: null
steps:
- intent: transfer_money
- action: transfer_money_form
- active_loop: transfer_money_form
- rule: Activate transaction_search_form when no other form is active
condition:
# this condition allows stories to handle form switching
- active_loop: null
steps:
- or:
- intent: search_transactions
- intent: check_earnings
- action: transaction_search_form
- active_loop: transaction_search_form
- rule: Submit cc_payment_form while not switched from previous form
condition:
- active_loop: cc_payment_form
- slot_was_set:
- previous_form_name: null
steps:
- action: cc_payment_form
- active_loop: null
- slot_was_set:
- requested_slot: null
- action: action_pay_cc
- rule: Submit transfer_money_form while not switched from previous form
condition:
- active_loop: transfer_money_form
- slot_was_set:
- previous_form_name: null
steps:
- action: transfer_money_form
- active_loop: null
- slot_was_set:
- requested_slot: null
- action: action_transfer_money
- rule: Submit transaction_search_form while not switched from previous form
condition:
- active_loop: transaction_search_form
- slot_was_set:
- previous_form_name: null
steps:
- action: transaction_search_form
- active_loop: null
- slot_was_set:
- requested_slot: null
- action: action_transaction_search
- rule: Show user money options
steps:
- intent: prompt
- action: utter_prompt
- rule: show weather
steps:
- intent: weather
- action: action_weather_api
- rule: Activate Form
steps:
- intent: request_names
- action: name_form
- active_loop: name_form
- rule: Submit Form
condition:
- active_loop: name_form
steps:
- action: name_form
- active_loop: null
- slot_was_set:
- requested_slot: null
- action: utter_submit
- action: utter_slots_values
domain.yml
version: '2.0'
session_config:
session_expiration_time: 0
carry_over_slots_to_new_session: true
intents:
- request_names:
use_entities: []
- nlu_fallback
- check_human
- transfer_money:
use_entities: []
- inform
- pay_cc:
use_entities: []
- greet
- goodbye
- affirm
- deny
- thankyou
- ask_transfer_charge
- search_transactions:
use_entities: []
- check_balance:
use_entities:
- account_type
- credit_card
- check_earnings:
use_entities: []
- check_recipients
- out_of_scope
- session_start
- restart
- trigger_handoff
- handoff
- human_handoff
- help
- prompt:
use_entities: []
- weather:
use_entities: []
entities:
- PERSON
- account_type
- amount-of-money
- credit_card
- handoff_to
- number
- payment_date
- search_type
- time
- vendor_name
slots:
first_name:
type: text
influence_conversation: true
last_name:
type: text
influence_conversation: true
AA_CONTINUE_FORM:
type: any
influence_conversation: false
PERSON:
type: any
influence_conversation: false
account_type:
type: any
influence_conversation: false
amount-of-money:
type: any
influence_conversation: false
amount_transferred:
type: any
initial_value: 0
influence_conversation: false
credit_card:
type: any
influence_conversation: false
currency:
type: any
initial_value: $
influence_conversation: false
end_time:
type: any
influence_conversation: false
end_time_formatted:
type: any
influence_conversation: false
grain:
type: any
influence_conversation: false
handoff_to:
type: any
influence_conversation: false
next_form_name:
type: text
influence_conversation: true
number:
type: any
influence_conversation: false
payment_amount_type:
type: any
initial_value: ''
influence_conversation: false
previous_form_name:
type: text
influence_conversation: true
repeated_validation_failures:
type: any
influence_conversation: false
requested_slot:
type: any
influence_conversation: false
search_type:
type: any
influence_conversation: false
start_time:
type: any
influence_conversation: false
start_time_formatted:
type: any
influence_conversation: false
time:
type: any
influence_conversation: false
time_formatted:
type: any
influence_conversation: false
vendor_name:
type: any
influence_conversation: false
zz_confirm_form:
type: any
influence_conversation: false
responses:
utter_out_of_scope:
- text: Sorry, I'm not sure how to respond to that. Type "help" for assistance.
utter_ask_transfer_money_form_amount-of-money:
- text: How much money do you want to transfer?
utter_ask_transfer_money_form_PERSON:
- text: Who do you want to transfer money to?
utter_goodbye:
- text: Bye
utter_noworries:
- text: You're welcome :)
utter_transfer_complete:
- text: Successfully transferred {currency}{amount-of-money} to {PERSON}.
utter_transfer_charge:
- text: You are entitled to six transfers within a statement cycle before being charged. For subsequent transfers you will be charged {currency}10 per transaction.
utter_ask_cc_payment_form_amount-of-money:
- text: How much do you want to pay?
utter_ask_cc_payment_form_credit_card:
- text: Towards which credit card account do you want to make a payment?
utter_ask_cc_payment_form_time:
- text: For which date would you like to schedule the payment?
utter_ask_transaction_search_form_vendor_name:
- text: For which vendor do you want to see transactions? e.g Starbucks, Target, Amazon
utter_ask_transaction_search_form_time:
- text: In which timeframe would you like to search for transactions?
utter_ask_transaction_search_form_search_type:
- buttons:
- payload: /inform{"search_type":"deposit"}'
title: Incoming (earnings)
- payload: /inform{"search_type":"spend"}'
title: Outgoing (spending)
text: Would you like to search incoming or outgoing transactions?
utter_no_payment_amount:
- text: Sorry, I don't understand that payment amount.
utter_no_paymentdate:
- text: Sorry, that is not a valid payment date.
utter_no_creditcard:
- text: Sorry, that is not a valid credit card account to make payments towards.
utter_no_vendor_name:
- text: Sorry, that's not a recognized vendor.
utter_no_transactdate:
- text: Sorry, that's not a recognized time frame.
utter_cc_pay_scheduled:
- text: Payment of {currency}{amount-of-money}{payment_amount_type} towards your {credit_card} account scheduled to be paid at {time_formatted}.
utter_searching_spend_transactions:
- text: Searching transactions{vendor_name} between {start_time_formatted} and {end_time_formatted}...
utter_found_spend_transactions:
- text: I found {numtransacts} transactions{vendor_name} totalling {currency}{total}.
utter_searching_deposit_transactions:
- text: Searching deposits made to your account between {start_time_formatted} and {end_time_formatted}...
utter_found_deposit_transactions:
- text: I found {numtransacts} deposits made to your account totalling {currency}{total}
utter_ask_rephrase:
- text: I didn't quite understand that. Can you rephrase?
utter_ok:
- text: 👍
utter_ask_continue:
- text: Would you like to continue?
utter_default:
- text: I didn't quite understand that. Could you rephrase?
utter_ask_cc_payment_form_AA_CONTINUE_FORM:
- buttons:
- payload: /affirm
title: Yes
- payload: /deny
title: No, cancel the transaction
text: Would you like to continue scheduling the credit card payment?
utter_ask_transfer_money_form_AA_CONTINUE_FORM:
- buttons:
- payload: /affirm
title: Yes
- payload: /deny
title: No, cancel the transfer
text: Would you like to continue scheduling the money transfer?
utter_ask_transaction_search_form_AA_CONTINUE_FORM:
- buttons:
- payload: /affirm
title: Yes
- payload: /deny
title: No, cancel the search
text: Would you like to continue the transaction search?
utter_ask_cc_payment_form_zz_confirm_form:
- buttons:
- payload: /affirm
title: Yes
- payload: /deny
title: No, cancel the transaction
text: Would you like to schedule a payment of {currency}{amount-of-money}{payment_amount_type} towards your {credit_card} account for {time_formatted}?
utter_ask_transfer_money_form_zz_confirm_form:
- buttons:
- payload: /affirm
title: Yes
- payload: /deny
title: No, cancel the transaction
text: Would you like to transfer {currency}{amount-of-money} to {PERSON}?
utter_cc_pay_cancelled:
- text: Credit card account payment cancelled.
utter_transfer_cancelled:
- text: Transfer cancelled.
utter_transaction_search_cancelled:
- text: Transaction search cancelled.
utter_account_balance:
- text: Your bank account balance is {currency}{init_account_balance}.
utter_changed_account_balance:
- text: Your bank account balance was {currency}{init_account_balance} and is now {currency}{account_balance} after transfers and payments.
utter_unknown_recipient:
- text: Sorry, {PERSON} is not in your list of known recipients.
utter_insufficient_funds:
- text: Sorry, you don't have enough money to do that!
utter_insufficient_funds_specific:
- text: The {payment_amount_type} on your {credit_card} credit card is {amount-of-money}, so you have insufficient funds to pay it off.
utter_credit_card_balance:
- text: The current balance for your {credit_card} account is {currency}{credit_card_balance}.
utter_nothing_due:
- text: Your don't owe any money on your {credit_card} credit card bill.
utter_recipients:
- text: These are your known recpients to whom you can send money:{formatted_recipients}
utter_greet:
- text: Hi! I'm your Financial Assistant!
utter_ask_handoff:
- text: It looks like you want to be transferred to a human agent.
utter_handoff:
- text: Alright, I'll try to transfer you.
utter_wouldve_handed_off:
- text: If you were talking to me via chatroom, I would have handed you off to {handoffhost}.
utter_no_handoff:
- text: Since you haven't configured a host to hand off to, I can't send you anywhere!
utter_ask_whatelse:
- text: What else can I help you with?
utter_bot:
- text: I'm a virtual assistant made with Rasa.
utter_help:
- text: |-
I can help you with your financial accounts.
You can ask me things like:
- What's my account balance?
- Pay off my credit card
- What did I spend at Target last month?
- I need to transfer money
utter_prompt:
- text: |-
- Savings
- Investing
- Insurance
- Buying a House
- Work Benefits
- Life Event
utter_weather:
- text: The temperature is {temp}.[Learn more about Rasa here.](www.rasa.com)
utter_ask_first_name:
- text: What is your first name?
utter_submit:
- text: Ok, Thanks!
utter_slots_values:
- text: I'll remember that your name is {first_name} {last_name}!
actions:
- action_ask_last_name
- action_ask_transaction_search_form_zz_confirm_form
- action_handoff
- action_handoff_options
- action_pay_cc
- action_restart
- action_session_start
- action_show_balance
- action_show_recipients
- action_show_transfer_charge
- action_switch_back_ask
- action_switch_forms_affirm
- action_switch_forms_ask
- action_switch_forms_deny
- action_transaction_search
- action_transfer_money
- action_weather_api
- utter_slots_values
- utter_submit
- validate_cc_payment_form
- validate_transaction_search_form
- validate_transfer_money_form
forms:
name_form:
required_slots:
first_name:
- type: from_text
last_name:
- type: from_text
cc_payment_form:
required_slots:
AA_CONTINUE_FORM:
- intent: affirm
type: from_intent
value: yes
- intent: deny
type: from_intent
value: no
- intent:
- inform
- cc_payment_form
type: from_text
amount-of-money:
- entity: amount-of-money
not_intent:
- check_balance
- check_earnings
type: from_entity
- entity: number
not_intent:
- check_balance
- check_earnings
type: from_entity
- intent:
- inform
- cc_payment_form
type: from_text
credit_card:
- entity: credit_card
type: from_entity
- intent:
- inform
- cc_payment_form
type: from_text
time:
- entity: time
type: from_entity
- intent:
- inform
- cc_payment_form
type: from_text
zz_confirm_form:
- intent: affirm
type: from_intent
value: yes
- intent: deny
type: from_intent
value: no
- intent:
- inform
- cc_payment_form
type: from_text
transfer_money_form:
required_slots:
AA_CONTINUE_FORM:
- intent: affirm
type: from_intent
value: yes
- intent: deny
type: from_intent
value: no
- intent:
- inform
- transfer_money_form
type: from_text
PERSON:
- entity: PERSON
type: from_entity
- intent:
- inform
- transfer_money_form
type: from_text
amount-of-money:
- entity: amount-of-money
not_intent:
- check_balance
- check_earnings
type: from_entity
- entity: number
not_intent:
- check_balance
- check_earnings
type: from_entity
- intent:
- inform
- transfer_money_form
type: from_text
zz_confirm_form:
- intent: affirm
type: from_intent
value: yes
- intent: deny
type: from_intent
value: no
- intent:
- inform
- transfer_money_form
type: from_text
transaction_search_form:
required_slots:
AA_CONTINUE_FORM:
- intent: affirm
type: from_intent
value: yes
- intent: deny
type: from_intent
value: no
- intent:
- inform
- transaction_search_form
type: from_text
search_type:
- intent: search_transactions
type: from_trigger_intent
value: spend
- intent: check_earnings
type: from_trigger_intent
value: deposit
- entity: search_type
type: from_entity
time:
- entity: time
type: from_entity
- intent:
- inform
- transaction_search_form
type: from_text
zz_confirm_form:
- intent: affirm
type: from_intent
value: yes
- intent: deny
type: from_intent
value: no
- intent:
- inform
- transaction_search_form
type: from_text
actions.py
"""Custom actions"""
import os
from typing import Dict, Text, Any, List
import logging
from dateutil import parser
import sqlalchemy as sa
import requests
# import pyodbc
from rasa_sdk.interfaces import Action
from rasa_sdk.events import (
SlotSet,
EventType,
ActionExecuted,
SessionStarted,
Restarted,
FollowupAction,
UserUtteranceReverted,
)
from rasa_sdk import Tracker
from rasa_sdk.executor import CollectingDispatcher
from actions.parsing import (
parse_duckling_time_as_interval,
parse_duckling_time,
get_entity_details,
parse_duckling_currency,
)
from actions.profile_db import create_database, ProfileDB
from actions.custom_forms import CustomFormValidationAction
logger = logging.getLogger(__name__)
# The profile database is created/connected to when the action server starts
# It is populated the first time `ActionSessionStart.run()` is called .
PROFILE_DB_NAME = os.environ.get("PROFILE_DB_NAME", "profile")
PROFILE_DB_URL = os.environ.get("PROFILE_DB_URL", f"sqlite:///{PROFILE_DB_NAME}.db")
ENGINE = sa.create_engine(PROFILE_DB_URL)
create_database(ENGINE, PROFILE_DB_NAME)
profile_db = ProfileDB(ENGINE)
# driver = 'ODBC Driver 17 for SQL Server'
# server = 'rasa-test.database.windows.net'
# database = 'rasa-test'
# username = 'wealthBuild'
# password = 'azureSql!23'
# connectionString = pyodbc.connect("DRIVER=" + driver
# + ";SERVER=" + server
# + ";DATABASE=" + database
# + ";UID=" + username
# + ";PWD=" + password)
# cursor = connectionString.cursor()
NEXT_FORM_NAME = {
"pay_cc": "cc_payment_form",
"transfer_money": "transfer_money_form",
"search_transactions": "transaction_search_form",
"check_earnings": "transaction_search_form",
}
FORM_DESCRIPTION = {
"cc_payment_form": "credit card payment",
"transfer_money_form": "money transfer",
"transaction_search_form": "transaction search",
}
class ActionPayCC(Action):
"""Pay credit card."""
def name(self) -> Text:
"""Unique identifier of the action"""
return "action_pay_cc"
async def run(
self,
dispatcher: CollectingDispatcher,
tracker: Tracker,
domain: Dict[Text, Any],
) -> List[Dict]:
"""Executes the action"""
slots = {
"AA_CONTINUE_FORM": None,
"zz_confirm_form": None,
"credit_card": None,
"account_type": None,
"amount-of-money": None,
"time": None,
"time_formatted": None,
"start_time": None,
"end_time": None,
"start_time_formatted": None,
"end_time_formatted": None,
"grain": None,
"number": None,
}
if tracker.get_slot("zz_confirm_form") == "yes":
credit_card = tracker.get_slot("credit_card")
amount_of_money = float(tracker.get_slot("amount-of-money"))
amount_transferred = float(tracker.get_slot("amount_transferred"))
profile_db.pay_off_credit_card(
tracker.sender_id, credit_card, amount_of_money
)
dispatcher.utter_message(response="utter_cc_pay_scheduled")
slots["amount_transferred"] = amount_transferred + amount_of_money
else:
dispatcher.utter_message(response="utter_cc_pay_cancelled")
return [SlotSet(slot, value) for slot, value in slots.items()]
class ValidatePayCCForm(CustomFormValidationAction):
"""Validates Slots of the cc_payment_form"""
def name(self) -> Text:
"""Unique identifier of the action"""
return "validate_cc_payment_form"
def amount_from_balance(
self, dispatcher, tracker, credit_card_name, balance_type
) -> Dict[Text, Any]:
amount_balance = profile_db.get_credit_card_balance(
tracker.sender_id, credit_card_name, balance_type
)
account_balance = profile_db.get_account_balance(tracker.sender_id)
if account_balance < float(amount_balance):
dispatcher.utter_message(response="utter_insufficient_funds")
return {"amount-of-money": None}
return {
"amount-of-money": f"{amount_balance:.2f}",
"payment_amount_type": f"(your {balance_type})",
}
async def validate_amount_of_money(
self,
value: Text,
dispatcher: CollectingDispatcher,
tracker: Tracker,
domain: Dict[Text, Any],
) -> Dict[Text, Any]:
"""Validates value of 'amount-of-money' slot"""
if not value:
return {"amount-of-money": None}
account_balance = profile_db.get_account_balance(tracker.sender_id)
# check if user asked to pay the full or the minimum balance
if type(value) is str:
credit_card_name = tracker.get_slot("credit_card")
if credit_card_name:
credit_card = profile_db.get_credit_card(
tracker.sender_id, credit_card_name
)
else:
credit_card = None
balance_types = profile_db.list_balance_types()
if value and value.lower() in balance_types:
balance_type = value.lower()
if not credit_card:
dispatcher.utter_message(
f"I see you'd like to pay the {balance_type}."
)
return {"amount-of-money": balance_type}
slots_to_set = self.amount_from_balance(
dispatcher, tracker, credit_card_name, balance_type
)
if float(slots_to_set.get("amount-of-money")) == 0:
dispatcher.utter_message(
response="utter_nothing_due", **slots_to_set
)
return {
"amount-of-money": None,
"credit_card": None,
"payment_amount_type": None,
}
return slots_to_set
try:
entity = get_entity_details(
tracker, "amount-of-money"
) or get_entity_details(tracker, "number")
amount_currency = parse_duckling_currency(entity)
if not amount_currency:
raise TypeError
if account_balance < float(amount_currency.get("amount-of-money")):
dispatcher.utter_message(response="utter_insufficient_funds")
return {"amount-of-money": None}
return amount_currency
except (TypeError, AttributeError):
pass
dispatcher.utter_message(response="utter_no_payment_amount")
return {"amount-of-money": None}
async def validate_credit_card(
self,
value: Text,
dispatcher: CollectingDispatcher,
tracker: Tracker,
domain: Dict[Text, Any],
) -> Dict[Text, Any]:
"""Validates value of 'credit_card' slot"""
if value and value.lower() in profile_db.list_credit_cards(tracker.sender_id):
amount = tracker.get_slot("amount-of-money")
credit_card_slot = {"credit_card": value.title()}
balance_types = profile_db.list_balance_types()
if amount and amount.lower() in balance_types:
updated_amount = self.amount_from_balance(
dispatcher, tracker, value.lower(), amount
)
if float(updated_amount.get("amount-of-money")) == 0:
dispatcher.utter_message(
response="utter_nothing_due", **updated_amount
)
return {
"amount-of-money": None,
"credit_card": None,
"payment_amount_type": None,
}
account_balance = profile_db.get_account_balance(tracker.sender_id)
if account_balance < float(updated_amount.get("amount-of-money")):
dispatcher.utter_message(
response="utter_insufficient_funds_specific", **updated_amount
)
return {"amount-of-money": None}
return {**credit_card_slot, **updated_amount}
return credit_card_slot
dispatcher.utter_message(response="utter_no_creditcard")
return {"credit_card": None}
async def explain_credit_card(
self,
value: Text,
dispatcher: CollectingDispatcher,
tracker: Tracker,
domain: Dict[Text, Any],
) -> Dict[Text, Any]:
"""Explains 'credit_card' slot"""
dispatcher.utter_message("You have the following credits cards:")
for credit_card in profile_db.list_credit_cards(tracker.sender_id):
current_balance = profile_db.get_credit_card_balance(
tracker.sender_id, credit_card
)
dispatcher.utter_message(
response="utter_credit_card_balance",
**{
"credit_card": credit_card.title(),
"amount-of-money": f"{current_balance:.2f}",
},
)
return {}
async def validate_time(
self,
value: Text,
dispatcher: CollectingDispatcher,
tracker: Tracker,
domain: Dict[Text, Any],
) -> Dict[Text, Any]:
"""Validates value of 'time' slot"""
timeentity = get_entity_details(tracker, "time")
parsedtime = timeentity and parse_duckling_time(timeentity)
if not parsedtime:
dispatcher.utter_message(response="utter_no_transactdate")
return {"time": None}
return parsedtime
async def validate_zz_confirm_form(
self,
value: Text,
dispatcher: CollectingDispatcher,
tracker: Tracker,
domain: Dict[Text, Any],
) -> Dict[Text, Any]:
"""Validates value of 'zz_confirm_form' slot"""
if value in ["yes", "no"]:
return {"zz_confirm_form": value}
return {"zz_confirm_form": None}
class ActionTransactionSearch(Action):
"""Searches for a transaction"""
def name(self) -> Text:
"""Unique identifier of the action"""
return "action_transaction_search"
async def run(
self,
dispatcher: CollectingDispatcher,
tracker: Tracker,
domain: Dict[Text, Any],
) -> List[Dict]:
"""Executes the action"""
slots = {
"AA_CONTINUE_FORM": None,
"zz_confirm_form": None,
"time": None,
"time_formatted": None,
"start_time": None,
"end_time": None,
"start_time_formatted": None,
"end_time_formatted": None,
"grain": None,
"search_type": None,
"vendor_name": None,
}
if tracker.get_slot("zz_confirm_form") == "yes":
search_type = tracker.get_slot("search_type")
deposit = search_type == "deposit"
vendor = tracker.get_slot("vendor_name")
vendor_name = f" at {vendor.title()}" if vendor else ""
start_time = parser.isoparse(tracker.get_slot("start_time"))
end_time = parser.isoparse(tracker.get_slot("end_time"))
transactions = profile_db.search_transactions(
tracker.sender_id,
start_time=start_time,
end_time=end_time,
deposit=deposit,
vendor=vendor,
)
aliased_transactions = transactions.subquery()
total = profile_db.session.query(
sa.func.sum(aliased_transactions.c.amount)
)[0][0]
if not total:
total = 0
numtransacts = transactions.count()
slotvars = {
"total": f"{total:.2f}",
"numtransacts": numtransacts,
"start_time_formatted": tracker.get_slot("start_time_formatted"),
"end_time_formatted": tracker.get_slot("end_time_formatted"),
"vendor_name": vendor_name,
}
dispatcher.utter_message(
response=f"utter_searching_{search_type}_transactions",
**slotvars,
)
dispatcher.utter_message(
response=f"utter_found_{search_type}_transactions", **slotvars
)
else:
dispatcher.utter_message(response="utter_transaction_search_cancelled")
return [SlotSet(slot, value) for slot, value in slots.items()]
class ValidateTransactionSearchForm(CustomFormValidationAction):
"""Validates Slots of the transaction_search_form"""
def name(self) -> Text:
"""Unique identifier of the form"""
return "validate_transaction_search_form"
async def run(
self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: Dict
) -> List[EventType]:
"""Custom validates the filled slots"""
events = await super().run(dispatcher, tracker, domain)
# For 'spend' type transactions we need to know the vendor_name
search_type = tracker.get_slot("search_type")
if search_type == "spend":
vendor_name = tracker.get_slot("vendor_name")
if not vendor_name:
events.append(SlotSet("requested_slot", "vendor_name"))
return events
async def validate_search_type(
self,
value: Text,
dispatcher: CollectingDispatcher,
tracker: Tracker,
domain: Dict[Text, Any],
) -> Dict[Text, Any]:
"""Validates value of 'search_type' slot"""
if value in ["spend", "deposit"]:
return {"search_type": value}
return {"search_type": None}
async def validate_vendor_name(
self,
value: Text,
dispatcher: CollectingDispatcher,
tracker: Tracker,
domain: Dict[Text, Any],
) -> Dict[Text, Any]:
"""Validates value of 'vendor_name' slot"""
if value and value.lower() in profile_db.list_vendors():
return {"vendor_name": value}
dispatcher.utter_message(response="utter_no_vendor_name")
return {"vendor_name": None}
async def validate_time(
self,
value: Text,
dispatcher: CollectingDispatcher,
tracker: Tracker,
domain: Dict[Text, Any],
) -> Dict[Text, Any]:
"""Validates value of 'time' slot"""
timeentity = get_entity_details(tracker, "time")
parsedinterval = timeentity and parse_duckling_time_as_interval(timeentity)
if not parsedinterval:
dispatcher.utter_message(response="utter_no_transactdate")
return {"time": None}
return parsedinterval
class ActionTransferMoney(Action):
"""Transfers Money."""
def name(self) -> Text:
"""Unique identifier of the action"""
return "action_transfer_money"
async def run(
self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: Dict
) -> List[EventType]:
"""Executes the action"""
slots = {
"AA_CONTINUE_FORM": None,
"zz_confirm_form": None,
"PERSON": None,
"amount-of-money": None,
"number": None,
}
if tracker.get_slot("zz_confirm_form") == "yes":
amount_of_money = float(tracker.get_slot("amount-of-money"))
from_account_number = profile_db.get_account_number(
profile_db.get_account_from_session_id(tracker.sender_id)
)
to_account_number = profile_db.get_account_number(
profile_db.get_recipient_from_name(
tracker.sender_id, tracker.get_slot("PERSON")
)
)
profile_db.transact(
from_account_number,
to_account_number,
amount_of_money,
)
dispatcher.utter_message(response="utter_transfer_complete")
amount_transferred = float(tracker.get_slot("amount_transferred"))
slots["amount_transferred"] = amount_transferred + amount_of_money
else:
dispatcher.utter_message(response="utter_transfer_cancelled")
return [SlotSet(slot, value) for slot, value in slots.items()]
class ValidateTransferMoneyForm(CustomFormValidationAction):
"""Validates Slots of the transfer_money_form"""
def name(self) -> Text:
"""Unique identifier of the action"""
return "validate_transfer_money_form"
async def validate_PERSON(
self,
value: Text,
dispatcher: CollectingDispatcher,
tracker: Tracker,
domain: Dict[Text, Any],
) -> Dict[Text, Any]:
"""Validates value of 'PERSON' slot"""
# It is possible that both Spacy & DIET extracted the PERSON
# Just pick the first one
if isinstance(value, list):
value = value[0]
name = value.lower() if value else None
known_recipients = profile_db.list_known_recipients(tracker.sender_id)
first_names = [name.split()[0] for name in known_recipients]
if name is not None and name in known_recipients:
return {"PERSON": name.title()}
if name in first_names:
index = first_names.index(name)
fullname = known_recipients[index]
return {"PERSON": fullname.title()}
dispatcher.utter_message(response="utter_unknown_recipient", PERSON=value)
return {"PERSON": None}
async def explain_PERSON(
self,
value: Text,
dispatcher: CollectingDispatcher,
tracker: Tracker,
domain: Dict[Text, Any],
) -> Dict[Text, Any]:
"""Explains 'PERSON' slot"""
recipients = profile_db.list_known_recipients(tracker.sender_id)
formatted_recipients = "\n" + "\n".join(
[f"- {recipient.title()}" for recipient in recipients]
)
dispatcher.utter_message(
response="utter_recipients",
formatted_recipients=formatted_recipients,
)
return {}
async def validate_amount_of_money(
self,
value: Text,
dispatcher: CollectingDispatcher,
tracker: Tracker,
domain: Dict[Text, Any],
) -> Dict[Text, Any]:
"""Validates value of 'amount-of-money' slot"""
account_balance = profile_db.get_account_balance(tracker.sender_id)
try:
entity = get_entity_details(
tracker, "amount-of-money"
) or get_entity_details(tracker, "number")
amount_currency = parse_duckling_currency(entity)
if not amount_currency:
raise TypeError
if account_balance < float(amount_currency.get("amount-of-money")):
dispatcher.utter_message(response="utter_insufficient_funds")
return {"amount-of-money": None}
return amount_currency
except (TypeError, AttributeError):
dispatcher.utter_message(response="utter_no_payment_amount")
return {"amount-of-money": None}
async def validate_zz_confirm_form(
self,
value: Text,
dispatcher: CollectingDispatcher,
tracker: Tracker,
domain: Dict[Text, Any],
) -> Dict[Text, Any]:
"""Validates value of 'zz_confirm_form' slot"""
if value in ["yes", "no"]:
return {"zz_confirm_form": value}
return {"zz_confirm_form": None}
class ActionShowBalance(Action):
"""Shows the balance of bank or credit card accounts"""
def name(self) -> Text:
"""Unique identifier of the action"""
return "action_show_balance"
async def run(
self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: Dict
) -> List[EventType]:
"""Executes the custom action"""
account_type = tracker.get_slot("account_type")
if account_type == "credit":
# show credit card balance
credit_card = tracker.get_slot("credit_card")
available_cards = profile_db.list_credit_cards(tracker.sender_id)
if credit_card and credit_card.lower() in available_cards:
current_balance = profile_db.get_credit_card_balance(
tracker.sender_id, credit_card
)
dispatcher.utter_message(
response="utter_credit_card_balance",
**{
"credit_card": credit_card.title(),
"credit_card_balance": f"{current_balance:.2f}",
},
)
else:
for credit_card in profile_db.list_credit_cards(tracker.sender_id):
current_balance = profile_db.get_credit_card_balance(
tracker.sender_id, credit_card
)
dispatcher.utter_message(
response="utter_credit_card_balance",
**{
"credit_card": credit_card.title(),
"credit_card_balance": f"{current_balance:.2f}",
},
)
else:
# show bank account balance
account_balance = profile_db.get_account_balance(tracker.sender_id)
amount = tracker.get_slot("amount_transferred")
if amount:
amount = float(tracker.get_slot("amount_transferred"))
init_account_balance = account_balance + amount
dispatcher.utter_message(
response="utter_changed_account_balance",
init_account_balance=f"{init_account_balance:.2f}",
account_balance=f"{account_balance:.2f}",
)
else:
dispatcher.utter_message(
response="utter_account_balance",
init_account_balance=f"{account_balance:.2f}",
)
events = []
active_form_name = tracker.active_form.get("name")
if active_form_name:
# keep the tracker clean for the predictions with form switch stories
events.append(UserUtteranceReverted())
# trigger utter_ask_{form}_AA_CONTINUE_FORM, by making it the requested_slot
events.append(SlotSet("AA_CONTINUE_FORM", None))
# avoid that bot goes in listen mode after UserUtteranceReverted
events.append(FollowupAction(active_form_name))
return events
class ActionShowRecipients(Action):
"""Lists the contents of then known_recipients slot"""
def name(self) -> Text:
"""Unique identifier of the action"""
return "action_show_recipients"
async def run(
self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: Dict
) -> List[EventType]:
"""Executes the custom action"""
recipients = profile_db.list_known_recipients(tracker.sender_id)
formatted_recipients = "\n" + "\n".join(
[f"- {recipient.title()}" for recipient in recipients]
)
dispatcher.utter_message(
response="utter_recipients",
formatted_recipients=formatted_recipients,
)
events = []
active_form_name = tracker.active_form.get("name")
if active_form_name:
# keep the tracker clean for the predictions with form switch stories
events.append(UserUtteranceReverted())
# trigger utter_ask_{form}_AA_CONTINUE_FORM, by making it the requested_slot
events.append(SlotSet("AA_CONTINUE_FORM", None))
# # avoid that bot goes in listen mode after UserUtteranceReverted
events.append(FollowupAction(active_form_name))
return events
class ActionShowTransferCharge(Action):
"""Lists the transfer charges"""
def name(self) -> Text:
"""Unique identifier of the action"""
return "action_show_transfer_charge"
async def run(
self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: Dict
) -> List[EventType]:
"""Executes the custom action"""
dispatcher.utter_message(response="utter_transfer_charge")
events = []
active_form_name = tracker.active_form.get("name")
if active_form_name:
# keep the tracker clean for the predictions with form switch stories
events.append(UserUtteranceReverted())
# trigger utter_ask_{form}_AA_CONTINUE_FORM, by making it the requested_slot
events.append(SlotSet("AA_CONTINUE_FORM", None))
# # avoid that bot goes in listen mode after UserUtteranceReverted
events.append(FollowupAction(active_form_name))
return events
class ActionSessionStart(Action):
"""Executes at start of session"""
def name(self) -> Text:
"""Unique identifier of the action"""
return "action_session_start"
@staticmethod
def _slot_set_events_from_tracker(
tracker: "Tracker",
) -> List["SlotSet"]:
"""Fetches SlotSet events from tracker and carries over keys and values"""
# when restarting most slots should be reset
relevant_slots = ["currency"]
return [
SlotSet(
key=event.get("name"),
value=event.get("value"),
)
for event in tracker.events
if event.get("event") == "slot" and event.get("name") in relevant_slots
]
async def run(
self,
dispatcher: CollectingDispatcher,
tracker: Tracker,
domain: Dict[Text, Any],
) -> List[EventType]:
"""Executes the custom action"""
# the session should begin with a `session_started` event
events = [SessionStarted()]
events.extend(self._slot_set_events_from_tracker(tracker))
# create a mock profile by populating database with values specific to tracker.sender_id
profile_db.populate_profile_db(tracker.sender_id)
currency = profile_db.get_currency(tracker.sender_id)
# initialize slots from mock profile
events.append(SlotSet("currency", currency))
# add `action_listen` at the end
events.append(ActionExecuted("action_listen"))
return events
class ActionRestart(Action):
"""Executes after restart of a session"""
def name(self) -> Text:
"""Unique identifier of the action"""
return "action_restart"
async def run(
self,
dispatcher: CollectingDispatcher,
tracker: Tracker,
domain: Dict[Text, Any],
) -> List[EventType]:
"""Executes the custom action"""
return [Restarted(), FollowupAction("action_session_start")]
class ActionAskTransactionSearchFormConfirm(Action):
"""Asks for the 'zz_confirm_form' slot of 'transaction_search_form'
A custom action is used instead of an 'utter_ask' response because a different
question is asked based on 'search_type' and 'vendor_name' slots.
"""
def name(self) -> Text:
return "action_ask_transaction_search_form_zz_confirm_form"
async def run(
self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: Dict
) -> List[EventType]:
"""Executes the custom action"""
search_type = tracker.get_slot("search_type")
vendor_name = tracker.get_slot("vendor_name")
start_time_formatted = tracker.get_slot("start_time_formatted")
end_time_formatted = tracker.get_slot("end_time_formatted")
if vendor_name:
vendor_name = f" with {vendor_name}"
else:
vendor_name = ""
if search_type == "spend":
text = (
f"Do you want to search for transactions{vendor_name} between "
f"{start_time_formatted} and {end_time_formatted}?"
)
elif search_type == "deposit":
text = (
f"Do you want to search deposits made to your account between "
f"{start_time_formatted} and {end_time_formatted}?"
)
buttons = [
{"payload": "/affirm", "title": "Yes"},
{"payload": "/deny", "title": "No"},
]
dispatcher.utter_message(text=text, buttons=buttons)
return []
class ActionSwitchFormsAsk(Action):
"""Asks to switch forms"""
def name(self) -> Text:
return "action_switch_forms_ask"
async def run(
self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: Dict
) -> List[EventType]:
"""Executes the custom action"""
active_form_name = tracker.active_form.get("name")
intent_name = tracker.latest_message["intent"]["name"]
next_form_name = NEXT_FORM_NAME.get(intent_name)
if (
active_form_name not in FORM_DESCRIPTION.keys()
or next_form_name not in FORM_DESCRIPTION.keys()
):
logger.debug(
f"Cannot create text for `active_form_name={active_form_name}` & "
f"`next_form_name={next_form_name}`"
)
next_form_name = None
else:
text = (
f"We haven't completed the {FORM_DESCRIPTION[active_form_name]} yet. "
f"Are you sure you want to switch to {FORM_DESCRIPTION[next_form_name]}?"
)
buttons = [
{"payload": "/affirm", "title": "Yes"},
{"payload": "/deny", "title": "No"},
]
dispatcher.utter_message(text=text, buttons=buttons)
return [SlotSet("next_form_name", next_form_name)]
class ActionSwitchFormsDeny(Action):
"""Does not switch forms"""
def name(self) -> Text:
return "action_switch_forms_deny"
async def run(
self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: Dict
) -> List[EventType]:
"""Executes the custom action"""
active_form_name = tracker.active_form.get("name")
if active_form_name not in FORM_DESCRIPTION.keys():
logger.debug(
f"Cannot create text for `active_form_name={active_form_name}`."
)
else:
text = f"Ok, let's continue with the {FORM_DESCRIPTION[active_form_name]}."
dispatcher.utter_message(text=text)
return [SlotSet("next_form_name", None)]
class ActionSwitchFormsAffirm(Action):
"""Switches forms"""
def name(self) -> Text:
return "action_switch_forms_affirm"
async def run(
self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: Dict
) -> List[EventType]:
"""Executes the custom action"""
active_form_name = tracker.active_form.get("name")
next_form_name = tracker.get_slot("next_form_name")
if (
active_form_name not in FORM_DESCRIPTION.keys()
or next_form_name not in FORM_DESCRIPTION.keys()
):
logger.debug(
f"Cannot create text for `active_form_name={active_form_name}` & "
f"`next_form_name={next_form_name}`"
)
else:
text = (
f"Great. Let's switch from the {FORM_DESCRIPTION[active_form_name]} "
f"to {FORM_DESCRIPTION[next_form_name]}. "
f"Once completed, you will have the option to switch back."
)
dispatcher.utter_message(text=text)
return [
SlotSet("previous_form_name", active_form_name),
SlotSet("next_form_name", None),
]
class ActionSwitchBackAsk(Action):
"""Asks to switch back to previous form"""
def name(self) -> Text:
return "action_switch_back_ask"
async def run(
self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: Dict
) -> List[EventType]:
"""Executes the custom action"""
previous_form_name = tracker.get_slot("previous_form_name")
if previous_form_name not in FORM_DESCRIPTION.keys():
logger.debug(
f"Cannot create text for `previous_form_name={previous_form_name}`"
)
previous_form_name = None
else:
text = (
f"Would you like to go back to the "
f"{FORM_DESCRIPTION[previous_form_name]} now?."
)
buttons = [
{"payload": "/affirm", "title": "Yes"},
{"payload": "/deny", "title": "No"},
]
dispatcher.utter_message(text=text, buttons=buttons)
return [SlotSet("previous_form_name", None)]
class ActionShowWeather(Action):
"""show weather"""
def name(self) -> Text:
return "action_weather_api"
async def run(
self,
dispatcher: CollectingDispatcher,
tracker: Tracker,
domain: Dict[Text, Any],
) -> List[Dict[Text, Any]]:
address = "https://int-api.mx.com/users/USR-8fe5e260-fe63-47dd-b8cd-438cd5a48b94/accounts?page=1&records_per_page=10"
headers = {
"Accept": "application/vnd.mx.api.v1+json",
"Content-Type": "application/json",
}
userName = "653f7dba-265c-4d70-ba21-3b94dc126361"
password = "0effbf747bfe42043998c4510489cf39d39c30ed"
json = requests.get(
address, verify=False, headers=headers, auth=(userName, password)
).json()["accounts"][0]["balance"]
buttons = [
{"payload": "/affirm", "title": "Yes"},
{"payload": "/deny", "title": "No"},
]
dispatcher.utter_message(response="utter_weather", temp=json, buttons=buttons)
return []
class AskForSlotAction(Action):
def name(self) -> Text:
return "action_ask_last_name"
def run(
self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: Dict
) -> List[EventType]:
first_name = tracker.get_slot("first_name")
dispatcher.utter_message(text=f"So {first_name}, what is your last name?")
return []
