class SmzdmSearchApi:
def init(self, proxy_pool: List[str] = None, cookie: str = ""):
self.base_url = "https://www.smzdm.com/search/"
self.ua = UserAgent()
self.proxy_pool = proxy_pool # 代理池列表，如["http://ip:port", ...]
self.cookie = cookie # 登录态Cookie（用于获取完整优惠信息）

    # 分类ID映射（简化版）
    self.category_map = {
        "耳机音箱": "157",
        "家用电器": "74",
        "母婴用品": "118",
        "美妆个护": "145"
    }
    # 平台编码映射（简化版）
    self.mall_map = {
        "京东": "jd",
        "天猫": "tmall",
        "淘宝": "taobao",
        "拼多多": "pinduoduo"
    }

def _get_headers(self) -> Dict[str, str]:
    """生成随机请求头"""
    headers = {
        "User-Agent": self.ua.random,
        "Referer": "https://www.smzdm.com/",
        "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 _parse_history_lowest(self, history_str: str) -> Dict[str, str]:
    """解析历史低价（如“历史最低¥799（2023-11）”）"""
    if not history_str:
        return {"price": 0.0, "date": ""}
    price_match = re.search(r"¥(\d+\.?\d*)", history_str)
    date_match = re.search(r"\((\d{4}-\d{2})\)", history_str)
    return {
        "price": self._clean_price(price_match.group(1)) if price_match else 0.0,
        "date": date_match.group(1) if date_match else ""
    }

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

    # 提取优惠标签（可能多个）
    tags = [tag.text.strip() for tag in item_soup.select(".tag")]

    return {
        "item_id": item_id,
        "title": item_soup.select_one(".feed-title")?.text.strip() or "",
        "main_image": item_soup.select_one(".feed-img img")?.get("src") or "",
        "url": f"https://www.smzdm.com{link}" if link.startswith("/") else link,
        "price": {
            "current": self._clean_price(item_soup.select_one(".z-price")?.text or ""),
            "history_lowest": self._parse_history_lowest(item_soup.select_one(".history-lowest")?.text or "")
        },
        "user_feedback": {
            "rating": float(item_soup.select_one(".rating")?.text or "0"),
            "comment_count": int(re.search(r"\d+", item_soup.select_one(".comment-count")?.text or "0").group()) if item_soup.select_one(".comment-count") else 0
        },
        "platform": {
            "source": item_soup.select_one(".mall-label")?.text.strip() or "",
            "buy_url": item_soup.select_one(".buy-link")?.get("href") or ""
        },
        "tags": tags,  # 优惠标签（如“满减”“百亿补贴”）
        "post_time": item_soup.select_one(".time")?.text.strip() or ""  # 爆料时间
    }

def _parse_page(self, html: str) -> List[Dict]:
    """解析页面的商品列表"""
    soup = BeautifulSoup(html, "lxml")
    # 商品列表容器（需根据实际页面结构调整）
    item_list = soup.select("li.feed-row")
    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, 
               keyword: str = "", 
               category: str = "", 
               price_min: float = None, 
               price_max: float = None, 
               mall: str = "", 
               coupon: int = 0, 
               sort: str = "", 
               page_limit: int = 5) -> Dict:
    """
    搜索值得买商品列表
    :param keyword: 搜索关键词
    :param category: 分类名称（如“耳机音箱”）或分类ID
    :param price_min: 最低价格（元）
    :param price_max: 最高价格（元）
    :param mall: 来源平台名称（如“京东”）或编码（如“jd”）
    :param coupon: 是否含优惠券（1=是，0=否）
    :param sort: 排序方式（price/rating等）
    :param page_limit: 最大页数（默认5）
    :return: 标准化搜索结果
    """
    try:
        # 1. 参数预处理
        if not keyword and not category:
            return {"success": False, "error_msg": "关键词（keyword）和分类（category）至少需提供一个"}
        # 转换分类名称为ID
        if category in self.category_map:
            cid = self.category_map[category]
        else:
            cid = category if category else ""
        # 转换平台名称为编码
        if mall in self.mall_map:
            mall_code = self.mall_map[mall]
        else:
            mall_code = mall if mall else ""
        # 编码关键词
        encoded_keyword = urllib.parse.quote(keyword, encoding="utf-8") if keyword else ""

        all_items = []
        current_page = 1

        while current_page <= page_limit:
            # 构建参数
            params = {
                "p": current_page
            }
            if encoded_keyword:
                params["s"] = encoded_keyword
            if cid:
                params["c"] = cid
            if price_min is not None:
                params["price_min"] = price_min
            if price_max is not None:
                params["price_max"] = price_max
            if mall_code:
                params["mall"] = mall_code
            if coupon in (0, 1):
                params["coupon"] = coupon
            if sort:
                params["v"] = sort

            # 发送请求（带随机延迟）
            time.sleep(random.uniform(2, 3))  # 控制频率，避免反爬
            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和20）
                total_pages = min(total_pages, page_limit, 20)
                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}
        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 = "sess=xxx; user_id=xxx; device_id=xxx"

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

