-
Notifications
You must be signed in to change notification settings - Fork 1
/
bot.py
124 lines (104 loc) · 4.14 KB
/
bot.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
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
from threading import Thread
from typing import List
from flask import Flask, send_file
from logzero import logger as log
from redis import StrictRedis
import settings
from clients.facebook import FacebookClient
from clients.nluclients import DialogflowClient
from clients.telegram import TelegramClient
from clients.telegramsupport import TelegramSupportChannel
from clients.voice import VoiceRecognitionClient
from core import ChatAction
from core.context import ContextManager, States
from core.dialogmanager import DialogManager
from core.recorder import ConversationRecorder
from corpus.media import get_file_by_media_id
from logic.planning import PlanningAgent
from logic.rules.dialogcontroller import application_router
def main():
app = Flask(__name__)
@app.route('/ping', methods=['GET'])
def ping():
return "Pong"
# Register a media endpoint for clients that cannot upload images, videos etc. but instead only work with URLs
@app.route('/media/<mimetype>/<media_id>.<ext>', methods=['GET'])
def media_endpoint(mimetype, media_id, ext):
filepath = get_file_by_media_id(media_id)
return send_file(filepath, mimetype=f'{mimetype}/{ext}')
# Initialize redis cache
redis = StrictRedis.from_url(settings.REDIS_URL)
# Initialize bot api client for Facebook Messenger
facebook_client = FacebookClient(
app=app,
token=settings.FACEBOOK_ACCESS_TOKEN
)
facebook_client.initialize()
# Initialize bot api client for Telegram Messenger
telegram_client = TelegramClient(
app=app,
webhook_url=settings.APP_URL,
token=settings.TELEGRAM_ACCESS_TOKEN,
test_mode=settings.DEBUG_MODE
)
telegram_client.initialize()
# sms_client = SMSClient(
# settings.TWILIO_ACCESS_TOKEN,
# settings.TWILIO_ACCOUNT_SID
# )
# sms_client.initialize()
# Register a conversation logging channel on Telegram. This is an actual "channel" in Telegram terminology.
support_client = TelegramSupportChannel(
telegram_bot=telegram_client.bot,
channel_id=settings.SUPPORT_CHANNEL_ID
)
# Initialize NLU and SLR clients
dialogflow_client = DialogflowClient(settings.DIALOGFLOW_ACCESS_TOKEN)
voice_client = VoiceRecognitionClient()
# Create recorder for all conversations with the bot. This will publish to the support channel from above if the
# custom `publish_trigger` condition is met.
conversation_recorder = None
if settings.ENABLE_CONVERSATION_RECORDING:
def publish_trigger(_update, actions: List[ChatAction], _dialog_states) -> bool:
if any('preview_claim' in a.intents for a in actions):
return True
return False
conversation_recorder = ConversationRecorder(
telegram_client.bot,
support_channel=support_client,
publish_trigger=publish_trigger
)
# Initialize planning agent that holds the core logic of creating context-sensitive responses
planning_agent = PlanningAgent(router=application_router)
# Initialize dialog management collection that will steer the main control flow when users interact with the bots
DialogManager(
context_manager=ContextManager(initial_state=States.SMALLTALK, redis=redis),
bot_clients=[telegram_client, facebook_client],
nlu_client=dialogflow_client,
planning_agent=planning_agent,
recorder=conversation_recorder,
voice_recognition_client=voice_client,
support_channel=support_client
)
# Actually start the application
if settings.DEBUG_MODE:
host = 'localhost'
log.info(f"Listening in debug mode on {host}:{settings.PORT}...")
Thread(
target=app.run,
name='Debug App',
kwargs=dict(
host=host,
port=settings.PORT,
debug=True,
use_reloader=False
)
).start()
# long polling
telegram_client.updater.idle()
else:
# webhooks
log.info("Listening...")
app.run(host='0.0.0.0', port=settings.PORT)
if __name__ == '__main__':
main()