forked from devicehive/AlexaDevice
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathalexa_http_config.py
executable file
·153 lines (147 loc) · 6.06 KB
/
alexa_http_config.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
#!/usr/bin/env python3
import os
import alexa_params
import alexa_control
import json
import requests
from http.server import BaseHTTPRequestHandler
from cgi import parse_header
from urllib.parse import parse_qs, urlencode, urlparse
threshold = alexa_params.DEFAULT_VOICE_THRESHOLD
productId = alexa_params.DEFAULT_PRODUCT_ID
clientId = alexa_params.DEFAULT_CLIEND_ID
deviceSerial = alexa_params.DEFAULT_DEVICE_SERIAL
clientSecret = alexa_params.DEFAULT_CLIENT_SECRET
def load_config():
global clientId, clientSecret, threshold, deviceSerial, productId
try:
with open(alexa_params.ALEXA_CREDENTIALS_FILE, 'r') as infile:
config = json.load(infile)
clientId = config['Client_ID']
clientSecret = config['Client_Secret']
threshold = config['threshold']
deviceSerial = config['deviceSerial']
productId = config['productId']
return config
except IOError:
pass
return None
class AlexaConfig(BaseHTTPRequestHandler):
def do_POST(self):
global threshold
global productId
global clientId
global deviceSerial
global clientSecret
ctype, pdict = parse_header(self.headers['content-type'])
if ctype == 'multipart/form-data':
postvars = parse_multipart(self.rfile, pdict)
elif ctype == 'application/x-www-form-urlencoded':
length = int(self.headers['content-length'])
postvars = parse_qs(self.rfile.read(length), keep_blank_values=1)
else:
postvars = {}
threshold = float(b''.join(postvars.get(bytes("threshold", "utf-8"))).decode("utf-8"))
productId = b''.join(postvars.get(bytes("productid", "utf-8"))).decode("utf-8")
clientId = b''.join(postvars.get(bytes("clientid", "utf-8"))).decode("utf-8")
deviceSerial = b''.join(postvars.get(bytes("serial", "utf-8"))).decode("utf-8")
clientSecret = b''.join(postvars.get(bytes("secret", "utf-8"))).decode("utf-8")
self.send_response(200)
self.send_header("Content-type", "text/html")
self.end_headers()
url = "https://www.amazon.com/ap/oa?" + urlencode({
"client_id": clientId,
"scope": "alexa:all",
"scope_data": json.dumps({
"alexa:all": {
"productID": productId,
"productInstanceAttributes": {
"deviceSerialNumber": deviceSerial
}
}
}).replace(" ", ""),
"response_type": "code",
"redirect_uri": alexa_params.BASE_URL + "authresponse"
})
self.wfile.write(bytes("<html><head><title>Alexa</title>", "utf-8"))
self.wfile.write(bytes("<meta http-equiv=\"refresh\" content=\"3; url=" + url + "\"/></head>", "utf-8"))
self.wfile.write(bytes("<body><p>Redirecting to amazon login service...</p>", "utf-8"))
self.wfile.write(bytes("</body></html>", "utf-8"))
def do_GET(self):
self.send_response(200)
self.send_header("Content-type", "text/html")
self.end_headers()
if self.path.startswith("/authresponse"):
parsed = urlparse(self.path)
code = ''.join(parse_qs(parsed.query).get("code"))
data = {
"grant_type": "authorization_code",
"code": code,
"client_id": clientId,
"client_secret": clientSecret,
"redirect_uri": alexa_params.BASE_URL + "authresponse"
}
headers = {'Content-Type': 'application/x-www-form-urlencoded'}
response = requests.post("https://api.amazon.com/auth/o2/token", data=urlencode(data), headers=headers)
if response.status_code == 200:
refresh_token = json.loads(response.text)["refresh_token"]
access_token = json.loads(response.text)["access_token"]
print("---------Access Token----------------")
print(access_token)
print("-------------------------------------")
with open(alexa_params.ALEXA_CREDENTIALS_FILE, 'w') as outfile:
data = {
"threshold": threshold,
"refresh_token": refresh_token,
"Client_ID": clientId,
"Client_Secret": clientSecret,
"deviceSerial": deviceSerial,
"productId": productId
}
json.dump(data, outfile)
alexa_control.start()
self.authorizedAnswer()
else:
self.wfile.write(bytes("<html><head><title>Alexa</title>", "utf-8"))
self.wfile.write(bytes("<body><p>Authorization error. <a href='/'>Click here to try again</a></p>", "utf-8"))
self.wfile.write(bytes("</body></html>", "utf-8"))
elif self.path == "/logout":
try:
os.remove(alexa_params.ALEXA_CREDENTIALS_FILE)
except FileNotFoundError:
pass
alexa_control.close()
self.wfile.write(bytes("<html><head><title>Alexa</title>", "utf-8"))
self.wfile.write(bytes("<meta http-equiv=\"refresh\" content=\"3; url=/\"/></head>", "utf-8"))
self.wfile.write(bytes("<body><p>Done. Reloading...</p>", "utf-8"))
self.wfile.write(bytes("</body></html>", "utf-8"))
elif self.path == "/restart":
alexa_control.close()
alexa_control.start()
self.authorizedAnswer();
elif os.path.isfile(alexa_params.ALEXA_CREDENTIALS_FILE):
self.authorizedAnswer();
else:
self.wfile.write(bytes("""
<html><head><title>Alexa</title>
<style type='text/css'>input[type=text],input[type=password]{width:100%;}input[type=submit]{width:30%;}</style>
</head><body><form method='post'>
Voice detection threshold(float value):<br>
<input type='text' name='threshold' value='""" + str(threshold) + """'><br><br>
Product ID(Device Type ID):<br>
<input type='text' name='productid' value='""" + productId + """'><br><br>
Client ID:<br>
<input type='text' name='clientid' value='""" + clientId + """'><br><br>
Device Serial Number:<br>
<input type='text' name='serial' value='""" + deviceSerial + """'><br><br>
Client Secret:<br>
<input type='password' name='secret' value='""" + clientSecret + """'><br><br>
<input type='submit' value='Log in'>
</form></body></html>
""", "utf-8"))
def authorizedAnswer(self):
self.wfile.write(bytes("<html><head><title>Alexa</title></head>", "utf-8"))
self.wfile.write(bytes("<body onload=\"if(window.location.pathname.length > 1) window.location.href='" + alexa_params.BASE_URL + "'\"><p>Alexa is configured and started.</p>", "utf-8"))
self.wfile.write(bytes("<p><a href=/logout>Click here to log out.</a></p>", "utf-8"))
self.wfile.write(bytes("<p><a href=/restart>Click here to restart Alexa.</a></p>", "utf-8"))
self.wfile.write(bytes("</body></html>", "utf-8"))