基于Python与Playwright的闲鱼商品监控助手：自动化抓取与实时通知实现

1. 项目概述：一个能帮你自动“捡漏”的闲鱼助手

最近几年，二手闲置交易平台越来越火，闲鱼作为其中的头部玩家，每天都有海量的商品上架和下架。对于很多喜欢“捡漏”的玩家或者有特定购物需求的人来说，手动刷新、蹲守心仪的商品，不仅耗时耗力，还常常因为手速慢而错失良机。今天要聊的这个项目shaxiu/XianyuAutoAgent，就是一个专门为解决这个问题而生的自动化工具。

简单来说，它就像一个不知疲倦的“数字助理”，能够7x24小时不间断地帮你监控闲鱼平台上的商品动态。你可以设定好你想要的关键词、价格区间、发布地区等条件，一旦有符合条件的新商品上架，或者有商品降价，这个助手就能在第一时间通知你，甚至可以根据预设的规则进行一些自动化的操作，比如自动发送询价消息。这对于抢购热门电子产品、限量潮玩，或者寻找特定型号的工业配件、二手家具的用户来说，无疑是一个效率倍增器。

这个项目在GitHub上开源，由开发者“shaxiu”维护。它本质上是一个基于Python编写的自动化脚本集合，通过模拟浏览器操作或调用平台接口（在合规范围内）的方式，来实现对闲鱼数据的抓取、筛选和响应。接下来，我会从一个实际使用者的角度，深度拆解这个项目的实现思路、技术细节、实操方法以及那些官方文档里不会写的“坑”和技巧。

2. 核心思路与技术选型解析

2.1 为什么选择自动化方案？

在深入代码之前，我们首先要理解为什么需要这样一个自动化工具。闲鱼本身有“提醒”功能，但通常只限于单个商品的价格变动或下架提醒，无法实现复杂的多条件、主动式搜索。手动搜索的弊端很明显：效率低下、无法实时监控、容易错过“秒杀”商品。因此，一个本地的、可高度自定义的自动化程序就成了刚性需求。

项目的核心思路可以概括为“感知-决策-执行”循环：

1. 感知：定期或实时地从闲鱼获取商品列表数据。
2. 决策：根据用户预设的规则（关键词、价格、成色、卖家信用等）过滤数据，判断是否有“目标”出现。
3. 执行：对符合条件的商品采取行动，如发送通知（微信、邮件）、自动发起聊天、记录信息等。

2.2 技术栈的权衡与选择

XianyuAutoAgent的技术选型体现了在效率、稳定性和合规性之间的平衡。

2.2.1 数据获取层：爬虫 vs 模拟操作

这是最核心也最敏感的一环。直接调用未公开的API（爬虫）效率最高，但风险也最大，容易触发平台的反爬机制导致IP被封。因此，更稳健的做法是使用浏览器自动化工具，如Selenium或Playwright，来模拟真实用户的操作行为（打开网页、输入搜索词、点击翻页）。XianyuAutoAgent很可能采用了类似方案，因为这样更贴近正常用户行为，规避了直接攻击API接口的风险。Playwright相比Selenium更新，在异步处理和浏览器上下文隔离方面更有优势，适合需要同时维护多个监控任务的情景。

2.2.2 规则引擎与过滤逻辑

获取到原始商品数据（通常是JSON格式）后，需要一套灵活的规则引擎进行过滤。这里通常会用到Python的字典和列表操作，结合条件判断语句。更复杂的项目可能会引入规则配置文件（如YAML或JSON），让用户无需修改代码就能调整监控条件。例如，规则可能包括：

• 关键词匹配：支持多关键词、排除词、模糊匹配。
• 价格区间：设定最低价和最高价，甚至可以考虑“比历史均价低X%”的动态规则。
• 卖家筛选：过滤芝麻信用分低的卖家、专业贩子（通过商品发布频率判断）等。
• 商品属性：对于手机数码类，可以筛选内存容量、版本；对于家具，可以筛选材质、尺寸。

2.2.3 通知与执行层

