标签，需关联价格接口获取对应阶梯价。
三、接口调用流程（基于页面解析）
以 “获取某卡通贴纸商品详情” 为例，核心流程为URL 构建→主页面解析→动态数据补充→规格与价格关联→数据结构化：
URL 与请求头构建
目标 URL： https://www.yiwugou.com/product/detail/{item_id}.html（替换item_id为实际值）；
请求头：模拟采购商浏览器行为，需包含User-Agent、Referer，部分批发价需携带登录态Cookie：
python
运行
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/118.0.0.0 Safari/537.36",
"Referer": " https://www.yiwugou.com/category/100120/", # 对应类目页
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9, /;q=0.8",
"Cookie": "PHPSESSID=xxx; user_id=xxx" # 登录后从浏览器获取
}
主页面静态数据解析从主页面 HTML 中提取基础信息，重点关注批发特有字段：
字段 解析方式（CSS 选择器 / XPath） 示例值
商品标题 h1.product-title（CSS 选择器） “3cm 卡通贴纸 儿童玩具配件 100 张 / 包”
图片列表 div.gallery img的src属性（主图、细节图） [" https://img.yiwugou.com/xxx.jpg", ...]
基础批发价 div.price-wholesale的文本（提取单阶价格） “1.2 元 / 件”
起订量 div.moq的文本（提取数字） “≥100 件” 提取 “100”
供应商 div.shop-name a的文本 “义乌市 XX 玩具商行”
商位信息 div.shop-location的文本 “二区 45 号门 3 楼 12345 商位”
诚信等级 div.credit-level的文本 “5 年诚信商铺 已实名认证”
规格选项 div.spec-select .spec-item的data-spec属性 ["3cm-卡通", "5cm-动物"]
动态数据补充（AJAX 接口）阶梯价、实时库存、物流信息等通过动态接口加载，需单独调用：
阶梯价接口示例： https://www.yiwugou.com/ajax/product/price?item_id={item_id}；
响应示例（多阶梯价格）：
json
{
"data": {
"price_ladder": [
{"quantity": 100, "price": 1.2, "unit": "件"},
{"quantity": 500, "price": 1.0, "unit": "件"},
{"quantity": 1000, "price": 0.8, "unit": "件"}
]
}
}
库存接口示例： https://www.yiwugou.com/ajax/product/stock?item_id={item_id}，返回实时库存数量及状态。
规格与价格关联多规格商品（如不同尺寸 / 图案）的价格和库存可能不同，需将静态规格选项与动态数据关联：
从静态页面提取所有规格（如["3cm-卡通", "5cm-动物"]）；
从阶梯价接口获取每个规格对应的多阶梯价格（若规格独立定价）；
示例：3cm - 卡通（100 件 1.2 元 / 件）、5cm - 动物（100 件 1.5 元 / 件）。
数据整合与结构化合并静态、动态数据，形成标准化字典，突出批发特性：
python
运行
standardized_data = {
"item_id": item_id,
"title": title,
"images": {
"main": main_images, # 主图
"detail": detail_images, # 细节图
"package": package_images # 包装图
},
"price": {
"ladder": price_ladder, # 多阶梯价格
"retail": retail_price, # 市场价（零售参考）
"discount": discount_info # 优惠活动
},
"specs": [
{
"name": spec_name, # 如“3cm-卡通”
"stock": spec_stock, # 该规格库存
"ladder_price": spec_ladder # 该规格的阶梯价（若独立定价）
} for spec_name, spec_stock, spec_ladder in spec_list
],
"trade": {
"moq": moq, # 起订量
"mix_batch": mix_batch_rule, # 混批规则
"delivery_time": delivery_time # 发货时效
},
"supplier": {
"name": shop_name,
"location": shop_location, # 商位地址
"credit": credit_level, # 诚信等级
"contact": contact_info # 联系方式
},
"logistics": {
"default_express": default_express, # 默认快递
"freight_rule": freight_rule # 运费规则
},
"url": detail_url
}
四、代码实现示例（Python）
以下是item_get接口的完整实现，包含主页面解析、动态接口调用、规格价格关联及反爬处理：
import requests
import time
import random
import re
import json
from bs4 import BeautifulSoup
from fake_useragent import UserAgent
from typing import Dict, List

class YiwugouItemApi:
def init(self, proxy_pool: List[str] = None, cookie: str = ""):
self.base_url = "https://www.yiwugou.com/product/detail/{item_id}.html"
self.price_api = "https://www.yiwugou.com/ajax/product/price" # 阶梯价接口
self.stock_api = "https://www.yiwugou.com/ajax/product/stock" # 库存接口
self.ua = UserAgent()
self.proxy_pool = proxy_pool # 代理池列表
self.cookie = cookie # 登录态Cookie（用于完整价格）

