Python 调用旺店通/淘宝top接口例子最近项目需要使用python来连接旺店通，并且需要调用top的接口，包含签名的规则，创建表，分页获取等等。
话不多说，上代码。
下面的例子是获取ERP销售订单的出库单信息可以通过按照时间来获取，也可以通过最后修改时间来获取增量数据获取的结果1.引入各种库import requestsimport jsonimport Connectionimport timeimport hashlibimport datetimeimport mathfrom urllib import parse2.自定义初始化参数def __init__(self):self.sid = '***'self.appkey = '***'self.appsecret = '****'self.page_size = 100self.appkey_tb = '***'self.target_app_key = "***"self.appsecret_tb = '***'self.timestamp_tb = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))3.创建计算旺店通签名函数def get_sign(self, timestamp, start_time, end_time, page_no):str_appkey = '06-appkey:' + ('0000' + str(len(self.appkey)))[-4:] + '-' + self.appkeystr_end_time = '08-end_time:0019-' + str(end_time)str_page_no = '07-page_no:' + ('0000' + str(len(str(page_no))))[-4:] + '-' + str(page_no)str_page_size = '09-page_size:' + ('0000' + str(len(str(self.page_size))))[-4:] + '-' + str(self.page_size)str_sid = '03-sid:' + ('0000' + str(len(self.sid)))[-4:] + '-' + self.sidstr_timestamp = '09-timestamp:' + '0010-' + str(timestamp)str_start_time = '10-start_time:0019-' + str(start_time)str_last = str_appkey + ';' + str_end_time + ';' + str_page_no + ';' + str_page_size + ';' + str_sid + ';' + str_start_time + ';' + str_timestamp + self.appsecretsign_md5 = hashlib.md5(str_last.encode(encoding='UTF-8')).hexdigest()# print("明文：%s" % str_last)return sign_md54.创建计算淘宝签名def get_sign_taobao(self, dict_params=None):list_params = sorted(dict_params.items(), key=lambda d: d[0])destr = self.appsecret_tbfor i in range(len(list_params)):destr = destr + list_params[i][0] + str(list_params[i][1])tempstr = destr + self.appsecret_tb# print("tb明文：%s" % tempstr)result = hashlib.md5(tempstr.encode(encoding='UTF-8')).hexdigest().upper()# print("tb_sign:%s" % result)return result5.创建接口请求参数def parse_url(self, url, timestamp, start_time, end_time, page_no, method=None): if method is None:sign = self.get_sign(timestamp, start_time, end_time, page_no)values = {'sid': self.sid,'appkey': self.appkey,'page_no': page_no,'page_size': self.page_size,'start_time': start_time,'end_time': end_time,'timestamp': timestamp,'sign': sign}b = requests.post(url, data=values)else:body_dict1 = {"sid": self.sid,"app_key": self.appkey_tb,"timestamp": self.timestamp_tb,"method": method,"target_app_key": self.target_app_key,"format": "json","sign_method": "md5","v": "2.0","page_size": self.page_size,"start_time": start_time,"end_time": end_time}sign = self.get_sign_taobao(body_dict1)body_dict1['sign'] = signurl_data = parse.urlencode(body_dict1)# print(url_data)url = url.format(method, url_data) + "&sign=" + sign # quote()将字符串进行编码b = requests.get(url)result = json.loads(b.content.decode())return result6.创建响应结果接收def get_condition(self, str_sql, url, table_dict, days, method=None):sql = Connection.connect_sql(str_sql, "sqlserver_Allove", 'read')start_date_ori = sql[0][0].strftime("%Y-%m-%d %H:%M:%S") if sql[0][0] is not None else (datetime.datetime.now() - datetime.timedelta(days=1)).strftime("%Y-%m-%d %H:%M:%S")if days != '':start_date_ori = (datetime.datetime.now() -datetime.timedelta(days=days)).strftime("%Y-%m-%d %H:%M:%S")start_date_ori = datetime.datetime.strptime(start_date_ori, "%Y-%m-%d %H:%M:%S") + datetime.timedelta(seconds=1)end_date_ori = (datetime.datetime.strptime(datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "%Y-%m-%d %H:%M:%S"))num = (end_date_ori - start_date_ori).days + 1for d in range(0, num):start_date = start_date_ori + datetime.timedelta(days=d)if end_date_ori > start_date_ori + datetime.timedelta(days=d + 1):end_date = start_date_ori + datetime.timedelta(days=d + 1)else:end_date = end_date_oriself.get_query(url, table_dict, start_date, end_date, method)7.创建响应结果内容def get_query(self, url, table_list, start_time, end_time, method=None):timestamp = int(time.time())print('表[%s][%s]，查找时间范围：【%s--%s】' % (method, table_list['table_name'], start_time, end_time)) response = self.parse_url(url, timestamp, start_time, end_time, 0, method)if method is None:response_code = str(response["code"])else:response_code = str(response["response"]["errorcode"])if response_code == '0':if method is None:total_count = response['total_count']else:total_count = response["response"]['total_count']response = response["response"]if total_count != '0':page_all = math.ceil(int(total_count) / self.page_size)self.insert_data_first(table_list, response, total_count, 0) # 首页数据for i in range(1, page_all - 1):response_page = self.parse_url(url, timestamp, start_time, end_time, i, method)if response_code == 0:self.insert_data_first(table_list, response_page, total_count, i) # 分页数据8.第一层数据结果写入def insert_data_first(self, table_list, data_dict, total_count, page):# print(data_dict)table_name = table_list['table_name']first_table = table_list['first']second_table = table_list['second']sum = len(data_dict[first_table])sum_one = 0for data_list in data_dict[first_table]:str_sql = self.get_sql(first_table, data_list, table_name)sql = Connection.connect_sql(str_sql, "sqlserver_Allove", 'insert')# print('-' * 100)ss = '成功写入' if sql > 0 else '已存在'print('【insert_data_first】【%s】记录【%s】条。
