标签内，每条商品信息包含以下核心字段：
字段 解析方式（CSS 选择器示例） 说明
商品 ID 从a标签的href中提取（如/product/detail/123456.html得123456） 唯一标识
标题 .product-title的文本 如 “10cm 塑料玩具 儿童赠品 混批”
主图 .product-img img的src属性 商品主图 URL
批发价 .price-wholesale的文本（提取数值） 如 “1.5 元 / 件” 提取 “1.5”
起订量 .moq的文本（提取数字） 如 “≥50 件” 提取 “50”
30 天成交 .trade-count的文本（提取数字） 如 “成交 12000 件” 提取 “12000”
供应商 .shop-name的文本 如 “义乌市 XX 玩具商行”
商位地址 .shop-location的文本 如 “一区 12 号门 2 楼 3456 商位”
混批标识 .mix-tag的文本 如 “支持混批”
分页处理
分页通过page参数控制，前 100 页为有效数据，超过则返回空结果；
终止条件：当前页商品数量 < 60（最后一页）或页码≥100；
分页间隔：每页请求间隔 3-5 秒（随机波动），避免触发反爬（义乌购对批发搜索的频率限制严格）。
四、代码实现示例（Python）
以下是item_search接口的完整实现，包含多条件筛选、分页遍历、数据解析及反爬处理：
import requests
import time
import random
import re
import urllib.parse
from bs4 import BeautifulSoup
from fake_useragent import UserAgent
from typing import List, Dict

class YiwugouSearchApi:
def init(self, proxy_pool: List[str] = None, cookie: str = ""):
self.base_url = "https://www.yiwugou.com/search/"
self.ua = UserAgent()
self.proxy_pool = proxy_pool # 代理池列表，如["http://ip:port", ...]
self.cookie = cookie # 登录态Cookie（用于完整价格和起订量）

    # 分类ID映射（简化版）
    self.category_map = {
        "玩具-塑料玩具": "100120_305",
        "饰品-耳环": "100230_412",
        "家居-收纳盒": "100340_521"
    }
    # 区域映射（义乌国际商贸城）
    self.market_map = {
        "一区": 1,
        "二区": 2,
        "三区": 3,
        "四区": 4,
        "五区": 5
    }

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_quantity(self, quantity_str: str) -> int:
    """清洗起订量（提取数字）"""
    if not quantity_str:
        return 0
    quantity_num = re.search(r"\d+", quantity_str)
    return int(quantity_num.group()) if quantity_num else 0

def _clean_trade_count(self, trade_str: str) -> int:
    """清洗30天成交量（提取数字）"""
    if not trade_str:
        return 0
    trade_num = re.search(r"\d+", trade_str)
    return int(trade_num.group()) if trade_num else 0

def _parse_item(self, item_soup) -> Dict[str, str]:
    """解析单条商品数据"""
    # 提取商品ID
    link = item_soup.select_one("a.product-title")["href"] if item_soup.select_one("a.product-title") else ""
    item_id = re.search(r"/product/detail/(\d+)\.html", link).group(1) if link else ""

    # 提取商位信息（拆分区域、门号、楼层、商位号）
    location = item_soup.select_one(".shop-location")?.text.strip() or ""
    location_info = {"full": location}
    if "号门" in location and "楼" in location:
        # 示例："一区12号门2楼3456商位" → 拆分区域、门号、楼层、商位号
        area_match = re.search(r"(一|二|三|四|五)区", location)
        door_match = re.search(r"(\d+)号门", location)
        floor_match = re.search(r"(\d+)楼", location)
        stall_match = re.search(r"(\d+)商位", location)
        if area_match:
            location_info["area"] = area_match.group()
        if door_match:
            location_info["door"] = door_match.group()
        if floor_match:
            location_info["floor"] = floor_match.group()
        if stall_match:
            location_info["stall"] = stall_match.group()

    return {
        "item_id": item_id,
        "title": item_soup.select_one(".product-title")?.text.strip() or "",
        "main_image": item_soup.select_one(".product-img img")?.get("src") or "",
        "url": f"https://www.yiwugou.com{link}" if link.startswith("/") else link,
        "price": {
            "wholesale": self._clean_price(item_soup.select_one(".price-wholesale")?.text or ""),
            "price_str": item_soup.select_one(".price-wholesale")?.text.strip() or ""  # 原始价格文本（含阶梯信息）
        },
        "trade": {
            "moq": self._clean_quantity(item_soup.select_one(".moq")?.text or ""),
            "moq_str": item_soup.select_one(".moq")?.text.strip() or "",  # 原始起订量文本
            "mix_batch": "支持混批" in (item_soup.select_one(".mix-tag")?.text or ""),
            "trade_count": self._clean_trade_count(item_soup.select_one(".trade-count")?.text or "")
        },
        "supplier": {
            "name": item_soup.select_one(".shop-name")?.text.strip() or "",
            "location": location_info,
            "credit": item_soup.select_one(".credit-level")?.text.strip() or ""
        }
    }

