-
Notifications
You must be signed in to change notification settings - Fork 0
/
defs.py
331 lines (253 loc) · 10.1 KB
/
defs.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
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
import asyncio
import twilio.rest
from twilio.base.exceptions import TwilioRestException
import credentials
import requests
import json
import mimetypes
import time
# create base twilio client. Put in your own credentials
twilio_client = twilio.rest.Client(credentials.twilio_get_sid(), credentials.twilio_get_auth())
dummy_list = [] # DO NOT REMOVE
# main message class
class Message:
message_type = "" # sms or mms
content = "" # message body
sid = "" # ID of message
number = "" # number of who sent message
message_to = "" # number of whom the msg was sent to
account_auth = "" # auth key of account
account_sid = "" # sid key of account
# this is for internal use only
beta_uri = ""
# respond with sms
def send_sms(self, content: str):
twilio_client.messages.create(
body=content,
from_=self.message_to,
to=self.number
)
# respond with mms
def send_mms(self, content: str, path: str):
twilio_client.messages.create(
body=content,
from_=self.message_to,
media_url=[path],
to=self.number
)
def send_multiple_mms(self, content: str, paths: list):
twilio_client.messages.create(
body=content,
from_=self.message_to,
media_url=paths,
to=self.number
)
# delete message
def mark_as_read(self):
try:
# if server has not received changes
twilio_client.messages(self.sid).delete()
except TwilioRestException:
if self.sid not in dummy_list:
# append to delete list
dummy_list.append(self.sid)
# download message
def MMS_mv(self, file_name: str):
# get uri link
string1 = f"https://api.twilio.com{self.beta_uri}"
# parse
string1 = f"{string1[:-5]}/Media{string1[-5:]}"
# get media link
r = requests.get(string1, auth=(self.account_sid, self.account_auth))
r = json.loads(r.text)
# parse
r = r["media_list"][0] # gets first media found, TODO: make multiple MMS receiving - Message
sid = r["sid"]
mime_type = r["content_type"]
media_url = f"{string1[:-5]}/{sid}"
# gets media data
r2 = requests.get(media_url)
# finally download
with open(f'{file_name}{mimetypes.guess_extension(mime_type, strict=True)}', 'wb') as handler:
handler.write(r2.content)
def MMS_raw_data(self):
# get uri link
string1 = f"https://api.twilio.com{self.beta_uri}"
# parse
string1 = f"{string1[:-5]}/Media{string1[-5:]}"
# get media link
r = requests.get(string1, auth=(self.account_sid, self.account_auth))
r = json.loads(r.text)
# parse
r = r["media_list"][0] # first media found, see above
sid = r["sid"]
mime_type = r["content_type"]
# get data
media_url = f"{string1[:-5]}/{sid}"
r2 = requests.get(media_url)
return r2.content, mime_type
def ask(self, question: str, timeout: int = 60, default: str = "", advanced: bool = False):
# init
client = Client(credentials.twilio_get_number(), credentials.twilio_get_sid(), credentials.twilio_get_auth())
timer_timeout = time.perf_counter()
self.send_sms(question)
# while timer not out
while time.perf_counter() - timer_timeout <= timeout:
time.sleep(1)
new_messages = client.get_unread_messages(
6) # idk why I make it this number, it seems best for high capacity
for message in new_messages:
# if from actual user
if message.number == self.number:
message.mark_as_read()
if advanced:
# return message class
return message
else:
# return message data
return message.content
# timeout error messages - delete if u want
time.sleep(1)
if default != "":
self.send_sms(f'ERROR:TIMEOUT. User took too long to respond. Default response: {default}.')
else:
self.send_sms("ERROR:TIMEOUT. User took too long to respond. Please use command again to retry.")
return default
async def async_ask(self, question: str, timeout: int = 60, default: str = "", advanced: bool = False):
# init
client = Client(credentials.twilio_get_number(), credentials.twilio_get_sid(), credentials.twilio_get_auth())
timer_timeout = time.perf_counter()
self.send_sms(question)
# while message not received
while time.perf_counter() - timer_timeout <= timeout:
await asyncio.sleep(1)
new_messages = client.get_unread_messages(
6) # idk why I make it this number, it seems best for high capacity
for message in new_messages:
# if from actual user
if message.number == self.number:
message.mark_as_read()
if advanced:
# return message class
return message
else:
# return message data
return message.content
# timeout error messages, again see above
await asyncio.sleep(1)
if default != "":
self.send_sms(f'ERROR:TIMEOUT. User took too long to respond. Default response: {default}.')
else:
self.send_sms("ERROR:TIMEOUT. User took too long to respond. Please use command again to retry.")
return default
class Client:
number = ""
sid = ""
auth = ""
def __init__(self, number: str, sid: str, auth: str):
self.number = number
self.sid = sid
self.auth = auth
def get_unread_messages(self, list_count: int = 20):
response = []
messages = twilio_client.messages.list(
to=self.number,
limit=list_count
)
for message in messages:
if message.sid not in dummy_list: # if the message should be deleted, see mark_as_read()
# assign values to Message class
msg = Message()
msg.content = message.body
msg.number = message.from_
msg.sid = message.sid
msg.message_to = message.to
msg.account_auth = self.auth
msg.account_sid = self.sid
# message type
msg.beta_uri = message.uri
if message.num_media != "0":
msg.message_type = "mms"
else:
msg.message_type = "sms"
response.append(msg)
else:
dummy_list.remove(message.sid)
self.mark_as_read(message.sid)
# return list of Message objects
return response
def send_sms(self, content: str, to: str):
twilio_client.messages.create(
body=content,
from_=self.number,
to=to
)
def send_mms(self, content: str, to: str, path: str):
twilio_client.messages.create(
body=content,
from_=self.number,
media_url=[path],
to=to
)
def send_multiple_mms(self, content: str, to: str, paths: list):
twilio_client.messages.create(
body=content,
from_=self.number,
media_url=paths,
to=to
)
def mark_as_read(self, sid):
try:
# if server has not received changes
twilio_client.messages(sid).delete()
except TwilioRestException:
if sid not in dummy_list:
# add to delete/ignore list until changes received by server
dummy_list.append(sid)
def mark_all_read(self, number: int = 200):
messages = self.get_unread_messages(number)
for message in messages:
message.mark_as_read()
async def async_ask(self, question: str, msg: Message, timeout: int = 60, default: str = "",
advanced: bool = False):
timer_timeout = time.perf_counter()
msg.send_sms(question)
while time.perf_counter() - timer_timeout <= timeout:
await asyncio.sleep(1)
new_messages = self.get_unread_messages(
6) # idk why I make it this number, it seems best for high capacity
for message in new_messages:
if message.number == msg.number:
message.mark_as_read()
if advanced:
return message
else:
return message.content
# timeout error messages
await asyncio.sleep(1)
if default != "":
msg.send_sms(f'ERROR:TIMEOUT. User took too long to respond. Default response: {default}.')
else:
msg.send_sms("ERROR:TIMEOUT. User took too long to respond. Please use command again to retry.")
return default
def ask(self, question: str, msg: Message, timeout: int = 60, default: str = "", advanced: bool = False):
timer_timeout = time.perf_counter()
msg.send_sms(question)
while time.perf_counter() - timer_timeout <= timeout:
time.sleep(1)
new_messages = self.get_unread_messages(
6) # idk why I make it this number, it seems best for high capacity
for message in new_messages:
if message.number == msg.number:
message.mark_as_read()
if advanced:
return message
else:
return message.content
# timeout error messages
time.sleep(1)
if default != "":
msg.send_sms(f'ERROR:TIMEOUT. User took too long to respond. Default response: {default}.')
else:
msg.send_sms("ERROR:TIMEOUT. User took too long to respond. Please use command again to retry.")
return default