def _get_headers(self) -> Dict[str, str]:
    """生成随机请求头"""
    headers = {
        "User-Agent": self.ua.random,
        "Referer": "https://www.yiwugou.com/category/",
        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8"
    }
    if self.cookie:
        headers["Cookie"] = self.cookie
    return headers

def _get_proxy(self) -> Dict[str, str]:
    """随机获取代理"""
    if self.proxy_pool and len(self.proxy_pool) > 0:
        proxy = random.choice(self.proxy_pool)
        return {"http": proxy, "https": proxy}
    return None

def _clean_price(self, price_str: str) -> float:
    """清洗价格字符串（去除元/件、¥等）"""
    if not price_str:
        return 0.0
    price_str = re.sub(r"[^\d.]", "", price_str)
    return float(price_str) if price_str else 0.0

def _clean_moq(self, moq_str: str) -> int:
    """清洗起订量（提取数字）"""
    if not moq_str:
        return 1
    moq_num = re.search(r"\d+", moq_str)
    return int(moq_num.group()) if moq_num else 1

def _parse_static_data(self, html: str) -> Dict:
    """解析主页面静态数据"""
    soup = BeautifulSoup(html, "lxml")

    # 提取规格选项
    spec_list = []
    for spec_item in soup.select("div.spec-select .spec-item"):
        spec_name = spec_item.get("data-spec") or spec_item.text.strip()
        spec_list.append(spec_name)

    # 提取供应商联系方式（电话/微信）
    contact_info = {}
    phone_tag = soup.select_one("div.contact-phone")
    if phone_tag:
        contact_info["phone"] = re.sub(r"\D", "", phone_tag.text.strip())
    wechat_tag = soup.select_one("div.contact-wechat")
    if wechat_tag:
        contact_info["wechat"] = wechat_tag.text.strip().replace("微信：", "")

    return {
        "title": soup.select_one("h1.product-title")?.text.strip() or "",
        "images": {
            "main": [img.get("src") for img in soup.select("div.main-gallery img") if img.get("src")],
            "detail": [img.get("src") for img in soup.select("div.detail-images img") if img.get("src")],
            "package": [img.get("src") for img in soup.select("div.package-images img") if img.get("src")]
        },
        "price": {
            "wholesale_basic": self._clean_price(soup.select_one("div.price-wholesale")?.text or ""),
            "retail": self._clean_price(soup.select_one("div.price-retail")?.text or "")
        },
        "trade": {
            "moq": self._clean_moq(soup.select_one("div.moq")?.text or ""),
            "mix_batch": soup.select_one("div.mix-batch")?.text.strip() or "",
            "delivery_time": soup.select_one("div.delivery-time")?.text.strip() or ""
        },
        "supplier": {
            "name": soup.select_one("div.shop-name a")?.text.strip() or "",
            "location": soup.select_one("div.shop-location")?.text.strip() or "",
            "credit": soup.select_one("div.credit-level")?.text.strip() or "",
            "contact": contact_info
        },
        "logistics": {
            "default_express": soup.select_one("div.default-express")?.text.strip() or "",
            "freight_rule": soup.select_one("div.freight-rule")?.text.strip() or ""
        },
        "specs": {
            "options": spec_list  # 规格选项列表
        }
    }

def _fetch_dynamic_data(self, item_id: str, headers: Dict[str, str], proxy: Dict[str, str]) -> Dict:
    """调用动态接口获取阶梯价和库存"""
    dynamic_data = {
        "price_ladder": [],  # 多阶梯价格
        "stock": 0,  # 总库存
        "spec_stock": {}  # 各规格库存（如{"3cm-卡通": 5000}）
    }
    try:
        # 1. 获取阶梯价
        price_params = {"item_id": item_id}
        price_resp = requests.get(
            self.price_api,
            params=price_params,
            headers=headers,
            proxies=proxy,
            timeout=10
        )
        price_data = price_resp.json()
        if price_data.get("status") == 1 and "data" in price_data:
            dynamic_data["price_ladder"] = price_data["data"].get("price_ladder", [])

        # 2. 获取库存
        stock_params = {"item_id": item_id}
        stock_resp = requests.get(
            self.stock_api,
            params=stock_params,
            headers=headers,
            proxies=proxy,
            timeout=10
        )
        stock_data = stock_resp.json()
        if stock_data.get("status") == 1 and "data" in stock_data:
            dynamic_data["stock"] = stock_data["data"].get("total_stock", 0)
            dynamic_data["spec_stock"] = stock_data["data"].get("spec_stock", {})

    except Exception as e:
        print(f"动态数据获取失败: {str(e)}")
    return dynamic_data