def _parse_page(self, html: str) -> List[Dict]:
    """解析页面的商品列表"""
    soup = BeautifulSoup(html, "lxml")
    # 商品列表容器（需根据实际页面结构调整）
    item_list = soup.select("div.product-item")
    return [self._parse_item(item) for item in item_list if item]

def _get_total_pages(self, html: str) -> int:
    """获取总页数"""
    soup = BeautifulSoup(html, "lxml")
    page_box = soup.select_one(".pagination")
    if not page_box:
        return 1
    # 提取最后一页页码
    last_page = page_box.select("a")[-1].text.strip()
    return int(last_page) if last_page.isdigit() else 1

def item_search(self, 
               keywords: str = "", 
               category: str = "", 
               price_min: float = None, 
               price_max: float = None, 
               quantity_min: int = None, 
               quantity_max: int = None, 
               market: str = "", 
               mix_batch: int = 0, 
               sort: str = "", 
               page_limit: int = 5) -> Dict:
    """
    搜索义乌购商品列表
    :param keywords: 搜索关键词
    :param category: 分类名称（如“玩具-塑料玩具”）或分类ID
    :param price_min: 最低批发价（元）
    :param price_max: 最高批发价（元）
    :param quantity_min: 最低起订量（件）
    :param quantity_max: 最高起订量（件）
    :param market: 供应商区域（如“一区”“二区”）
    :param mix_batch: 是否支持混批（1=仅支持，0=全部）
    :param sort: 排序方式（volume_desc/price_asc等）
    :param page_limit: 最大页数（默认5）
    :return: 标准化搜索结果
    """
    try:
        # 1. 参数预处理
        if not keywords and not category:
            return {"success": False, "error_msg": "关键词（keywords）和分类（category）至少需提供一个"}
        # 转换分类名称为ID
        if category in self.category_map:
            cat_id = self.category_map[category]
        else:
            cat_id = category if category else ""
        # 转换区域名称为数字
        market_id = self.market_map.get(market, "") if market else ""
        # 编码关键词
        encoded_keywords = urllib.parse.quote(keywords, encoding="utf-8") if keywords else ""

        all_items = []
        current_page = 1

        while current_page <= page_limit:
            # 构建参数
            params = {
                "page": current_page
            }
            if encoded_keywords:
                params["keywords"] = encoded_keywords
            if cat_id:
                params["cat_id"] = cat_id
            if price_min is not None:
                params["price_min"] = price_min
            if price_max is not None:
                params["price_max"] = price_max
            if quantity_min is not None:
                params["quantity_min"] = quantity_min
            if quantity_max is not None:
                params["quantity_max"] = quantity_max
            if market_id:
                params["market"] = market_id
            if mix_batch in (0, 1):
                params["mix_batch"] = mix_batch
            if sort:
                params["sort"] = sort

            # 发送请求（带随机延迟）
            time.sleep(random.uniform(3, 5))  # 义乌购反爬严格，延迟需更高
            headers = self._get_headers()
            proxy = self._get_proxy()

            response = requests.get(
                url=self.base_url,
                params=params,
                headers=headers,
                proxies=proxy,
                timeout=10
            )
            response.raise_for_status()
            html = response.text

            # 解析当前页商品
            items = self._parse_page(html)
            if not items:
                break  # 无数据，终止分页

            all_items.extend(items)

            # 获取总页数（仅第一页需要）
            if current_page == 1:
                total_pages = self._get_total_pages(html)
                # 修正最大页数（不超过page_limit和100）
                total_pages = min(total_pages, page_limit, 100)
                if total_pages < current_page:
                    break

            # 若当前页是最后一页，终止
            if current_page >= total_pages:
                break

            current_page += 1

        # 去重（基于item_id）
        seen_ids = set()
        unique_items = []
        for item in all_items:
            if item["item_id"] not in seen_ids:
                seen_ids.add(item["item_id"])
                unique_items.append(item)

        return {
            "success": True,
            "total": len(unique_items),
            "page_processed": current_page,
            "items": unique_items
        }

    except requests.exceptions.HTTPError as e:
        if "403" in str(e):
            return {"success": False, "error_msg": "触发反爬，建议更换代理或Cookie", "code": 403}
        if "401" in str(e):
            return {"success": False, "error_msg": "需要登录，请提供有效Cookie", "code": 401}
        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客户端
search_api = YiwugouSearchApi(proxy_pool=PROXIES, cookie=COOKIE)

# 搜索“塑料玩具”，分类“玩具-塑料玩具”，价格1-5元，起订量50-500件，支持混批，按成交降序，最多3页
result = search_api.item_search(
    keywords="塑料玩具",
    category="玩具-塑料玩具",
    price_min=1,
    price_max=5,
    quantity_min=50,
    quantity_max=500,
    mix_batch=1,
    sort="volume_desc",
    page_limit=3
)

