小红书商品详情签名算法Python

?

在电商开发的战场上摸爬滚打多年，每一次对接新平台的 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": "//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 开发上踩过的坑，都成了宝贵的经验财富。每一次攻克难题，都像是解锁了新技能。

?审核编辑 黄宇