废旧物资 item_search 接口是按关键字检索再生资源商品列表的核心入口，支持通过关键词、品类、价格区间、地区、质检标准等多维度筛选，返回分页商品基础信息（含 ID、标题、品类、基准价、卖家资质等），可联动 item_get 接口获取详情。该接口普遍采用 HTTPS + HMAC‑SHA256 签名认证，权限与数据分级严格，适配回收企业采购、固废处置比价、环保数据统计等场景。本攻略从接口认知、权限获取、实操对接、调试排错到生产级优化，提供结构化全链路指导，兼顾合规性与稳定性。
一、接口核心认知：功能与适配场景
1. 接口定位与核心价值
2. 核心参数与返回字段
（1）请求参数（公共参数 + 私有参数，POST 提交）
（2）返回核心字段（按业务场景分类）
3. 接口限制与注意事项
二、对接前准备：权限与环境搭建
1. 获取接口权限（官方唯一合规路径）
2. 技术环境准备
（1）支持语言与协议
（2）必备工具与依赖
三、实操步骤：接口对接全流程（Python 示例）
步骤 1：理解签名认证规则（核心）
步骤 2：完整代码实现
（1）依赖安装
（2）Python 代码
import requests
import hmac
import hashlib
import time
import pandas as pd
import logging
from urllib.parse import urlencode
# 封装好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("waste_item_search.log"), logging.StreamHandler()]
)
# 配置信息（替换为你的平台信息）
CONFIG = {
"app_key": "你的app_key",
"app_secret": "你的app_secret",
"api_url": "https://open.wasteplat.com/api/v2/waste/item/search",
"version": "v2"
}
def generate_sign(params: dict, app_secret: str) -> str:
"""生成HMAC‑SHA256签名"""
filtered_params = {k: v for k, v in params.items() if v and k != "sign"}
sorted_params = sorted(filtered_params.items(), key=lambda x: x[0])
param_str = urlencode(sorted_params, encoding="utf-8") + f"&app_secret={app_secret}"
sign = hmac.new(
app_secret.encode("utf-8"),
param_str.encode("utf-8"),
hashlib.sha256
).hexdigest().lower()
return sign
def standardize_search_data(raw_item: dict, keyword: str) -> dict:
"""标准化搜索结果"""
price_info = raw_item.get("price_info", {})
seller_info = raw_item.get("seller_info", {})
return {
"搜索关键词": keyword,
"商品ID": raw_item.get("item_id", ""),
"商品标题": raw_item.get("title", ""),
"物资品类": raw_item.get("category", ""),
"细分品类": raw_item.get("sub_category", ""),
"规格参数": str(raw_item.get("spec", {})),
"计价单位": raw_item.get("unit", ""),
"基准价(元/单位)": float(price_info.get("base_price", 0)),
"区域补贴(元/单位)": float(price_info.get("region_subsidy", 0)),
"预估成交价(元/单位)": float(price_info.get("final_price", 0)),
"价格有效期": price_info.get("price_valid_time", ""),
"卖家名称": seller_info.get("seller_name", ""),
"卖家资质": seller_info.get("seller_qualification", ""),
"服务区域": seller_info.get("location", ""),
"请求时间": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
}
def waste_item_search(
keyword: str,
category: str = None,
start_price: float = None,
end_price: float = None,
region: str = None,
sort: str = "price_asc",
page_no: int = 1,
page_size: int = 20
) -> dict:
"""调用废旧物资item_search接口"""
params = {
"app_key": CONFIG["app_key"],
"method": "waste.item.search",
"timestamp": str(int(time.time() * 1000)),
"version": CONFIG["version"],
"q": keyword,
"sort": sort,
"page_no": page_no,
"page_size": min(page_size, 50) # 单页最大50条
}
# 补充分筛参数
if category:
params["category"] = category
if start_price:
params["start_price"] = start_price
if end_price:
params["end_price"] = end_price
if region:
params["region"] = region
# 生成签名
params["sign"] = generate_sign(params, CONFIG["app_secret"])
try:
response = requests.post(
url=CONFIG["api_url"],
data=params,
headers={"Content-Type": "application/x-www-form-urlencoded; charset=utf-8"},
timeout=10,
verify=True
)
response.raise_for_status()
result = response.json()
if result.get("error_response"):
error_msg = f"{result['error_response']['code']}: {result['error_response']['msg']}"
logging.error(f"调用失败（关键词：{keyword}）：{error_msg}")
return {"success": False, "error_msg": error_msg, "data": [], "pagination": {}}
search_resp = result.get("item_search_response", {})
raw_items = search_resp.get("items", {}).get("item", [])
if not raw_items:
logging.warning(f"无数据（关键词：{keyword}）")
return {"success": False, "error_msg": "无数据", "data": [], "pagination": {}}
# 标准化数据
standard_items = [standardize_search_data(item, keyword) for item in raw_items]
pagination = {
"total_results": int(search_resp.get("total_results", 0)),
"page_no": page_no,
"page_size": page_size,
"has_more": search_resp.get("has_more", 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__":
keyword = "上海废钢 Q235"
category = "废金属"
region = "上海"
page_size = 20
result = waste_item_search(
keyword=keyword,
category=category,
region=region,
page_size=page_size
)
if result["success"]:
print(f"搜索成功：共 {result['pagination']['total_results']} 条，当前页 {len(result['data'])} 条")
for item in result["data"][:5]:
print(f"商品ID：{item['商品ID']} | 标题：{item['商品标题']} | 预估成交价：{item['预估成交价(元/单位)']} 元")
# 保存为Excel
df = pd.DataFrame(result["data"])
df.to_excel(f"waste_search_{keyword}.xlsx", index=False)
# 翻页示例
if result["pagination"]["has_more"]:
next_page = waste_item_search(
keyword=keyword,
category=category,
region=region,
page_no=2,
page_size=page_size
)
print(f"下一页获取 {len(next_page['data'])} 条")
else:
print(f"失败：{result['error_msg']}")