-
Notifications
You must be signed in to change notification settings - Fork 0
/
twilio.py
179 lines (151 loc) · 6.09 KB
/
twilio.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
# Simple REST interface to the Twilio phone SMS service.
import urllib
import urllib2
import base64
import json
import os
import config
# How often, in seconds, to fetch messages.
if config.PARTY_MODE:
FETCH_PERIOD_S = 1
else:
FETCH_PERIOD_S = 60
USER_AGENT = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/78.0.3904.108 Safari/537.36"
AUTHORIZATION = "Basic " + base64.b64encode(config.TWILIO_SID + ":" + config.TWILIO_TOKEN)
# A downloaded image.
class Image(object):
def __init__(self, pathname, content_type, phone_number, date_sent):
# Pathname of where the image was saved.
self.pathname = pathname
# Content type of the image.
self.content_type = content_type
# Phone number that sent the image.
self.phone_number = phone_number
# When the image was sent (string).
self.date_sent = date_sent
# All information about this fetch.
class DownloadResults(object):
def __init__(self, messages, images):
# JSON structure of all messages available, for debugging.
self.messages = messages
# List of Image objects that were downloaded.
self.images = images
# Returns the raw content of the specified Twilio-relative URL (after the
# domain name), or None of an error occurred. Set auth to True for most
# requests, False for JPEG requests.
def get_raw(url, auth, logger):
url = "https://api.twilio.com" + url
headers = {
# We get a 403 fetching images without this:
"User-Agent": USER_AGENT,
}
# Our auth information.
if auth:
headers["Authorization"] = AUTHORIZATION
request = urllib2.Request(url, None, headers)
try:
f = urllib2.urlopen(request)
data = f.read()
f.close()
except urllib2.HTTPError as e:
if logger:
logger.warning("Can't fetch \"%s\" from Twilio (%s)" % (url, e,))
return None
return data
# Returns the JSON content of the specified Twilio-relative URL (after the
# domain name), or None of an error occurred.
def get_json(url, logger):
j = get_raw(url, True, logger)
return json.loads(j) if j else None
# Save the image at the specified Twilio-relative URL (after the
# domain name), or None of an error occurred. Returns whether
# successful.
def save_image(uri, pathname, logger):
logger.info("Fetching photo to " + pathname)
raw_image = get_raw(uri, False, logger)
if not raw_image:
return False
try:
f = open(pathname, "wb")
f.write(raw_image)
f.close()
except IOError as e:
if logger:
logger.warning("Can't save \"%s\" (%s)" % (pathname, e,))
return False
return True
# Return all message information, or None if it can't be fetched.
def list_messages(logger):
return get_json("/2010-04-01/Accounts/%s/Messages.json" % config.TWILIO_SID, logger)
# Delete the specified resource (message or media).
def delete_resources(uri, logger):
url = "https://api.twilio.com" + uri
headers = {
"Authorization": "Basic " + base64.b64encode(config.TWILIO_SID + ":" + config.TWILIO_TOKEN),
}
request = urllib2.Request(url, None, headers)
request.get_method = lambda: "DELETE"
try:
f = urllib2.urlopen(request)
f.read()
f.close()
except urllib2.HTTPError as e:
if logger:
logger.warning("Can't delete Twilio resource \"%s\" (%s)" % (uri, e,))
# Connects to Twilio, downloads the first few message, downloads their
# images to files on disk, and optionally deletes the images and messages.
#
# image_path: directory where to save the image
#
# Returns a DownloadResults object, or None if the messages can't be fetched.
def download_images(image_path, delete, logger):
images = []
messages = list_messages(logger)
if not messages:
return None
for message in messages["messages"]:
source_phone_number = message["from"]
date_sent = message["date_sent"]
# logger.info("Fetching message from %s" % source_phone_number)
direction = message["direction"]
num_media = int(message["num_media"])
if direction == "inbound" and num_media > 0:
subresource_uris = message.get("subresource_uris")
if subresource_uris:
media_list = subresource_uris.get("media")
if media_list:
medias = get_json(media_list, logger)
for media in medias["media_list"]:
content_type = media["content_type"]
photo_sid = media["sid"]
uri = media["uri"]
if uri.endswith(".json") and content_type == "image/jpeg":
image_uri = uri[:-5]
pathname = os.path.join(image_path, photo_sid + ".jpg")
save_image(image_uri, pathname, logger)
images.append(Image(pathname, content_type, source_phone_number, date_sent))
if delete:
delete_resources(uri, logger)
if delete:
# We've seen cases where messages appeared in this list a few
# seconds before their photos were available. If we catch it in
# that window, we'll delete the message before we have a chance
# to get the photo. So don't delete messages, just let them live
# indefinitely.
## delete_resources(message["uri"], logger)
pass
return DownloadResults(messages, images)
def main():
# As a test, this connects to the Twilio service, downloads all
# the available photos, and delete all media and messages.
delete = False
import logging, pprint
logger = logging.getLogger()
logger.setLevel(logging.DEBUG)
logger.addHandler(logging.StreamHandler())
results = download_images(".", delete, logger)
pprint.pprint(results.messages)
for image in results.images:
print("%s (%s): %s" % (image.phone_number, image.date_sent, image.pathname))
if __name__ == "__main__":
main()