废旧物资item_get接口是获取废旧物资回收 / 交易平台商品详情的核心接口,支持通过商品 ID查询废旧物资的品类、规格、成色、回收价格、交易模式、质检标准、物流方案等核心数据,适配废旧物资回收平台、企业固废处置、再生资源交易、环保数据统计等场景。该接口通常采用HTTPS + 签名认证机制,数据具有强行业属性、定价规则复杂、权限分级严格的特点。本攻略从接口认知、权限获取、实操对接、调试排错到生产级优化,提供结构化全链路指导,兼顾入门易用性与企业级稳定性。
一、接口核心认知:功能与适配场景
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_get.log"), logging.StreamHandler()]
)
# 配置信息(替换为你的平台信息)
CONFIG = {
"app_key": "你的app_key",
"app_secret": "你的app_secret",
"api_url": "https://open.wasteplat.com/api/v2/waste/item/get", # 平台接口地址
"version": "v2"
}
def generate_sign(params: dict, app_secret: str) -> str:
"""生成废旧物资接口HMAC-SHA256签名"""
# 1. 排除sign字段,筛选非空参数
filtered_params = {k: v for k, v in params.items() if v and k != "sign"}
# 2. 按参数名ASCII升序排序
sorted_params = sorted(filtered_params.items(), key=lambda x: x[0])
# 3. 拼接参数字符串(UTF-8编码)
param_str = urlencode(sorted_params, encoding="utf-8") + f"&app_secret={app_secret}"
# 4. HMAC-SHA256加密,生成小写签名
sign = hmac.new(
app_secret.encode("utf-8"),
param_str.encode("utf-8"),
hashlib.sha256
).hexdigest().lower()
return sign
def standardize_item_data(raw_item: dict, region: str) -> dict:
"""标准化废旧物资商品详情数据,统一输出格式"""
price_info = raw_item.get("price_info", {})
quality_info = raw_item.get("quality_info", {})
seller_info = raw_item.get("seller_info", {})
return {
"商品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", ""),
"查询地区": region,
"基准价(元/单位)": float(price_info.get("base_price", 0)),
"品质溢价(元/单位)": float(price_info.get("quality_premium", 0)),
"品质扣减(元/单位)": float(price_info.get("quality_deduction", 0)),
"区域补贴(元/单位)": float(price_info.get("region_subsidy", 0)),
"最终回收价(元/单位)": float(price_info.get("final_price", 0)),
"价格有效期": price_info.get("price_valid_time", ""),
"质检标准": quality_info.get("quality_standard", ""),
"交易模式": raw_item.get("transaction_mode", ""),
"物流方式": raw_item.get("logistics_mode", ""),
"物流费用(元/单位)": float(raw_item.get("logistics_fee", 0)),
"卖家名称": seller_info.get("seller_name", ""),
"卖家资质": seller_info.get("seller_qualification", ""),
"服务区域": seller_info.get("seller_location", ""),
"请求时间": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
}
def waste_item_get(item_id: str, region: str = None, fields: str = None) -> dict:
"""调用废旧物资item_get接口获取商品详情"""
# 1. 构建请求参数
params = {
"app_key": CONFIG["app_key"],
"method": "waste.item.get",
"timestamp": str(int(time.time() * 1000)),
"version": CONFIG["version"],
"item_id": item_id
}
# 补充分筛参数
if region:
params["region"] = region
if fields:
params["fields"] = fields
# 2. 生成签名
params["sign"] = generate_sign(params, CONFIG["app_secret"])
try:
# 3. 发送POST请求
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()
# 4. 解析响应结果
if result.get("code") != 0:
error_msg = f"{result.get('code')}: {result.get('msg')}"
logging.error(f"接口调用失败(商品ID:{item_id}):{error_msg}")
return {"success": False, "error_msg": error_msg, "data": {}}
raw_item = result.get("data", {})
if not raw_item:
logging.warning(f"无商品数据返回(商品ID:{item_id})")
return {"success": False, "error_msg": "无商品数据", "data": {}}
# 5. 标准化数据
standard_data = standardize_item_data(raw_item, region or "全国")
return {
"success": True,
"data": standard_data,
"error_msg": ""
}
except requests.exceptions.RequestException as e:
logging.error(f"网络请求异常(商品ID:{item_id}):{str(e)}")
return {"success": False, "error_msg": f"网络异常:{str(e)}", "data": {}}
except Exception as e:
logging.error(f"数据解析异常(商品ID:{item_id}):{str(e)}")
return {"success": False, "error_msg": f"解析异常:{str(e)}", "data": {}}
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# 调用示例
if __name__ == "__main__":
# 替换为真实的废旧物资商品ID
item_id = "WS20250101001"
# 查询地区(影响区域补贴)
region = "上海"
# 按需指定返回字段
fields = "title,category,price_info,quality_info,seller_info"
result = waste_item_get(item_id=item_id, region=region, fields=fields)
if result["success"]:
print("废旧物资商品详情:")
for k, v in result["data"].items():
print(f"{k}: {v}")
# 保存为Excel
df = pd.DataFrame([result["data"]])
df.to_excel(f"waste_item_detail_{item_id}.xlsx", index=False)
else:
print(f"获取失败:{result['error_msg']}")