当发现目标商品后，如何及时通知用户是关键。简单的方式是打印日志到控制台，但这显然不够。成熟的项目会集成多种通知渠道：

• 桌面通知：使用plyer或win10toast库。
• 邮件通知：通过SMTP协议发送邮件到指定邮箱。
• 即时通讯：这是最常用的方式。通过调用企业微信、钉钉的机器人Webhook，或者使用itchat、wxpy（已停止维护）等库直接操作个人微信（风险高且不稳定）。目前更推荐使用Server酱、PushPlus等第三方推送服务，它们提供了将消息转发到微信的稳定通道。
• 自动化交互：自动发送闲鱼内置的“我想要”或预设的询价话术。这部分需要极其谨慎，必须严格遵守平台规则，模拟人类操作节奏（如随机延迟），避免被判定为营销机器人。

2.2.4 调度与持久化

监控任务需要定时执行。轻量级方案可以用Python内置的time.sleep循环，但更健壮的做法是使用APScheduler这样的调度库，它能更好地管理任务周期、并发和错误处理。此外，为了避免重复通知同一个商品，程序需要将已处理过的商品ID记录到本地文件（如SQLite数据库）或内存中，实现简单的持久化。

3. 环境搭建与核心依赖详解

要运行或二次开发这样一个项目，首先需要搭建好Python环境并理解其核心依赖。

3.1 Python环境与包管理

建议使用 Python 3.8 或以上版本。使用虚拟环境（venv或conda）是绝对的最佳实践，它能隔离项目依赖，避免污染系统环境。

# 创建虚拟环境 python -m venv xianyu_agent_env # 激活虚拟环境 (Windows) xianyu_agent_env\Scripts\activate # 激活虚拟环境 (Linux/macOS) source xianyu_agent_env/bin/activate

项目通常会提供一个requirements.txt文件。假设其内容包含以下核心库：

playwright>=1.40.0 apscheduler>=3.10.0 requests>=2.28.0 python-dotenv>=0.19.0 schedule>=1.2.0

使用pip install -r requirements.txt一键安装。对于Playwright，还需要安装浏览器内核：

playwright install chromium

这里选择Chromium而非Chrome，因为它是开源的，更适合自动化场景。

3.2 核心依赖库的作用与配置要点

• Playwright：负责浏览器自动化。相比Selenium，它的API更现代，自动等待机制更智能，能有效减少因页面加载不全导致的脚本错误。在初始化时，建议使用headless=False模式进行调试，观察其操作是否正常，稳定后再改为headless=True无头模式在后台运行。
• APScheduler：任务调度器。你可以设置每5分钟或10分钟执行一次搜索任务。关键是要设置合理的coalesce（合并多次错过的任务）和max_instances（最大并发实例数），防止因任务执行时间过长导致的任务堆积。
• requests：用于发送HTTP请求，可能用于调用推送服务的API，或者在某些简化版中直接请求闲鱼的数据接口（需谨慎）。
• python-dotenv：用于管理配置。将敏感信息如推送服务的Token、监控关键词等放在.env文件中，避免硬编码在代码里，提高安全性。

注意：直接使用个人微信的库（如itchat）进行通知，目前极不稳定且存在账号风险。强烈建议使用企业微信机器人、Server酱等合规的中转服务。

4. 项目架构与核心模块拆解

一个典型的XianyuAutoAgent项目会包含以下几个核心模块，我们来逐一拆解其实现逻辑和代码要点。

4.1 配置管理模块 (config.py)

这个模块负责统一加载和管理所有配置。使用.env文件配合python-dotenv是标准做法。

.env文件示例：

SEARCH_KEYWORDS=“iPhone 13,小米手机” MAX_PRICE=3000 MIN_PRICE=1000 PUSHPLUS_TOKEN=your_pushplus_token_here SCAN_INTERVAL_MINUTES=5

config.py代码示例：

