比比网 item_search - 关键词搜索接口对接全攻略：从入门到精通

python

import requests
import hashlib
import time
import logging
import pandas as pd
from urllib.parse import quote
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# 日志配置：记录调用日志，便于问题排查
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.FileHandler("bibi_item_search.log"), logging.StreamHandler()]
)

# 配置信息（替换为你的开放平台信息）
CONFIG = {
    "key": "你的接口key",
    "secret": "你的接口secret",
    "api_url": "https://api.bibi.com/bibi/item_search",
    "result_type": "json",
    "cache": "yes"
}

def generate_sign(params: dict, secret: str) -> str:
    """生成比比网接口签名（MD5加密）"""
    # 1. 按参数名ASCII升序排序
    sorted_params = sorted(params.items(), key=lambda x: x[0])
    # 2. 拼接参数为key1value1key2value2格式
    param_str = "".join([f"{k}{v}" for k, v in sorted_params])
    # 3. 拼接secret并MD5加密
    sign_str = param_str + secret
    sign = hashlib.md5(sign_str.encode("utf-8")).hexdigest().lower()
    return sign

def standardize_item_data(raw_item: dict) -> dict:
    """标准化商品数据，统一输出格式"""
    return {
        "商品ID": raw_item.get("item_id", ""),
        "商品标题": raw_item.get("item_name", ""),
        "商品品牌": raw_item.get("brand", ""),
        "所属品类": raw_item.get("category", ""),
        "商品主图URL": raw_item.get("item_pic", ""),
        "原价(元)": raw_item.get("original_price", 0.0),
        "现价(元)": raw_item.get("current_price", 0.0),
        "历史最低价(元)": raw_item.get("history_low_price", 0.0),
        "降价幅度(%)": raw_item.get("price_drop_rate", 0.0),
        "月销量": raw_item.get("monthly_sales", 0),
        "商家名称": raw_item.get("merchant_name", ""),
        "商家类型": raw_item.get("merchant_type", ""),
        "商家评分": raw_item.get("merchant_score", 0.0),
        "请求时间": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
    }

def bibi_item_search(
    keyword: str,
    category_id: str = None,
    price_min: float = None,
    price_max: float = None,
    sales_min: int = None,
    merchant_type: str = None,
    sort_type: str = "price_asc",
    page_num: int = 1,
    page_size: int = 20
) -> dict:
    """调用比比网item_search接口获取商品列表"""
    # 1. 校验必填参数
    if not keyword:
        return {"success": False, "error_msg": "关键词不能为空", "data": [], "pagination": {}}

    # 2. 构建公共参数
    params = {
        "key": CONFIG["key"],
        "api_name": "item_search",
        "result_type": CONFIG["result_type"],
        "cache": CONFIG["cache"],
        "q": quote(keyword, encoding="utf-8"),
        "sort_type": sort_type,
        "page_num": page_num,
        "page_size": min(page_size, 50)  # 单页最大50条
    }

    # 3. 补充分筛参数
    if category_id:
        params["category_id"] = category_id
    if price_min is not None:
        params["price_min"] = price_min
    if price_max is not None:
        params["price_max"] = price_max
    if sales_min is not None:
        params["sales_min"] = sales_min
    if merchant_type:
        params["merchant_type"] = merchant_type

    # 4. 生成签名
    sign = generate_sign(params, CONFIG["secret"])
    params["sign"] = sign

    try:
        # 5. 发送HTTPS请求
        response = requests.get(
            url=CONFIG["api_url"],
            params=params,
            timeout=10,
            verify=True  # 生产环境建议开启证书验证
        )
        response.raise_for_status()  # 抛出HTTP状态码异常
        result = response.json()

        # 6. 解析响应结果
        if result.get("error_response"):
            error = result["error_response"]
            error_msg = f"{error.get('code', '未知错误')}: {error.get('msg', '无错误信息')}"
            logging.error(f"搜索失败（关键词：{keyword}）：{error_msg}")
            return {"success": False, "error_msg": error_msg, "data": [], "pagination": {}}

        search_result = result.get("item_search_response", {}).get("items", {})
        raw_items = search_result.get("item", [])
        if not raw_items:
            logging.warning(f"无匹配商品（关键词：{keyword}）")
            return {"success": False, "error_msg": "无匹配商品", "data": [], "pagination": {}}

        # 7. 标准化数据与分页信息
        standard_items = [standardize_item_data(item) for item in raw_items]
        pagination = {
            "total": int(search_result.get("total", 0)),
            "page_num": page_num,
            "page_size": page_size,
            "has_next": search_result.get("has_next", False)
        }
        return {"success": True, "data": standard_items, "pagination": pagination, "error_msg": ""}

    except requests.exceptions.RequestException as e:
        logging.error(f"网络请求异常（关键词：{keyword}）：{str(e)}")
        return {"success": False, "error_msg": f"网络异常：{str(e)}", "data": [], "pagination": {}}
    except Exception as e:
        logging.error(f"数据解析异常（关键词：{keyword}）：{str(e)}")
        return {"success": False, "error_msg": f"解析异常：{str(e)}", "data": [], "pagination": {}}
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# 调用示例
if __name__ == "__main__":
    # 搜索条件：智能手机 5G 256G，价格2000-5000元，月销≥500，自营商家，按价格升序
    keyword = "智能手机 5G 256G"
    price_min = 2000.0
    price_max = 5000.0
    sales_min = 500
    merchant_type = "self"
    sort_type = "price_asc"

    # 调用接口
    result = bibi_item_search(
        keyword=keyword,
        price_min=price_min,
        price_max=price_max,
        sales_min=sales_min,
        merchant_type=merchant_type,
        sort_type=sort_type,
        page_size=20
    )

    if result["success"]:
        print(f"搜索成功：共找到 {result['pagination']['total']} 件商品")
        for item in result["data"][:5]:  # 打印前5条数据
            print(f"商品ID：{item['商品ID']} | 标题：{item['商品标题']} | 现价：{item['现价(元)']} | 月销：{item['月销量']}")
        # 保存为Excel
        df = pd.DataFrame(result["data"])
        df.to_excel(f"比比网商品搜索_{keyword}.xlsx", index=False)
        # 翻页示例
        if result["pagination"]["has_next"]:
            next_page_result = bibi_item_search(
                keyword=keyword,
                price_min=price_min,
                price_max=price_max,
                sales_min=sales_min,
                merchant_type=merchant_type,
                sort_type=sort_type,
                page_num=2,
                page_size=20
            )
            print(f"下一页获取到 {len(next_page_result['data'])} 件商品")
    else:
        print(f"搜索失败：{result['error_msg']}")

