当前位置：首页 > news >正文

震坤行工业超市开放平台接口实战：工业品精准检索与详情解析全方案

news 2025/9/18 6:31:25

震坤行工业超市作为国内领先的工业用品采购平台，其开放接口为制造业企业提供了便捷的工业用品数据获取渠道。与消费品电商平台不同，震坤行接口专注于 MRO（维护、维修、运营）工业品，包含丰富的技术参数、认证信息和供应保障数据。本文将系统讲解震坤行开放平台核心接口的技术实现，重点解决接口认证、复杂参数构造、数据解析和高并发处理等关键问题，提供一套可直接应用于生产环境的完整解决方案。

一、接口基础信息与应用场景

接口核心信息

震坤行开放平台主要接口的关键技术参数：

接口域名：https://openapi.zkh360.com
认证方式：OAuth 2.0（Access Token）
请求格式：HTTP GET/POST，JSON 格式参数
响应格式：JSON
编码格式：UTF-8
调用限制：单应用 QPS=10，日调用上限 10 万次

核心接口列表

接口名称	接口地址	请求方式	功能描述
令牌获取	`/oauth/token`	POST	获取访问令牌
商品搜索	`/api/v2/products/search`	GET	按关键词搜索商品
商品详情	`/api/v2/products/{sku}`	GET	获取商品详细信息
分类列表	`/api/v2/categories`	GET	获取商品分类体系
库存查询	`/api/v2/inventory`	POST	查询商品库存状态

典型应用场景

企业采购系统：集成工业品数据实现快速采购
生产管理系统：自动匹配生产所需工业物料
MRO 供应商系统：多平台商品数据同步
供应链管理工具：监控工业物料价格与库存波动
智能制造平台：实现生产物料的智能推荐与自动补货

接口调用流程

plaintext

应用注册 → 令牌获取 → 签名生成 → 请求构造 → 接口调用 → 响应解析 → 数据存储 → 令牌刷新

二、接口认证与参数详解

OAuth 2.0 认证流程

震坤行采用 OAuth 2.0 认证机制，具体流程：

开发者在平台注册应用，获取 client_id 和 client_secret
通过客户端凭证模式（client_credentials）获取 access_token
每次接口调用在 HTTP 头部携带 access_token
令牌过期前（默认 2 小时）需重新获取

商品搜索核心参数

参数名	类型	说明	是否必须
keyword	String	搜索关键词	是
categoryId	String	分类 ID	否
brandId	String	品牌 ID	否
spec	String	规格参数，JSON 格式	否
priceMin	Float	最低价格	否
priceMax	Float	最高价格	否
certification	String	认证类型，多个用逗号分隔	否
inStock	Boolean	是否有库存	否
page	Integer	页码，默认 1	否
size	Integer	每页条数，1-50	否
sort	String	排序方式：price_asc, price_desc, popularity	否

响应结果结构

商品搜索接口的核心响应字段：

json

{"code": 200,"message": "success","data": {"total": 1250,"pages": 63,"pageNum": 1,"pageSize": 20,"list": [{"sku": "ZKH12345678","name": "内六角螺栓 M8×30 8.8级 镀锌","brandName": "震坤行","categoryName": "紧固件/螺栓","price": 1.56,"originalPrice": 1.80,"minOrderQuantity": 100,"packagingQuantity": 1000,"spec": "M8×30mm，8.8级，镀锌","certification": ["ISO9001", "CE"],"stockStatus": "IN_STOCK","stockQuantity": 56800,"deliveryCycle": "1-3天","imageUrl": "https://img.zkh360.com/product/ZKH12345678.jpg"}]}
}

三、核心技术实现

1. 认证令牌管理工具

python

运行

