forked from Vaakash89/Rasa-Restaurant-Chatbot
-
Notifications
You must be signed in to change notification settings - Fork 0
/
rasa_slack_connector.py
71 lines (47 loc) · 2.08 KB
/
rasa_slack_connector.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
from __future__ import unicode_literals
import logging
from builtins import str
from flask import Blueprint, request, jsonify
from rasa_core.channels.channel import UserMessage, OutputChannel
from rasa_core.channels.rest import HttpInputComponent
logger = logging.getLogger(__name__)
class SlackBot(OutputChannel):
def __init__(self, slack_verification_token, channel):
self.slack_verification_token = slack_verification_token
self.channel = channel
def send_text_message(self, recipient_id, message):
from slackclient import SlackClient
text = message
recipient = recipient_id
CLIENT = SlackClient(self.slack_verification_token)
CLIENT.api_call('chat.postMessage', channel = self.channel, text = text, as_user = True)
class SlackInput(HttpInputComponent):
def __init__(self, slack_dev_token, slack_verification_token, slack_client, debug_mode):
self.slack_dev_token = slack_dev_token
self.debug_mode = debug_mode
self.slack_client = slack_client
self.slack_verification_token = slack_verification_token
def blueprint(self, on_new_message):
from flask import Flask, request, Response
slack_webhook = Blueprint('slack_webhook', __name__)
@slack_webhook.route('/', methods = ['GET'])
def health():
return jsonify({'status':'ok'})
@slack_webhook.route('/slack/events', methods = ['POST'])
def event():
if request.json.get('type') == 'url_verification':
return request.json.get('challenge'), 200
if request.json.get('token') == self.slack_client and request.json.get('type') == 'event_callback':
data = request.json
messaging_events = data.get('event')
channel = messaging_events.get('channel')
user = messaging_events.get('user')
text = messaging_events.get('text')
bot = messaging_events.get('bot_id')
if bot == None:
on_new_message(UserMessage(text, SlackBot(self.slack_verification_token, channel)))
return Response(headers={"X-Slack-No-Retry":1}), 200
return slack_webhook