四、调试与问题排查：快速解决对接异常

1. 优先用官方工具调试（排除签名与参数问题）

1. 登录比比网开放平台调试工具，选择 bibi.item.search 接口；

2. 输入关键词、品类编码、价格区间等参数，工具自动生成签名并发送请求；

3. 若官方工具调用成功 → 问题出在代码的签名生成或参数拼接逻辑；

4. 若官方工具调用失败 → 问题出在权限配置或参数有效性（如品类编码错误、权限不足）。

2. 高频问题排查表

五、进阶优化：生产级稳定性提升

1. 性能与配额优化

• 批量翻页优化：通过 has_next 字段判断是否继续翻页，避免无效请求；多关键词 / 多品类查询时采用 异步并发（aiohttp），并发数≤权限允许的频率上限（如企业基础权限 5 次 / 秒）；

• 智能缓存策略：用 Redis 缓存搜索结果，缓存 key 为 bibi_search_关键词_品类_价格区间_页码，价格 / 销量等动态数据缓存 5 分钟，基础信息缓存 15 分钟，减少重复调用；

• 字段精简：仅保留业务必需字段（如商品 ID、价格、销量），避免返回冗余数据，提升响应速度。

2. 数据质量优化

• 数据去重：按 item_id 去重，避免同一商品重复出现在不同页码或不同关键词搜索结果中；

• 异常值过滤：过滤价格≤0、销量≤0 的商品，剔除降价幅度异常（如超过 100%）的数据，提升数据有效性；

• 关键词优化：对长关键词进行分词处理（如 “256G 5G 智能手机 全网通”→“智能手机 5G 256G”），提升搜索覆盖率。

3. 合规与安全优化

• 密钥管理：生产环境禁止硬编码 key/secret，存储在配置中心（如 Nacos、Apollo），并定期轮换密钥（每 3 个月一次）；

• 重试机制：对 403（频率超限）、504（超时）等错误添加 指数退避重试 策略，首次重试间隔 1 秒，之后间隔翻倍，最多重试 3 次；

• 日志审计：记录每次调用的关键词、筛选条件、响应状态、数据条数，日志保留至少 30 天，满足商业合规审计要求。

六、扩展场景：接口联动与功能升级

1. 联动 item_get 接口：通过 item_search 获取商品 ID 列表，批量调用 item_get 接口获取商品详情、规格参数、用户评价，实现 “搜索 - 详情” 全链路数据采集；

2. 价格预警系统：定时调用 item_search 接口，监控目标商品价格波动，当价格低于设定阈值时，通过邮件 / 短信自动推送降价提醒；

3. 竞品分析平台：聚合多维度搜索结果，统计竞品商品的价格分布、销量变化、降价频率，生成可视化分析报告，辅助企业定价决策；

4. 电商选品工具：按 “高销量 + 低价格 + 高降价幅度” 筛选潜力商品，生成选品清单，为商家提供进货参考。