import requests
import time
import json
from datetime import datetime, timedeltaclass ZkhAuthManager:"""震坤行平台认证令牌管理工具"""def __init__(self, client_id, client_secret, token_url="https://openapi.zkh360.com/oauth/token"):self.client_id = client_idself.client_secret = client_secretself.token_url = token_urlself.access_token = Noneself.token_type = Noneself.expires_at = 0  # 令牌过期时间戳self.expires_in = 0  # 令牌有效期（秒）def get_access_token(self, force_refresh=False):"""获取访问令牌，自动处理令牌刷新:param force_refresh: 是否强制刷新令牌:return: 访问令牌字符串"""# 检查令牌是否有效if not force_refresh and self.access_token and time.time() < self.expires_at - 60:return self.access_token# 准备请求参数params = {"grant_type": "client_credentials","client_id": self.client_id,"client_secret": self.client_secret}try:response = requests.post(self.token_url,data=params,headers={"Content-Type": "application/x-www-form-urlencoded"},timeout=10)result = response.json()if "access_token" in result:self.access_token = result["access_token"]self.token_type = result.get("token_type", "bearer")self.expires_in = int(result.get("expires_in", 7200))self.expires_at = time.time() + self.expires_inprint(f"令牌获取成功，有效期至: {datetime.fromtimestamp(self.expires_at)}")return self.access_tokenelse:raise Exception(f"获取令牌失败: {result.get('error_description', '未知错误')}")except Exception as e:print(f"令牌获取异常: {str(e)}")return Nonedef get_authorization_header(self):"""获取认证头部信息"""token = self.get_access_token()if token and self.token_type:return {"Authorization": f"{self.token_type.capitalize()} {token}","Content-Type": "application/json"}return Nonedef is_token_valid(self):"""检查令牌是否有效"""return self.access_token is not None and time.time() < self.expires_at - 60

2. 商品搜索接口客户端

python

运行

import requests
import time
import json
from urllib.parse import urlencode
from threading import Lock
from datetime import datetimeclass ZkhProductSearchClient:"""震坤行商品搜索接口客户端"""def __init__(self, auth_manager):self.auth_manager = auth_managerself.base_url = "https://openapi.zkh360.com/api/v2"self.timeout = 15  # 超时时间(秒)self.qps_limit = 10  # QPS限制self.last_request_time = 0self.request_lock = Lock()  # 线程锁控制QPSdef _check_qps(self):"""检查并控制QPS"""with self.request_lock:current_time = time.time()interval = 1.0 / self.qps_limit  # 每次请求最小间隔elapsed = current_time - self.last_request_timeif elapsed < interval:# 需要等待的时间time.sleep(interval - elapsed)self.last_request_time = time.time()def search_products(self, keyword, **kwargs):"""搜索商品:param keyword: 搜索关键词:param kwargs: 其他搜索参数:return: 搜索结果字典"""# 检查QPS限制self._check_qps()# 1. 构造请求URLurl = f"{self.base_url}/products/search"# 2. 构建请求参数params = {"keyword": keyword}# 添加其他参数valid_params = ["categoryId", "brandId", "spec", "priceMin", "priceMax","certification", "inStock", "page", "size", "sort"]for param in valid_params:if param in kwargs and kwargs[param] is not None:# 处理规格参数（JSON格式）if param == "spec" and isinstance(kwargs[param], dict):params[param] = json.dumps(kwargs[param], ensure_ascii=False)else:params[param] = kwargs[param]# 3. 获取认证头部headers = self.auth_manager.get_authorization_header()if not headers:raise Exception("获取认证信息失败")# 4. 发送请求try:# 构建完整URLfull_url = f"{url}?{urlencode(params)}"response = requests.get(full_url,headers=headers,timeout=self.timeout)# 5. 解析响应result = response.json()# 6. 处理响应结果if result.get("code") == 200:return self._parse_search_result(result.get("data", {}))else:raise Exception(f"搜索商品失败: {result.get('message', '未知错误')} "f"(错误码: {result.get('code', '未知')})")except Exception as e:print(f"商品搜索接口调用异常: {str(e)}")return Nonedef _parse_search_result(self, raw_data):"""解析搜索结果"""if not raw_data or "list" not in raw_data:return None# 处理商品列表products = []for item in raw_data["list"]:# 处理规格参数spec_info = {}try:if "spec" in item and isinstance(item["spec"], str):# 尝试解析JSON格式的规格spec_info = json.loads(item["spec"])elif isinstance(item["spec"], dict):spec_info = item["spec"]except:# 如果解析失败，保留原始字符串spec_info = {"raw_spec": item.get("spec", "")}products.append({"sku": item.get("sku", ""),"name": item.get("name", ""),"brand_name": item.get("brandName", ""),"category_name": item.get("categoryName", ""),"category_id": item.get("categoryId", ""),"brand_id": item.get("brandId", ""),"price": float(item.get("price", 0)),"original_price": float(item.get("originalPrice", 0)),"min_order_quantity": int(item.get("minOrderQuantity", 1)),"packaging_quantity": int(item.get("packagingQuantity", 1)),"spec": spec_info,"certification": item.get("certification", []),"stock_status": item.get("stockStatus", ""),"stock_quantity": int(item.get("stockQuantity", 0)),"delivery_cycle": item.get("deliveryCycle", ""),"image_url": self._complete_image_url(item.get("imageUrl", "")),"is_promotion": item.get("isPromotion", False),"promotion_info": item.get("promotionInfo", ""),"search_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")})return {"search_info": {"total": raw_data.get("total", 0),"page_num": raw_data.get("pageNum", 1),"page_size": raw_data.get("pageSize", 20),"total_pages": raw_data.get("pages", 0)},"products": products}def _complete_image_url(self, url):"""补全图片URL"""if url and not url.startswith(("http://", "https://")):return f"https:{url}" if url.startswith("//") else f"https://img.zkh360.com{url}"return url

