-
Notifications
You must be signed in to change notification settings - Fork 3
/
api.py
executable file
·122 lines (95 loc) · 3.45 KB
/
api.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
import requests
import json
def is_online(b):
url = "https://tapi.telstra.com/v2/messages/sms/heathcheck"
headers = {
'accept': "application/json",
'content-type': "application/json",
'authorization': "Bearer {}".format(b),
'cache-control': "no-cache",
}
x = requests.request("GET", url, headers=headers)
return x.json()["status"] == "up"
def get_status(b, msg_id):
url = "https://tapi.telstra.com/v2/messages/sms/{}/status".format(msg_id)
headers = {
'accept': "application/json",
'content-type': "application/json",
'authorization': "Bearer {}".format(b),
'cache-control': "no-cache",
}
return requests.request("GET", url, headers=headers)
def get_bearer(key, secret):
url = "https://tapi.telstra.com/v2/oauth/token"
headers = {
'content-type': "application/x-www-form-urlencoded",
'cache-control': "no-cache"
}
body = {
'grant_type': "client_credentials",
'client_id': key,
'client_secret': secret
}
return requests.request("POST", url, data=body, headers=headers)
def get_message(b):
url = "https://tapi.telstra.com/v2/messages/sms"
headers = {
'accept': "application/json",
'content-type': "application/json",
'authorization': "Bearer {}".format(b),
'cache-control': "no-cache",
}
return requests.request("GET", url, headers=headers)
def get_number(b):
url = "https://tapi.telstra.com/v2/messages/provisioning/subscriptions"
headers = {
'accept': "application/json",
'content-type': "application/json",
'authorization': "Bearer {}".format(b),
'cache-control': "no-cache",
}
return requests.request("GET", url, headers=headers)
def new_number(b):
url = "https://tapi.telstra.com/v2/messages/provisioning/subscriptions"
payload = ""
headers = {
'accept': "application/json",
'content-type': "application/x-www-form-urlencoded",
'authorization': "Bearer {}".format(b),
'cache-control': "no-cache"
}
return requests.request("POST", url, data=payload, headers=headers)
def send_message(b, dest, msg, validity=5, wait=0):
url = "https://tapi.telstra.com/v2/messages/sms"
headers = {
'accept': "application/json",
'content-type': "application/json",
'authorization': "Bearer {}".format(b),
'cache-control': "no-cache",
}
body = {
'to': dest,
'body': msg,
'validity': validity
}
if wait > 0:
body['scheduledDelivery'] = wait
payload = json.dumps(body)
return requests.request("POST", url, data=payload, headers=headers)
if __name__ == "__main__":
import argparse
import readline
import code
parser = argparse.ArgumentParser(description="Interactive console to interface with the Telstra SMS API.")
parser.add_argument("key", help="client key")
parser.add_argument("secret", help="client secret")
args = parser.parse_args()
try:
b = get_bearer(args.key, args.secret).json()['access_token']
num_info = get_number(b).json()
number = num_info['destinationAddress']
print("{} expiring in {} day(s)".format(number, num_info['activeDays']))
print("bearer stored in `b` variable")
except:
print("Error requesting bearer")
code.InteractiveConsole({**globals(), **locals()}).interact()