?

在電商開發的戰場上摸爬滾打多年，每一次對接新平臺的 API 都是一場硬仗。要說最讓人印象深刻的，小紅書商品詳情 API 接口的對接經歷絕對能排上前三。從申請權限時的層層關卡，到數據抓取時的各種 “幺蛾子”，今天就掰開了揉碎了，把這段實戰經歷和能用得上的代碼技巧全分享出來！

剛開始接觸小紅書商品詳情 API，天真地以為和其他平臺差不多，按文檔操作就能輕松拿到數據。現實卻給我潑了一盆冷水 —— 小紅書的開發者平臺審核堪稱 “魔鬼級別”，不僅要詳細說明應用用途、使用場景，還得提供完整的業務規劃。光是提交申請材料就反復修改了四五次，等了整整一周才收到通過通知。好不容易拿到了client_id和client_secret，簽名算法又成了攔路虎。小紅書采用的HMAC-SHA256加密方式，對參數順序、編碼格式要求極為嚴格，稍有差錯就返回401 Unauthorized。那段時間對著文檔和官方示例代碼反復研究，終于啃出了簽名生成函數：

python

運行

import hashlib
import hmac
import time
import urllib.parse


def generate_signature(params, client_secret):
    sorted_params = sorted(params.items(), key=lambda x: x[0])
    query_str = ""
    for k, v in sorted_params:
        if isinstance(v, list):
            v = ",".join(map(str, v))
        query_str += f"{k}{v}"
    sign_str = f"{client_secret}{query_str}{client_secret}"
    return hmac.new(
        client_secret.encode('utf-8'),
        sign_str.encode('utf-8'),
        hashlib.sha256
    ).hexdigest().upper()


# 使用示例
api_params = {
    "method": "xhs.product.detail.get",
    "client_id": "your_client_id",
    "product_id": "123456789",
    "timestamp": str(int(time.time()))
}
signed_params = generate_signature(api_params, "your_client_secret")

python返回結果