3. 商品详情与库存接口客户端

python

运行

class ZkhProductDetailClient:"""震坤行商品详情与库存接口客户端"""def __init__(self, auth_manager):self.auth_manager = auth_managerself.base_url = "https://openapi.zkh360.com/api/v2"self.timeout = 15  # 超时时间(秒)self.qps_limit = 10  # QPS限制self.last_request_time = 0self.request_lock = Lock()  # 线程锁控制QPSdef _check_qps(self):"""检查并控制QPS"""with self.request_lock:current_time = time.time()interval = 1.0 / self.qps_limit  # 每次请求最小间隔elapsed = current_time - self.last_request_timeif elapsed < interval:# 需要等待的时间time.sleep(interval - elapsed)self.last_request_time = time.time()def get_product_detail(self, sku):"""获取商品详情:param sku: 商品SKU编码:return: 商品详情字典"""# 检查QPS限制self._check_qps()# 1. 构造请求URLurl = f"{self.base_url}/products/{sku}"# 2. 获取认证头部headers = self.auth_manager.get_authorization_header()if not headers:raise Exception("获取认证信息失败")# 3. 发送请求try:response = requests.get(url,headers=headers,timeout=self.timeout)# 4. 解析响应result = response.json()# 5. 处理响应结果if result.get("code") == 200:return self._parse_product_detail(result.get("data", {}))else:raise Exception(f"获取商品详情失败: {result.get('message', '未知错误')} "f"(错误码: {result.get('code', '未知')})")except Exception as e:print(f"商品详情接口调用异常: {str(e)}")return Nonedef batch_get_inventory(self, sku_list, warehouse_ids=None):"""批量查询库存:param sku_list: SKU列表:param warehouse_ids: 仓库ID列表，None表示所有仓库:return: 库存信息字典"""if not sku_list or len(sku_list) > 100:raise ValueError("SKU列表不能为空且最多包含100个SKU")# 检查QPS限制self._check_qps()# 1. 构造请求URLurl = f"{self.base_url}/inventory"# 2. 构建请求参数data = {"skus": sku_list}if warehouse_ids:data["warehouseIds"] = warehouse_ids# 3. 获取认证头部headers = self.auth_manager.get_authorization_header()if not headers:raise Exception("获取认证信息失败")# 4. 发送请求try:response = requests.post(url,json=data,headers=headers,timeout=self.timeout)# 5. 解析响应result = response.json()# 6. 处理响应结果if result.get("code") == 200:return self._parse_inventory_result(result.get("data", {}))else:raise Exception(f"查询库存失败: {result.get('message', '未知错误')} "f"(错误码: {result.get('code', '未知')})")except Exception as e:print(f"库存查询接口调用异常: {str(e)}")return Nonedef _parse_product_detail(self, raw_data):"""解析商品详情"""if not raw_data:return None# 处理规格参数spec_info = {}try:if "spec" in raw_data and isinstance(raw_data["spec"], str):spec_info = json.loads(raw_data["spec"])elif isinstance(raw_data["spec"], dict):spec_info = raw_data["spec"]except:spec_info = {"raw_spec": raw_data.get("spec", "")}# 处理技术参数technical_params = []if "technicalParams" in raw_data and isinstance(raw_data["technicalParams"], list):technical_params = [{"name": param.get("name", ""),"value": param.get("value", ""),"unit": param.get("unit", "")} for param in raw_data["technicalParams"]]# 处理图片images = []if "images" in raw_data and isinstance(raw_data["images"], list):images = [self._complete_image_url(img) for img in raw_data["images"]]# 处理包装信息packaging_info = {}if "packaging" in raw_data:packaging = raw_data["packaging"]packaging_info = {"quantity": int(packaging.get("quantity", 1)),"unit": packaging.get("unit", ""),"weight": float(packaging.get("weight", 0)),"weightUnit": packaging.get("weightUnit", "kg"),"dimensions": packaging.get("dimensions", {})}return {"sku": raw_data.get("sku", ""),"name": raw_data.get("name", ""),"full_name": raw_data.get("fullName", ""),"brand": {"id": raw_data.get("brandId", ""),"name": raw_data.get("brandName", "")},"category": {"id": raw_data.get("categoryId", ""),"name": raw_data.get("categoryName", ""),"path": raw_data.get("categoryPath", "")},"price": {"current": float(raw_data.get("price", 0)),"original": float(raw_data.get("originalPrice", 0)),"currency": raw_data.get("currency", "CNY"),"price_levels": raw_data.get("priceLevels", [])  # 阶梯价格},"min_order_quantity": int(raw_data.get("minOrderQuantity", 1)),"packaging": packaging_info,"spec": spec_info,"technical_params": technical_params,"certification": raw_data.get("certification", []),"certification_files": raw_data.get("certificationFiles", []),"application": raw_data.get("application", ""),"description": raw_data.get("description", ""),"usage": raw_data.get("usage", ""),"storage": raw_data.get("storage", ""),"images": images,"videos": raw_data.get("videos", []),"warranty": raw_data.get("warranty", ""),"manufacturer": raw_data.get("manufacturer", ""),"production_date": raw_data.get("productionDate", ""),"expiry_date": raw_data.get("expiryDate", ""),"updated_at": raw_data.get("updatedAt", ""),"fetch_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")}def _parse_inventory_result(self, raw_data):"""解析库存查询结果"""if not raw_data or "inventoryList" not in raw_data:return Noneinventory_map = {}for item in raw_data["inventoryList"]:warehouses = []if "warehouses" in item and isinstance(item["warehouses"], list):warehouses = [{"id": wh.get("id", ""),"name": wh.get("name", ""),"location": wh.get("location", ""),"quantity": int(wh.get("quantity", 0)),"lockQuantity": int(wh.get("lockQuantity", 0)),"availableQuantity": int(wh.get("availableQuantity", 0)),"updateTime": wh.get("updateTime", "")} for wh in item["warehouses"]]inventory_map[item.get("sku", "")] = {"sku": item.get("sku", ""),"total_quantity": int(item.get("totalQuantity", 0)),"total_available": int(item.get("totalAvailableQuantity", 0)),"warehouses": warehouses,"update_time": item.get("updateTime", ""),"fetch_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")}return {"total_skus": len(inventory_map),"inventory_map": inventory_map,"query_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")}def _complete_image_url(self, url):"""补全图片URL"""if url and not url.startswith(("http://", "https://")):return f"https:{url}" if url.startswith("//") else f"https://img.zkh360.com{url}"return url

