?

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

剛?cè)胄心菚?huì)，接了個(gè)給電商商家做競(jìng)品分析工具的活兒，核心就是調(diào)用淘寶評(píng)論 API 抓取競(jìng)品評(píng)論。本以為按文檔流程走，申請(qǐng)個(gè)接口權(quán)限，寫幾行代碼調(diào)用就行，結(jié)果現(xiàn)實(shí)給了我狠狠一擊。

申請(qǐng)接口權(quán)限就不順利，淘寶開放平臺(tái)對(duì)開發(fā)者資質(zhì)審核嚴(yán)格，我把公司營(yíng)業(yè)執(zhí)照、應(yīng)用使用場(chǎng)景說明改了又改，提交后還等了整整三個(gè)工作日才通過。好不容易拿到App Key和App Secret，調(diào)用接口時(shí)又卡在???????簽名驗(yàn)證上。官方文檔里簽名算法寫得晦澀難懂，參數(shù)排序、加密方式稍有差錯(cuò)，就返回40001簽名錯(cuò)誤。為了搞懂這算法，我對(duì)著文檔研究了兩天，還在 Stack Overflow 和國(guó)內(nèi)技術(shù)論壇瘋狂搜索，終于寫出了正確的簽名生成函數(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ā)送請(qǐng)求，結(jié)果又碰上接口調(diào)用頻率限制。當(dāng)時(shí)為了快速采集大量數(shù)據(jù)，沒控制好請(qǐng)求頻率，短時(shí)間內(nèi)發(fā)送太多請(qǐng)求，直接被淘寶封了 IP，還收到警告郵件。無奈之下，只能研究淘寶的限流規(guī)則，用???????漏桶算法寫了個(gè)頻率控制類：

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()
        # 補(bǔ)充令牌
        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，每秒補(bǔ)充20個(gè)令牌
if bucket.consume():
    # 調(diào)用API
    response = requests.get(api_url)
else:
    time.sleep(0.1)  # 等待令牌補(bǔ)充

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

import jieba
from textblob import TextBlob
def clean_and_analyze_comment(comment):
    # 分詞
    words = jieba.lcut(comment)
    # 去除停用詞（可根據(jù)需求擴(kuò)展停用詞表）
    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
# 示例調(diào)用
comment = "這款手機(jī)拍照效果超棒，就是電池續(xù)航不太給力"
sentiment, clean_comment = clean_and_analyze_comment(comment)
print(f"情感傾向: {sentiment}, 清洗后評(píng)論: {clean_comment}")

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

import requests
import time
# 假設(shè)last_query_time是上次查詢時(shí)間戳
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"錯(cuò)誤碼: {data['code']}, 消息: {data.get('msg', '未知錯(cuò)誤')}")
        return []
# 示例調(diào)用
last_query_time = 1690000000  # 假設(shè)初始時(shí)間戳
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']}, 評(píng)論: {comment['rate_content']}")

還有一次，項(xiàng)目要做一個(gè)商品評(píng)論可視化大屏，展示不同商品好評(píng)率、差評(píng)關(guān)鍵詞云圖等信息。為了保證數(shù)據(jù)實(shí)時(shí)性和準(zhǔn)確性，我用???????Flask框架搭建了一個(gè)后端服務(wù)，定時(shí)調(diào)用淘寶評(píng)論 API 更新數(shù)據(jù)，再通過???????Echarts在前端展示可視化圖表。這過程中，數(shù)據(jù)緩存又成了難題，頻繁調(diào)用 API 不僅浪費(fèi)資源，還可能觸發(fā)限流。于是引入???????Redis緩存，先從緩存里讀取數(shù)據(jù)，如果緩存沒有再調(diào)用 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:
        # 調(diào)用淘寶評(píng)論API獲取數(shù)據(jù)，此處省略具體調(diào)用代碼
        api_data = get_comments_from_api(num_iid)
        redis_client.setex(num_iid, 3600, str(api_data))  # 緩存1小時(shí)
        return jsonify(api_data)
if __name__ == "__main__":
    app.run(debug=True)

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


審核編輯 黃宇

?