import os from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的变量到环境变量 class Config: # 搜索相关 KEYWORDS = [k.strip() for k in os.getenv('SEARCH_KEYWORDS', '').split(',') if k.strip()] MAX_PRICE = float(os.getenv('MAX_PRICE', 999999)) MIN_PRICE = float(os.getenv('MIN_PRICE', 0)) # 推送相关 PUSHPLUS_TOKEN = os.getenv('PUSHPLUS_TOKEN') # 任务调度相关 INTERVAL = int(os.getenv('SCAN_INTERVAL_MINUTES', 5)) * 60 # 转换为秒

这样，主程序通过from config import Config来获取配置，修改监控条件只需编辑.env文件，无需改动代码。

4.2 闲鱼爬虫/采集模块 (crawler.py)

这是项目的引擎。我们以使用Playwright模拟搜索为例。

import asyncio from playwright.async_api import async_playwright from bs4 import BeautifulSoup # 用于解析HTML class XianyuCrawler: def __init__(self, headless=True): self.headless = headless async def search_by_keyword(self, keyword, max_pages=3): """根据关键词搜索商品，返回商品信息列表""" async with async_playwright() as p: # 启动浏览器，建议使用Chromium browser = await p.chromium.launch(headless=self.headless, slow_mo=100) # slow_mo 让操作变慢，便于观察和防检测 context = await browser.new_context( viewport={'width': 1920, 'height': 1080}, user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36...' # 使用真实UA ) page = await context.new_page() items_list = [] try: # 1. 导航到闲鱼搜索页 search_url = f"https://s.2.taobao.com/list/list.htm?q={keyword}" await page.goto(search_url, wait_until='networkidle') # 等待网络空闲 for page_num in range(1, max_pages + 1): # 2. 等待商品列表加载 await page.wait_for_selector('.item-lists .card', timeout=10000) # 3. 获取页面HTML并用BeautifulSoup解析 content = await page.content() soup = BeautifulSoup(content, 'html.parser') cards = soup.select('.item-lists .card') # 根据实际页面结构调整选择器 for card in cards: item_info = self._parse_card(card) if item_info: items_list.append(item_info) # 4. 尝试点击“下一页” next_button = page.locator('text=下一页') if await next_button.count() > 0 and await next_button.is_enabled(): await next_button.click() await page.wait_for_timeout(2000) # 等待页面加载 else: break # 没有下一页则退出循环 except Exception as e: print(f"搜索过程发生错误: {e}") finally: await browser.close() return items_list def _parse_card(self, card): """解析单个商品卡片，提取标题、价格、链接等信息""" # 这是一个示例解析函数，闲鱼页面结构可能经常变动，需要定期更新 try: title_elem = card.select_one('.title') price_elem = card.select_one('.price') link_elem = card.select_one('a') if not all([title_elem, price_elem, link_elem]): return None title = title_elem.get_text(strip=True) # 价格处理，可能包含“¥”符号 price_text = price_elem.get_text(strip=True).replace('¥', '').strip() price = float(price_text) if price_text.replace('.', '', 1).isdigit() else 0 link = link_elem.get('href') if link and not link.startswith('http'): link = 'https:' + link return { 'title': title, 'price': price, 'url': link, 'item_id': link.split('id=')[-1] if 'id=' in link else None # 提取商品ID } except Exception as e: print(f"解析商品卡片失败: {e}") return None

实操心得：闲鱼的前端结构并非一成不变，_parse_card函数中的CSS选择器（如.title,.price）很可能需要根据实际情况调整。最稳妥的方法是先用浏览器的开发者工具（F12）手动分析目标页面的HTML结构。此外，slow_mo参数和随机的wait_for_timeout对于模拟人类操作、避免被反爬系统识别至关重要。

4.3 规则过滤与决策模块 (filter.py)

采集到原始数据后，需要根据用户配置进行精细过滤。