4. 工业品数据管理工具（缓存与批量处理）

python

运行

import os
import json
import sqlite3
from datetime import datetime, timedelta
import time
import hashlib
from concurrent.futures import ThreadPoolExecutorclass ZkhProductManager:"""震坤行工业品数据管理工具，支持缓存与批量处理"""def __init__(self, client_id, client_secret, cache_dir="./zkh_cache"):self.auth_manager = ZkhAuthManager(client_id, client_secret)self.search_client = ZkhProductSearchClient(self.auth_manager)self.detail_client = ZkhProductDetailClient(self.auth_manager)self.cache_dir = cache_dirself.db_path = os.path.join(cache_dir, "zkh_product_cache.db")self._init_cache()def _init_cache(self):"""初始化缓存数据库"""if not os.path.exists(self.cache_dir):os.makedirs(self.cache_dir)# 连接数据库conn = sqlite3.connect(self.db_path)cursor = conn.cursor()# 创建商品详情缓存表cursor.execute('''CREATE TABLE IF NOT EXISTS product_cache (sku TEXT PRIMARY KEY,data TEXT,fetch_time TEXT)''')# 创建搜索缓存表cursor.execute('''CREATE TABLE IF NOT EXISTS search_cache (cache_key TEXT PRIMARY KEY,data TEXT,fetch_time TEXT,keyword TEXT)''')# 创建库存缓存表cursor.execute('''CREATE TABLE IF NOT EXISTS inventory_cache (sku TEXT PRIMARY KEY,data TEXT,fetch_time TEXT)''')conn.commit()conn.close()def search_products(self, keyword, use_cache=True, cache_ttl=1800, **kwargs):"""搜索商品，支持缓存:param keyword: 搜索关键词:param use_cache: 是否使用缓存:param cache_ttl: 缓存有效期(秒):param kwargs: 其他搜索参数:return: 搜索结果"""# 生成缓存键cache_key = self._generate_cache_key(keyword,** kwargs)# 尝试从缓存获取if use_cache:cached_data = self._get_cached_data("search_cache", cache_key, cache_ttl)if cached_data:print(f"使用缓存数据，关键词: {keyword}，页码: {kwargs.get('page', 1)}")return cached_data# 从接口获取print(f"调用接口搜索，关键词: {keyword}，页码: {kwargs.get('page', 1)}")result = self.search_client.search_products(keyword, **kwargs)# 更新缓存if result:self._update_search_cache(cache_key, result, keyword)return resultdef get_product_detail(self, sku, use_cache=True, cache_ttl=3600):"""获取商品详情:param sku: 商品SKU:param use_cache: 是否使用缓存:param cache_ttl: 缓存有效期(秒):return: 商品详情"""# 尝试从缓存获取if use_cache:cached_data = self._get_cached_data("product_cache", sku, cache_ttl)if cached_data:print(f"使用缓存数据，SKU: {sku}")return cached_data# 从接口获取print(f"调用接口获取商品详情，SKU: {sku}")result = self.detail_client.get_product_detail(sku)# 更新缓存if result:self._update_product_cache(sku, result)return resultdef get_inventory(self, sku_list, use_cache=True, cache_ttl=300, warehouse_ids=None):"""获取库存信息:param sku_list: SKU列表:param use_cache: 是否使用缓存:param cache_ttl: 缓存有效期(秒):param warehouse_ids: 仓库ID列表:return: 库存信息"""if not sku_list:return None# 分离需要从缓存和接口获取的SKUfrom_cache = {}need_query = []if use_cache:for sku in sku_list:cached = self._get_cached_data("inventory_cache", sku, cache_ttl)if cached:from_cache[sku] = cachedelse:need_query.append(sku)else:need_query = sku_list# 从接口获取需要查询的SKUapi_result = Noneif need_query:# 分批查询，每批最多100个batch_size = 100all_inventory = {}for i in range(0, len(need_query), batch_size):batch = need_query[i:i+batch_size]batch_result = self.detail_client.batch_get_inventory(batch, warehouse_ids)if batch_result and "inventory_map" in batch_result:all_inventory.update(batch_result["inventory_map"])# 更新缓存for sku, data in batch_result["inventory_map"].items():self._update_inventory_cache(sku, data)api_result = all_inventory# 合并结果result = from_cache.copy()if api_result:result.update(api_result)return {"inventory_map": result,"query_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")}def batch_get_product_details(self, sku_list, use_cache=True, cache_ttl=3600, max_workers=3):"""批量获取商品详情:param sku_list: SKU列表:param use_cache: 是否使用缓存:param cache_ttl: 缓存有效期:param max_workers: 线程池大小:return: 商品详情列表"""if not sku_list:return []# 使用线程池并行获取with ThreadPoolExecutor(max_workers=max_workers) as executor:# 提交任务futures = [executor.submit(self.get_product_detail, sku, use_cache=use_cache, cache_ttl=cache_ttl) for sku in sku_list]# 获取结果results = []for future in futures:try:result = future.result()if result:results.append(result)except Exception as e:print(f"获取商品详情异常: {str(e)}")return resultsdef _generate_cache_key(self, keyword, **params):"""生成搜索缓存键"""sorted_params = sorted(params.items(), key=lambda x: x[0])params_str = "&".join([f"{k}={v}" for k, v in sorted_params])return hashlib.md5(f"search_{keyword}_{params_str}".encode()).hexdigest()def _get_cached_data(self, table, key, ttl):"""从缓存获取数据"""conn = sqlite3.connect(self.db_path)cursor = conn.cursor()# 确定查询字段名field_name = "cache_key" if table == "search_cache" else f"{table[:-6]}_id"cursor.execute(f"SELECT data, fetch_time FROM {table} WHERE {field_name} = ?",(key,))result = cursor.fetchone()conn.close()if result:data_str, fetch_time = result# 检查缓存是否过期fetch_time_obj = datetime.strptime(fetch_time, "%Y-%m-%d %H:%M:%S")if (datetime.now() - fetch_time_obj).total_seconds() <= ttl:try:return json.loads(data_str)except:return Nonereturn Nonedef _update_product_cache(self, sku, data):"""更新商品详情缓存"""if not data or "sku" not in data:returnconn = sqlite3.connect(self.db_path)cursor = conn.cursor()data_str = json.dumps(data, ensure_ascii=False)fetch_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")cursor.execute('''INSERT OR REPLACE INTO product_cache (sku, data, fetch_time)VALUES (?, ?, ?)''', (sku, data_str, fetch_time))conn.commit()conn.close()def _update_search_cache(self, cache_key, data, keyword):"""更新搜索缓存"""if not data:returnconn = sqlite3.connect(self.db_path)cursor = conn.cursor()data_str = json.dumps(data, ensure_ascii=False)fetch_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")cursor.execute('''INSERT OR REPLACE INTO search_cache (cache_key, data, fetch_time, keyword)VALUES (?, ?, ?, ?)''', (cache_key, data_str, fetch_time, keyword))conn.commit()conn.close()def _update_inventory_cache(self, sku, data):"""更新库存缓存"""if not data or "sku" not in data:returnconn = sqlite3.connect(self.db_path)cursor = conn.cursor()data_str = json.dumps(data, ensure_ascii=False)fetch_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")cursor.execute('''INSERT OR REPLACE INTO inventory_cache (sku, data, fetch_time)VALUES (?, ?, ?)''', (sku, data_str, fetch_time))conn.commit()conn.close()def clean_expired_cache(self, max_age=86400):"""清理过期缓存，默认保留24小时内的数据"""conn = sqlite3.connect(self.db_path)cursor = conn.cursor()# 计算过期时间expire_time = (datetime.now() - timedelta(seconds=max_age)).strftime("%Y-%m-%d %H:%M:%S")# 清理商品缓存cursor.execute("DELETE FROM product_cache WHERE fetch_time < ?", (expire_time,))deleted_products = cursor.rowcount# 清理搜索缓存cursor.execute("DELETE FROM search_cache WHERE fetch_time < ?", (expire_time,))deleted_searches = cursor.rowcount# 清理库存缓存cursor.execute("DELETE FROM inventory_cache WHERE fetch_time < ?", (expire_time,))deleted_inventory = cursor.rowcountconn.commit()conn.close()print(f"清理过期缓存完成，删除商品缓存 {deleted_products} 条，搜索缓存 {deleted_searches} 条，库存缓存 {deleted_inventory} 条")return {"deleted_products": deleted_products,"deleted_searches": deleted_searches,"deleted_inventory": deleted_inventory}

