快手电商作为主流的内容电商平台,item_search 接口是通过关键词批量获取快手商品列表的核心工具,支持按关键词、价格区间、销量、佣金比例、店铺类型等多维度筛选,返回商品标题、价格、佣金、销量、店铺信息等关键数据。该接口广泛应用于快手小店选品、电商数据分析、分销推广系统搭建等场景。
一、接口核心认知:功能与适配场景
1. 接口定位与核心价值
2. 核心参数与返回字段
(1)请求参数(必填 + 可选,按优先级排序)
(2)返回核心字段(按业务场景分类)
3. 接口限制与注意事项
二、对接前准备:3 步搞定前置条件
1. 获取接口权限(两种接入方式)
2. 技术环境准备
(1)支持语言与协议
(2)必备工具与依赖
3. 业务需求梳理
三、实操步骤:接口对接全流程(Python 示例)
步骤 1:理解签名生成规则(关键!快手官方规则)
签名示例
步骤 2:完整代码实现(Python)
(1)依赖安装
(2)核心代码(含签名生成、接口调用、数据保存)
import requests
import hashlib
import time
import json
import pandas as pd
from urllib.parse import urlencode
import logging
# 日志配置(记录接口调用日志,便于排查问题)
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s",
handlers=[logging.FileHandler("kuaishou_item_search.log"), logging.StreamHandler()]
)
# 接口配置(替换为自己的 app_key、app_secret、API 地址)
CONFIG = {
"app_key": "你的app_key",
"app_secret": "你的app_secret",
"api_url": "https://open.kuaishou.com/api/open/item_search", # 官方/服务商接口地址
"save_path": "快手商品列表.xlsx"
}
def generate_sign(params: dict, app_secret: str) -> str:
"""生成快手接口签名(MD5 32位大写)"""
# 1. 剔除 sign 字段(如果存在)
params.pop("sign", None)
# 2. 按参数名 ASCII 升序排序
sorted_params = sorted(params.items(), key=lambda x: x[0])
# 3. 拼接参数字符串
param_str = urlencode(sorted_params, encoding="utf-8") + f"&app_secret={app_secret}"
# 4. MD5 加密并转大写
md5 = hashlib.md5()
md5.update(param_str.encode("utf-8"))
return md5.hexdigest().upper()
def kuaishou_item_search(
keyword: str,
price_min: float = None,
price_max: float = None,
sort: str = "relevance",
commission_rate_min: int = None,
shop_type: str = "all",
page_no: int = 1,
page_size: int = 20
) -> dict:
"""
调用快手 item_search 接口,获取商品列表
:param keyword: 搜索关键词
:param price_min: 最低价格
:param price_max: 最高价格
:param sort: 排序方式
:param commission_rate_min: 最低佣金比例
:param shop_type: 店铺类型
:param page_no: 页码
:param page_size: 每页条数
:return: 标准化后的商品数据
"""
# 1. 构建基础参数
params = {
"app_key": CONFIG["app_key"],
"keyword": keyword,
"sort": sort,
"shop_type": shop_type,
"page_no": page_no,
"page_size": page_size,
"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 commission_rate_min is not None:
params["commission_rate_min"] = commission_rate_min
# 3. 生成签名
params["sign"] = generate_sign(params, CONFIG["app_secret"])
try:
# 4. 发送 POST 请求(快手接口推荐 POST)
response = requests.post(
url=CONFIG["api_url"],
data=json.dumps(params),
headers={"Content-Type": "application/json"},
timeout=15,
verify=True
)
response.raise_for_status() # 抛出 HTTP 错误(如 401/403)
result = response.json()
# 5. 解析响应结果(以快手官方返回格式为例)
if result.get("code") == 0:
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 = []
for item in item_list:
standard_items.append({
"关键词": keyword,
"商品ID": item.get("item_id", ""),
"商品标题": item.get("title", ""),
"主图链接": item.get("cover_url", ""),
"售价(元)": item.get("price", 0),
"原价(元)": item.get("original_price", 0),
"近30天销量": item.get("sales", 0),
"佣金比例(%)": item.get("commission_rate", 0),
"单件佣金(元)": round(item.get("price", 0) * item.get("commission_rate", 0) / 100, 2),
"店铺名称": item.get("shop_name", ""),
"店铺类型": item.get("shop_type", ""),
"店铺评分": item.get("shop_score", 0),
"商品状态": item.get("status", ""),
"发货地": item.get("location", ""),
"推广链接": item.get("promotion_link", ""),
"请求时间": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
})
return {
"success": True,
"data": standard_items,
"total": total,
"page_no": page_no,
"page_total": page_total
}
else:
error_msg = result.get("msg", "接口调用失败")
logging.error(f"接口返回错误:{error_msg}(code={result.get('code')})")
return {"success": False, "error_msg": error_msg}
except requests.exceptions.RequestException as e:
logging.error(f"网络请求异常:{str(e)}")
return {"success": False, "error_msg": f"网络异常:{str(e)}"}
except Exception as e:
logging.error(f"数据解析异常:{str(e)}")
return {"success": False, "error_msg": f"解析异常:{str(e)}"}
def batch_get_items(
keyword: str,
max_page: int = 5,
**kwargs
) -> list:
"""批量获取多页商品列表(控制频率,避免超限)"""
all_items = []
page_no = 1
while True:
logging.info(f"正在获取关键词「{keyword}」第 {page_no} 页商品")
result = kuaishou_item_search(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 >= max_page or page_no >= result["page_total"]:
break
page_no += 1
# 控制调用频率(个人用户间隔 6 秒,企业用户间隔 1 秒)
time.sleep(6)
return all_items
def save_to_excel(items: list, save_path: str = CONFIG["save_path"]):
"""将商品列表保存为 Excel 文件"""
if not items:
logging.warning("无商品数据可保存")
return
df = pd.DataFrame(items)
# 去重(按商品ID)
df = df.drop_duplicates(subset=["商品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)} 条)")
# 调用示例
if __name__ == "__main__":
# 单页获取:搜索“家居清洁 神器”,筛选价格 9.9-99.9 元,销量优先
single_page_result = kuaishou_item_search(
keyword="家居清洁 神器",
price_min=9.9,
price_max=99.9,
sort="sales",
page_size=20
)
if single_page_result["success"]:
print(f"单页获取 {len(single_page_result['data'])} 条商品")
print("第一条商品信息:", single_page_result["data"][0])
# 批量获取:获取前 5 页商品,保存到 Excel
# batch_items = batch_get_items(
# keyword="家居清洁 神器",
# price_min=9.9,
# price_max=99.9,
# sort="sales",
# max_page=5
# )
# save_to_excel(batch_items)