tb运营系统接口实战指南：从商品管理到订单履约的全链路接入

┌──────────────────────────────────────────────────────────────┐
│                  淘宝运营系统接口化架构                        │
├─────────────┬─────────────┬─────────────┬──────────────────┤
│   商品管理    │   订单管理    │   物流履约    │   数据运营       │
├─────────────┼─────────────┼─────────────┼──────────────────┤
│ • 商品发布    │ • 订单同步    │ • 电子面单    │ • 评价采集       │
│ • 上下架      │ • 自动发货    │ • 物流跟踪    │ • 销售报表       │
│ • 库存同步    │ • 退款处理    │ • 地址库管理  │ • 竞品监控       │
│ • SKU管理     │ • 订单备注    │ • 运费模板    │ • 用户画像       │
└─────────────┴─────────────┴─────────────┴──────────────────┘
         ↓                    ↓
┌──────────────────────────────────────────────────────────────┐
│  聚石塔 ECS（订单/商品接口必须在聚石塔内调用）                 │
│  淘宝开放平台 API Gateway → OAuth2.0 授权 → 数据返回         │
└──────────────────────────────────────────────────────────────┘

import hashlib
import time
import requests
APP_KEY = "你的AppKey"
APP_SECRET = "你的AppSecret"
ACCESS_TOKEN = "店铺AccessToken"
GATEWAY = "https://eco.taobao.com/router/rest"
def top_sign(params, app_secret):
    """淘宝开放平台 TOP 签名（商品/订单/物流通用）"""
    filtered = {k: v for k, v in params.items() 
                if v is not None and v != "" and k != "sign"}
    sorted_kv = sorted(filtered.items(), key=lambda x: x[0])
    raw = app_secret + "".join([f"{k}{v}" for k, v in sorted_kv]) + app_secret
    return hashlib.md5(raw.encode("utf-8")).hexdigest().upper()
def get_onsale_items(page_no=1, page_size=50):
    """获取出售中商品列表"""
    params = {
        "method": "taobao.items.onsale.get",
        "app_key": APP_KEY,
        "session": ACCESS_TOKEN,
        "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
        "v": "2.0",
        "format": "json",
        "sign_method": "md5",
        "fields": "num_iid,title,price,num,sold_quantity,pic_url,list_time",
        "page_no": page_no,
        "page_size": page_size
    }
    params["sign"] = top_sign(params, APP_SECRET)
    
    resp = requests.get(GATEWAY, params=params, timeout=15)
    data = resp.json()
    if "items_onsale_get_response" in data:
        items = data["items_onsale_get_response"].get("items", {}).get("item", [])
        return items
    return []
def update_item_quantity(num_iid, quantity, sku_id=None):
    """更新商品库存"""
    params = {
        "method": "taobao.item.quantity.update",
        "app_key": APP_KEY,
        "session": ACCESS_TOKEN,
        "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
        "v": "2.0",
        "format": "json",
        "sign_method": "md5",
        "num_iid": num_iid,
        "quantity": quantity
    }
    if sku_id:
        params["sku_id"] = sku_id
    
    params["sign"] = top_sign(params, APP_SECRET)
    resp = requests.get(GATEWAY, params=params, timeout=15)
    return resp.json()

def get_recent_orders(start_time, end_time, page_no=1):
    """获取指定时间段的订单列表"""
    params = {
        "method": "taobao.trades.sold.get",
        "app_key": APP_KEY,
        "session": ACCESS_TOKEN,
        "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
        "v": "2.0",
        "format": "json",
        "sign_method": "md5",
        "fields": "tid,buyer_nick,created,payment,status,receiver_name,receiver_mobile,receiver_address,orders",
        "start_created": start_time,
        "end_created": end_time,
        "page_no": page_no,
        "page_size": 50
    }
    params["sign"] = top_sign(params, APP_SECRET)
    
    resp = requests.get(GATEWAY, params=params, timeout=15)
    data = resp.json()
    if "trades_sold_get_response" in data:
        trades = data["trades_sold_get_response"].get("trades", {}).get("trade", [])
        return trades
    return []