import re from config import Config class ItemFilter: def __init__(self, config: Config): self.config = config self.seen_items = set() # 内存中的已见商品ID集合，防止重复通知 def filter_items(self, items): """过滤商品列表，返回符合条件的商品""" qualified_items = [] for item in items: # 1. 去重检查 if item['item_id'] in self.seen_items: continue # 2. 价格过滤 if not (self.config.MIN_PRICE <= item['price'] <= self.config.MAX_PRICE): continue # 3. 关键词过滤 (支持简单逻辑) title = item['title'].lower() keyword_matched = False for kw in self.config.KEYWORDS: if kw.lower() in title: keyword_matched = True break if not keyword_matched: continue # 4. 高级过滤：可以在这里添加更多规则，如排除“贩子”、“批发”等词 exclude_words = ['批发', '大量', '代理', '勿拍'] if any(word in title for word in exclude_words): continue # 5. 通过所有过滤，加入结果列表并标记为已见 qualified_items.append(item) self.seen_items.add(item['item_id']) return qualified_items def load_seen_items(self, filepath='seen_items.txt'): """从文件加载已见商品ID，用于持久化""" try: with open(filepath, 'r') as f: self.seen_items = set(line.strip() for line in f) except FileNotFoundError: self.seen_items = set() def save_seen_items(self, filepath='seen_items.txt'): """保存已见商品ID到文件""" with open(filepath, 'w') as f: for item_id in self.seen_items: f.write(f"{item_id}\n")

这个过滤逻辑是核心，你可以根据自己的需求扩展。例如，加入对商品描述文本的分析，或者引入更复杂的正则表达式匹配。

4.4 消息推送模块 (notifier.py)

当有商品通过过滤时，需要及时通知用户。这里以PushPlus推送加为例。

import requests import json from config import Config class Notifier: def __init__(self, config: Config): self.config = config self.pushplus_token = config.PUSHPLUS_TOKEN def send_pushplus(self, title, content, template='html'): """通过PushPlus发送通知到微信""" if not self.pushplus_token: print("未配置PushPlus Token，无法发送通知") return False url = 'http://www.pushplus.plus/send' data = { "token": self.pushplus_token, "title": title, "content": content, "template": template } try: response = requests.post(url, data=json.dumps(data), headers={'Content-Type': 'application/json'}) result = response.json() if result.get('code') == 200: print("PushPlus消息发送成功") return True else: print(f"PushPlus消息发送失败: {result.get('msg')}") return False except Exception as e: print(f"发送PushPlus请求时出错: {e}") return False def format_item_message(self, items): """将商品列表格式化为推送消息内容""" if not items: return "" lines = ["🎯 发现符合条件的闲鱼商品："] for idx, item in enumerate(items, 1): lines.append(f"{idx}. 【{item['price']}元】{item['title']}") lines.append(f" 链接：{item['url']}") lines.append("") # 空行分隔 return "\n".join(lines)

将过滤后的商品列表传入format_item_message生成易读的消息，再调用send_pushplus即可推送到微信。PushPlus的免费额度对于个人监控完全足够。

4.5 主程序与任务调度 (main.py)

最后，我们需要一个主程序来串联所有模块，并实现定时任务。

import asyncio import sys from apscheduler.schedulers.asyncio import AsyncIOScheduler from apscheduler.triggers.interval import IntervalTrigger from datetime import datetime from config import Config from crawler import XianyuCrawler from filter import ItemFilter from notifier import Notifier class XianyuAutoAgent: def __init__(self): self.config = Config() self.crawler = XianyuCrawler(headless=True) # 生产环境用无头模式 self.filter = ItemFilter(self.config) self.notifier = Notifier(self.config) # 加载历史记录 self.filter.load_seen_items() async def scan_task(self): """一次完整的扫描任务""" print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] 开始扫描任务...") all_qualified_items = [] for keyword in self.config.KEYWORDS: if not keyword: continue print(f" 正在搜索关键词: {keyword}") try: items = await self.crawler.search_by_keyword(keyword, max_pages=2) # 每关键词搜2页 qualified = self.filter.filter_items(items) if qualified: all_qualified_items.extend(qualified) print(f" 发现 {len(qualified)} 个符合条件的商品。") except Exception as e: print(f" 搜索关键词 '{keyword}' 时出错: {e}") # 发送通知 if all_qualified_items: message = self.notifier.format_item_message(all_qualified_items) self.notifier.send_pushplus("闲鱼监控发现新商品！", message) # 保存已见记录 self.filter.save_seen_items() else: print(" 本次扫描未发现新商品。") print("扫描任务结束。\n") async def run(self): """启动定时任务""" scheduler = AsyncIOScheduler() # 每隔配置的间隔时间执行一次扫描任务 trigger = IntervalTrigger(seconds=self.config.INTERVAL) scheduler.add_job(self.scan_task, trigger, id='scan_job') print(f"闲鱼自动监控助手已启动，扫描间隔：{self.config.INTERVAL//60} 分钟") print(f"监控关键词：{self.config.KEYWORDS}") scheduler.start() try: # 保持主程序运行 await asyncio.Event().wait() except (KeyboardInterrupt, SystemExit): print("\n收到停止信号，正在清理...") scheduler.shutdown() self.filter.save_seen_items() # 退出前保存记录 print("程序已退出。") sys.exit(0) if __name__ == '__main__': agent = XianyuAutoAgent() asyncio.run(agent.run())

