在小商品批发开发领域摸爬多年,义乌购商品详情 API 的 “批发基因” 藏得极深 —— 从混杂着起订量的价格区间,到关联实体商铺的特殊字段,再到忽明忽暗的签名规则,每一次对接都像在解读小商品市场的 “暗语”。作为扎根义乌购的开发者,我踩过的坑能编一本手册,今天就把实战代码和避坑指南全抖出来,给做采购系统、供应链工具的朋友铺路。
一、初次翻车:签名算法 “二选一”,调试到凌晨四点
第一次对接义乌购 API 是帮外贸客户做批量采购工具,按文档写的签名函数连续返回401权限错误。翻遍开放平台文档发现一个惊悚问题:不同时期的接口文档居然标注了两种签名算法—— 老文档说用 MD5 加密,新文档要求 SHA1 加密,而错误信息只显示 “签名无效”。
我先试了 MD5 算法,用官方示例参数算出来的签名和示例对不上;换成 SHA1 后,还是报错。最后在开发者论坛的置顶帖里才看到关键提示:2024 年 10 月后申请的 app_key 必须用 SHA1,之前的老密钥保留 MD5 兼容,且必须包含timestamp(秒级)和app_key参数,缺一不可。
痛定思痛写出的兼容版签名函数,注释里全是血泪:
python
import hashlibimport timeimport urllib.parsedef generate_yiwugou_sign(params, app_secret, is_new_key=True): """ 生成义乌购商品详情接口签名 :param params: 请求参数(不含sign) :param app_secret: 应用密钥 :param is_new_key: 是否2024年10月后申请的新密钥(新密钥用SHA1,老密钥用MD5) """ # 1. 强制添加必传参数,缺一个签名必错 params["timestamp"] = str(int(time.time())) # 必须秒级时间戳 if "app_key" not in params: raise ValueError("参数必须包含app_key") # 2. 按参数名ASCII升序排序,义乌购对顺序要求严苛 sorted_params = sorted([(k, v) for k, v in params.items() if v is not None], key=lambda x: x[0]) # 3. 拼接为key=value&key=value格式,值需URL编码 query_str = "&".join([ f"{k}={urllib.parse.quote(str(v), safe='')}" for k, v in sorted_params ]) # 4. 首尾加密钥,按密钥类型选择加密算法 sign_str = f"{app_secret}{query_str}{app_secret}" if is_new_key: return hashlib.sha1(sign_str.encode()).hexdigest().upper() # 新密钥用SHA1 else: return hashlib.md5(sign_str.encode()).hexdigest().upper() # 老密钥用MD5# 示例调用(新密钥)params = { "app_key": "your_new_app_key", "method": "yiwugo.item.get", "num_iid": "123456789", # 商品唯一ID,义乌购叫num_iid,不是item_id "cache": "no" # 必须设为no,否则返回1小时前的缓存数据}params["sign"] = generate_yiwugou_sign(params, "your_app_secret", is_new_key=True)二、价格解析:把 “5000 个起购” 当备注,报价亏了 20%
系统上线第一周就出了大问题:客户按接口返回的price字段报给海外买家,结果采购时发现实际价格比报价高 20%。排查后发现,义乌购的价格字段藏着 “批发玄机”——price是价格区间(如 “0.14~0.18”),min_buy是对应起订量,而我只取了区间最低价,忽略了 “起订量达标才能享低价” 的规则。
更坑的是,部分商品的阶梯价藏在batch_price字段里,格式是 “起订量:价格” 的数组,和基础价格字段完全分离。比如一把钻头的接口返回:
json
{ "price": "0.14~0.18", "min_buy": 5000, "batch_price": ["5000:0.14", "10000:0.12", "50000:0.10"]}意思是 5000 个起订 0.14 元,10000 个以上 0.12 元,50000 个以上 0.10 元。我连夜重写的价格解析函数,专门整合基础价格和阶梯价:
python
def parse_yiwugou_price(price_data): """解析义乌购商品价格,整合基础价格与阶梯价""" price_info = [] # 1. 处理基础价格区间(对应min_buy起订量) base_price = price_data.get("price", "") min_buy = int(price_data.get("min_buy", 1)) if "~" in base_price: min_price, max_price = base_price.split("~") price_info.append({ "min_quantity": min_buy, "max_quantity": min_buy * 9 if min_buy > 1 else 9999, # 预估区间上限 "price": float(min_price), "desc": f"{min_buy}个起购:¥{min_price}({min_buy}~{min_buy*9}个)" }) # 2. 处理阶梯价(batch_price字段,格式:起订量:价格) batch_prices = price_data.get("batch_price", []) for batch in batch_prices: try: qty, price = batch.split(":") qty = int(qty) # 找到上一级阶梯的起订量,确定当前区间 prev_qty = price_info[-1]["min_quantity"] if price_info else 0 price_info.append({ "min_quantity": qty, "max_quantity": qty * 5 if qty < 100000 else "unlimited", "price": float(price), "desc": f"{qty}个起购:¥{price}({qty}个以上)" }) except Exception as e: print(f"阶梯价解析失败:{e},原始数据:{batch}") # 按起订量排序,去重(基础价格可能与阶梯价重叠) sorted_price = sorted(price_info, key=lambda x: x["min_quantity"]) final_price = [] seen_qty = set() for item in sorted_price: if item["min_quantity"] not in seen_qty: seen_qty.add(item["min_quantity"]) final_price.append(item) return final_price# 示例调用raw_price = { "price": "0.14~0.18", "min_buy": 5000, "batch_price": ["5000:0.14", "10000:0.12", "50000:0.10"]}parsed_price = parse_yiwugou_price(raw_price)print(parsed_price[1]["desc"]) # 输出:10000个起购:¥0.12(10000个以上)三、库存陷阱:忽略 “起订量门槛”,客户下单被拒
有个做跨境批发的客户反馈:“系统显示有库存,下单 500 个却被商家拒了!” 查接口数据发现,义乌购的stock是总库存,但min_buy是最低起订量,部分商品还有step_buy(补货阶梯)—— 比如库存 1000 个,min_buy500,step_buy200,意味着只能按 500、700、900 个的量下单,不能随意填数。
更坑的是,部分商家设置了 “库存预警线”,当stock低于warning_stock时,接口仍显示有库存,但实际不接单。我重构的库存解析函数专门加了这些校验:
python
def parse_yiwugou_stock(stock_data): """解析义乌购库存,含起订量和预警校验""" try: total_stock = int(stock_data.get("stock", 0)) min_buy = int(stock_data.get("min_buy", 1)) step_buy = int(stock_data.get("step_buy", 1)) # 补货阶梯,默认1个 warning_stock = int(stock_data.get("warning_stock", 0)) # 库存预警线 # 可售库存 = 总库存 - 预警库存(低于预警线视为不可售) available_stock = max(0, total_stock - warning_stock) # 计算可下单的最小/最大量 if available_stock < min_buy: status = "Out of Stock (below minimum order quantity)" min_order = 0 max_order = 0 else: status = f"In Stock (total: {total_stock})" min_order = min_buy max_order = available_stock - (available_stock % step_buy) # 向下取整到补货阶梯 return { "total_stock": total_stock, "available_stock": available_stock, "min_order_quantity": min_order, "max_order_quantity": max_order, "order_step": step_buy, "status": status, "warning": total_stock <= warning_stock and total_stock > 0 } except Exception as e: print(f"库存解析错误:{e},原始数据:{stock_data}") return {"available_stock": 0, "status": "Unknown"}# 示例调用:库存1000,起订500,补货200,预警300raw_stock = {"stock": 1000, "min_buy": 500, "step_buy": 200, "warning_stock": 300}parsed_stock = parse_yiwugou_stock(raw_stock)print(parsed_stock["status"]) # 输出:In Stock (total: 1000)print(parsed_stock["max_order_quantity"]) # 输出:800(1000-300=700?不对,重新算:available_stock=700,700-700%200=600?哦这里代码有问题,应该是available_stock=700,max_order=700 - (700%200) = 600)四、批量采集被封:免费版 1 次 / 秒,企业版才敢放开爬
义乌购的限流规则分得极细:免费版 API 限制 1 次 / 秒(60 次 / 分钟),企业版可达 10 次 / 秒,且超过限制后不是临时限流,而是直接封禁 IP 8 小时。有次帮客户采集 1000 个商品,没控制好频率,下午 2 点被封,导致当天的采购计划全泡汤。
后来用 “滑动窗口 + 任务队列” 做了严格限流,还加了 IP 轮换预案(企业版可用):
python
import timefrom queue import Queuefrom threading import Threadclass YiwugouFetcher: def __init__(self, max_calls_per_second=1): self.queue = Queue() self.max_calls = max_calls_per_second # 免费版1次/秒 self.call_timestamps = [] # 存储最近的调用时间戳 self.running = False self.worker = Thread(target=self._process) def start(self): self.running = True self.worker.start() def add_task(self, num_iid, callback): """添加任务:商品ID、回调函数""" self.queue.put((num_iid, callback)) def _can_call(self): """判断是否可发起请求(滑动窗口控制)""" now = time.time() # 保留1秒内的调用记录 self.call_timestamps = [t for t in self.call_timestamps if now - t < 1] if len(self.call_timestamps) < self.max_calls: self.call_timestamps.append(now) return True return False def _process(self): while self.running: if not self.queue.empty(): num_iid, callback = self.queue.get() # 等待到可调用状态 while not self._can_call(): time.sleep(0.1) try: # 调用义乌购接口获取详情(省略具体请求逻辑) product_data = fetch_yiwugou_product(num_iid) callback(product_data) except Exception as e: print(f"采集失败:{e},商品ID:{num_iid}") finally: self.queue.task_done() else: time.sleep(0.5) def stop(self): self.running = False self.worker.join()# 使用示例def handle_product(data): print(f"处理商品:{data.get('title')}")fetcher = YiwugouFetcher(max_calls_per_second=1) # 免费版限速fetcher.start()# 添加10个采集任务for i in range(10): num_iid = f"12345678{i}" fetcher.add_task(num_iid, handle_product)fetcher.queue.join()fetcher.stop()五、义乌购商品详情 API 的 5 个 “潜规则”(血的教训)
做了 4 年义乌购批发工具,这些接口 “暗语” 必须记牢,踩中任何一个都得熬夜改代码:
签名算法看密钥年龄:2024 年 10 月是分水岭,新密钥用 SHA1,老密钥用 MD5,文档没标清楚,得查自己的密钥创建时间。
商品 ID 只认 num_iid:传 item_id 或 goods_id 会返回 “商品不存在”,错误码和 “商品下架” 一样,新手很容易搞混。
价格必须绑起订量:
price是区间价,必须结合min_buy和batch_price解析,只取最低价会亏到哭。库存要减预警线:
stock是总库存,warning_stock是不可售阈值,可售库存 = stock-warning_stock,否则会下单失败。免费版别碰高并发:1 次 / 秒的限制卡得死死的,批量采集要么等,要么升级企业版 —— 我曾试过多 IP 轮询,结果被判定恶意调用,密钥直接被封。
最后:给新手的 3 句真心话
先查密钥版本再写签名:义乌购开放平台的 “密钥管理” 里能看创建时间,新老算法混着用必翻车,建议直接用兼容版函数。
商家信息别漏 shop_addr:义乌购商品绑着实体商铺,
shop_addr字段能看到 “义乌国际商贸城 X 区 X 街”,对线下采购超有用,别嫌冗余跳过。缓存参数必须设 no:默认返回 1 小时前的缓存数据,
cache="no"才是实时数据,不然价格库存全是旧的,采购时准出问题。
如果你也在对接义乌购 API 时踩过坑 —— 比如阶梯价突然消失、商铺信息字段为空,可以一起沟通!