if result["success"]:
    print(f"搜索成功：共找到 {result['total']} 件商品，处理 {result['page_processed']} 页")
    for i, item in enumerate(result["items"][:5]):  # 打印前5条
        print(f"\n商品 {i+1}:")
        print(f"标题：{item['title'][:50]}...")  # 截断长标题
        print(f"价格：{item['price']['price_str']} | 起订量：{item['trade']['moq_str']}")
        print(f"交易：30天成交{item['trade']['trade_count']}件 | 混批：{'是' if item['trade']['mix_batch'] else '否'}")
        print(f"供应商：{item['supplier']['name']} | 诚信等级：{item['supplier']['credit']}")
        print(f"商位：{item['supplier']['location']['full']}")
        print(f"详情页：{item['url']}")
else:
    print(f"搜索失败：{result['error_msg']}（错误码：{result.get('code')}）")
五、关键技术难点与解决方案

多维度筛选参数整合
问题：义乌购搜索依赖批发场景特有的筛选条件（如起订量、混批、区域），参数格式多样，易出现拼接错误导致筛选失效。
解决方案：
建立参数映射表（如market_map将 “一区” 转换为1），统一参数格式；
对起订量、价格等数值参数进行类型校验，避免非数字值导致请求错误；
示例代码中item_search函数对category和market参数的预处理，确保多条件筛选生效。
商位地址结构化解析
问题：义乌购商品与实体商位强绑定，地址格式为 “区域 + 门号 + 楼层 + 商位号”（如 “一区 12 号门 2 楼 3456 商位”），需拆分为结构化数据便于线下对接。
解决方案：
用正则匹配关键词（如 “号门”“楼”“商位”），拆分区域、门号、楼层、商位号；
存储原始地址和拆分后的结构化信息（如location_info字段），兼顾展示与分析需求；
示例代码中_parse_item函数的地址解析逻辑，支持按区域筛选供应商（如 “仅看一区商品”）。
批发价与起订量提取
问题：列表页的价格和起订量常包含文本描述（如 “1.5 元 / 件（≥100 件）”“混批 50 件起”），需提取纯数值用于筛选和计算。
解决方案：
用正则去除非数字字符（如re.sub(r"[^\d.]", "", price_str)），提取价格和起订量的数值；
同时保留原始文本（如price_str“1.5 元 / 件”），便于用户理解；
示例代码中_clean_price和_clean_quantity函数实现数值提取，支持后续按价格区间筛选。
反爬机制对抗
问题：义乌购作为批发平台，对异常搜索行为敏感，高频请求会触发 IP 封锁（403 错误），尤其针对热门小商品类目。
解决方案：
代理 IP 策略：使用高匿代理池，每 1 页切换一次 IP，优先选择浙江地区 IP（义乌购用户集中区域）；
请求频率控制：单 IP 每分钟请求≤1 次，两次请求间隔 3-5 秒（模拟采购商仔细对比商品的行为）；
Cookie 池管理：维护多个登录态 Cookie（通过不同采购账号获取），随机携带以降低单一账号风险；
动态 UA：每次请求使用fake_useragent生成随机 User-Agent，避免固定标识被识别。
六、最佳实践与合规要点
系统架构设计采用 “分布式低频采集” 架构，适配批发场景特性：
任务调度层：通过定时任务（如 Celery）分发搜索任务，控制单任务并发数≤1；
采集层：多节点分散请求，每个节点绑定独立代理池，节点间请求间隔≥15 秒；
存储层：用 Redis 缓存热门搜索结果（6 小时过期，批发价波动较慢），MySQL 存储供应商信息（长期有效，便于筛选）；
监控层：实时监控代理存活率、商位信息完整度，异常时通过短信告警。
性能优化策略
异步批量搜索：使用aiohttp并发处理多关键词（如 “塑料玩具”“毛绒玩具”），控制并发数≤1，适配低频率限制；
按需解析：列表页优先提取item_id、批发价、起订量等核心字段，详细阶梯价通过后续item_get接口补充；
热点抑制：对同一关键词 + 参数组合的搜索，2 小时内仅处理 1 次（返回缓存结果）。
合规性与风险控制
访问限制：单 IP 日搜索请求≤100 次，避免对平台服务器造成压力，符合 robots 协议；
数据使用边界：不得将批发价、供应商联系方式用于恶意竞价或商业泄露，需注明数据来源 “义乌购”；
商位信息保护：实体商位地址属于供应商隐私，使用时需遵守《电子商务法》，不得擅自公开或用于非法用途。
七、总结
义乌购item_search接口的对接核心在于批发场景参数的精准整合、商位地址的结构化解析及低频率高稳定性的采集策略。开发者需重点关注：
起订量、混批等 B2B 特有参数的处理（确保筛选条件生效）；
商位地址的拆分逻辑（便于线下采购对接）；
代理池与请求频率的精细化控制（应对严格反爬）。
通过本文的技术方案，可构建稳定的商品搜索系统，为批发采购、供应商筛选等场景提供可靠数据支持。实际应用中，需根据平台最新页面结构动态调整解析规则，平衡数据获取效率与合规性。
需要进一步了解义乌购分类 ID 完整映射表或反爬代理池搭建细节，可以告诉我，我会补充相关内容