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 授权机制，保证了数据的安全性和用户隐私。
