taobao_order_api.py
import requests import time import hashlib import json class TaobaoAPI: def __init__(self, app_key, app_secret, redirect_uri): """初始化淘宝API客户端""" self.app_key = app_key self.app_secret = app_secret self.redirect_uri = redirect_uri self.access_token = None self.refresh_token = None self.token_expire_time = 0 # 淘宝开放平台API端点 self.auth_url = "https://oauth.taobao.com/authorize" self.token_url = "https://oauth.taobao.com/token" self.api_url = "https://eco.taobao.com/router/rest" def get_authorize_url(self, state=None): """生成授权URL，引导商家登录授权""" params = { "response_type": "code", "client_id": self.app_key, "redirect_uri": self.redirect_uri, "view": "web" # 可选值: web/wap，决定授权页面的样式 } if state: params["state"] = state # 构建授权URL query_string = '&'.join([f"{k}={v}" for k, v in params.items()]) return f"{self.auth_url}?{query_string}" def get_access_token(self, code): """通过授权码换取Access Token""" params = { "grant_type": "authorization_code", "client_id": self.app_key, "client_secret": self.app_secret, "code": code, "redirect_uri": self.redirect_uri } response = requests.post(self.token_url, data=params) result = response.json() if "access_token" in result: self.access_token = result["access_token"] self.refresh_token = result.get("refresh_token") self.token_expire_time = int(time.time()) + result.get("expires_in", 0) return result else: raise Exception(f"获取AccessToken失败: {result}") def refresh_access_token(self): """刷新Access Token""" if not self.refresh_token: raise Exception("没有刷新令牌，无法刷新AccessToken") if time.time() + 600 > self.token_expire_time: # 提前10分钟刷新 params = { "grant_type": "refresh_token", "client_id": self.app_key, "client_secret": self.app_secret, "refresh_token": self.refresh_token } response = requests.post(self.token_url, data=params) result = response.json() if "access_token" in result: self.access_token = result["access_token"] self.refresh_token = result.get("refresh_token", self.refresh_token) self.token_expire_time = int(time.time()) + result.get("expires_in", 0) return result else: raise Exception(f"刷新AccessToken失败: {result}") return {"message": "Token未过期，无需刷新"} def generate_sign(self, params): """生成API请求签名""" # 按参数名升序排序 sorted_params = sorted(params.items(), key=lambda x: x[0]) # 拼接参数 sign_str = self.app_secret for k, v in sorted_params: if k != "sign" and v is not None: sign_str += f"{k}{v}" sign_str += self.app_secret # MD5加密 return hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper() def call(self, method, params=None): """调用淘宝API""" if not self.access_token: raise Exception("请先获取Access Token") # 检查并刷新token self.refresh_access_token() # 构建公共参数 public_params = { "method": method, "app_key": self.app_key, "timestamp": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()), "format": "json", "v": "2.0", "sign_method": "md5", "session": self.access_token } # 合并参数 all_params = {**public_params, **(params or {})} # 生成签名 all_params["sign"] = self.generate_sign(all_params) # 发送请求 response = requests.post(self.api_url, data=all_params) return response.json() def get_orders(self, start_time, end_time, status="WAIT_SELLER_SEND_GOODS", page_no=1, page_size=20): """获取淘宝订单列表 :param start_time: 订单创建开始时间，格式: yyyy-MM-dd HH:mm:ss :param end_time: 订单创建结束时间，格式: yyyy-MM-dd HH:mm:ss :param status: 订单状态，默认等待发货 :param page_no: 页码 :param page_size: 每页数量 """ method = "taobao.trades.sold.get" params = { "fields": "tid,title,type,status,payment,discount_fee,adjust_fee,post_fee,total_fee," "pay_time,end_time,created,seller_nick,buyer_nick,buyer_message," "receiver_name,receiver_state,receiver_city,receiver_district," "receiver_address,receiver_zip,receiver_mobile,receiver_phone", "start_created": start_time, "end_created": end_time, "status": status, "page_no": page_no, "page_size": page_size, "use_has_next": True } return self.call(method, params) # 使用示例 if __name__ == "__main__": # 替换为你的应用信息 APP_KEY = "your_app_key" APP_SECRET = "your_app_secret" REDIRECT_URI = "https://yourdomain.com/callback" # 初始化API客户端 taobao = TaobaoAPI(APP_KEY, APP_SECRET, REDIRECT_URI) # 1. 生成授权URL，引导商家授权 auth_url = taobao.get_authorize_url() print(f"请访问以下URL进行授权: {auth_url}") # 2. 用户授权后，会跳转至REDIRECT_URI，并携带code参数 # 这里需要开发者自己实现接收code的逻辑 code = input("请输入授权码: ") # 3. 获取Access Token token_result = taobao.get_access_token(code) print(f"获取Token成功: {json.dumps(token_result, indent=2)}") # 4. 获取订单数据 start_time = "2023-01-01 00:00:00" end_time = "2023-01-31 23:59:59" orders_result = taobao.get_orders(start_time, end_time) if "trades_sold_get_response" in orders_result: trades = orders_result["trades_sold_get_response"].get("trades", {}).get("trade", []) print(f"成功获取{len(trades)}个订单") for trade in trades: print(f"订单号: {trade['tid']}, 状态: {trade['status']}, 金额: {trade['payment']}") else: print(f"获取订单失败: {json.dumps(orders_result, indent=2)}")
上述代码实现了一个完整的淘宝订单接口客户端，包括授权流程、Token 管理和订单数据获取。使用时需要替换为自己的应用信息。商家首先需要访问生成的授权 URL 进行登录授权，授权通过后可以获取订单数据。接口使用接口使用OAuth2.0 授权机制，保证了数据的安全性和用户隐私。