def _merge_specs_and_price(self, static_specs: List[str], price_ladder: List[Dict], spec_stock: Dict) -> List[Dict]:
    """合并规格与价格、库存（若规格独立定价则单独处理，否则共用阶梯价）"""
    merged_specs = []
    for spec in static_specs:
        # 检查是否有规格独立的阶梯价（部分商品支持）
        spec_price_ladder = next((pl for pl in price_ladder if pl.get("spec") == spec), None)
        if not spec_price_ladder:
            spec_price_ladder = price_ladder  # 共用默认阶梯价

        merged_specs.append({
            "name": spec,
            "stock": spec_stock.get(spec, 0),
            "ladder_price": spec_price_ladder
        })
    return merged_specs

def item_get(self, item_id: str, timeout: int = 10) -> Dict:
    """
    获取义乌购商品详情
    :param item_id: 商品ID（如12345678）
    :param timeout: 超时时间
    :return: 标准化商品数据
    """
    try:
        # 1. 主页面请求
        url = self.base_url.format(item_id=item_id)
        headers = self._get_headers()
        proxy = self._get_proxy()

        # 随机延迟，避免反爬
        time.sleep(random.uniform(2, 4))
        response = requests.get(
            url=url,
            headers=headers,
            proxies=proxy,
            timeout=timeout
        )
        response.raise_for_status()
        main_html = response.text

        # 2. 解析主页面数据
        static_data = self._parse_static_data(main_html)
        if not static_data["title"]:
            return {"success": False, "error_msg": "商品不存在或已下架"}

        # 3. 获取动态数据（阶梯价、库存）
        dynamic_data = self._fetch_dynamic_data(item_id, headers, proxy)

        # 4. 合并规格与价格、库存
        merged_specs = self._merge_specs_and_price(
            static_data["specs"]["options"],
            dynamic_data["price_ladder"],
            dynamic_data["spec_stock"]
        )

        # 5. 整合结果
        result = {
            "success": True,
            "data": {
                "item_id": item_id,** static_data,
                "price": {
                    **static_data["price"],
                    "ladder": dynamic_data["price_ladder"]  # 覆盖为完整阶梯价
                },
                "specs": merged_specs,
                "total_stock": dynamic_data["stock"],
                "url": url,
                "update_time": time.strftime("%Y-%m-%d %H:%M:%S")
            }
        }
        return result

    except requests.exceptions.HTTPError as e:
        if "403" in str(e):
            return {"success": False, "error_msg": "触发反爬，建议更换代理或Cookie", "code": 403}
        return {"success": False, "error_msg": f"HTTP错误: {str(e)}", "code": response.status_code}
    except Exception as e:
        return {"success": False, "error_msg": f"获取失败: {str(e)}", "code": -1}

使用示例

if name == "main":

# 代理池（替换为有效代理）
PROXIES = [
    "http://123.45.67.89:8888",
    "http://98.76.54.32:8080"
]
# 登录态Cookie（从浏览器获取，用于完整价格）
COOKIE = "PHPSESSID=xxx; user_id=xxx"

# 初始化API客户端
api = YiwugouItemApi(proxy_pool=PROXIES, cookie=COOKIE)

# 获取商品详情（示例item_id）
item_id = "12345678"  # 替换为实际商品ID
result = api.item_get(item_id)

if result["success"]:
    data = result["data"]
    print(f"商品标题: {data['title']}")
    print(f"价格信息: 基础批发价￥{data['price']['wholesale_basic']} | 市场价￥{data['price']['retail']}")
    print(f"阶梯价:")
    for ladder in data['price']['ladder'][:3]:
        print(f"  {ladder['quantity']}件及以上: ￥{ladder['price']}/{ladder['unit']}")
    print(f"交易规则: 起订量{data['trade']['moq']}件 | {data['trade']['mix_batch']} | 发货时效: {data['trade']['delivery_time']}")
    print(f"供应商: {data['supplier']['name']} | 商位: {data['supplier']['location']} | 诚信等级: {data['supplier']['credit']}")
    print(f"联系方式: 电话{data['supplier']['contact'].get('phone', '未知')} | 微信{data['supplier']['contact'].get('wechat', '未知')}")
    print(f"规格与库存:")
    for spec in data['specs'][:3]:
        print(f"  {spec['name']}: 库存{spec['stock']}件 | 阶梯价: {[f'{l["quantity"]}件￥{l["price"]}' for l in spec['ladder_price'][:2]]}")
    print(f"物流: 默认快递{data['logistics']['default_express']} | 运费规则: {data['logistics']['freight_rule']}")