# 搜索“无线耳机”，分类“耳机音箱”，价格500-1000元，来源京东，含优惠券，按历史低价排序，最多3页
result = search_api.item_search(
    keyword="无线耳机",
    category="耳机音箱",
    price_min=500,
    price_max=1000,
    mall="京东",
    coupon=1,
    sort="price",
    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']['current']} | 历史最低￥{item['price']['history_lowest']['price']}（{item['price']['history_lowest']['date']}）")
        print(f"用户评分：{item['user_feedback']['rating']}分 | 评价数：{item['user_feedback']['comment_count']}条")
        print(f"来源平台：{item['platform']['source']} | 优惠标签：{', '.join(item['tags'])}")
        print(f"爆料时间：{item['post_time']} | 详情页：{item['url']}")
else:
    print(f"搜索失败：{result['error_msg']}（错误码：{result.get('code')}）")
五、关键技术难点与解决方案

消费决策字段解析（历史低价、评分）
问题：历史低价（如 “历史最低 ¥799（2023-11）”）和用户评分是值得买的核心决策数据，但格式不统一，提取难度大。
解决方案：
历史低价通过正则匹配价格（¥(\d+.?\d*)）和日期（(\d{4}-\d{2})），结构化输出 “价格 + 时间”；
用户评分从.rating标签提取，处理 “暂无评分” 等异常值（默认 0 分）；
示例代码中_parse_history_lowest函数专门解析历史低价，确保决策数据准确性。
优惠标签提取（多标签场景）
问题：商品可能包含多个优惠标签（如 “满减”“百亿补贴”“分期免息”），分散在多个标签中。
解决方案：
通过select(".tag")批量提取所有标签，用列表存储；
对标签进行分类（如 “价格优惠”“服务优惠”），便于后续筛选（如仅保留 “满减” 标签的商品）；
示例代码中tags字段整合所有优惠标签，突出商品的优惠力度。
反爬机制对抗
问题：值得买对数据爬取限制严格，高频请求会触发 IP 封锁（403 错误），甚至要求验证码验证。
解决方案：
代理 IP 轮换：使用高匿代理池，每 2 页切换一次 IP，优先选择与用户真实地域匹配的 IP；
请求频率控制：单 IP 每分钟请求≤3 次，两次请求间隔 2-3 秒（模拟用户浏览消费内容的节奏）；
Cookie 池策略：维护多个登录态 Cookie（通过不同账号获取），随机携带以降低单一账号风险；
动态参数伪装：在 URL 后添加随机时间戳（如&t=1620000000），避免请求缓存被识别为爬虫。
多平台数据融合处理
问题：搜索结果包含多平台商品（京东、天猫等），价格、优惠规则存在差异，需统一结构化。
解决方案：
提取mall-label字段标记来源平台，便于后续按平台筛选；
价格统一转换为 “元” 单位，去除各平台特有符号（如天猫的 “¥”、拼多多的 “￥”）；
对跳转链接进行有效性校验（如检查buy_url是否包含http），过滤无效链接。
六、最佳实践与合规要点
系统架构设计采用 “低频率、高价值” 采集架构，适配消费决策数据特性：
任务分发层：通过消息队列（如 RabbitMQ）分发搜索任务，控制单任务并发数≤2；
采集层：多节点并行采集，每个节点绑定独立代理池，节点间请求间隔≥5 秒；
存储层：用 Redis 缓存热门搜索结果（30 分钟过期，优惠信息更新快），MySQL 存储历史数据（用于消费趋势分析）；
监控层：实时监控代理存活率、优惠标签提取完整度，异常时通过企业微信告警。
性能优化策略
异步批量搜索：使用aiohttp并发处理多关键词（如 “无线耳机”“蓝牙耳机”），控制并发数≤2；
按需解析：列表页优先提取item_id、价格、评分等核心字段，详细评价和历史价格通过后续item_get接口补充；
热点抑制：对同一关键词 + 条件的搜索，30 分钟内仅处理 1 次（返回缓存结果）。
合规性与风险控制
频率限制：单 IP 日搜索请求≤200 次，避免对平台服务器造成压力，符合 robots 协议；
数据使用边界：不得将用户评价、评分用于商业售卖或恶意竞争，需注明数据来源 “什么值得买”；
法律风险规避：优惠信息可能涉及平台版权，使用时需遵守《电子商务法》，不得篡改优惠规则或误导消费。
七、总结
值得买item_search接口的对接核心在于消费决策字段的精准解析（历史低价、评分）、优惠标签的有效提取及低频率高稳定性的采集策略。开发者需重点关注：
历史价格与评分的结构化处理（确保决策数据可用性）；
多平台商品的统一格式整合（便于跨平台对比）；
代理池与请求频率的精细化控制（应对严格反爬）。
通过本文的技术方案，可构建稳定的消费导向型商品搜索系统，为比价决策、优惠监控等场景提供可靠数据支持。实际应用中，需根据平台最新页面结构动态调整解析规则，平衡数据获取效率与合规性