Python获取速卖通授权

先准备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()