else:
    print(f"获取失败: {result['error_msg']}（错误码: {result.get('code')}）")
五、关键技术难点与解决方案

多阶梯批发价解析
问题：义乌购商品价格按采购量分阶梯（如 “100 件 1.2 元、500 件 1.0 元”），格式多样（文本 / 动态接口），需结构化提取。
解决方案：
优先调用动态价格接口（返回结构化price_ladder），若接口不可用则解析页面文本（用正则匹配 “X 件 - Y 元”）；
统一转换为阶梯列表（[{"quantity": 100, "price": 1.2}, ...]），支持采购成本计算；
示例代码中_fetch_dynamic_data函数获取完整阶梯价，适配批发场景的批量采购决策。
多规格与价格 / 库存关联
问题：小商品常含多规格（如不同尺寸 / 图案），且各规格可能有独立价格和库存，需精准关联避免混淆。
解决方案：
从静态页面提取规格名称列表（如["3cm-卡通", "5cm-动物"]）；
从动态接口获取spec_stock（规格 - 库存映射）和规格专属阶梯价；
遍历规格列表，为每个规格匹配对应库存和价格（共用默认阶梯价或使用专属阶梯价）；
示例代码中_merge_specs_and_price函数实现关联逻辑，确保规格 - 价格 - 库存一一对应。
供应商信息与商位解析
问题：义乌购商品与实体商位强绑定，商位地址（如 “二区 45 号门 3 楼 12345”）格式特殊，需结构化提取以便线下对接。
解决方案：
用正则拆分商位地址（如提取 “区域 = 二区”“门号 = 45 号门”“楼层 = 3 楼”“商位号 = 12345”）；
关联供应商诚信等级（如 “5 年诚信商铺”）和联系方式（电话 / 微信），辅助供应商筛选；
示例代码中_parse_static_data函数提取并结构化商位信息，便于后续线下采购对接。
反爬机制对抗
问题：义乌购对批发数据爬取限制严格，高频请求会触发 IP 封锁（403 错误），尤其针对热门小商品类目。
解决方案：
代理 IP 轮换：使用高匿代理池，每 2 次请求切换 IP，优先选择浙江地区 IP（义乌购用户集中区域）；
请求频率控制：单 IP 每分钟请求≤1 次，两次请求间隔 2-4 秒（模拟采购商对比商品的节奏）；
Cookie 池策略：维护多个登录态 Cookie（通过不同采购账号获取），随机携带以降低单一账号风险；
动态 UA：每次请求使用fake_useragent生成随机 User-Agent，避免固定标识被识别。
六、最佳实践与合规要点
系统架构设计采用 “低频精准采集 + 商位关联存储” 架构，适配批发场景特性：
采集层：集成代理池、Cookie 池，控制请求频率，优先获取动态接口数据；
解析层：重点处理阶梯价、规格关联逻辑，确保批发核心数据准确；
存储层：用 Redis 缓存商品基础信息（4 小时过期，库存波动较快），MySQL 存储供应商信息（长期有效，便于筛选）；
监控层：实时监控请求成功率、商位信息完整度，异常时通过企业微信告警。
性能优化策略
异步批量获取：使用aiohttp并发处理多个item_id（控制并发数≤2），提升效率；
按需解析：列表页优先提取item_id、基础批发价、供应商名称等核心字段，详细阶梯价通过后续item_get接口补充；
热点抑制：对同一商品 ID 的请求，1 小时内仅处理 1 次（返回缓存结果）。
合规性与风险控制
频率限制：单 IP 日请求量≤200 次，避免对平台服务器造成压力，符合 robots 协议；
数据使用边界：不得将批发价、供应商联系方式用于恶意竞价或商业泄露，需注明数据来源 “义乌购”；
商位信息保护：实体商位地址属于供应商隐私，使用时需遵守《电子商务法》，不得擅自公开或用于非法用途。
七、总结
义乌购item_get接口的对接核心在于多阶梯批发价的结构化解析、规格与价格 / 库存的精准关联及供应商商位信息的提取。开发者需重点关注：
阶梯价的动态接口获取与格式标准化（支撑批量采购决策）；
多规格商品的独立价格 / 库存匹配（避免采购错误）；
代理池与请求频率的精细化控制（应对严格反爬）。
通过本文的技术方案，可构建稳定的商品详情获取系统，为批发采购、供应链分析等场景提供可靠数据支持。实际应用中，需根据平台最新页面结构动态调整解析规则，平衡数据获取效率与合规性。