?

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

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

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返回結(jié)果

Result Object:
---------------------------------------
{
	"item": {
		"num_iid": "685d0fe3c16548001691655c",
		"title": "奶白釉面包貓卡通可愛陶瓷餐具碗碟家用一二人食碗盤飯盤子釉下彩 · 4.75寸新佩碗/奶白釉面包貓 無規(guī)格",
		"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" ],

解決了簽名問題，新的挑戰(zhàn)又接踵而至。小紅書對 API 調(diào)用頻率限制嚴(yán)格，而且不同接口的限流規(guī)則還不一樣。為了避免觸發(fā)封禁，我用 Python 的asyncio庫結(jié)合隊列實現(xiàn)了異步請求和限流控制。比如設(shè)置每秒最多發(fā)起 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


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

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

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 {}


# 示例調(diào)用
raw_response = {...}  # API返回數(shù)據(jù)
parsed_data = parse_product_data(raw_response)

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

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)

    # 處理接收到的商品更新數(shù)據(jù)
    webhook_data = request.json
    # 業(yè)務(wù)邏輯處理...
    return "OK", 200


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

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

?審核編輯 黃宇