这个主程序创建了一个异步调度器，定期执行scan_task函数。它遍历所有配置的关键词，执行爬取、过滤和通知的完整流程。

5. 部署与长期运行方案

开发完成后，你需要让这个脚本在服务器或旧电脑上7x24小时稳定运行。

5.1 本地运行与调试

初期建议在本地电脑上以非无头模式 (headless=False) 运行，观察整个自动化流程是否顺畅，检查CSS选择器是否能正确抓到数据。确认无误后，再改为无头模式。

5.2 服务器部署（以Linux为例）

1. 环境准备：在云服务器（如腾讯云、阿里云ECS）或家中的树莓派上安装Python和项目依赖。
2. 进程守护：使用systemd或supervisor来管理进程，确保脚本崩溃后能自动重启。
   • systemd服务文件示例(/etc/systemd/system/xianyu-agent.service)：

     [Unit] Description=Xianyu Auto Agent Service After=network.target [Service] Type=simple User=your_username WorkingDirectory=/path/to/XianyuAutoAgent Environment="PATH=/path/to/venv/bin" ExecStart=/path/to/venv/bin/python /path/to/XianyuAutoAgent/main.py Restart=on-failure RestartSec=10 [Install] WantedBy=multi-user.target

   然后执行sudo systemctl daemon-reload,sudo systemctl enable xianyu-agent,sudo systemctl start xianyu-agent。
3. 日志管理：将程序中的print语句改为使用Python的logging模块，并将日志输出到文件，方便后期排查问题。

5.3 使用Docker容器化（高级）

对于更复杂的部署，可以使用Docker，它能更好地解决环境一致性问题。

# Dockerfile FROM python:3.9-slim # 安装Playwright的依赖和浏览器 RUN apt-get update && apt-get install -y \ wget \ gnupg \ && rm -rf /var/lib/apt/lists/* RUN pip install playwright && playwright install chromium WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD ["python", "main.py"]

构建并运行：docker build -t xianyu-agent .和docker run -d --name xianyu-agent xianyu-agent。

6. 常见问题、优化与避坑指南

在实际运行中，你一定会遇到各种问题。以下是我总结的一些常见坑点和优化建议。

6.1 反爬虫与风控应对

这是最大的挑战。闲鱼等平台有完善的反爬机制。

• 症状：爬取不到数据、跳出验证码、IP被临时封锁。
• 应对策略：
  1. 降低请求频率：这是最重要的。将扫描间隔设置为5分钟甚至更长。在代码中随机化等待时间，await page.wait_for_timeout(random.randint(1000, 3000))。
  2. 模拟真人行为：使用slow_mo参数，并让鼠标移动、点击操作带有随机轨迹。
  3. 使用高质量代理IP池：如果监控需求非常强烈，可以考虑使用付费的住宅代理IP，并轮换使用。但这会大大增加复杂性和成本。
  4. 接受验证码：最根本的方案是接入打码平台，但这对个人项目来说性价比不高。一旦频繁遇到验证码，最好的办法是暂停一段时间，或者手动处理一次。
• 核心原则：你的行为越像真人，就越安全。不要试图挑战平台的底线。

6.2 页面结构变更导致爬虫失效

闲鱼前端偶尔会改版，导致CSS选择器失效。

