forked from NacirB/StackOverFlowBot
-
Notifications
You must be signed in to change notification settings - Fork 0
/
main_bot.py
executable file
·115 lines (89 loc) · 3.72 KB
/
main_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
#!/usr/bin/env python3
import requests
import time
import argparse
import os
import json
from requests.compat import urljoin
from dialogue_manager import DialogueManager
from utils import RESOURCE_PATH
class BotHandler(object):
"""
BotHandler is a class which implements all back-end of the bot.
It has tree main functions:
'get_updates' — checks for new messages
'send_message' – posts new message to user
'get_answer' — computes the most relevant on a user's question
"""
def __init__(self, token, dialogue_manager):
self.token = token
self.api_url = "https://api.telegram.org/bot{}/".format(token)
self.dialogue_manager = dialogue_manager
def get_updates(self, offset=None, timeout=30):
params = {"timeout": timeout, "offset": offset}
raw_resp = requests.get(urljoin(self.api_url, "getUpdates"), params)
try:
resp = raw_resp.json()
except json.decoder.JSONDecodeError as e:
print("Failed to parse response {}: {}.".format(raw_resp.content, e))
return []
if "result" not in resp:
return []
return resp["result"]
def send_message(self, chat_id, text):
params = {"chat_id": chat_id, "text": text}
return requests.post(urljoin(self.api_url, "sendMessage"), params)
def get_answer(self, question):
if question == '/start':
return "Hi, I am your project bot. How can I help you today?"
return self.dialogue_manager.generate_answer(question)
def parse_args():
parser = argparse.ArgumentParser()
parser.add_argument('--token', type=str, default='')
return parser.parse_args()
def is_unicode(text):
return len(text) == len(text.encode())
class SimpleDialogueManager(object):
"""
This is the simplest dialogue manager to test the telegram bot.
Your task is to create a more advanced one in dialogue_manager.py."
"""
def generate_answer(self, question):
return "Hello, world!"
def main():
args = parse_args()
token = args.token
if not token:
if not "TELEGRAM_TOKEN" in os.environ:
print("Please, set bot token through --token or TELEGRAM_TOKEN env variable")
return
token = os.environ["TELEGRAM_TOKEN"]
#################################################################
# Your task is to complete dialogue_manager.py and use your
# advanced DialogueManager instead of SimpleDialogueManager.
# This is the point where you plug it into the Telegram bot.
# Do not forget to import all needed dependencies when you do so.
#simple_manager = SimpleDialogueManager()
#bot = BotHandler(token, simple_manager)
dialogue_manager = DialogueManager(RESOURCE_PATH)
bot = BotHandler(token, dialogue_manager)
###############################################################
print("Ready to talk!")
offset = 0
while True:
updates = bot.get_updates(offset=offset)
for update in updates:
print("An update received.")
if "message" in update:
chat_id = update["message"]["chat"]["id"]
if "text" in update["message"]:
text = update["message"]["text"]
if is_unicode(text):
print("Update content: {}".format(update))
bot.send_message(chat_id, bot.get_answer(update["message"]["text"]))
else:
bot.send_message(chat_id, "Hmm, you are sending some weird characters to me...")
offset = max(offset, update['update_id'] + 1)
time.sleep(1)
if __name__ == "__main__":
main()