四、完整使用示例

1. 工业品搜索与详情获取示例

python

运行

def industrial_product_search_demo():# 替换为实际的应用信息CLIENT_ID = "your_client_id"CLIENT_SECRET = "your_client_secret"# 初始化产品管理器product_manager = ZkhProductManager(CLIENT_ID, CLIENT_SECRET)# 搜索参数：查找特定规格的螺栓search_params = {"keyword": "内六角螺栓","spec": {"直径": "M8","长度": "30mm","强度等级": "8.8级"},"priceMin": 0.5,"priceMax": 5.0,"inStock": True,"page": 1,"size": 10,"sort": "price_asc"}# 执行搜索result = product_manager.search_products(**search_params,use_cache=True,cache_ttl=1800  # 缓存30分钟)if result:print(f"===== 工业品搜索结果: {search_params['keyword']} =====")print(f"共搜索到 {result['search_info']['total']} 个商品，第 {result['search_info']['page_num']}/{result['search_info']['total_pages']} 页")# 打印搜索结果for i, product in enumerate(result["products"], 1):print(f"\n{i}. {product['name']}")print(f"品牌: {product['brand_name']}")print(f"规格: {json.dumps(product['spec'], ensure_ascii=False)}")print(f"价格: ¥{product['price']} (起订量: {product['min_order_quantity']})")print(f"库存: {product['stock_quantity']} {product['stock_status']}")print(f"交付周期: {product['delivery_cycle']}")# 获取第一个商品的详细信息if result["products"]:first_sku = result["products"][0]["sku"]product_detail = product_manager.get_product_detail(first_sku)if product_detail:print(f"\n===== 商品详情: {product_detail['name']} =====")print(f"完整名称: {product_detail['full_name']}")print(f"分类路径: {product_detail['category']['path']}")print(f"阶梯价格: {json.dumps(product_detail['price']['price_levels'], ensure_ascii=False)}")print(f"包装信息: {json.dumps(product_detail['packaging'], ensure_ascii=False)}")# 打印技术参数print("\n主要技术参数:")for param in product_detail["technical_params"][:5]:  # 只显示前5个print(f"- {param['name']}: {param['value']} {param['unit'] or ''}")# 打印认证信息if product_detail["certification"]:print(f"\n认证信息: {', '.join(product_detail['certification'])}")# 清理过期缓存product_manager.clean_expired_cache()if __name__ == "__main__":industrial_product_search_demo()

