小红书作为 “内容 + 电商” 双驱动平台，汇聚了海量美妆、穿搭、家居等垂直领域商品，item_search 接口（非官方原生接口，由第三方合规服务商基于平台规则封装）是通过关键词批量获取商品列表的核心工具。该接口支持按关键词、价格区间、销量、评分等多维度筛选，返回商品基础信息、交易数据、店铺信息等关键内容，广泛应用于电商选品、竞品监测、市场分析、导购平台搭建等场景。
本攻略结合小红书电商生态特性（内容驱动消费、商品与笔记强关联、反爬规则严格），从接口认知、前置准备、实操落地、调试优化到进阶技巧，全方位拆解对接流程，兼顾入门易用性与生产级稳定性，帮助开发者高效完成接口对接。
一、接口核心认知：先明确 “能做什么”“适配什么场景”

1. 接口定位与核心价值
   核心功能：输入关键词（支持多关键词组合），筛选小红书商品列表，支持按价格区间、销量、评分、店铺类型、商品标签等条件过滤，返回分页商品数据（含商品 ID、标题、价格、销量、店铺信息等），可联动 item_get 接口获取商品详情；
   平台特性：
   数据覆盖：收录小红书 90% 以上公开电商商品，涵盖美妆、护肤、穿搭、家居、食品等全品类，新商品收录延迟 5-15 分钟；
   筛选能力：支持精确 / 模糊匹配、价格区间（最低价 - 最高价）、销量排序、评分筛选、店铺类型（品牌店 / 普通店 / 旗舰店）、商品标签（如 “正品保障”“运费险”）；
   关联数据：部分服务商支持同步返回商品关联笔记数、笔记关键词，适配 “内容 + 商品” 联动分析场景；
   无需官方授权：依赖第三方服务商合规采集能力，无需小红书商家 / 开放平台账号，仅需服务商提供的密钥；
   典型应用：
   电商选品：按垂直领域关键词（如 “平价粉底液”“ins 风家居”）批量采集爆款商品，分析用户偏好；
   竞品监测：输入竞品品牌 / 产品关键词，追踪竞品商品价格、销量、评分变化；
   市场分析：按行业关键词（如 “国货美妆”“露营装备”）采集商品数据，统计市场规模、价格分布、热门卖点；
   导购平台：合规前提下整合垂直领域优质商品，搭建小红书商品导购聚合平台；
   舆情监测：通过商品关键词 + 负面评价关键词，监测商品用户反馈与投诉情况。
2. 核心参数与返回字段（关键词搜索场景适配版）
   （1）请求参数（必填 + 可选，按优先级排序）
   参数名称 类型 是否必填 说明 应用场景示例
   appkey string 是 接口调用密钥，由合规服务商分配（注册并通过审核后获取） xhs_search_abc123
   secret string 是 签名密钥，用于请求合法性校验（不可泄露，定期轮换） xhs_search_def456
   keyword string 是 搜索关键词（支持多关键词组合，AND 用空格分隔，OR 用 分隔） 平价粉底液 持妆（AND）、露营装备 户外帐篷（OR）
   price_min float 否 最低价格（单位：元），默认无下限 50（筛选 50 元以上商品）
   price_max float 否 最高价格（单位：元），默认无上限 200（筛选 200 元以下商品）
   sort_type string 否 排序方式，默认 “relevance”（相关度优先） relevance（相关度）、sales（销量倒序）、price_asc（价格升序）、price_desc（价格降序）、score（评分倒序）、newest（最新上架）
   shop_type string 否 店铺类型筛选，默认 “all”（全部） all（全部）、brand（品牌店）、flagship（旗舰店）、ordinary（普通店）
   tag string 否 商品标签筛选（多个标签用逗号分隔） 正品保障，运费险，7 天无理由
   page_no int 否 页码，默认 1（支持分页，最大页码通常≤100，不同服务商限制不同） 1、2、3
   page_size int 否 每页商品条数，默认 20（最大支持 50 条 / 页，企业版可申请 100 条） 20、30、50
   match_type string 否 关键词匹配类型，默认 “fuzzy”（模糊匹配） fuzzy（模糊匹配，商品标题 / 标签含关键词）、exact（精确匹配，商品标题完整包含关键词）
   need_note int 否 是否返回商品关联笔记数，默认 0（不返回） 1（返回笔记数、热门笔记关键词）、0（仅返回商品信息）
   timestamp long 是 请求时间戳（毫秒级，有效期 5 分钟，避免重复请求） 1735689600000
   sign string 是 签名值（按服务商规则加密生成，核心校验项） 32 位 MD5 大写串（如 A8F7C3D2E1B0967453120FEDCBA9876）
   注：
   价格筛选规则：仅当price_min和price_max同时传入时，按区间筛选；仅传其一则仅限制单一下限 / 上限；
   关键词组合规则：空格表示 “同时包含”，|表示 “包含任一”，如 “平价 粉底液 | 气垫” 表示 “包含平价和粉底液，或包含气垫”；
   部分服务商支持高级筛选参数（如 “商品产地”“发货地”），需以实际文档为准。
   （2）返回核心字段（按业务场景分类，关键词搜索重点标注）
   商品基础信息：商品 ID（num_iid，小红书唯一标识，用于item_get接口）、商品标题（title，含商品卖点、规格）、商品主图（pic_url，高清主图 URL）、商品图集（pic_list，多个详情图 URL 列表）、商品标签（tags，如 “正品保障”“运费险”）、商品规格摘要（spec_abstract，如 “30ml/50ml”）、上架时间（pub_time，精确到秒）；
   交易数据：原价（price，单位：元）、优惠价（promotion_price，单位：元）、销量（sales，累计销量）、库存（stock，部分服务商支持）、价格区间（price_range，多规格商品价格范围）、是否有优惠（has_promotion，true/false）、优惠活动（promotion_info，如 “满减”“折扣”）；
   店铺信息：店铺名称（shop_name）、店铺 ID（shop_id）、店铺类型（shop_type，如 “旗舰店”）、店铺评分（shop_score，综合评分 0-5 分）、店铺粉丝数（shop_fans，部分服务商支持）、是否官方认证（is_verified，true/false）；
   评价与内容数据：商品评分（item_score，0-5 分）、评价数（comment_count）、好评率（positive_rate，部分服务商支持）、关联笔记数（note_count，need_note=1时返回）、热门笔记关键词（note_keywords，need_note=1时返回）；
   扩展信息：是否包邮（free_shipping，true/false）、发货地（shipping_place，部分服务商支持）、售后保障（after_sales，如 “7 天无理由”）、相关度评分（relevance_score，0-100 分）、是否违规（is_illegal，true/false，违规 / 下架商品标记）。
