forked from lchagnoleau/Tissegram
-
Notifications
You must be signed in to change notification settings - Fork 0
/
telegram.py
147 lines (118 loc) · 4.97 KB
/
telegram.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
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
# coding=UTF-8
import requests
from log import get_logger
class BotHandlerMixin:
def __init__(self, token, webhook_ip):
#init logs
self.logger = get_logger(self.__class__.__name__)
self.logger.info("Init {}".format(self.__class__.__name__))
self.url = 'https://api.telegram.org/bot{}/'.format(token)
self.logger.debug("Telegram URL is {}".format(self.url))
self.data = {}
self.chat_id = ""
self.webhook_ip = webhook_ip
self.logger.debug("Telegram webhook IP is {}".format(self.webhook_ip))
def set_data(self, data):
self.data.clear()
self.data = dict(data)
self.logger.debug("Set data dict at \n{}".format(self.data))
def is_callback(self):
if 'callback_query' in self.data.keys():
self.logger.debug("Message received is a callback")
return True
else:
self.logger.debug("Message received is NOT a callback")
return False
def get_chat_id(self):
chat_id = ""
if self.is_callback():
chat_id = self.data['callback_query']['message']['chat']['id']
else:
chat_id = self.data['message']['chat']['id']
self.logger.debug("Chat ID is {}".format(chat_id))
return chat_id
def get_callback_id(self):
callback_query_id = self.data['callback_query']['id']
self.logger.debug("Callback query ID is {}".format(callback_query_id))
return callback_query_id
def get_chat_info(self):
chat_info = {}
if self.is_callback():
chat_info = dict(self.data['callback_query']['message']['chat'])
else:
chat_info = dict(self.data['message']['chat'])
chat_info.pop('id', None)
self.logger.debug("Chat info is {}".format(chat_info))
return chat_info
def get_user_info(self):
user_info = {}
if self.is_callback():
user_info = dict(self.data['callback_query']['from'])
else:
user_info = dict(self.data['message']['from'])
user_info.pop('id', None)
self.logger.debug("User info is {}".format(user_info))
return user_info
def get_user_name(self):
user_name = self.get_user_info()['first_name']
self.logger.debug("User name is {}".format(user_name))
return user_name
def get_message(self):
message = self.data['message']['text']
self.logger.debug("Message is {}".format(message))
return message
def get_message_id(self):
message_id = self.data['message']['message_id']
self.logger.debug("Message ID is {}".format(message_id))
return message_id
def get_callback(self):
callback = str(self.data['callback_query']['data'])
self.logger.debug("Callback is {}".format(callback))
return callback
def send_message(self, chat_id, message):
message_url = self.url + 'sendMessage'
json = {'chat_id':chat_id, 'text':message}
r = self.send(url=message_url, body=json)
return r.json()['result']['message_id']
def send_callback(self, chat_id, message, callback):
message_url = self.url + 'sendMessage'
json = {'chat_id':chat_id, 'reply_markup':callback, 'text':message}
r = self.send(url=message_url, body=json)
return r.json()['result']['message_id']
def get_Updates(self, offset=0):
message_url = self.url + 'getUpdates'
json = {'offset':offset}
r = self.send(url=message_url, body=json)
return r.json()
def answer_callback(self, callback_id):
message_url = self.url + 'answerCallbackQuery'
json = {"callback_query_id": callback_id}
self.send(url=message_url, body=json)
def delete_message(self, chat_id, message_id):
self.logger.debug("List of message ID to delete : \n{}".format(message_id))
for m in message_id:
message_url = self.url + 'deleteMessage'
json = {'chat_id':chat_id, 'message_id':m}
self.send(url=message_url, body=json)
def reset_messages(self):
message_url = self.url + 'deleteWebhook'
json = {}
self.send(url=message_url, body=json)
updates = self.get_Updates()
if len(updates['result']) > 0:
update_id = 0
for results in updates['result']:
if results['update_id'] >= update_id:
update_id = results['update_id'] + 1
updates = self.get_Updates(offset=update_id)
message_url = self.url + 'setWebhook'
json = {'url':self.webhook_ip}
self.send(url=message_url, body=json)
def send(self, url, body):
r = requests.post(url, json=body)
self.logger.debug("Send message at URL : {}\nBody : {}".format(url, body))
if r.json()['ok'] is True:
self.logger.debug("Response is : \n{}".format(r.json()))
else:
self.logger.warning("Response is : \n{}".format(r.json()))
return r