?

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

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

python

運(yùn)行

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寸新佩碗/奶白釉面包貓 無(wú)規(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" ],

解決了簽名問(wèn)題，新的挑戰(zhàn)又接踵而至。小紅書對(duì) API 調(diào)用頻率限制嚴(yán)格，而且不同接口的限流規(guī)則還不一樣。為了避免觸發(fā)封禁，我用 Python 的asyncio庫(kù)結(jié)合隊(duì)列實(shí)現(xiàn)了異步請(qǐng)求和限流控制。比如設(shè)置每秒最多發(fā)起 10 次請(qǐng)求，超出部分自動(dòng)排隊(duì)：

python

運(yùn)行

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ù)雜。商品信息、用戶評(píng)價(jià)、關(guān)聯(lián)筆記、促銷活動(dòng)等數(shù)據(jù)嵌套多層，光是解析 JSON 就要費(fèi)不少功夫。而且小紅書會(huì)不定期調(diào)整接口返回字段，有次突然發(fā)現(xiàn)原本正常的 “商品規(guī)格” 字段消失了，害得我緊急排查了半天。后來(lái)學(xué)聰明了，寫了個(gè)通用的數(shù)據(jù)解析函數(shù)，還加了字段缺失的容錯(cuò)處理：

python

運(yùn)行

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返回?cái)?shù)據(jù)
parsed_data = parse_product_data(raw_response)

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

python

運(yùn)行

# 處理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ā)上踩過(guò)的坑，都成了寶貴的經(jīng)驗(yàn)財(cái)富。每一次攻克難題，都像是解鎖了新技能。

?審核編輯 黃宇