• 预防：定期（比如每周）手动运行一次脚本，检查是否能正常抓取数据。
• 应对：及时更新_parse_card函数中的选择器。可以编写一个简单的测试脚本，只抓取第一个商品的标题和价格，用于快速验证。

6.3 误报与漏报优化

• 问题：关键词匹配太宽泛，导致推送很多不相关的商品（误报）；或者匹配太严格，漏掉了真正的好货（漏报）。
• 优化：
  • 使用更精确的关键词组合：例如，搜“iPhone 13”会出来很多壳膜，可以尝试“iPhone 13 自用”、“iPhone 13 国行 转让”。
  • 引入排除词：在过滤器中增加排除词列表，如“模型”、“手机壳”、“包装盒”、“仅限”。
  • 价格合理性判断：对于明显低于市场价（如原价5000的手机卖500）的商品，可以加入判断，可能是虚假信息或引流，可以选择性过滤或标记。
  • 卖家信用筛选：虽然爬取卖家信用信息更复杂，但可以通过商品描述中的“芝麻信用极高”等文本进行简单筛选。

6.4 性能与资源优化

• 问题：同时监控多个关键词，浏览器实例开得太多，导致内存占用高。
• 优化：
  • 复用浏览器上下文：在XianyuCrawler中，不要为每次搜索都开启和关闭一个浏览器。可以在一个浏览器实例内创建多个独立的页面（context）来执行任务。
  • 异步并发：利用asyncio.gather并发执行多个关键词的搜索任务，而不是顺序执行，可以显著缩短单次扫描总时间。
  • 限制扫描页数：对于刷新快的品类，只扫描前2-3页通常就够了，因为新商品一般出现在前面。

6.5 通知信息过载

• 问题：一旦监控条件设置得比较宽，可能会在短时间内收到大量推送，造成干扰。
• 优化：
  • 设置“冷静期”：对于同一个商品ID，即使多次扫描到，也只通知一次。我们的ItemFilter中的seen_items集合就是干这个的。
  • 消息聚合：不要发现一个商品就推送一次。可以等单次扫描任务全部结束后，将本次所有符合条件的商品汇总成一条消息推送。
  • 设置推送阈值：比如只推送价格低于设定值80%的“好价”商品。

7. 进阶玩法与扩展思路

基础功能稳定后，你可以考虑以下扩展方向，让这个助手更加强大：

1. 多平台监控：将架构抽象化，使其不仅能监控闲鱼，还能监控其他二手平台（如转转）。只需要为每个平台实现一个特定的Crawler子类即可。
2. 价格追踪与趋势分析：将抓取到的商品价格和时间存入数据库（如SQLite或MySQL）。长期积累数据后，可以分析某个型号商品的价格走势，帮你判断当前是否是入手的好时机。
3. 图像识别辅助：对于某些商品，标题信息可能不全。可以尝试使用OCR技术（如pytesseract）识别商品主图中的文字，或者用简单的图像分类判断商品成色（全新、二手、破损）。
4. 自动化询价与沟通：在确认规则极其严格（如“iPhone 13 256G 国行 在保 价格<4000”）后，可以尝试自动发送预设的礼貌询价语。但必须极其谨慎：话术要像真人，发送频率要极低（如一天只发1-2条），并且最好加入人工确认环节。滥用此功能极易导致账号被封。
5. Web仪表板：使用Flask或Streamlit搭建一个简单的Web界面，方便在手机上随时修改监控关键词、查看历史记录和统计图表。

这个项目从技术上看，是Web爬虫、自动化测试、任务调度和消息推送等多个领域的综合应用。从实用角度看，它精准地解决了一个高频痛点。自己动手搭建和优化这样一个系统的过程，不仅能让你真正“捡到漏”，更能深入理解现代Web交互、反爬对抗和自动化系统的设计思路。记住，工具是为你服务的，务必在效率和合规之间找到平衡点，愉快地享受技术带来的便利吧。如果在搭建过程中遇到任何具体问题，比如Playwright的某个API不会用，或者推送服务配置不成功，随时可以针对那个点进行更深入的探讨。