?

在電商開發(fā)這行摸爬滾打了八年，和淘寶評論 API 打交道的過程，簡直就是一部 “血淚奮斗史”。從最初對接時被各種報錯折磨得夜不能寐，到現(xiàn)在能熟練用它搭建起各種實用功能，其中的酸甜苦辣，今天就來給大伙好好嘮嘮，順便分享些超實用的代碼干貨！

剛入行那會，接了個給電商商家做競品分析工具的活兒，核心就是調用淘寶評論 API 抓取競品評論。本以為按文檔流程走，申請個接口權限，寫幾行代碼調用就行，結果現(xiàn)實給了我狠狠一擊。

申請接口權限就不順利，淘寶開放平臺對開發(fā)者資質審核嚴格，我把公司營業(yè)執(zhí)照、應用使用場景說明改了又改，提交后還等了整整三個工作日才通過。好不容易拿到App Key和App Secret，調用接口時又卡在???????簽名驗證上。官方文檔里簽名算法寫得晦澀難懂，參數(shù)排序、加密方式稍有差錯，就返回40001簽名錯誤。為了搞懂這算法，我對著文檔研究了兩天，還在 Stack Overflow 和國內技術論壇瘋狂搜索，終于寫出了正確的簽名生成函數(shù)：

import hashlib
import hmac
import time
import urllib.parse
def generate_sign(params, app_secret):
    sorted_params = sorted(params.items(), key=lambda x: x[0])
    sign_str = app_secret
    for k, v in sorted_params:
        sign_str += f"{k}{v}"
    sign_str += app_secret
    return hmac.new(
        app_secret.encode(), sign_str.encode(), hashlib.sha256
    ).hexdigest().upper()

解決了簽名問題，滿心歡喜發(fā)送請求，結果又碰上接口調用頻率限制。當時為了快速采集大量數(shù)據(jù)，沒控制好請求頻率，短時間內發(fā)送太多請求，直接被淘寶封了 IP，還收到警告郵件。無奈之下，只能研究淘寶的限流規(guī)則，用???????漏桶算法寫了個頻率控制類：

import time
class LeakyBucket:
    def __init__(self, capacity, rate):
        self.capacity = capacity
        self.rate = rate
        self.tokens = capacity
        self.last_update = time.time()
    def consume(self, tokens=1):
        now = time.time()
        # 補充令牌
        self.tokens = min(
            self.capacity, self.tokens + (now - self.last_update) * self.rate
        )
        self.last_update = now
        if self.tokens >= tokens:
            self.tokens -= tokens
            return True
        return False
# 使用示例
bucket = LeakyBucket(capacity=100, rate=20)  # 容量100，每秒補充20個令牌
if bucket.consume():
    # 調用API
    response = requests.get(api_url)
else:
    time.sleep(0.1)  # 等待令牌補充

數(shù)據(jù)到手后，也不是一帆風順。淘寶評論數(shù)據(jù)格式復雜，有文字評論、圖片評論、評分、追評等多種類型，不同類型數(shù)據(jù)結構差異大。就拿文字評論來說，有的用戶會寫一大段話，標點符號、表情符號混用，要準確提取關鍵信息，得用???????自然語言處理技術。為了處理這些數(shù)據(jù)，我引入了???????jieba分詞庫和???????TextBlob情感分析庫，寫了個數(shù)據(jù)清洗和情感分析函數(shù)：

import jieba
from textblob import TextBlob
def clean_and_analyze_comment(comment):
    # 分詞
    words = jieba.lcut(comment)
    # 去除停用詞（可根據(jù)需求擴展停用詞表）
    stopwords = {"的", "了", "是", "在"}
    clean_words = [word for word in words if word not in stopwords]
    clean_comment = " ".join(clean_words)
    # 情感分析
    blob = TextBlob(clean_comment)
    sentiment = blob.sentiment.polarity
    return sentiment, clean_comment
# 示例調用
comment = "這款手機拍照效果超棒，就是電池續(xù)航不太給力"
sentiment, clean_comment = clean_and_analyze_comment(comment)
print(f"情感傾向: {sentiment}, 清洗后評論: {clean_comment}")

有一回，客戶要求實時監(jiān)控自家商品評論，一有新評論就及時推送通知。這可難不倒我，通過設置合適的時間間隔，不斷調用淘寶評論 API 獲取最新評論。但很快又出現(xiàn)新問題，每次獲取評論都要從第一頁開始查，效率極低。后來我發(fā)現(xiàn)接口可以通過評論時間戳來篩選，只獲取上次查詢時間之后的評論，大大提高了效率：

import requests
import time
# 假設last_query_time是上次查詢時間戳
def get_new_comments(last_query_time, app_key, app_secret, num_iid):
    params = {
        "method": "taobao.item.reviews.get",
        "app_key": app_key,
        "num_iid": num_iid,
        "start_date": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(last_query_time)),
        "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
        "format": "json",
        "v": "2.0",
    }
    params["sign"] = generate_sign(params, app_secret)
    response = requests.get(
        "https://eco.taobao.com/router/rest", params=params
    )
    data = response.json()
    if data["code"] == "200":
        new_comments = data["item_reviews_get_response"]["reviews"]
        return new_comments
    else:
        print(f"錯誤碼: {data['code']}, 消息: {data.get('msg', '未知錯誤')}")
        return []
# 示例調用
last_query_time = 1690000000  # 假設初始時間戳
app_key = "your_app_key"
app_secret = "your_app_secret"
num_iid = "123456789"  # 商品ID
new_comments = get_new_comments(last_query_time, app_key, app_secret, num_iid)
if new_comments:
    for comment in new_comments:
        print(f"用戶: {comment['user_nick']}, 評論: {comment['rate_content']}")

還有一次，項目要做一個商品評論可視化大屏，展示不同商品好評率、差評關鍵詞云圖等信息。為了保證數(shù)據(jù)實時性和準確性，我用???????Flask框架搭建了一個后端服務，定時調用淘寶評論 API 更新數(shù)據(jù)，再通過???????Echarts在前端展示可視化圖表。這過程中，數(shù)據(jù)緩存又成了難題，頻繁調用 API 不僅浪費資源，還可能觸發(fā)限流。于是引入???????Redis緩存，先從緩存里讀取數(shù)據(jù)，如果緩存沒有再調用 API 獲取，獲取后存入緩存：

import redis
from flask import Flask, jsonify
app = Flask(__name__)
redis_client = redis.Redis(host="localhost", port=6379, db=0)
@app.route("/product_comments/", methods=["GET"])
def get_product_comments(num_iid):
    cached_data = redis_client.get(num_iid)
    if cached_data:
        return jsonify(eval(cached_data.decode("utf-8")))
    else:
        # 調用淘寶評論API獲取數(shù)據(jù)，此處省略具體調用代碼
        api_data = get_comments_from_api(num_iid)
        redis_client.setex(num_iid, 3600, str(api_data))  # 緩存1小時
        return jsonify(api_data)
if __name__ == "__main__":
    app.run(debug=True)

這些年，靠著不斷踩坑、填坑，在???????淘寶評論 API開發(fā)上也算積累了些經驗。


審核編輯 黃宇

?