闲鱼智能监控机器人：关键词轮询与自动通知实战

简介这是一套面向闲鱼找货与二手交易场景的智能监控分析系统源码适合希望用自动化手段替代人工刷新的开发者、副业卖家与选品运营者。项目以 Playwright 抓取结合 AI 分析为核心通过可视化 Web 界面完成多任务实时监控、自然语言筛选规则编辑与结果浏览无需命令行和配置文件即可上手。压缩包共 37 个文件、约 12.31MB包含 11 个 Python 模块爬虫、AI 处理、解析与提示词工具、2 个 yml 与 docker-compose 部署配置、2 个 html 与配套 js、css 前端页面以及 txt 提示词、png/jpg 截图、md 说明和 docx 教程等覆盖采集、分析、调度与通知全链路。已有 326 人学习。读者可据此搭建多任务并发监控用自然语言生成带复杂逻辑的筛选条件结合商品图文与卖家画像排除低质信息并通过 Cron 表达式定制执行计划、多渠道即时通知同时参考内置随机延迟与行为策略降低账号风险快速获得一套可二次开发的完整方案。1. 闲鱼智能监控机器人从关键词到任务流的自动化拆解做二手电商运营的人都有一个共识闲鱼上的好货不是搜出来的是蹲出来的。一款热门相机镜头、一张低价显卡、一批清仓的母婴用品往往在发布后几分钟内就被拍下等手动刷新看到时早已下架。闲鱼智能监控机器人要解决的就是这个时间差问题——它把“人盯着屏幕刷”变成“程序按关键词轮询、命中规则后自动通知甚至自动执行后续动作”。这套任务监控分析系统的核心链路并不复杂关键词配置、定时抓取、数据解析、规则匹配、通知推送、日志留存。适合谁用做闲鱼无货源选品的运营、需要批量监控竞品价格的小团队、以及想把自己从重复刷新里解放出来的个人卖家。但先泼一盆冷水闲鱼没有开放商品搜索 API所有监控方案都绕不开页面解析或接口模拟稳定性和合规边界是必须提前想清楚的事。下面按落地顺序把选型、实现、参数和踩坑一次讲透。2. 监控任务的核心链路关键词、轮询与数据解析怎么串起来2.1 为什么不能直接调官方接口以及替代方案怎么选闲鱼作为阿里系产品对外没有公开的商品搜索或订单查询接口。市面上常见的做法有三类一是模拟移动端请求抓取返回的 JSON 数据二是用无头浏览器渲染页面后提取 DOM 文本三是通过抓包工具分析 PC 端或 App 端的请求结构复现关键参数。三种方案里模拟请求效率最高但参数签名容易变无头浏览器最稳但资源消耗大抓包复现介于两者之间。我一般会先用抓包工具把搜索接口的请求头、查询参数、签名算法摸清楚再用 Python 的 requests 库做最小复现。这里的关键是理解闲鱼搜索接口的翻页机制和排序参数否则监控范围会漏掉大量新上架商品。import requests import time import hashlib # 模拟闲鱼搜索请求的最小示例实际参数需根据抓包结果替换 def fetch_xianyu_search(keyword, page1): base_url https://h5api.m.goofish.com/h5/mtop.taobao.idlemtopsearch.pc.search/1.0/ params { q: keyword, page: page, sort: new, # 按最新发布排序监控场景必选 rows: 20, # 每页条数建议不超过30 } headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) ..., Referer: https://www.goofish.com/, Cookie: 你的cookie字符串, # 必须携带否则返回登录页 } # 实际请求中可能还需要 sign 参数需根据抓包结果计算 resp requests.get(base_url, paramsparams, headersheaders, timeout10) if resp.status_code 200: return resp.json() return None # 轮询调度每30秒拉取一次第一页对比上次结果 last_ids set() while True: data fetch_xianyu_search(索尼A7M4) if data: items data.get(data, {}).get(resultList, []) for item in items: item_id item.get(itemId) if item_id not in last_ids: print(f新商品: {item.get(title)} 价格: {item.get(price)}) last_ids.add(item_id) time.sleep(30)这段代码的逻辑是用关键词请求搜索接口按最新排序取第一页把商品 ID 存入集合做去重新出现的 ID 就是刚上架或刚被搜到的商品。参数方面sort必须设为按时间排序否则拿到的是综合排序结果新商品会被淹没rows不宜过大20 到 30 条足够覆盖几分钟内的新增Cookie是硬门槛没有登录态请求会被重定向。轮询间隔 30 秒是经验值再快容易触发风控再慢可能错过秒杀价商品。2.2 数据解析从 JSON 到结构化字段的映射拿到接口返回的 JSON 后需要提取标题、价格、发布时间、卖家 ID、商品链接等字段。闲鱼的返回结构层级较深不同端PC 端、App 端、国外版 Goofish的字段名还有差异。我一般会先打印完整 JSON 的前两层结构确认resultList里每个 item 的字段路径再写映射函数。价格字段有时是字符串带“¥”符号有时是分单位整数需要统一清洗。发布时间如果是相对时间如“3分钟前”要转换成绝对时间戳才能做时效性过滤。import re from datetime import datetime, timedelta def parse_item(raw_item): 将闲鱼返回的原始 item 解析为结构化字典 result {} # 标题通常在 title 或 mainTitle 字段 result[title] raw_item.get(title, ).strip() # 价格清洗去掉货币符号统一为浮点数 price_str str(raw_item.get(price, 0)) price_match re.search(r[\d.], price_str) result[price] float(price_match.group()) if price_match else 0.0 # 商品 ID 和链接 result[item_id] raw_item.get(itemId, ) result[url] fhttps://www.goofish.com/item?id{result[item_id]} # 发布时间处理相对时间转绝对时间 pub_text raw_item.get(publishTime, ) if 分钟前 in pub_text: minutes int(re.search(r(\d), pub_text).group(1)) result[publish_ts] datetime.now() - timedelta(minutesminutes) elif 小时前 in pub_text: hours int(re.search(r(\d), pub_text).group(1)) result[publish_ts] datetime.now() - timedelta(hourshours) else: result[publish_ts] datetime.now() return result解析环节最容易翻车的地方是字段缺失。闲鱼不同类目的商品返回字段并不完全一致比如虚拟商品可能没有物流信息拍卖商品的价格字段结构不同。我的习惯是在解析函数里对每个字段做防御性取值缺失时给默认值而不是直接抛异常否则一条脏数据就能让整个监控循环挂掉。另外价格清洗时要注意“面议”这类非数字文本正则匹配不到就归零后续规则里再单独处理。3. 规则引擎与通知推送让监控机器人真正“智能”起来3.1 规则配置价格阈值、关键词排除与卖家过滤监控机器人如果只是把所有新商品都推给你那和手动刷新没有本质区别。真正的价值在于规则过滤只推价格低于某个阈值的、标题包含特定型号的、排除掉商家的重复铺货。我一般把规则设计成 JSON 配置每条规则包含匹配字段、操作符和阈值。比如“价格小于 2000 且标题包含‘全新’且卖家不是商家”这样一条规则就能过滤掉大量噪音。规则引擎的实现可以用简单的 if-else 链也可以用策略模式做可扩展结构。对于大多数个人使用场景配置化加简单匹配就够用。# 规则配置示例JSON 结构便于动态修改 rules [ { name: 低价相机镜头, conditions: [ {field: price, op: lt, value: 2000}, {field: title, op: contains, value: 镜头}, {field: title, op: not_contains, value: 维修} ], action: notify }, { name: 显卡秒杀, conditions: [ {field: price, op: lt, value: 1500}, {field: title, op: regex, value: RTX\\s*30[6-9]0} ], action: notify_and_log } ] def match_rules(item, rules): 对单个商品应用所有规则返回命中的规则列表 matched [] for rule in rules: ok True for cond in rule[conditions]: field_val item.get(cond[field], ) if cond[op] lt and not (field_val cond[value]): ok False elif cond[op] contains and cond[value] not in str(field_val): ok False elif cond[op] not_contains and cond[value] in str(field_val): ok False elif cond[op] regex and not re.search(cond[value], str(field_val)): ok False if not ok: break if ok: matched.append(rule[name]) return matched规则匹配的顺序有讲究先做价格过滤因为价格是数值比较最快再做标题文本匹配正则表达式开销较大放在后面。如果规则数量超过 20 条建议按关键词分组先用关键词缩小候选集再逐条匹配否则每轮轮询都要对几百条商品做全量规则计算CPU 占用会明显上升。另外not_contains条件要慎用比如排除“维修”可能会误伤“维修工具套装”这类正常商品最好结合多个条件做与运算。3.2 通知渠道从 Server酱 到企业微信机器人的落地选择命中规则后通知推送是最后一公里。常见的免费渠道有 Server酱、PushPlus、企业微信机器人、钉钉机器人、BarkiOS。选择依据是是否需要多设备同步、是否在意延迟、是否能接受第三方中转。我一般用企业微信机器人做主力因为消息到达率高、支持 Markdown 格式、可以直接在手机上点开商品链接。配置方式很简单在企业微信群里添加机器人拿到 Webhook 地址用 POST 请求发送 JSON 消息即可。import requests import json def send_wecom_notify(webhook_url, item, rule_name): 通过企业微信机器人推送商品信息 content f**闲鱼监控命中** 规则: {rule_name} 标题: {item[title]} 价格: ¥{item[price]} 链接: [点击查看]({item[url]}) 时间: {item[publish_ts].strftime(%H:%M:%S)} payload { msgtype: markdown, markdown: {content: content} } resp requests.post(webhook_url, jsonpayload, timeout5) return resp.status_code 200通知环节的坑在于频率控制。如果一轮轮询命中十几条商品瞬间推送十几条消息会被渠道限流甚至封禁 Webhook。我的做法是加一个聚合窗口30 秒内的命中商品合并成一条消息发送或者设置每条规则每分钟最多推送一次。另外通知内容里带上商品链接很重要看到消息后能直接跳转否则还要手动去搜体验差很多。对于需要自动下单的场景通知之后还可以接一个自动化脚本但这就涉及更复杂的风控对抗后面避坑章节会展开。4. 避坑与排查闲鱼监控系统最容易翻车的五个地方4.1 Cookie 失效导致监控静默停止现象程序运行几个小时后不再输出任何新商品日志里没有报错但通知也停了。原因闲鱼的登录态 Cookie 有效期通常只有几小时到一天过期后接口返回的是登录页 HTML 而不是 JSON解析函数拿不到resultList字段返回空列表程序误以为“没有新商品”。解决在请求函数里加状态码和返回内容类型判断如果返回的是 HTML 或状态码为 302立即触发 Cookie 失效告警。更稳妥的做法是接入扫码登录的自动化流程定期刷新 Cookie但这部分实现复杂度较高个人使用建议手动更新加失效提醒。4.2 请求频率过高触发滑块验证现象运行一段时间后接口开始返回验证码页面或 403 状态码换 IP 后恢复。原因闲鱼对同一 IP 或同一账号的请求频率有阈值超过后触发风控。轮询间隔低于 10 秒、单次请求翻页过多、短时间内大量关键词并发都会加速触发。解决轮询间隔设置在 20 到 60 秒之间单账号同时监控的关键词不超过 5 个翻页只取第一页。如果确实需要更大规模考虑多账号轮换或降低频率不要试图用代理 IP 硬扛成本高且不稳定。4.3 商品 ID 去重逻辑在翻页时误判现象同一商品被重复推送多次或者新商品被漏掉。原因去重集合只存了当前页的 ID翻页后上一页的商品滑到第二页又被当成新商品或者商品重新上架后 ID 不变但发布时间更新去重逻辑直接跳过。解决去重集合要持久化到本地文件或数据库并且记录每个 ID 的最后一次见到时间。对于重新上架的商品可以结合发布时间判断如果发布时间比上次记录晚即使 ID 相同也视为新商品。4.4 价格解析遇到“面议”或区间价直接崩溃现象程序抛出ValueError或TypeError监控循环中断。原因闲鱼部分商品价格显示为“面议”或“¥100-200”正则匹配[\d.]会拿到多个数字或空值直接转 float 失败。解决解析函数里用 try-except 包裹类型转换失败时价格设为 -1 或 None规则匹配时对无效价格做特殊处理。区间价可以取最低值作为参考或者单独标记为“区间价”不参与阈值比较。4.5 无头浏览器方案的内存泄漏现象用 Selenium 或 Playwright 跑监控运行几小时后内存占用飙升最终被系统杀掉。原因每次请求都新建浏览器实例或者页面对象没有正确关闭导致内存持续增长。解决复用同一个浏览器实例每个请求开新标签页而不是新窗口处理完后关闭标签页。设置定时重启策略比如每处理 500 个请求重启一次浏览器。如果对稳定性要求高建议用 requests 模拟请求替代无头浏览器资源消耗低一个数量级。5. 进阶技巧用历史数据做价格趋势与选品分析监控系统跑起来之后积累的数据本身就是资产。我习惯把每次命中的商品写入 SQLite 数据库字段包括商品 ID、标题、价格、发布时间、卖家 ID、命中规则。跑上一两周后就可以做简单的趋势分析某个关键词下的商品均价是多少、价格分布集中在哪个区间、哪些卖家频繁上架同类商品。这些分析结果反过来可以优化监控规则比如把价格阈值调到均价以下 20%或者把频繁铺货的卖家加入黑名单。import sqlite3 import pandas as pd # 建表语句记录每次命中的商品快照 conn sqlite3.connect(xianyu_monitor.db) conn.execute( CREATE TABLE IF NOT EXISTS hits ( id INTEGER PRIMARY KEY AUTOINCREMENT, item_id TEXT, title TEXT, price REAL, publish_ts TEXT, seller_id TEXT, rule_name TEXT, created_at TEXT DEFAULT CURRENT_TIMESTAMP ) ) # 分析示例查询某关键词下近7天的价格分布 df pd.read_sql_query( SELECT price, COUNT(*) as cnt FROM hits WHERE title LIKE %索尼% AND created_at datetime(now, -7 days) GROUP BY CAST(price / 100 AS INTEGER) * 100 ORDER BY price , conn) print(df)这个分析表能直观看到价格集中在哪些区间。如果发现大量商品集中在 1500 到 1800 元而你的阈值设在 1200那可能一周都命中不了一次需要调整预期。另一个实用技巧是监控“降价”行为同一商品 ID 如果第二次出现时价格更低说明卖家在调价这类商品往往有议价空间。实现方式是在数据库里对同一 item_id 保留多条记录查询时取最新价格和上次价格做对比。最后说一个我踩过的坑不要试图用监控系统做自动下单。闲鱼的订单接口有更严格的风控自动下单脚本很容易被判定为异常行为轻则订单被取消重则账号受限。监控系统的边界是“发现”和“通知”把决策和操作留给人这样既安全又可持续。我现在的习惯是每天早上花十分钟看一遍夜间积累的命中记录比整天盯着手机刷效率高得多。希望帮到你。本文还有配套的精品资源点击获取