3. 接口限制与注意事项
   调用频率：普通用户 10-30 次 / 分钟，企业用户 50-200 次 / 分钟（不同服务商限制不同，需提前确认）；
   数据缓存：搜索结果缓存 30 分钟 - 1 小时，热门关键词缓存可能缩短至 10 分钟，企业版支持refresh=1强制刷新（需申请权限）；
   权限差异：
   权限类型 核心差异 适用场景
   普通用户 基础筛选（价格、销量、店铺类型）、基础字段、单页最大 20 条、低调用频率 个人选品、小型分析
   企业用户 高级筛选（标签、产地、发货地）、完整字段（库存、好评率、笔记数据）、单页最大 100 条、高调用频率、强制刷新 商业选品、竞品监测、市场分析
   内容限制：
   已下架 / 违规商品、隐私店铺商品、未被服务商收录的商品无法返回；
   部分品牌店商品可能因版权限制，仅返回基础信息（标题、价格），详情需通过item_get接口进一步获取；
   合规要求：
   必须使用合规服务商接口，禁止自行爬取小红书商品数据（违反《反不正当竞争法》及小红书平台规则）；
   数据仅可用于自身合规业务，禁止商业化售卖、恶意传播或侵犯品牌 / 店铺知识产权；
   引用数据时需标注 “数据来源：小红书”，商品信息二次使用需获得相关品牌 / 店铺授权；
   不得用于批量抓取用户隐私、敏感信息或从事违法活动（如刷单、恶意比价）。
   二、对接前准备：3 步搞定前置条件
4. 获取接口密钥（核心步骤）
   小红书关键词商品列表接口无官方公开申请渠道，需通过合规第三方服务商（如聚合数据、极速数据、APISpace、阿里云 API 市场等）获取，步骤如下：
   选择合规服务商：优先选择有资质、口碑好、支持企业认证的服务商，要求提供《数据合规承诺书》《服务协议》，避免使用非法爬虫接口（存在账号封禁、法律风险）；
   注册与认证：
   个人用户：注册账号，完成实名认证（提供身份证信息），获取基础接口权限；
   企业用户：注册账号，完成企业认证（提供营业执照、法人信息、业务用途说明），获取高级接口权限；
   申请接口权限：搜索 “小红书商品关键词搜索接口” 或 “小红书 item_search 接口”，选择对应接口，提交申请（需明确业务用途，如 “电商选品分析”“竞品监测系统开发”）；
   获取密钥：审核通过后，在服务商后台 “应用管理” 中获取 appkey 和 secret（部分服务商按调用次数收费，需提前充值或购买套餐）；
   查阅文档：下载服务商提供的接口文档，重点确认：
   参数格式（如sort_type取值、关键词组合规则）；
   签名生成规则（MD5/SHA256、参数排序要求）；
   返回字段含义（不同服务商字段命名可能差异）；
   调用频率、分页上限、权限限制等细节。
5. 技术环境准备
   （1）支持语言与协议
   接口采用 HTTPS 协议，支持所有主流开发语言：Python、Java、PHP、Go、Node.js 等，无框架限制，推荐使用 Python（数据处理、批量请求效率高，适配关键词搜索的多条件筛选场景）。
   （2）必备工具与依赖
   调试工具：
   Postman（快速验证接口可用性，批量测试参数组合）；
   curl（命令行调试，快速排查网络问题）；
   浏览器开发者工具（分析小红书商品搜索结果页结构，辅助字段映射）；
   在线 MD5/SHA256 工具（验证签名生成正确性）；
   开发依赖：
   网络请求：requests（Python）、OkHttp（Java）、axios（Node.js）；
   加密工具：语言内置 MD5/SHA256 库（签名生成用，按服务商规则选择）；
   数据处理：json（解析响应数据）、pandas（批量整理商品列表）、datetime（时间格式转换）；
   辅助工具：
   日志库（如 logging，记录请求 / 响应 / 错误，便于追溯）；
   Redis（缓存搜索结果，减少重复请求）；
   定时任务框架（如 APScheduler，实时关键词监测）；
   异常监控工具（如 Sentry，生产级报错追踪）；
   多线程 / 异步库（如 aiohttp、concurrent.futures，提升批量分页请求效率）。