2. 批量库存查询与分析示例

python

运行

def inventory_query_demo():# 替换为实际的应用信息CLIENT_ID = "your_client_id"CLIENT_SECRET = "your_client_secret"# 初始化产品管理器product_manager = ZkhProductManager(CLIENT_ID, CLIENT_SECRET)# 要查询的商品SKU列表target_skus = ["ZKH12345678", "ZKH12345679", "ZKH12345680","ZKH12345681", "ZKH12345682"]# 查询库存（指定仓库ID，可从震坤行平台获取仓库列表）warehouse_ids = ["WH001", "WH003"]  # 示例仓库IDinventory_result = product_manager.get_inventory(target_skus,use_cache=True,cache_ttl=300,  # 库存数据缓存5分钟warehouse_ids=warehouse_ids)if inventory_result and "inventory_map" in inventory_result:print(f"===== 库存查询结果 =====")for sku, inventory in inventory_result["inventory_map"].items():print(f"\nSKU: {sku}")print(f"总库存: {inventory['total_quantity']}，可用库存: {inventory['total_available']}")# 打印各仓库库存print("仓库库存分布:")for wh in inventory["warehouses"]:print(f"- {wh['name']}({wh['location']}): "f"库存 {wh['quantity']}, 可用 {wh['availableQuantity']}")# 批量获取商品详情details = product_manager.batch_get_product_details(target_skus, max_workers=3)if details:print(f"\n===== 批量商品详情获取完成，共 {len(details)} 个商品 =====")# 可以在这里添加价格对比、规格分析等逻辑if __name__ == "__main__":inventory_query_demo()

五、常见问题与优化建议

1. 常见错误码及解决方案

错误码	说明	解决方案
200	成功	-
400	参数错误	检查参数格式和取值范围，特别是 spec 参数的 JSON 格式
401	未授权	检查 access_token 是否有效，尝试重新获取令牌
403	权限不足	检查应用是否已申请该接口权限
404	资源不存在	检查 SKU 或其他 ID 是否正确
429	频率超限	降低调用频率，确保不超过 QPS 限制
500	服务器错误	记录错误信息，稍后重试
503	服务不可用	等待服务恢复或联系平台技术支持