-
Notifications
You must be signed in to change notification settings - Fork 0
/
bot.py
106 lines (83 loc) · 3.33 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
import requests
from config import TELEGRAM_SEND_MESSAGE_URL, TELEGRAM_SEND_MESSAGE_URL_BASE, TELEGRAM_SEND_PHOTO_URL, TELEGRAM_SEND_AUDIO_URL, CHAT_PROCESSOR_URL, TELEGRAM_SEND_REPLY_MARKUP_URL, TELEGRAM_SEND_TYPING_ACTION
import json, urllib.parse
def send_message_to_chat_processor(req):
data = {}
if 'message' in req:
message = req['message']
data['idChat'] = message['chat']['id'] # ID of Telegram chat
data['idUser'] = message['from']['id'] # ID of Telegram user
# Message received on Telegram chat
if 'text' in message:
data['msg'] = str(message['text'].encode('utf-8'), encoding='utf-8')
else:
data['msg'] = ''
# Name of the user
data['name'] = message['from']['first_name']
if 'last_name' in message['from']:
data['name'] += " " + message['from']['last_name']
# Timestamp
data['timestamp'] = message['date']
# Location
if 'location' in message:
loc = message['location']
data['location'] = {'lon': loc['longitude'], 'lat': loc['latitude']}
else:
data['location'] = None
else:
message = req['callback_query']
data['idChat'] = message['message']['chat']['id']
data['idUser'] = message['from']['id']
data['msg'] = str(message['data'].encode('utf-8'), encoding='utf-8')
data['name'] = message['from']['first_name']
if 'last_name' in message['from']:
data['name'] += " " + message['from']['last_name']
data['timestamp'] = message['message']['date']
if 'location' in message['message']:
loc = message['message']['location']
data['location'] = {'lon': loc['longitude'], 'lat': loc['latitude']}
else:
data['location'] = None
#Avisa chat_processor que chegou uma nova msg
res = requests.post(CHAT_PROCESSOR_URL + "/getResponse", json=data)
def send_message_to_user(idChat, msg, disable_notification):
msg = urllib.parse.quote(msg, safe='')
res = requests.get(TELEGRAM_SEND_MESSAGE_URL.format(idChat, msg, 'HTML', disable_notification))
if res.status_code == 200:
return True
else:
return False
def send_photo_to_user(idChat, photo_url, caption):
caption = urllib.parse.quote(caption, safe='')
res = requests.get(TELEGRAM_SEND_PHOTO_URL.format(idChat, photo_url, caption, 'HTML'))
if res.status_code == 200:
return True
else:
return False
def send_keyboard_to_user(idChat, text, keyboard):
res = requests.get(TELEGRAM_SEND_REPLY_MARKUP_URL.format(idChat, text, keyboard))
if res.status_code == 200:
return True
else:
return False
def send_typing_action(idChat):
res = requests.get(TELEGRAM_SEND_TYPING_ACTION.format(idChat, 'typing'))
def get_location(idChat):
reply_markup={
'keyboard': [
[{
'text': 'Enviar Localização',
'request_location': True
}]
]
}
payload = {
'chat_id': idChat,
'text': 'Clica no botão para me enviares a tua localização',
'reply_markup': json.dumps(reply_markup)
}
res = requests.post(TELEGRAM_SEND_MESSAGE_URL_BASE, data=payload)
if res.status_code == 200:
return True
else:
return False