6. 业务需求梳理
   关键词策略：明确核心关键词及组合规则（如选品需 “垂直品类 + 卖点” 组合，竞品监测需 “品牌名 + 产品名” 组合），避免关键词过于宽泛导致结果冗余；
   筛选条件定义：
   价格区间：按目标用户群体定位（如平价商品筛选 “price_min=30&price_max=100”，高端商品筛选 “price_min=500”）；
   店铺类型：品牌合作调研选 “brand/flagship”，平价选品选 “ordinary”；
   排序方式：爆款分析选 “sales”（销量倒序），新品调研选 “newest”（最新上架）；
   字段筛选：仅保留业务必需字段（如选品需 “title、price、sales、shop_name、item_score”，竞品监测需 “promotion_price、promotion_info、shop_score”），减少数据传输量；
   分页与批量需求：确认是否需要获取全部搜索结果（如市场分析），或仅需前 N 页（如爆款选品），避免无效分页请求；
   异常场景预设：关键词无匹配结果、频率超限、网络波动、商品已下架等场景，需设计降级方案（如返回 “无相关商品” 提示、缓存历史搜索结果）。
   三、实操步骤：从调试到落地（Python 示例）
   步骤 1：理解请求流程
   拼接关键词、筛选条件等核心参数；
   按服务商规则生成签名（sign），确保请求合法性；
   发送 POST/GET 请求（多数服务商推荐 POST，参数传输更安全；部分支持 GET，参数拼接在 URL 中）；
   接收响应数据，解析 JSON 格式结果，提取商品列表；
   数据标准化处理（时间格式统一、价格单位统一、字段映射）；
   按需分页请求（遍历所有页码，获取完整结果）；
   异常处理（签名错误、权限不足、无结果等）。
   步骤 2：签名生成规则（关键！不同服务商可能差异，以实际文档为准）
   多数合规服务商采用以下通用签名规则（若有差异，需按服务商文档调整）：
   按参数名ASCII 升序排序所有请求参数（不含sign字段）；
   将排序后的参数拼接为 “key1=value1&key2=value2&...” 格式（中文 / 特殊字符需 URL 编码）；
   在拼接字符串末尾追加 &secret=你的secret；
   对拼接后的字符串进行MD5 加密（32 位大写） 或 SHA256 加密（64 位大写），结果即为sign。
   签名示例（参数排序与拼接）
   假设请求参数：
   appkey=xhs_search_abc123
   keyword = 平价粉底液 持妆
   price_min=50
   price_max=200
   sort_type=sales
   shop_type=all
   page_no=1
   page_size=20
   timestamp=1735689600000
   secret=xhs_search_def456
   排序后参数：appkey、keyword、page_no、page_size、price_max、price_min、shop_type、sort_type、timestamp；
   拼接字符串（中文已 URL 编码）：appkey=xhs_search_abc123&keyword=%E5%B9%B3%E4%BB%B7%E7%B2%89%E4%B8%B9%E6%B6%B2%20%E6%8C%81%E5%A6%86&page_no=1&page_size=20&price_max=200&price_min=50&shop_type=all&sort_type=sales×tamp=1735689600000&secret=xhs_search_def456；
   MD5 加密后 sign：A8F7C3D2E1B0967453120FEDCBA9876543210ABCDEF（32 位大写）。
   步骤 3：完整代码实现（Python）
   （1）依赖安装
   bash
   运行
   pip install requests pandas aiohttp apscheduler # requests：网络请求；pandas：数据整理；aiohttp：异步请求；APScheduler：定时任务
   （2）完整代码（含签名生成、接口调用、批量分页、数据保存）
   ```jsimport requests
   import hashlib
   import time
   import json
   import pandas as pd
   from urllib.parse import urlencode, unquote
   from typing import Dict, List, Optional
   import logging
   from apscheduler.schedulers.blocking import BlockingScheduler
   import aiohttp
   import asyncio

配置日志（记录接口调用、错误信息，便于合规追溯）

logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s",
handlers=[logging.FileHandler("xiaohongshu_item_search.log"), logging.StreamHandler()]
)

接口核心配置（替换为服务商提供的appkey、secret、API_URL）

APP_KEY = "你的appkey"
SECRET = "你的secret"
API_URL = "https://api.xiaohongshu-search.com/item_search" # 服务商接口地址（以实际为准）
SAVE_PATH = "小红书商品列表.xlsx" # 数据保存路径

def generate_sign(params: Dict) -> str:
"""生成接口签名（通用MD5版，若服务商要求SHA256可调整）"""

# 1. 按参数名ASCII升序排序（排除sign字段）
sorted_params = sorted(params.items(), key=lambda x: x[0])
# 2. 拼接参数字符串（urlencode自动处理中文、特殊字符）
param_str = urlencode(sorted_params, encoding="utf-8") + f"&secret={SECRET}"
# 3. MD5加密（32位大写）
md5 = hashlib.md5()
md5.update(param_str.encode("utf-8"))
return md5.hexdigest().upper()

def standardize_data(raw_item: Dict) -> Dict:
"""标准化商品数据（统一字段格式，适配业务展示/分析）"""

# 解析交易数据
promotion_info = raw_item.get("promotion_info", {}) or {}
# 标准化时间格式
pub_time = raw_item.get("pub_time", 0)
pub_time_str = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(pub_time/1000)) if pub_time else ""
# 处理价格（兼容多规格价格区间）
price = raw_item.get("price", 0)
promotion_price = raw_item.get("promotion_price", price)
price_range = raw_item.get("price_range", f"{price}-{price}") if price else "0"

return {
    "请求时间": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
    "关键词": raw_item.get("keyword", ""),
    "商品ID": raw_item.get("num_iid", ""),
    "商品标题": raw_item.get("title", ""),
    "商品主图": raw_item.get("pic_url", ""),
    "商品标签": ",".join(raw_item.get("tags", [])) if raw_item.get("tags") else "",
    "原价（元）": price,
    "优惠价（元）": promotion_price,
    "价格区间（元）": price_range,
    "销量": raw_item.get("sales", 0),
    "库存": raw_item.get("stock", "未知"),
    "商品评分": raw_item.get("item_score", 0),
    "评价数": raw_item.get("comment_count", 0),
    "好评率（%）": raw_item.get("positive_rate", 0),
    "店铺名称": raw_item.get("shop_name", ""),
    "店铺类型": raw_item.get("shop_type", ""),
    "店铺评分": raw_item.get("shop_score", 0),
    "上架时间": pub_time_str,
    "是否包邮": "是" if raw_item.get("free_shipping", False) else "否",
    "售后保障": raw_item.get("after_sales", ""),
    "优惠活动": promotion_info.get("activity", ""),
    "关联笔记数": raw_item.get("note_count", 0),
    "热门笔记关键词": ",".join(raw_item.get("note_keywords", [])) if raw_item.get("note_keywords") else "",
    "相关度评分": raw_item.get("relevance_score", 0),
    "商品状态": "正常" if not raw_item.get("is_illegal", False) else "已下架/违规"
}

def get_item_list_sync(
keyword: str,
price_min: Optional[float] = None,
price_max: Optional[float] = None,
sort_type: str = "relevance",
shop_type: str = "all",
tag: Optional[str] = None,
page_no: int = 1,
page_size: int = 20,
match_type: str = "fuzzy",
need_note: int = 0
) -> Dict:
"""
同步调用item_search接口，获取单页小红书商品列表
:param keyword: 搜索关键词（支持多关键词组合）
:param price_min: 最低价格
:param price_max: 最高价格
:param sort_type: 排序方式
:param shop_type: 店铺类型
:param tag: 商品标签（多个用逗号分隔）
:param page_no: 页码
:param page_size: 每页条数
:param match_type: 关键词匹配类型
:param need_note: 是否返回关联笔记数据（1=是，0=否）
:return: 标准化后的商品列表结果
"""

# 1. 构建基础参数（必填项）
params = {
    "appkey": APP_KEY,
    "keyword": keyword,
    "sort_type": sort_type,
    "shop_type": shop_type,
    "page_no": page_no,
    "page_size": page_size,
    "match_type": match_type,
    "need_note": need_note,
    "timestamp": int(time.time() * 1000)
}

# 2. 补充可选参数
if price_min is not None:
    params["price_min"] = price_min
if price_max is not None:
    params["price_max"] = price_max
if tag:
    params["tag"] = tag

# 3. 生成签名
params["sign"] = generate_sign(params)

try:
    # 4. 发送POST请求（HTTPS协议，超时15秒）
    response = requests.post(
        url=API_URL,
        data=json.dumps(params),
        headers={"Content-Type": "application/json"},
        timeout=15,
        verify=True
    )
    response.raise_for_status()  # 抛出HTTP请求异常（如404、500）
    result = response.json()

    # 5. 处理响应（不同服务商返回格式可能不同，需按实际文档调整）
    # 假设服务商返回格式：{"code":200,"msg":"success","data":{"item_list":[],"total":0,"page_total":1}}
    if result.get("code") == 200:
        raw_data = result.get("data", {})
        item_list = raw_data.get("item_list", [])
        total = raw_data.get("total", 0)  # 匹配商品总条数
        page_total = raw_data.get("page_total", 1)  # 总页码

        # 标准化商品数据
        standard_items = [standardize_data(item) for item in item_list]
        for item in standard_items:
            item["关键词"] = keyword  # 补充关键词字段，便于多关键词筛选时区分

        return {
            "success": True,
            "data": standard_items,
            "total": total,
            "page_no": page_no,
            "page_total": page_total,
            "error_msg": ""
        }
    else:
        error_msg = result.get("msg", "接口调用失败")
        error_code = result.get("code", -2)
        logging.error(f"接口返回错误：code={error_code}, msg={error_msg}（关键词：{keyword}，页码：{page_no}）")
        return {
            "success": False,
            "data": [],
            "total": 0,
            "page_no": page_no,
            "page_total": 0,
            "error_code": error_code,
            "error_msg": error_msg
        }
except requests.exceptions.RequestException as e:
    logging.error(f"网络异常（关键词：{keyword}，页码：{page_no}）：{str(e)}")
    return {
        "success": False,
        "data": [],
        "total": 0,
        "page_no": page_no,
        "page_total": 0,
        "error_code": -3,
        "error_msg": f"网络异常：{str(e)}"
    }
except Exception as e:
    logging.error(f"数据处理异常（关键词：{keyword}，页码：{page_no}）：{str(e)}")
    return {
        "success": False,
        "data": [],
        "total": 0,
        "page_no": page_no,
        "page_total": 0,
        "error_code": -4,
        "error_msg": f"处理异常：{str(e)}"
    }

async def get_item_list_async(session, keyword: str, **kwargs) -> Dict:
"""异步调用item_search接口，提升批量请求效率"""
page_no = kwargs.get("page_no", 1)

# 构建参数（逻辑同同步方法）
params = {
    "appkey": APP_KEY,
    "keyword": keyword,
    "sort_type": kwargs.get("sort_type", "relevance"),
    "shop_type": kwargs.get("shop_type", "all"),
    "page_no": page_no,
    "page_size": kwargs.get("page_size", 20),
    "match_type": kwargs.get("match_type", "fuzzy"),
    "need_note": kwargs.get("need_note", 0),
    "timestamp": int(time.time() * 1000)
}
if kwargs.get("price_min") is not None:
    params["price_min"] = kwargs["price_min"]
if kwargs.get("price_max") is not None:
    params["price_max"] = kwargs["price_max"]
if kwargs.get("tag"):
    params["tag"] = kwargs["tag"]
# 生成签名
params["sign"] = generate_sign(params)

try:
    async with session.post(
        API_URL,
        json=params,
        headers={"Content-Type": "application/json"},
        timeout=15
    ) as response:
        response.raise_for_status()
        result = await response.json()
        if result.get("code") == 200:
            raw_data = result.get("data", {})
            item_list = raw_data.get("item_list", [])
            standard_items = [standardize_data(item) for item in item_list]
            for item in standard_items:
                item["关键词"] = keyword
            return {
                "success": True,
                "data": standard_items,
                "total": raw_data.get("total", 0),
                "page_no": page_no,
                "page_total": raw_data.get("page_total", 1),
                "error_msg": ""
            }
        else:
            error_msg = result.get("msg", "接口调用失败")
            logging.error(f"异步请求失败（关键词：{keyword}，页码：{page_no}）：{error_msg}")
            return {"success": False, "data": [], "error_msg": error_msg}
except Exception as e:
    logging.error(f"异步请求异常（关键词：{keyword}，页码：{page_no}）：{str(e)}")
    return {"success": False, "data": [], "error_msg": str(e)}

def batch_get_item_list_sync(
keyword: str,
max_page: int = 5, # 最大获取页码（避免无限制分页）
kwargs
) -> List[Dict]:
"""同步批量获取多页商品列表（遍历所有页码或指定max_page）"""
all_items = []
page_no = 1
while True:
logging.info(f"正在获取关键词「{keyword}」第{page_no}页商品列表")
result = get_item_list_sync(keyword=keyword, page_no=page_no, kwargs)
if not result["success"]:
logging.error(f"第{page_no}页获取失败：{result['error_msg']}")
break
page_items = result["data"]
if not page_items:
logging.info(f"第{page_no}页无匹配商品，批量获取结束")
break
all_items.extend(page_items)
logging.info(f"第{page_no}页获取成功，新增{len(page_items)}条数据（累计{len(all_items)}条）")

    # 终止条件：达到最大页码或总页码
    if page_no >= result["page_total"] or page_no >= max_page:
        break
    page_no += 1
    # 控制调用频率（普通用户10次/分钟，间隔6秒；企业用户50次/分钟，间隔1秒）
    time.sleep(6)
return all_items

async def batch_get_item_list_async(
keyword: str,
max_page: int = 5,
**kwargs
) -> List[Dict]:
"""异步批量获取多页商品列表（并行请求，提升效率）"""
all_items = []

# 先获取总页码
first_page_result = await get_item_list_async(
    session=None, keyword=keyword, page_no=1, **kwargs
)
if not first_page_result["success"]:
    logging.error(f"关键词「{keyword}」第1页获取失败：{first_page_result['error_msg']}")
    return all_items
all_items.extend(first_page_result["data"])
page_total = min(first_page_result["page_total"], max_page)
if page_total <= 1:
    logging.info(f"关键词「{keyword}」仅1页数据，批量获取结束")
    return all_items

# 并行请求剩余页码
async with aiohttp.ClientSession() as session:
    tasks = []
    for page_no in range(2, page_total + 1):
        tasks.append(get_item_list_async(session, keyword=keyword, page_no=page_no, **kwargs))
    # 控制并发数（避免频率超限）
    semaphore = asyncio.Semaphore(5)  # 最大并发5个请求
    async def bounded_task(task):
        async with semaphore:
            return await task
    results = await asyncio.gather(*[bounded_task(task) for task in tasks])

    for result in results:
        if result["success"]:
            all_items.extend(result["data"])
            logging.info(f"异步获取成功，新增{len(result['data'])}条数据")
logging.info(f"关键词「{keyword}」异步批量获取结束，累计{len(all_items)}条数据")
return all_items

def save_item_list(items: List[Dict], save_path: str = SAVE_PATH):
"""将商品列表保存为Excel文件（便于归档/分析）"""
if not items:
logging.warning("无商品数据可保存")
return

df = pd.DataFrame(items)
# 筛选常用字段，优化Excel可读性
columns = [
    "请求时间", "关键词", "商品ID", "商品标题", "原价（元）", "优惠价（元）",
    "销量", "商品评分", "评价数", "店铺名称", "店铺类型", "店铺评分",
    "上架时间", "是否包邮", "售后保障", "优惠活动", "关联笔记数",
    "热门笔记关键词", "商品状态", "商品主图", "商品标签"
]
df = df[columns].drop_duplicates(subset=["商品ID"])  # 按商品ID去重（避免重复保存）

# 增量保存（避免覆盖历史数据）
try:
    history_df = pd.read_excel(save_path, engine="openpyxl")
    df = pd.concat([history_df, df], ignore_index=True).drop_duplicates(subset=["商品ID"])
except FileNotFoundError:
    pass

df.to_excel(save_path, index=False, engine="openpyxl")
logging.info(f"商品列表已归档至：{save_path}（累计{len(df)}条数据）")

def keyword_monitor_task(keywords: List[str], **kwargs):
"""关键词定时监测任务（实时追踪商品动态）"""
logging.info("=== 开始执行关键词定时监测任务 ===")
all_items = []
for keyword in keywords:
logging.info(f"=== 开始监测关键词：{keyword} ===")

    # 同步批量获取（如需高效可改用异步）
    items = batch_get_item_list_sync(keyword=keyword, **kwargs)
    all_items.extend(items)
    logging.info(f"=== 关键词「{keyword}」监测完成，获取{len(items)}条数据 ===")
# 保存所有关键词的商品列表
save_item_list(all_items)
logging.info("=== 关键词定时监测任务执行完成 ===")

调用示例（支持单页/批量/异步/定时监测）

if name == "main":

# 模式1：获取单页商品列表（关键词“平价粉底液 持妆”，50-200元，按销量排序）
single_page_result = get_item_list_sync(
    keyword="平价粉底液 持妆",
    price_min=50,
    price_max=200,
    sort_type="sales",
    page_no=1,
    page_size=20,
    need_note=1
)
if single_page_result["success"]:
    print("="*80)
    print(f"关键词「平价粉底液 持妆」第1页商品列表（共{len(single_page_result['data'])}条）")
    print("="*80)
    for idx, item in enumerate(single_page_result["data"][:10], 1):  # 打印前10条
        print(f"{idx:2d}. 标题：{item['商品标题']}")
        print(f"   价格：{item['原价（元）']}元 → 优惠价：{item['优惠价（元）']}元")
        print(f"   销量：{item['销量']} | 评分：{item['商品评分']}分 | 店铺：{item['店铺名称']}（{item['店铺类型']}）")
        print(f"   上架时间：{item['上架时间']} | 关联笔记数：{item['关联笔记数']}")
        print(f"   状态：{item['商品状态']} | 相关度：{item['相关度评分']}分")
        print("-"*80)
else:
    print(f"单页商品列表获取失败：{single_page_result['error_msg']}（错误码：{single_page_result['error_code']}）")

# 模式2：同步批量获取多页商品列表（关键词“露营装备”，按最新上架排序，最多获取5页）
# batch_items = batch_get_item_list_sync(
#     keyword="露营装备",
#     sort_type="newest",
#     max_page=5,
#     page_size=30,
#     tag="正品保障,运费险"
# )
# save_item_list(batch_items)

# 模式3：异步批量获取多页商品列表（关键词“ins风家居”，无价格限制，按评分排序）
# asyncio.run(batch_get_item_list_async(
#     keyword="ins风家居",
#     sort_type="score",
#     max_page=10,
#     page_size=50,
#     need_note=1
# ))

# 模式4：启动定时监测任务（每2小时监测一次关键词“国货美妆”“平价护肤”）
# scheduler = BlockingScheduler()
# scheduler.add_job(
#     keyword_monitor_task,
#     'interval',
#     hours=2,
#     args=[["国货美妆", "平价护肤"]],
#     kwargs={"sort_type": "sales", "max_page": 3, "page_size": 20}
# )
# logging.info("关键词定时监测任务已启动，每2小时执行一次...")
# try:
#     scheduler.start()
# except (KeyboardInterrupt, SystemExit):
#     logging.info("关键词定时监测任务已停止")

四、调试与验证：快速定位问题

1. 调试步骤（优先用 Postman 验证，排除代码干扰）
   手动拼接参数：在 Postman 中创建 POST 请求，填写appkey、keyword、timestamp等必填项，按业务需求补充price_min、sort_type等可选参数；
   生成签名：按服务商规则手动计算sign（用在线 MD5 工具验证，输入拼接后的字符串）；
   配置请求头：设置Content-Type: application/json，将参数以 JSON 格式传入请求体；
   发送请求：点击发送，查看响应结果；
   验证结果：
   若返回 200 且商品列表完整：接口正常，可对接代码；
   若返回 401（签名无效）：检查参数排序、secret 是否正确、时间戳是否过期、中文关键词是否编码；
   若返回 403（权限不足）：确认账号类型（普通 / 企业）是否匹配筛选条件 / 字段需求，是否已付费或申请高级权限；
   若返回 400（参数错误）：核对sort_type/match_type/shop_type等参数取值是否合法、价格格式是否正确、标签是否用逗号分隔；
   若返回 429（频率超限）：降低调用频率，或申请提升配额；
   若返回 404（接口不存在）：核对 API_URL 是否正确，确认item_search接口已申请开通；
   若返回 500（服务器异常）：记录日志，稍后重试（可能是服务商接口问题）；
   若返回 503（服务不可用）：联系服务商确认接口是否维护；
   若返回 “无匹配商品”：调整关键词（如扩大范围）、价格区间或筛选条件，确认小红书官网是否有相关商品。
2. 常见调试问题排查（关键词搜索场景高频问题）
   问题现象
   常见原因
   排查方案

签名错误（401）

1. 参数排序错误；2. secret 错误；3. 时间戳过期；4. 中文关键词未编码；5. 服务商签名规则差异
2. 按 ASCII 升序排序参数并打印验证；2. 核对 secret 与服务商后台一致；3. 校准本地时间（误差≤5 分钟）；4. 用urlencode处理中文关键词；5. 重新查看服务商文档，确认是否需要 SHA256 加密或额外参数（如nonce随机数）

无匹配商品

1. 关键词过于精准 / 宽泛；2. 价格区间过窄；3. 筛选条件过于严格；4. 商品未被服务商收录；5. 关键词组合逻辑错误
2. 调整关键词（如 “持妆粉底液” 改为 “粉底液 持妆”）；2. 扩大价格区间（如从 50-100 元改为 30-200 元）；3. 放宽筛选条件（如tag从 “正品保障，运费险” 改为 “正品保障”）；4. 直接在小红书官网搜索关键词，确认是否有相关商品；5. 确认多关键词组合是否正确（AND 用空格，OR 用 ）
   权限不足（403）
3. 普通用户使用高级筛选（如tag筛选、need_note=1）；2. 调用次数耗尽；3. 业务用途未通过审核
4. 移除高级筛选条件，或升级为企业用户；2. 查看服务商后台调用次数剩余量，及时充值；3. 补充业务合规说明（如 “仅用于内部选品分析”），重新提交审核

频率超限（429）
单 IP / 账号调用次数超过服务商限制

1. 批量分页时增加间隔（普通用户 6 秒 / 次，企业用户 1 秒 / 次）；2. 异步请求控制并发数（≤5）；3. 分时段调用，避免集中请求；4. 用ratelimit库控制调用频率（Python 示例：@limits(calls=10, period=60)）；5. 联系服务商申请提升配额

参数错误（400）

1. 价格格式错误（如字符串类型）；2. sort_type/shop_type取值非法；3. 标签用其他分隔符（如分号）；4. 关键词为空
2. 确保price_min/price_max为数字类型（int/float）；2. 核对服务商文档，确认参数取值范围（如sort_type仅支持 “relevance/sales/price_asc/price_desc/score/newest”）；3. 商品标签必须用逗号分隔；4. 确保关键词非空且符合格式要求

商品列表重复

1. 分页时缓存不一致；2. 关键词多义性导致重复收录；3. 未去重处理
2. 批量获取时一次性请求所有页码，减少缓存影响；2. 优化关键词（如增加限定词 “平价”“高端”）；3. 按商品ID去重（代码中已实现）

商品状态为 “已下架 / 违规”

1. 商品已被店铺下架；2. 商品涉及违规内容被平台处罚；3. 服务商数据未及时同步
2. 直接访问商品链接（若有）确认商品状态；2. 过滤该类商品，避免影响数据质量；3. 企业版可申请 “实时状态同步” 权限，提升数据时效性

关联笔记数为 0（need_note=1）

1. 商品无关联笔记；2. 普通用户无权限获取笔记数据；3. 服务商未支持该字段
2. 在小红书官网确认商品是否有关联笔记；2. 升级为企业用户，申请笔记数据权限；3. 联系服务商确认是否支持need_note参数

五、进阶优化：提升效率与稳定性（生产级必备）

1. 性能优化（批量 / 实时监测场景重点）
   （1）批量请求优化
   异步并发请求：多页码 / 多关键词批量获取时，用异步请求并行拉取（控制并发数≤5，避免频率超限），如代码中batch_get_item_list_async方法，效率比同步提升 3-5 倍；
   分页智能停止：获取第 1 页后，根据page_total计算总页码，仅请求有效页码，避免无效分页（如page_no超过page_total）；
   字段筛选精准化：仅保留业务必需字段，减少数据传输量和解析耗时（需服务商支持字段筛选参数）。
   （2）缓存策略优化
   搜索结果缓存：用 Redis 缓存关键词 + 筛选条件组合对应的搜索结果（缓存 key = 关键词 + 价格区间 + 筛选条件，有效期 30 分钟 - 1 小时），避免重复请求相同关键词；
   缓存穿透防护：对无匹配结果的关键词，缓存空结果（有效期 10 分钟），避免频繁无效请求；
   热点关键词缓存：对高频监测的关键词（如爆款选品关键词），缩短缓存有效期（如 10 分钟），平衡实时性与效率。
   （3）关键词策略优化
   关键词分词与扩展：利用 jieba 分词对核心关键词进行扩展（如 “露营装备” 扩展为 “露营帐篷、露营睡袋、露营炊具”），提升匹配覆盖率；
   多关键词优先级：按业务重要性划分关键词优先级（如核心选品关键词、次要竞品关键词），核心关键词高频监测，次要关键词低频监测；
   关键词去重：多关键词监测时，避免关键词重复（如 “平价粉底液” 和 “粉底液 平价”），减少冗余请求。
2. 稳定性优化（生产级必备）
   异常重试机制：
   对 429（频率超限）、500（服务器异常）、503（服务不可用）错误，采用指数退避策略重试（5s→10s→20s）；
   重试次数≤3 次，避免无效重试导致更多错误；
   对 401（签名错误）、403（权限不足）、400（参数错误）错误，直接返回并日志告警（需人工介入）。
   密钥与权限安全：
   定期轮换secret（每 3 个月更新 1 次），在服务商后台操作；
   生产环境将appkey和secret存储在环境变量或配置中心（如 Nacos），避免硬编码；
   限制接口调用 IP（部分服务商支持 IP 白名单），防止密钥泄露后被滥用；
   企业用户可申请 “子账号” 权限，按业务模块分配不同appkey，便于权限管控。
   日志与监控：
   详细记录每次请求的参数、签名、响应结果、错误信息、调用耗时，便于问题追溯；
   配置日志告警（如通过邮件 / 钉钉推送高频错误、频率超限、权限不足提示），实时监控接口状态；
   监控数据质量（如连续 10 次请求无结果、高频关键词响应时间超过 5 秒），及时触发告警；
   统计关键词匹配数量、商品状态分布，评估接口使用效果。
3. 业务场景专属适配
   （1）电商选品场景
   爆款筛选：按 “销量≥1000 + 商品评分≥4.5 分” 筛选，结合note_keywords分析用户热门反馈（如 “持妆”“控油”）；
   价格带分析：按价格区间（如 0-50 元、50-100 元、100-200 元）统计商品数量，定位主流价格带；
   卖点提取：从商品标题和标签中提取高频关键词（如 “防水”“持久”“无刺激”），指导自身产品设计；
   竞品排除：筛选时排除自有品牌商品，聚焦竞品和潜在竞品。
   （2）竞品监测场景
   价格监控：定时采集竞品商品价格，记录价格变动趋势，当价格低于阈值时触发告警；
   优惠活动追踪：监控竞品promotion_info字段，分析竞品促销策略（如满减、折扣、赠品）；
   店铺表现对比：统计竞品店铺的商品评分、好评率、销量，评估竞品市场表现；
   新品追踪：按 “newest” 排序，实时捕获竞品新上架商品，及时调整自身产品策略。
   （3）市场分析场景
   品类规模统计：按核心关键词批量采集商品，统计总商品数、总销量，评估品类市场规模；
   品牌分布分析：统计 TOP10 品牌的商品数量、销量占比，识别行业头部品牌；
   趋势预测：按 “上架时间” 统计商品发布趋势，结合note_count分析品类热度变化；
   地域分布：部分服务商支持按发货地筛选，分析品类货源集中区域（如美妆集中在广州、义乌）。
   六、避坑指南：常见问题与解决方案（关键词搜索场景高频）
4. 签名错误（最高频问题）
   原因：参数排序错误、secret 错误、时间戳过期、中文关键词未编码、服务商签名规则差异；
   解决方案：
   严格按 ASCII 升序排序参数，打印sorted_params确认排序正确；
   用urlencode自动处理中文关键词和特殊字符，避免手动拼接编码错误；
   用在线 MD5/SHA256 工具验证签名（输入拼接后的字符串，对比代码生成结果）；
   确保时间戳为毫秒级（int(time.time()*1000)），本地时间与服务器时间误差≤5 分钟；
   若仍报错，重新查阅服务商文档，确认是否需要添加额外参数（如nonce）或采用其他加密方式。
5. 合规风险（核心注意事项）
   原因：使用非法爬虫接口、数据用于商业化售卖、侵犯品牌 / 店铺知识产权、抓取敏感信息；
   解决方案：
   仅使用合规服务商接口，要求服务商提供《数据合规承诺书》和《版权使用授权》；
   明确数据使用场景，禁止商业化售卖、恶意传播或用于违法活动（如刷单、恶意比价）；
   引用商品数据时，标注 “数据来源：小红书” 和 “商品版权归原品牌 / 店铺所有”；
   涉及品牌商品的二次使用（如导购展示），需提前获得相关品牌 / 店铺授权；
   避开敏感关键词（如涉密、违法违规商品关键词），不采集隐私店铺商品。
6. 频率超限与调用成本控制
   原因：批量请求未控制频率、关键词监测过于频繁、无效分页请求过多；
   解决方案：
   按服务商限制控制调用频率（普通用户 10 次 / 分钟，企业用户 50 次 / 分钟），批量请求增加间隔；
   按关键词优先级调整监测频率（核心关键词每 2 小时，次要关键词每 6 小时）；
   智能分页：仅请求前 N 页（如前 5 页），或根据total判断是否需要继续分页（如total≤100时请求全部）；
   企业用户可申请 “按结果计费” 套餐，避免无效请求（如无匹配商品不扣费）。
7. 数据质量优化
   原因：商品重复、状态异常（已下架 / 违规）、销量 / 评分不准确、收录延迟；
   解决方案：
   按商品ID去重，避免同一商品被多个关键词匹配重复收录；
   过滤 “已下架 / 违规” 商品，仅保留正常状态数据；
   销量 / 评分数据以小红书官网为准，若存在差异，联系服务商同步数据；
   新商品收录延迟 5-15 分钟，实时监测场景需预留缓冲时间；
   对比多个服务商的接口数据，选择数据准确性更高的服务商。
   七、上线前检查清单（生产级必查）
   密钥是否保密（未硬编码、未提交到代码仓库，用环境变量 / 配置中心存储）；
   异常处理是否完整（覆盖 401/403/404/429/500/503 等所有常见错误码）；
   频率控制是否到位（调用频率未超过服务商配额，批量请求有合理间隔 / 并发控制）；
   权限与付费是否合规（账号类型匹配筛选条件 / 字段需求，调用次数充足）；
   数据处理```