先准备app授权码
import tornado.ioloop
import tornado.web
import urllib.request
import urllib.parse
import urllib.error
class ALiExpressAuthorityModel(object):
app_key = None
app_signature = None
code = None
aliId = None
resource_owner = None
expires_in = None
access_token = None
refresh_token = None
refresh_token_timeout = None
class MainHandler(tornado.web.RequestHandler):
def get(self):
self.write("Hello, world")
class AuthorityHandler(tornado.web.RequestHandler):
redirect_uri = "http://localhost:888/Authority"
model = ALiExpressAuthorityModel()
def __init__(self, application, request, **kwargs):
super(AuthorityHandler, self).__init__(application, request, **kwargs)
def get(self):
self.model.app_key = "xxx"
self.model.app_signature = "xxx"
if "code" in self.request.arguments:
code = self.get_argument('code')
self.model.code = code
self.get_token(self.model)
else:
self.get_code(self.model)
def get_code(self, model):
"""获取code"""
app_key = model.app_key
app_signature = model.app_signature
# 基础参数
param = {
"client_id": app_key,
"site": "aliexpress",
"redirect_uri": self.redirect_uri,
"state": "",
}
# 参数连接并排序
param_list = [str(key) + str(value) for key, value in param.items()]
param_list.sort()
data = ''.join(param_list)
# 计算签名
from hashlib import sha1
import hmac
aop_signature = hmac.new(app_signature.encode('utf-8'), data.encode('utf-8'), sha1).hexdigest()
aop_signature = aop_signature.upper()
# 添加签名
param["_aop_signature"] = aop_signature
url = "https://authhz.alibaba.com/auth/authorize.htm"
url_data = urllib.parse.urlencode(param)
url += "?" + url_data
self.redirect(url)
def get_token(self, model):
"""获取token"""
# 基础参数
param = {
"grant_type":"authorization_code",
"need_refresh_token":"true",
"client_id":model.app_key,
"client_secret":model.app_signature,
"redirect_uri":self.redirect_uri,
"code":model.code
}
url_param = urllib.parse.urlencode(param)
url = "https://gw.api.alibaba.com/openapi/http/1/system.oauth2/getToken/"
url += model.app_key+ "?"+ url_param
headers = {
"UserAgent":"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/65.0.3325.181 Safari/537.36 OPR/52.0.2871.64"
}
# 提交获取token
result = None
try:
req = urllib.request.Request(url=url, headers=headers)
res = urllib.request.urlopen(req)
result = res.read().decode('utf-8')
except Exception as e:
print(e)
if result is None:
self.write('fail')
self.finish()
return
import json
result_json = json.loads(result)
# {"aliId":"8888888888","resource_owner":"xxx","memberId":"xxxxxxx","expires_in":"36000","refresh_token":"479f9564-1049-456e-ab62-29d3e82277d9","access_token":"f14da3b8-b0b1-4f73-a5de-9bed637e0188","refresh_token_timeout":"20121222222222+0800"}
model.access_token=result_json["access_token"]
model.aliId = result_json["aliId"]
model.refresh_token = result_json["refresh_token"]
model.resource_owner = result_json["resource_owner"]
model.expires_in = result_json["expires_in"]
model.refresh_token_timeout = result_json["refresh_token_timeout"]
self.write('success')
self.finish()
s = json.dumps(model,default=lambda obj:obj.__dict__,sort_keys=False,indent=4)
print(s)
def refresh_token(self, model):
"""刷新token"""
url="https://gw.api.alibaba.com/openapi/param2/1/system.oauth2/postponeToken/"
url += model.app_key
headers = {
"UserAgent":"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/65.0.3325.181 Safari/537.36 OPR/52.0.2871.64"
}
param = {
"client_id":model.app_key,
"client_secret":model.app_signature,
"refresh_token":model.refresh_token,
"access_token":model.access_token
}
post_data = urllib.parse.urlencode(param).encode('utf-8')
result = None
try:
req = urllib.request.Request(url=url,headers=headers,data=post_data)
res = urllib.request.urlopen(req)
result = res.read().decode('utf-8')
except Exception as e:
print(e)
if result is None:
self.write('fail')
self.finish()
return
import json
result_json = json.loads(result)
# {"aliId":"8888888888","resource_owner":"xxx","memberId":"xxxxxxx","expires_in":"36000","refresh_token":"479f9564-1049-456e-ab62-29d3e82277d9","access_token":"f14da3b8-b0b1-4f73-a5de-9bed637e0188","refresh_token_timeout":"20121222222222+0800"}
model.access_token=result_json["access_token"]
model.aliId = result_json["aliId"]
model.refresh_token = result_json["refresh_token"]
model.resource_owner = result_json["resource_owner"]
model.expires_in = result_json["expires_in"]
model.refresh_token_timeout = result_json["refresh_token_timeout"]
self.write('success')
self.finish()
s = json.dumps(model,default=lambda obj:obj.__dict__,sort_keys=False,indent=4)
print(s)
# 127.0.0.1:888/Authority
handlers = [
(r"/", MainHandler),
(r"/Authority", AuthorityHandler),
]
if __name__ == "__main__":
application = tornado.web.Application(handlers)
application.listen(888)
tornado.ioloop.IOLoop.instance().start()
网友评论