def get_order_detail(tid):
    """获取单笔订单详情（含物流、商品、买家信息）"""
    params = {
        "method": "taobao.trade.fullinfo.get",
        "app_key": APP_KEY,
        "session": ACCESS_TOKEN,
        "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
        "v": "2.0",
        "format": "json",
        "sign_method": "md5",
        "tid": tid,
        "fields": "tid,buyer_nick,created,pay_time,consign_time,end_time,status,payment,post_fee,receiver_name,receiver_mobile,receiver_state,receiver_city,receiver_district,receiver_address,buyer_message,orders,logistics_companies"
    }
    params["sign"] = top_sign(params, APP_SECRET)
    
    resp = requests.get(GATEWAY, params=params, timeout=15)
    data = resp.json()
    if "trade_fullinfo_get_response" in data:
        return data["trade_fullinfo_get_response"]["trade"]
    return None

def send_goods(tid, out_sid, company_code):
    """订单发货（回填物流单号）"""
    params = {
        "method": "taobao.logistics.online.send",
        "app_key": APP_KEY,
        "session": ACCESS_TOKEN,
        "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
        "v": "2.0",
        "format": "json",
        "sign_method": "md5",
        "tid": tid,
        "out_sid": out_sid,           # 物流单号
        "company_code": company_code   # 物流公司编码，如"yuantong"
    }
    params["sign"] = top_sign(params, APP_SECRET)
    
    resp = requests.get(GATEWAY, params=params, timeout=15)
    return resp.json()
def get_waybill(cp_code, trade_order_info):
    """获取电子面单"""
    params = {
        "method": "taobao.wlb.waybill.i.get",
        "app_key": APP_KEY,
        "session": ACCESS_TOKEN,
        "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
        "v": "2.0",
        "format": "json",
        "sign_method": "md5",
        "cp_code": cp_code,  # 物流公司CP码
        "trade_order_info": trade_order_info
    }
    params["sign"] = top_sign(params, APP_SECRET)
    
    resp = requests.get(GATEWAY, params=params, timeout=15)
    return resp.json()

def get_traderates(num_iid=None, rate_type="get", page_no=1):
    """获取评价列表"""
    params = {
        "method": "taobao.traderates.get",
        "app_key": APP_KEY,
        "session": ACCESS_TOKEN,
        "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
        "v": "2.0",
        "format": "json",
        "sign_method": "md5",
        "fields": "tid,oid,role,nick,result,created,item_title,item_price,content,reply",
        "rate_type": rate_type,  # "get"=收到的评价，"give"=给出的评价
        "page_no": page_no,
        "page_size": 20
    }
    if num_iid:
        params["num_iid"] = num_iid
    
    params["sign"] = top_sign(params, APP_SECRET)
    resp = requests.get(GATEWAY, params=params, timeout=15)
    data = resp.json()
    if "traderates_get_response" in data:
        return data["traderates_get_response"].get("trade_rates", {}).get("trade_rate", [])
    return []

┌─────────────────────────────────────────┐
│           商家自有系统（可选）             │
│    （ERP、WMS、BI 看板，可部署在本地）    │
└─────────────┬───────────────────────────┘
              │ HTTPS API
┌─────────────▼───────────────────────────┐
│         聚石塔 ECS（4核8G × 2台）        │
│  ┌─────────────┐  ┌─────────────────┐  │
│  │ 订单同步服务 │  │ 商品库存服务     │  │
│  │ (Python/Go) │  │ (Python/Go)     │  │
│  └─────────────┘  └─────────────────┘  │
│              ↓                          │
│  ┌─────────────────────────────────────┐│
│  │      聚石塔 RDS（订单数据存储）       ││
│  │   + Redis（缓存热点数据）            ││
│  └─────────────────────────────────────┘│
└─────────────────────────────────────────┘
              │ 内网调用
┌─────────────▼───────────────────────────┐
│        淘宝开放平台 API Gateway          │
└─────────────────────────────────────────┘