Result Object:
---------------------------------------
{
	"item": {
		"num_iid": "685d0fe3c16548001691655c",
		"title": "奶白釉面包貓卡通可愛陶瓷餐具碗碟家用一二人食碗盤飯盤子釉下彩 · 4.75寸新佩碗/奶白釉面包貓 無規格",
		"desc_short": "",
		"price": 11,
		"total_price": "",
		"suggestive_price": "",
		"orginal_price": 11,
		"nick": "喵植雜貨鋪的店",
		"num": 200,
		"min_num": 0,
		"detail_url": "https://www.xiaohongshu.com/goods-detail/685d0fe3c16548001691655c",
		"pic_url": "http://mall-i4.xhscdn.com/material_space/f6fbc3bd-443b-4f75-8909-fb5484dd3573?imageView2/2/w/1080/format/webp/q/80",
		"brand": null,
		"brandId": "",
		"rootCatId": "",
		"cid": "",
		"desc": "

", "desc_img": [ "https://file1.elecfans.com//web3/M00/2B/7B/wKgZO2isG4WAMPzMAAEm5rCqx28321.jpg", "https://file1.elecfans.com/web3/M00/2B/69/wKgZPGisGuaAPYbFAAACCUoruo8524.jpg", "https://file1.elecfans.com//web3/M00/2B/7B/wKgZO2isG4WAKsBJAAEJtD_LhPA492.jpg", "https://file1.elecfans.com//web3/M00/2B/7B/wKgZO2isG4WAEq4yAAEqphMFR0Q944.jpg", "https://file1.elecfans.com//web3/M00/2B/7B/wKgZO2isG4WAZN9bAADD0K_mB48577.jpg", "https://file1.elecfans.com/web3/M00/2B/7B/wKgZO2isGumAehmpAAFIujkshVM029.jpg", "https://file1.elecfans.com/web3/M00/2B/69/wKgZPGisGuqAKQqbAACbKgy43JM420.jpg", "https://file1.elecfans.com//web3/M00/2B/7B/wKgZO2isG4WAWdgEAADVfIZYMts530.jpg", "https://file1.elecfans.com/web3/M00/2B/7B/wKgZO2isGuuALxojAAClLj4Tf1g651.jpg", "https://file1.elecfans.com//web3/M00/2B/7B/wKgZO2isG4WAEgOlAAEa0DFkQl4588.jpg", "https://file1.elecfans.com/web3/M00/2B/69/wKgZPGisGu6AXfL6AACdkVCaOuU203.jpg", "https://file1.elecfans.com//web3/M00/2B/7B/wKgZO2isG4WAVIU6AADNCMR78Sc685.jpg", "https://file1.elecfans.com/web3/M00/2B/69/wKgZPGisGu-APiIQAAEfXSap07E396.jpg", "https://file1.elecfans.com//web3/M00/2B/7B/wKgZO2isG4WAMQ4qAACZ4hi2ARc880.jpg", "https://file1.elecfans.com//web3/M00/2B/7B/wKgZO2isG4WARxjNAACkHCZWawE467.jpg", "https://file1.elecfans.com/web3/M00/2B/7B/wKgZO2isGvKAZtJNAACpbgm7md0264.jpg", "https://file1.elecfans.com/web3/M00/2B/7B/wKgZO2isGvKAOUCTAAE8zK31p84679.jpg", "https://file1.elecfans.com/web3/M00/2B/69/wKgZPGisGvOAHp1fAACEGOjiWWg891.jpg", "https://file1.elecfans.com/web3/M00/2B/7B/wKgZO2isGvSAIRjQAAC_Af5YwY4776.jpg", "https://file1.elecfans.com/web3/M00/2B/69/wKgZPGisGvWARpd5AADPgSo_rHk381.jpg", "https://file1.elecfans.com/web3/M00/2B/7B/wKgZO2isGvaAa8WPAACys93Q9ts329.jpg", "https://file1.elecfans.com/web3/M00/2B/7B/wKgZO2isGvaAEiWaAACm4wmnEvM461.jpg" ],

解決了簽名問題，新的挑戰又接踵而至。小紅書對 API 調用頻率限制嚴格，而且不同接口的限流規則還不一樣。為了避免觸發封禁，我用 Python 的asyncio庫結合隊列實現了異步請求和限流控制。比如設置每秒最多發起 10 次請求，超出部分自動排隊：

python

運行

import asyncio
import ???????aiohttp


class RateLimiter:
    def __init__(self, rate_limit):
        self.semaphore = asyncio.Semaphore(rate_limit)

    async def limit(self):
        await self.semaphore.acquire()
        try:
            yield
        finally:
            self.semaphore.release()


async def fetch_product_detail(session, product_id, signed_params):
    async with RateLimiter(10).limit():
        url = "https://open.xiaohongshu.com/api/rest"
        params = {**signed_params, "product_id": product_id}
        async with session.get(url, params=params) as response:
            return await response.json()


async def main(product_ids, signed_params):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_product_detail(session, pid, signed_params) for pid in product_ids]
        results = await asyncio.gather(*tasks)
        return results


# 示例調用
product_ids = ["123456789", "987654321"]
signed_params = {...}  # 已簽名參數
asyncio.run(main(product_ids, signed_params))

拿到數據后，才發現小紅書商品詳情的結構遠比想象中復雜。商品信息、用戶評價、關聯筆記、促銷活動等數據嵌套多層，光是解析 JSON 就要費不少功夫。而且小紅書會不定期調整接口返回字段，有次突然發現原本正常的 “商品規格” 字段消失了，害得我緊急排查了半天。后來學聰明了，寫了個通用的數據解析函數，還加了字段缺失的容錯處理：

python

運行

def parse_product_data(raw_data):
    try:
        return {
            "product_name": raw_data.get("product_info", {}).get("name", ""),
            "price": raw_data.get("product_info", {}).get("price", 0),
            "sales_count": raw_data.get("product_info", {}).get("sales_count", 0),
            "user_rating": raw_data.get("evaluation_info", {}).get("score", 0),
            "main_image": raw_data.get("product_info", {}).get("main_image", "")
        }
    except KeyError as e:
        print(f"字段缺失: {e}")
        return {}


# 示例調用
raw_response = {...}  # API返回數據
parsed_data = parse_product_data(raw_response)

在開發一個小紅書商品比價工具時，需要實時監控商品價格變動。最開始我采用輪詢的方式定時調用 API，結果不僅浪費資源，還容易觸發限流。后來改用???????Webhook訂閱的方式，當商品信息更新時，小紅書主動推送通知，大大提高了效率。雖然接入 Webhook 的過程也不輕松，要處理簽名驗證、消息加密、重復消息過濾等問題，但最終實現的效果堪稱絲滑。

python

運行

# 處理Webhook消息示例
import hmac
import hashlib
from flask import Flask, request, abort

app = Flask(__name__)
SECRET_KEY = "your_webhook_secret"


@app.route('/webhook', methods=['POST'])
def handle_webhook():
    signature = request.headers.get('X-Xhs-Signature')
    if not signature:
        abort(401)

    data = request.data
    local_signature = hmac.new(
        SECRET_KEY.encode('utf-8'),
        data,
        hashlib.sha256
    ).hexdigest()

    if signature != local_signature:
        abort(401)

    # 處理接收到的商品更新數據
    webhook_data = request.json
    # 業務邏輯處理...
    return "OK", 200


if __name__ == '__main__':
    app.run(debug=True)

這些年在小紅書 API 開發上踩過的坑，都成了寶貴的經驗財富。每一次攻克難題，都像是解鎖了新技能。

?審核編輯 黃宇