def index(self):
scope="alexa_all"
sd = json.dumps({
"alexa:all": {
"productID": ProductID,
"productInstanceAttributes": {
"deviceSerialNumber": "001"
}
}
})
url = "https://www.amazon.com/ap/oa"
callback = cherrypy.url() + "code"
payload = {"client_id" : Client_ID, "scope" : "alexa:all", "scope_data" : sd, "response_type" : "code", "redirect_uri" : callback }
req = requests.Request('GET', url, params=payload)
p = req.prepare()
raise cherrypy.HTTPRedirect(p.url)
python类url()的实例源码
def index(self):
sd = json.dumps({
"alexa:all": {
"productID": config['alexa']['Device_Type_ID'],
"productInstanceAttributes": {
"deviceSerialNumber": hashlib.sha256(str(uuid.getnode()).encode()).hexdigest()
}
}
})
url = "https://www.amazon.com/ap/oa"
callback = cherrypy.url() + "code"
payload = {
"client_id": config['alexa']['Client_ID'],
"scope": "alexa:all",
"scope_data": sd,
"response_type": "code",
"redirect_uri": callback
}
req = requests.Request('GET', url, params=payload)
prepared_req = req.prepare()
raise cherrypy.HTTPRedirect(prepared_req.url)
def code(self, var=None, **params): # pylint: disable=unused-argument
code = quote(cherrypy.request.params['code'])
callback = cherrypy.url()
payload = {
"client_id": config['alexa']['Client_ID'],
"client_secret": config['alexa']['Client_Secret'],
"code": code,
"grant_type": "authorization_code",
"redirect_uri": callback
}
url = "https://api.amazon.com/auth/o2/token"
response = requests.post(url, data=payload)
resp = response.json()
alexapi.config.set_variable(['alexa', 'refresh_token'], resp['refresh_token'])
return "<h2>Success!</h2>" \
"<p>The refresh token has been added to your config file.</p>" \
"<p>Now:</p>" \
"<ul>" \
"<li>close your this browser window,</li>" \
"<li>exit the setup script as indicated,</li>" \
"<li>and follow the Post-installation steps.</li>" \
"</ul>"
def index(self):
scope="alexa_all"
sd = json.dumps({
"alexa:all": {
"productID": ProductID,
"productInstanceAttributes": {
"deviceSerialNumber": "001"
}
}
})
url = "https://www.amazon.com/ap/oa"
callback = cherrypy.url() + "code"
payload = {"client_id" : Client_ID, "scope" : "alexa:all", "scope_data" : sd, "response_type" : "code", "redirect_uri" : callback }
req = requests.Request('GET', url, params=payload)
p = req.prepare()
raise cherrypy.HTTPRedirect(p.url)
def index(self):
scope="alexa_all"
sd = json.dumps({
"alexa:all": {
"productID": ProductID,
"productInstanceAttributes": {
"deviceSerialNumber": "001"
}
}
})
url = "https://www.amazon.com/ap/oa"
callback = cherrypy.url() + "code"
payload = {"client_id" : Client_ID, "scope" : "alexa:all", "scope_data" : sd, "response_type" : "code", "redirect_uri" : callback }
req = requests.Request('GET', url, params=payload)
p = req.prepare()
raise cherrypy.HTTPRedirect(p.url)
def index(self):
scope="alexa_all"
sd = json.dumps({
"alexa:all": {
"productID": ProductID,
"productInstanceAttributes": {
"deviceSerialNumber": "001"
}
}
})
url = "https://www.amazon.com/ap/oa"
callback = cherrypy.url() + "code"
payload = {"client_id" : Client_ID, "scope" : "alexa:all", "scope_data" : sd, "response_type" : "code", "redirect_uri" : callback }
req = requests.Request('GET', url, params=payload)
p = req.prepare()
raise cherrypy.HTTPRedirect(p.url)
def index(self):
scope="alexa_all"
sd = json.dumps({
"alexa:all": {
"productID": ProductID,
"productInstanceAttributes": {
"deviceSerialNumber": "001"
}
}
})
url = "https://www.amazon.com/ap/oa"
# Establish AMZN callback URL
callback = cherrypy.url() + "code"
payload = { "client_id" : Client_ID,
"scope" : "alexa:all",
"scope_data" : sd,
"response_type" : "code",
"redirect_uri" : callback }
req = requests.Request('GET', url, params=payload)
p = req.prepare()
raise cherrypy.HTTPRedirect(p.url)
def index(self):
scope="alexa_all"
sd = json.dumps({
"alexa:all": {
"productID": ProductID,
"productInstanceAttributes": {
"deviceSerialNumber": "001"
}
}
})
url = "https://www.amazon.com/ap/oa"
callback = cherrypy.url() + "code"
payload = {"client_id" : Client_ID, "scope" : "alexa:all", "scope_data" : sd, "response_type" : "code", "redirect_uri" : callback }
req = requests.Request('GET', url, params=payload)
p = req.prepare()
raise cherrypy.HTTPRedirect(p.url)
def index(self):
scope = "alexa_all"
sd = json.dumps({
"alexa:all": {
"productID": ProductID,
"productInstanceAttributes": {
"deviceSerialNumber": "001"
}
}
})
url = "https://www.amazon.com/ap/oa"
callback = cherrypy.url() + "code"
payload = {"client_id": Client_ID, "scope": "alexa:all",
"scope_data": sd, "response_type": "code", "redirect_uri": callback}
req = requests.Request('GET', url, params=payload)
p = req.prepare()
raise cherrypy.HTTPRedirect(p.url)
def index(self):
scope="alexa_all"
sd = json.dumps({
"alexa:all": {
"productID": ProductID,
"productInstanceAttributes": {
"deviceSerialNumber": "001"
}
}
})
url = "https://www.amazon.com/ap/oa"
callback = cherrypy.url() + "code"
payload = {"client_id" : Client_ID, "scope" : "alexa:all", "scope_data" : sd, "response_type" : "code", "redirect_uri" : callback }
req = requests.Request('GET', url, params=payload)
p = req.prepare()
raise cherrypy.HTTPRedirect(p.url)
def get(self):
"""Return the current variant if in the cache, else None."""
request = cherrypy.serving.request
self.tot_gets += 1
uri = cherrypy.url(qs=request.query_string)
uricache = self.store.get(uri)
if uricache is None:
return None
header_values = [request.headers.get(h, '')
for h in uricache.selecting_headers]
variant = uricache.wait(key=tuple(sorted(header_values)),
timeout=self.antistampede_timeout,
debug=self.debug)
if variant is not None:
self.tot_hist += 1
return variant
def do_check(self):
"""Assert username. Raise redirect, or return True if request handled.
"""
sess = cherrypy.session
request = cherrypy.serving.request
response = cherrypy.serving.response
username = sess.get(self.session_key)
if not username:
sess[self.session_key] = username = self.anonymous()
self._debug_message('No session[username], trying anonymous')
if not username:
url = cherrypy.url(qs=request.query_string)
self._debug_message(
'No username, routing to login_screen with from_page %(url)r',
locals(),
)
response.body = self.login_screen(url)
if 'Content-Length' in response.headers:
# Delete Content-Length header so finalize() recalcs it.
del response.headers['Content-Length']
return True
self._debug_message('Setting request.login to %(username)r', locals())
request.login = username
self.on_check(username)
def log_hooks(debug=False):
"""Write request.hooks to the cherrypy error log."""
request = cherrypy.serving.request
msg = []
# Sort by the standard points if possible.
from cherrypy import _cprequest
points = _cprequest.hookpoints
for k in request.hooks.keys():
if k not in points:
points.append(k)
for k in points:
msg.append(' %s:' % k)
v = request.hooks.get(k, [])
v.sort()
for h in v:
msg.append(' %r' % h)
cherrypy.log('\nRequest Hooks for ' + cherrypy.url() +
':\n' + '\n'.join(msg), 'HTTP')
def trailing_slash(missing=True, extra=False, status=None, debug=False):
"""Redirect if path_info has (missing|extra) trailing slash."""
request = cherrypy.serving.request
pi = request.path_info
if debug:
cherrypy.log('is_index: %r, missing: %r, extra: %r, path_info: %r' %
(request.is_index, missing, extra, pi),
'TOOLS.TRAILING_SLASH')
if request.is_index is True:
if missing:
if not pi.endswith('/'):
new_url = cherrypy.url(pi + '/', request.query_string)
raise cherrypy.HTTPRedirect(new_url, status=status or 301)
elif request.is_index is False:
if extra:
# If pi == '/', don't redirect to ''!
if pi.endswith('/') and pi != '/':
new_url = cherrypy.url(pi[:-1], request.query_string)
raise cherrypy.HTTPRedirect(new_url, status=status or 301)
def __init__(self, path, query_string=''):
import cherrypy
self.request = cherrypy.serving.request
self.query_string = query_string
if '?' in path:
# Separate any params included in the path
path, self.query_string = path.split('?', 1)
# Note that urljoin will "do the right thing" whether url is:
# 1. a URL relative to root (e.g. "/dummy")
# 2. a URL relative to the current path
# Note that any query string will be discarded.
path = _urljoin(self.request.path_info, path)
# Set a 'path' member attribute so that code which traps this
# error can have access to it.
self.path = path
CherryPyException.__init__(self, path, self.query_string)
def index(self):
scope="alexa_all"
sd = json.dumps({
"alexa:all": {
"productID": ProductID,
"productInstanceAttributes": {
"deviceSerialNumber": "001"
}
}
})
url = "https://www.amazon.com/ap/oa"
callback = cherrypy.url() + "code"
payload = {"client_id" : Client_ID, "scope" : "alexa:all", "scope_data" : sd, "response_type" : "code", "redirect_uri" : callback }
req = requests.Request('GET', url, params=payload)
p = req.prepare()
raise cherrypy.HTTPRedirect(p.url)
def index(self):
scope = "alexa_all"
sd = json.dumps({
"alexa:all": {
"productID": ProductID,
"productInstanceAttributes": {
"deviceSerialNumber": "001"
}
}
})
url = "https://www.amazon.com/ap/oa"
callback = cherrypy.url() + "authresponse"
payload = {"client_id": Client_ID, "scope": "alexa:all", "scope_data": sd, "response_type": "code",
"redirect_uri": callback}
req = requests.Request('GET', url, params=payload)
p = req.prepare()
raise cherrypy.HTTPRedirect(p.url)
def index(self):
scope="alexa_all"
sd = json.dumps({
"alexa:all": {
"productID": ProductID,
"productInstanceAttributes": {
"deviceSerialNumber": "001"
}
}
})
url = "https://www.amazon.com/ap/oa"
callback = cherrypy.url() + "code"
payload = {"client_id" : Client_ID, "scope" : "alexa:all", "scope_data" : sd, "response_type" : "code", "redirect_uri" : callback }
req = requests.Request('GET', url, params=payload)
p = req.prepare()
raise cherrypy.HTTPRedirect(p.url)
def index(self):
sd = json.dumps({
"alexa:all": {
"productID": PRODUCT_ID,
"productInstanceAttributes": {
"deviceSerialNumber": uuid.getnode()
}
}
})
url = "https://www.amazon.com/ap/oa"
callback = cherrypy.url() + "authresponse"
payload = {
"client_id": CLIENT_ID,
"scope": "alexa:all",
"scope_data": sd,
"response_type": "code",
"redirect_uri": callback
}
req = requests.Request('GET', url, params=payload)
p = req.prepare()
raise cherrypy.HTTPRedirect(p.url)
def get(self):
"""Return the current variant if in the cache, else None."""
request = cherrypy.serving.request
self.tot_gets += 1
uri = cherrypy.url(qs=request.query_string)
uricache = self.store.get(uri)
if uricache is None:
return None
header_values = [request.headers.get(h, '')
for h in uricache.selecting_headers]
variant = uricache.wait(key=tuple(sorted(header_values)),
timeout=self.antistampede_timeout,
debug=self.debug)
if variant is not None:
self.tot_hist += 1
return variant
def do_check(self):
"""Assert username. Raise redirect, or return True if request handled.
"""
sess = cherrypy.session
request = cherrypy.serving.request
response = cherrypy.serving.response
username = sess.get(self.session_key)
if not username:
sess[self.session_key] = username = self.anonymous()
self._debug_message('No session[username], trying anonymous')
if not username:
url = cherrypy.url(qs=request.query_string)
self._debug_message(
'No username, routing to login_screen with from_page %(url)r',
locals(),
)
response.body = self.login_screen(url)
if "Content-Length" in response.headers:
# Delete Content-Length header so finalize() recalcs it.
del response.headers["Content-Length"]
return True
self._debug_message('Setting request.login to %(username)r', locals())
request.login = username
self.on_check(username)
def log_hooks(debug=False):
"""Write request.hooks to the cherrypy error log."""
request = cherrypy.serving.request
msg = []
# Sort by the standard points if possible.
from cherrypy import _cprequest
points = _cprequest.hookpoints
for k in request.hooks.keys():
if k not in points:
points.append(k)
for k in points:
msg.append(" %s:" % k)
v = request.hooks.get(k, [])
v.sort()
for h in v:
msg.append(" %r" % h)
cherrypy.log('\nRequest Hooks for ' + cherrypy.url() +
':\n' + '\n'.join(msg), "HTTP")
def trailing_slash(missing=True, extra=False, status=None, debug=False):
"""Redirect if path_info has (missing|extra) trailing slash."""
request = cherrypy.serving.request
pi = request.path_info
if debug:
cherrypy.log('is_index: %r, missing: %r, extra: %r, path_info: %r' %
(request.is_index, missing, extra, pi),
'TOOLS.TRAILING_SLASH')
if request.is_index is True:
if missing:
if not pi.endswith('/'):
new_url = cherrypy.url(pi + '/', request.query_string)
raise cherrypy.HTTPRedirect(new_url, status=status or 301)
elif request.is_index is False:
if extra:
# If pi == '/', don't redirect to ''!
if pi.endswith('/') and pi != '/':
new_url = cherrypy.url(pi[:-1], request.query_string)
raise cherrypy.HTTPRedirect(new_url, status=status or 301)
def test_redir_using_url(self):
for url in script_names:
prefix = self.script_name = url
# Test the absolute path to the parent (leading slash)
self.getPage('/redirect_via_url?path=./')
self.assertStatus(('302 Found', '303 See Other'))
self.assertHeader('Location', '%s/' % self.base())
# Test the relative path to the parent (no leading slash)
self.getPage('/redirect_via_url?path=./')
self.assertStatus(('302 Found', '303 See Other'))
self.assertHeader('Location', '%s/' % self.base())
# Test the absolute path to the parent (leading slash)
self.getPage('/redirect_via_url/?path=./')
self.assertStatus(('302 Found', '303 See Other'))
self.assertHeader('Location', '%s/' % self.base())
# Test the relative path to the parent (no leading slash)
self.getPage('/redirect_via_url/?path=./')
self.assertStatus(('302 Found', '303 See Other'))
self.assertHeader('Location', '%s/' % self.base())
def __init__(self, path, query_string=""):
import cherrypy
self.request = cherrypy.serving.request
self.query_string = query_string
if "?" in path:
# Separate any params included in the path
path, self.query_string = path.split("?", 1)
# Note that urljoin will "do the right thing" whether url is:
# 1. a URL relative to root (e.g. "/dummy")
# 2. a URL relative to the current path
# Note that any query string will be discarded.
path = _urljoin(self.request.path_info, path)
# Set a 'path' member attribute so that code which traps this
# error can have access to it.
self.path = path
CherryPyException.__init__(self, path, self.query_string)
def code(self, var=None, **params):
code = urllib.quote(cherrypy.request.params['code'])
callback = cherrypy.url()
payload = {"client_id" : Client_ID, "client_secret" : Client_Secret, "code" : code, "grant_type" : "authorization_code", "redirect_uri" : callback }
url = "https://api.amazon.com/auth/o2/token"
r = requests.post(url, data = payload)
resp = r.json()
line = 'refresh_token = "{}"'.format(resp['refresh_token'])
with open("creds.py", 'a') as f:
f.write(line)
return "Success!, refresh token has been added to your creds file, you may now reboot the Pi <br>{}".format(resp['refresh_token'])
def code(self, var=None, **params):
code = urllib.quote(cherrypy.request.params['code'])
callback = cherrypy.url()
payload = {"client_id" : Client_ID, "client_secret" : Client_Secret, "code" : code, "grant_type" : "authorization_code", "redirect_uri" : callback }
url = "https://api.amazon.com/auth/o2/token"
r = requests.post(url, data = payload)
resp = r.json()
line = 'refresh_token = "{}"'.format(resp['refresh_token'])
with open("creds.py", 'a') as f:
f.write(line)
return "Success!, refresh token has been added to your creds file, you may now reboot the Pi <br>{}".format(resp['refresh_token'])
def code(self, var=None, **params):
code = urllib.quote(cherrypy.request.params['code'])
callback = cherrypy.url()
payload = {"client_id" : Client_ID, "client_secret" : Client_Secret, "code" : code, "grant_type" : "authorization_code", "redirect_uri" : callback }
url = "https://api.amazon.com/auth/o2/token"
r = requests.post(url, data = payload)
resp = r.json()
line = 'refresh_token = "{}"'.format(resp['refresh_token'])
with open("creds.py", 'a') as f:
f.write(line)
return "Success!, refresh token has been added to your creds file, you may now reboot the Pi <br>{}".format(resp['refresh_token'])
def code(self, var=None, **params):
code = urllib.quote(cherrypy.request.params['code'])
callback = cherrypy.url()
payload = {"client_id" : Client_ID, "client_secret" : Client_Secret, "code" : code, "grant_type" : "authorization_code", "redirect_uri" : callback }
url = "https://api.amazon.com/auth/o2/token"
r = requests.post(url, data = payload)
resp = r.json()
line = 'refresh_token = "{}"'.format(resp['refresh_token'])
with open("creds.py", 'a') as f:
f.write(line)
return "Success!, refresh token has been added to your creds file, you may now reboot the Pi <br>{}".format(resp['refresh_token'])
def index(self):
""" The main page
"""
product_id = Config.get_config(Config.FIELD_PRODUCT_ID)
client_id = Config.get_config(Config.FIELD_CLIENT_ID)
scope_data = json.dumps(
{"alexa:all": {
"productID": product_id,
"productInstanceAttributes": {
"deviceSerialNumber": "001"}
}}
)
callback = cherrypy.url() + "code"
payload = {
"client_id": client_id,
"scope": "alexa:all",
"scope_data": scope_data,
"response_type": "code",
"redirect_uri": callback
}
req = requests.Request(
'GET', AlexaService.AMAZON_BASE_URL, params=payload
)
raise cherrypy.HTTPRedirect(req.prepare().url)
