基于Playwright与AI的闲鱼智能监控机器人：自动化抓取与语义分析实战 1. 项目概述当“捡漏”遇上自动化在二手交易市场里信息差就是最大的财富。无论是寻找心仪的绝版商品、监控特定型号的数码产品降价还是批量筛选高性价比的货源手动刷新、比价、沟通的效率低到令人发指。我花了大量时间在闲鱼上“淘金”深知其中的痛点错过一个秒杀可能就与好物失之交臂同时跟踪多个关键词和卖家精力根本不够用。于是一个想法诞生了能不能做一个机器人让它7x24小时不知疲倦地帮我盯着闲鱼不仅能自动刷新、发现新上架或降价的商品还能初步分析商品描述甚至帮我完成一些基础的沟通这就是“闲鱼智能监控机器人”项目的初衷。它不是一个简单的页面刷新工具而是一个集成了自动化浏览Playwright与智能分析AI的多任务监控分析系统。简单来说这个工具能帮你做三件事多维度监控同时监控多个关键词、多个卖家店铺、多个收藏夹的动态一旦有符合条件的商品上新或价格变动立即通知你。智能初筛利用AI大模型的能力自动解读商品标题和描述过滤掉“女生自用”、“99新”这类模糊表述识别出真正的“个人闲置”与“专业商家”甚至可以总结商品的核心瑕疵。自动化交互在设定好的规则下自动发送预设的询问话术帮你抢占沟通先机或者自动完成一些固定流程如获取快递模板。它适合谁如果你是二手爱好者、小型卖家、寻找特定配件的极客或者任何需要从信息洪流中高效提取价值的人这个工具都能显著提升你的效率。接下来我将详细拆解这个项目的设计思路、核心实现以及我踩过的那些坑。2. 技术选型与整体架构设计为什么是Playwright AI这个组合并非凭空想象而是经过了对多种技术方案对比后的选择。2.1 为什么选择 Playwright 而非 Selenium 或 Puppeteer自动化浏览器测试工具有很多Selenium 老当益壮Puppeteer 是 Chrome 亲儿子但 Playwright 是后起之秀其特性完美契合这个项目。对现代 Web 技术的原生支持闲鱼作为一款活跃的移动端优先的应用其 Web 端或 H5 端大量使用了动态加载、单页面应用SPA技术。Playwright 内置了对这些技术的等待策略比如wait_for_load_state(‘networkidle’)能更智能地判断页面何时真正加载完毕避免了因元素未加载而导致的定位失败。强大的浏览器上下文Browser Context与多页面管理这是实现“多任务监控”的核心。一个 Browser Context 可以看作一个独立的浏览器会话拥有独立的 cookies、localStorage。我可以为每个监控任务例如监控关键词“索尼微单”、监控某个卖家创建一个独立的 Context 甚至 Page它们完全隔离互不干扰。这比管理多个浏览器进程或标签页要优雅和高效得多。可靠的元素定位与丰富的 APIPlaywright 提供了text,has-text等非常灵活的文本定位器对于闲鱼这种结构可能变化的页面非常友好。其截图、录制视频、拦截网络请求等功能也为调试和更高级的监控如监控特定接口的数据提供了可能。跨浏览器支持虽然我们主要用 Chromium但 Playwright 同时支持 Firefox 和 WebKit为后续可能的兼容性测试留有余地。注意直接使用 Playwright 操作商业网站需要格外谨慎过于频繁或规律的请求可能触发反爬机制。我们的策略是模拟人类行为随机等待时间、使用真实的 User-Agent、通过浏览器上下文维持登录态而非直接调用接口。2.2 AI 能力的接入从关键词匹配到语义理解传统的监控工具只能做简单的关键词匹配。但闲鱼的商品描述千奇百怪“iPhone 13 自用一手”和“iPhone13 公司年会奖品未拆封”可能指向同一类商品但卖家性质不同。这时就需要 AI 出场。本地模型 vs. 云 API为了控制成本和保证响应速度项目采用了混合策略。轻量级任务分类、关键词提取使用本地运行的轻量模型例如通过transformers库加载小型的 BERT 变体如bert-tiny-chinese。虽然能力不如千亿大模型但对于“判断是否为商家”、“提取商品品牌型号”这类任务准确率足够高且零延迟、零费用。复杂任务描述总结、意图分析对于需要深度理解的任务如“总结这个相机的瑕疵描述”则调用云端大模型的 API如 OpenAI GPT、国内合规的同类大模型 API。通过任务队列异步处理避免阻塞监控主流程。提示词Prompt工程这是让 AI 正确工作的关键。我们不是简单地把商品描述扔给 AI而是设计结构化的提示词。例如你是一个二手商品分析助手。请分析以下商品描述 [商品描述原文] 请按以下 JSON 格式输出 { “seller_type”: “personal” 或 “business” “item_condition”: “new”, “like_new”, “good”, “fair”, “poor”, “key_flaws”: [“划痕描述1”, “功能问题描述2”], “is_urgent_sale”: true 或 false }这样AI 的输出就是结构化的数据方便程序后续处理和决策。2.3 系统架构图概念层面整个系统可以看作一个生产者-消费者模型配合任务调度中心。[任务配置中心] | | (生成监控任务) v [Playwright 集群] --(抓取页面数据)-- [原始数据队列] | | |(行为模拟、登录态维护) v | [AI 处理管道] | | | |--(轻量模型)-- [本地分析模块] | |--(复杂任务)-- [云 API 调用模块] | | v v [行为日志] [结构化结果队列] | v [策略引擎 通知中心] | v [Telegram/钉钉/邮件通知]核心流程配置任务后调度中心分配给 Playwright 工作节点。节点抓取数据原始数据HTML、JSON进入队列。AI 管道消费队列数据进行智能处理。处理后的结构化结果交给策略引擎判断是否触发通知例如价格低于阈值、卖家是个人、商品描述包含“急出”等。所有步骤异步执行互不阻塞。3. 核心模块拆解与实现细节3.1 Playwright 自动化模块稳定抓取的基石这个模块的目标是稳定、可靠、像真人一样地获取闲鱼页面数据。3.1.1 登录态维持与防检测策略直接处理登录流程复杂且易触发验证码。更优的做法是手动登录一次然后持久化浏览器上下文的状态。import asyncio from playwright.async_api import async_playwright async def get_logged_in_context(): async with async_playwright() as p: # 1. 启动浏览器建议使用 headedFalse 无头模式运行节省资源 browser await p.chromium.launch(headlessFalse) # 调试时可设为True # 2. 创建上下文并加载之前保存的存储状态 context await browser.new_context() try: await context.storage_state(path“auth/xy_state.json”) print(“已加载已有登录状态”) except FileNotFoundError: print(“未找到存储状态需手动登录”) page await context.new_page() await page.goto(“https://2.taobao.com”) # 闲鱼官网 # 这里阻塞等待用户手动扫码或密码登录 input(“请手动完成登录然后按回车继续...”) # 登录成功后保存状态 await context.storage_state(path“auth/xy_state.json”) print(“登录状态已保存”) await page.close() return context实操心得storage_state保存的是 cookies 和 localStorage。务必确保保存状态的页面域名与后续操作的域名一致。闲鱼可能涉及*.taobao.com等多个域名登录后最好在目标域名页面下再保存一次状态。此外登录态可能过期需要设计定期检查与更新的机制。3.1.2 页面导航与数据提取闲鱼列表页是动态加载的需要模拟滚动。async def monitor_keyword(context, keyword): page await context.new_page() await page.goto(f“https://s.2.taobao.com/list/?q{keyword}”) items_data [] max_scroll 5 # 控制滚动深度避免无限加载 for i in range(max_scroll): # 等待商品卡片元素出现 await page.wait_for_selector(“.Card”) # 提取当前视窗内的商品信息 current_items await page.eval_on_selector_all(“.Card”, “cards cards.map(card ({ ‘title’: card.querySelector(‘.Title’)?.innerText, ‘price’: card.querySelector(‘.Price’)?.innerText, ‘link’: card.querySelector(‘a’)?.href, ‘seller’: card.querySelector(‘.SellerInfo’)?.innerText, ‘item_id’: card.dataset.itemId // 尝试从数据属性获取ID }))”) items_data.extend(current_items) # 模拟人类滚动随机延迟滚动距离随机 import random scroll_height random.randint(500, 1000) await page.evaluate(f“window.scrollBy(0, {scroll_height})”) await page.wait_for_timeout(random.uniform(1000, 3000)) # 随机等待1-3秒 # 检查是否已到页面底部或加载出更多内容 # 可以通过检查特定元素或页面高度变化来判断 # ... await page.close() return items_data注意事项page.evaluate是在浏览器环境中执行 JavaScript非常适合提取复杂数据。但闲鱼的反爬可能会检测到自动化操作因此滚动、等待时间一定要加入随机性。此外商品 ID 是去重的关键最好能从数据属性或链接中解析出来而不是依赖易变的文本。3.1.3 多任务与资源管理使用asyncio管理多个并发监控任务。async def main_monitor_loop(config_list): 主监控循环管理多个配置任务 context await get_logged_in_context() tasks [] for config in config_list: task asyncio.create_task( run_single_monitor(context, config), namef“monitor-{config[‘name’]}” ) tasks.append(task) # 错开任务启动时间避免并发请求过于密集 await asyncio.sleep(random.uniform(1, 5)) # 等待所有任务完成或持续运行 await asyncio.gather(*tasks, return_exceptionsTrue) # 实际应用中这里应该是一个无限循环定期执行任务3.2 AI 智能分析模块从文本到洞察原始的商品标题和描述是字符串我们的目标是将其转化为结构化的、可编程的洞察。3.2.1 轻量级本地模型集成使用sentence-transformers或transformers进行语义相似度计算和简单分类。from sentence_transformers import SentenceTransformer, util import torch class LocalItemAnalyzer: def __init__(self): # 加载一个轻量级的中文语义模型 self.model SentenceTransformer(‘paraphrase-multilingual-MiniLM-L12-v2’) # 预定义一些关键概念的嵌入向量 self.business_keywords [“批发”, “代理”, “全新”, “大量”, “公司”, “实体店”] self.personal_keywords [“自用”, “闲置”, “换下”, “个人”, “搬家”, “年会奖品”] self.business_embeddings self.model.encode(self.business_keywords) self.personal_embeddings self.model.encode(self.personal_keywords) def analyze_seller_type(self, description): desc_embedding self.model.encode(description) # 计算与商家关键词和个人关键词的相似度 biz_sim util.pytorch_cos_sim(desc_embedding, self.business_embeddings).max().item() per_sim util.pytorch_cos_sim(desc_embedding, self.personal_embeddings).max().item() if biz_sim per_sim and biz_sim 0.5: # 设置一个阈值 return “business”, biz_sim elif per_sim 0.4: return “personal”, per_sim else: return “unknown”, max(biz_sim, per_sim)3.2.2 云端大模型 API 调用对于更复杂的分析如总结瑕疵调用云端 API。import openai # 或国内合规替代品 import json async def analyze_with_llm(item_title, item_desc): prompt f“”” 你是一个专业的二手商品评估助手。请分析以下商品信息 商品标题{item_title} 商品描述{item_desc} 请输出一个JSON对象包含以下字段 1. “summary”: 对商品状况的一句话总结30字内。 2. “flaws”: 一个数组列出描述中明确提到的所有瑕疵或问题。如果没有则为空数组。 3. “urgency”: 布尔值卖家是否表现出急于出售如使用“急出”、“今天必出”等词。 4. “authenticity_hints”: 数组描述中能佐证商品真伪或来源的线索。 “”” try: response await openai.ChatCompletion.acreate( model“gpt-3.5-turbo”, # 或 “gpt-4” messages[{“role”: “user”, “content”: prompt}], temperature0.2, # 低温度保证输出稳定 response_format{ “type”: “json_object” } # 要求返回JSON ) result json.loads(response.choices[0].message.content) return result except Exception as e: print(f“LLM分析失败: {e}”) return None踩坑记录大模型 API 有速率限制和成本。务必做好请求队列、失败重试和成本监控。不要在每个商品上都调用而是先通过本地模型过滤出高价值的商品如价格低于阈值、卖家是个人再对这些商品进行深度 AI 分析。另外提示词的设计需要反复调试才能让 AI 输出稳定、格式正确的结果。3.3 任务调度与通知模块这是机器人的大脑和嘴巴负责决定“什么时候监控什么”以及“发现了什么要告诉你”。3.3.1 基于时间的灵活调度使用apscheduler库实现复杂的定时任务。from apscheduler.schedulers.asyncio import AsyncIOScheduler from apscheduler.triggers.cron import CronTrigger from apscheduler.triggers.interval import IntervalTrigger scheduler AsyncIOScheduler() # 例子1每30分钟监控一次热门关键词白天频率高晚上低 scheduler.scheduled_job(IntervalTrigger(minutes30), id‘monitor_hot’, max_instances1) async def job_monitor_hot(): if 8 datetime.now().hour 22: # 只在早8点到晚10点运行 await monitor_keywords([“索尼 A7M3”, “Switch 续航版”]) # 例子2每天凌晨2点全面扫描一次收藏夹 scheduler.scheduled_job(CronTrigger(hour2, minute0), id‘monitor_fav’) async def job_monitor_fav(): await monitor_favorite_sellers() # 例子3对特定高价值商品每5分钟检查一次价格 scheduler.scheduled_job(IntervalTrigger(minutes5), id‘monitor_specific_item’) async def job_monitor_specific_item(): await check_price_for_item(“item_id_123456”)3.3.2 分级通知策略不是所有发现都值得立即推送。设计分级通知避免信息过载。class NotificationManager: def __init__(self): self.notification_channels {‘telegram’: ..., ‘dingtalk’: ...} # 初始化各个通知渠道 async def send_notification(self, item, priority‘medium’): 根据优先级发送通知 message self._format_message(item) if priority ‘critical’: # 关键通知价格远低于市场价、卖家描述“急出” # 同时发送所有渠道并特定人 for channel in self.notification_channels.values(): await channel.send(message, at_allTrue) elif priority ‘high’: # 高优先级心仪商品降价、新上架个人卖家 # 发送主要渠道如Telegram await self.notification_channels[‘telegram’].send(message) elif priority ‘medium’: # 中优先级普通上新、每日摘要 # 可以积累一批每小时或每天汇总发送一次 self._add_to_digest(message) # low 优先级可能只记录日志不主动推送3.3.3 数据持久化与去重使用 SQLite 或 PostgreSQL 存储历史数据核心是去重和比价。-- 简单的商品表结构 CREATE TABLE items ( id INTEGER PRIMARY KEY AUTOINCREMENT, platform_item_id TEXT UNIQUE, -- 平台商品ID用于去重 title TEXT, price REAL, seller_info TEXT, link TEXT, ai_analysis JSON, -- 存储AI分析的结构化结果 first_seen TIMESTAMP, last_updated TIMESTAMP, price_history JSON -- 记录价格变化如 [{“time”: “...”, “price”: 1000}] );每次抓取到新商品先根据platform_item_id查询是否存在。如果存在则比较最新价格与记录的价格如果变化超过设定比例如5%则触发价格变动通知并更新price_history。4. 部署、优化与实战心得4.1 部署方案从本地到服务器本地开发/轻量使用直接运行 Python 脚本。适合监控少数几个关键词。注意电脑休眠问题。服务器部署推荐使用 Linux 服务器如 Ubuntu。需要解决无图形界面的浏览器运行问题。# 安装 Playwright 系统依赖及浏览器 sudo apt-get update sudo apt-get install -y libgbm-dev libnss3 libatk-bridge2.0-0 libdrm-dev libxkbcommon-dev libasound2 pip install playwright playwright install chromium --with-deps进程管理使用systemd或supervisord管理 Python 进程保证崩溃后自动重启。日志配置logging模块将日志输出到文件便于排查问题。资源隔离使用 Docker 容器化部署是更干净的选择可以打包所有依赖。4.2 性能优化与反反爬策略并发控制虽然asyncio支持高并发但向同一网站发起过多并行请求是自杀行为。使用信号量asyncio.Semaphore限制并发任务数例如最多同时进行5个监控页面。请求间隔随机化在任务之间、滚动操作之间加入随机等待时间random.uniform(a, b)模拟人类的不规律操作。使用住宅代理IP高级如果监控频率极高考虑使用付费的住宅代理IP池为每个 Browser Context 分配不同的IP降低被封风险。Playwright 支持通过launch_persistent_context的args参数设置代理。浏览器指纹管理Playwright 可以通过context.new_context()时的viewport,user_agent等参数模拟不同设备。定期更换这些参数可以增加多样性。优雅降级与重试网络请求可能失败页面结构可能变化。代码中必须包含健壮的错误处理和重试逻辑。async def robust_fetch(page, url, max_retries3): for attempt in range(max_retries): try: response await page.goto(url, wait_until“networkidle”, timeout30000) if response.ok: return True except Exception as e: print(f“第{attempt1}次尝试失败: {e}”) await asyncio.sleep(2 ** attempt) # 指数退避 return False4.3 常见问题与排查技巧实录问题1Playwright 提示Target closed或Browser context closed。原因通常是由于浏览器实例意外崩溃或页面被垃圾回收。排查检查系统内存是否充足。无头浏览器很耗内存。确保代码中正确管理了 Page 和 Context 的生命周期每个new_page()都对应一个page.close()避免泄漏。在服务器上可能是缺少必要的库。确保已安装playwright install-deps列出的所有依赖。问题2抓取不到数据元素定位失败。原因页面结构已更新或元素加载太慢。排查使用 Playwright 的调试工具运行脚本时加上headlessFalse亲眼看看页面加载情况。使用page.pause()暂停脚本打开开发者工具检查元素。使用更宽松的定位器优先使用page.get_by_text()或page.get_by_role()等基于文本和语义的定位器它们比基于 CSS 选择器的page.locator(‘.class’)更健壮。增加等待时间或使用更智能的等待将wait_for_selector替换为wait_for_load_state(‘networkidle’)或locator.wait_for()。问题3AI 分析结果不稳定或格式错误。原因提示词不清晰或大模型“自由发挥”。排查结构化输出明确要求 AI 返回 JSON 格式并指定字段。如使用 OpenAI API设置response_format{ “type”: “json_object” }。提供示例在提示词中给出一个清晰的输入输出示例Few-shot Learning能极大提升输出稳定性。降低 Temperature将 API 调用的temperature参数设为较低值如 0.1-0.3减少随机性。后处理校验对 AI 返回的 JSON 进行解析和校验如果格式错误可以丢弃或重试。问题4监控一段时间后被限制访问或出现验证码。原因行为模式被识别为机器人。应对首要策略是降低频率这是最有效的方法。将监控间隔从几分钟调整为几十分钟甚至小时级。模拟更真实的行为在监控流程中加入随机点击、鼠标移动page.mouse.move()等无意义但像人的操作。考虑切换账号或设备指纹如果只有一个账号风险较高。有条件可以准备多个账号轮换使用。终极方案识别到验证码后暂停任务发送通知给人工处理。切勿尝试自动破解验证码这违反平台规则且技术难度极高。这个项目从构思到实现是一个典型的“用技术解决生活痛点”的过程。它不只是一个脚本的堆砌而是涉及了自动化、AI、系统设计、反反爬等多个层面的思考。最大的体会是在真实网络环境中鲁棒性远比功能性更重要。你的代码必须能处理各种网络异常、页面变化和平台策略。从“能跑通”到“能稳定运行”中间需要大量的测试、观察和调整。最后请务必在法律和平台规则的框架内合理使用此类工具尊重数据所有者的权益将其作为提升个人效率的助手而非恶意爬取或干扰的工具。