简介这是一套面向二手交易场景的闲鱼多任务实时监控与智能分析工具基于 Playwright 与多模态大语言模型构建适合希望自动追踪商品、按需筛选好价的开发者与进阶用户。系统提供完整的 Web 管理界面支持任务可视化配置、AI 标准在线编辑、运行日志实时查看与结果筛选无需直接操作命令行用户只需用自然语言描述购买需求即可一键生成包含复杂筛选逻辑的监控任务并借助 GPT-4o 结合商品图文与卖家画像做深度分析再通过 ntfy.sh、企业微信机器人或 Bark 即时推送。资源包共 37 个文件约 12.31MB以 Python 源码为主体辅以 HTML、JS、CSS 前端页面、YAML 与 Dockerfile 部署配置、JSON 示例及说明文档覆盖爬虫、AI 处理、提示词生成与定时调度等模块。目前已有 442 人学习可帮助读者快速搭建可定制、可容器化部署的监控分析系统。1. 闲鱼智能监控机器人到底在监控什么从一条低价漏网商品说起你盯了三天的那款二手镜头标价 1800你觉得贵没下手。第四天早上刷到一条 950 的点进去已经显示「已售出」卖家说凌晨两点就被拍走了。这不是运气问题是信息差问题——闲鱼上真正的好价商品从发布到成交往往只有几分钟窗口靠人肉刷新根本抢不过脚本。闲鱼智能监控机器人要解决的就是这件事用程序代替你 7×24 小时盯着指定关键词的新发布商品自动过滤、自动分析、自动通知让你在别人还在睡觉的时候就知道有货上架了。这套「闲鱼任务监控分析系统」的核心不是爬虫本身而是三层结构采集层负责按关键词和筛选条件拉取新商品分析层负责判断这条商品值不值得推给你通知层负责在最短时间内把消息送到你手机上。适合谁用想捡漏的二手玩家、做无货源选品的卖家、需要监控竞品定价的运营以及任何愿意花一个下午把环境搭起来、之后长期受益的人。下面从架构选型一路讲到参数调优和踩坑记录能抄的代码我都贴出来。2. 采集层怎么搭关键词轮询、频率控制与反爬边界2.1 为什么不用官方 API 而走页面轮询闲鱼没有开放的商品搜索 API 给个人开发者市面上能用的方案基本是模拟搜索请求或解析搜索结果页。常见做法是构造搜索 URL带上关键词、排序方式、价格区间等参数然后解析返回的 HTML 或 JSON 数据。这里有个关键选择用无头浏览器Playwright、Selenium还是直接发 HTTP 请求。无头浏览器更接近真实用户行为能拿到动态渲染的内容但资源消耗大、速度慢直接请求快但需要自己处理签名和 Cookie。我的建议是混合策略首次登录和获取 Cookie 用无头浏览器后续轮询用轻量 HTTP 请求复用会话。这样单次轮询可以压到 1-2 秒一台 2 核 4G 的云服务器能同时跑几十个关键词任务。import requests import time import random from urllib.parse import quote class XianyuMonitor: def __init__(self, cookie: str): self.session requests.Session() self.session.headers.update({ User-Agent: Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.0 Mobile/15E148 Safari/604.1, Cookie: cookie, Referer: https://www.goofish.com/, }) self.seen_ids set() # 已处理商品 ID避免重复推送 def search(self, keyword: str, price_max: int None, page: int 1): 按关键词搜索返回商品列表 params { q: keyword, sort: newOnSale, # 按最新发布排序捡漏关键 page: page, } if price_max: params[priceMax] price_max # 实际接口路径以当前页面为准这里用搜索页做示例 url fhttps://www.goofish.com/search?{ .join(f{k}{quote(str(v))} for k,v in params.items()) } resp self.session.get(url, timeout10) resp.raise_for_status() return self._parse(resp.text) def _parse(self, html: str): 解析搜索结果提取商品 ID、标题、价格、链接 # 实际解析依赖页面结构建议用正则或 lxml 提取 # 这里返回模拟结构真实场景需根据页面调整 import re items [] # 示例匹配商品卡片中的关键字段 pattern re.compile(ritemId:(\d).*?title:(.*?).*?price:([\d.])) for m in pattern.finditer(html): item_id, title, price m.group(1), m.group(2), float(m.group(3)) if item_id not in self.seen_ids: items.append({id: item_id, title: title, price: price}) return items def run_once(self, keywords: list, interval: tuple (3, 8)): 轮询一轮所有关键词 for kw in keywords: try: items self.search(kw) for item in items: self.seen_ids.add(item[id]) # 交给分析层处理 yield item except Exception as e: print(f[采集失败] {kw}: {e}) time.sleep(random.uniform(*interval)) # 随机间隔降低被封风险这段代码的核心逻辑是用持久化 Session 复用 Cookie按newOnSale排序只拿最新商品用seen_ids集合去重。参数上interval控制请求间隔建议设在 3-8 秒之间随机浮动固定间隔反而容易被识别。price_max用来做粗筛把明显超出预算的商品在采集阶段就过滤掉减少后续分析压力。2.2 频率控制与账号安全的三条红线采集层最容易翻车的地方不是代码写错而是触发风控。血泪经验总结下来有三条红线不能碰第一单账号请求频率不要超过每分钟 20 次超过这个量级很容易被限流第二不要用同一个 IP 高频请求家用宽带和云服务器 IP 段的风控策略不同云服务器更容易被标记第三Cookie 过期后不要反复重试登录接口应该停下来人工介入更新。实际部署时我一般会准备 2-3 个账号轮换每个账号负责不同关键词组单账号日请求量控制在 5000 次以内。如果只是个人用一个账号监控 5-10 个关键词完全够用没必要上多账号。另外请求头里的User-Agent要跟 Cookie 来源设备一致用 iPhone 的 Cookie 就配 iPhone 的 UA混用会触发异常检测。提示Cookie 有效期通常几天到几周不等建议在系统里加一个失效检测连续返回登录页就发通知提醒更新。3. 分析层怎么做从关键词匹配到多模态判断值不值得推3.1 规则过滤价格、成色、卖家信用三要素采集回来的商品不能直接推给你否则一天几百条消息跟没监控一样。分析层第一道关卡是规则过滤用硬性条件把明显不合适的筛掉。核心看三个维度价格是否低于你设定的阈值、商品描述里有没有关键排除词比如「维修」「配件」「尸体」、卖家信用分和历史评价是否达标。class ItemAnalyzer: def __init__(self, config: dict): self.price_threshold config.get(price_threshold, {}) self.exclude_words config.get(exclude_words, [维修, 配件, 尸体, 缺件]) self.min_seller_score config.get(min_seller_score, 4.8) def rule_filter(self, item: dict, keyword: str) - bool: 返回 True 表示通过规则过滤 # 价格检查 max_price self.price_threshold.get(keyword) if max_price and item[price] max_price: return False # 排除词检查 title item.get(title, ) if any(w in title for w in self.exclude_words): return False # 卖家信用需在采集时补充 seller_score 字段 if item.get(seller_score, 5.0) self.min_seller_score: return False return True参数说明price_threshold按关键词分别设置比如「索尼 A7M3」设 8000「适马 35 1.4」设 2500不同品类价格差异大不能一刀切。exclude_words根据品类调整卖相机镜头和卖手机要排除的词完全不同。min_seller_score建议不低于 4.7低于这个分数的卖家纠纷率明显上升。3.2 多模态分析用视觉模型判断商品真实成色规则过滤只能处理结构化信息但闲鱼商品的核心价值往往藏在图片里。同样是「95 新」的描述有的镜头前组有明显划痕有的只是轻微使用痕迹。这时候可以接一个视觉分析模块用多模态模型对商品主图做初步判断。常见做法是调用支持图像理解的模型接口传入商品主图和一段提示词让模型输出成色评估和风险点。比如提示词可以写「这是一张二手相机镜头的商品图请判断镜片是否有划痕、霉斑、进灰镜身是否有明显磕碰用一句话总结成色等级优秀/良好/一般/较差。」模型返回结果后再跟卖家描述做交叉验证如果描述说「无划痕」但模型判断有划痕这条商品就标记为高风险不推送或者推送时附带警告。def analyze_image(image_url: str, model_client) - dict: 调用多模态模型分析商品图片 prompt ( 这是一张二手商品图片请判断\n 1. 商品外观是否有明显损伤划痕、磕碰、变形\n 2. 功能相关部件是否完整\n 3. 用一句话给出成色等级优秀/良好/一般/较差\n 输出格式{condition: 良好, issues: [轻微划痕], summary: ...} ) resp model_client.chat_with_image(image_url, prompt) # 解析模型返回实际需处理 JSON 解析异常 import json try: return json.loads(resp) except json.JSONDecodeError: return {condition: 未知, issues: [], summary: resp[:100]}这个环节的坑在于模型调用有延迟和成本不是每条商品都值得跑。我的策略是先用规则过滤筛掉 80% 的垃圾商品剩下的 20% 里再挑价格低于阈值 90% 的「高潜力商品」做视觉分析。这样每天实际调用量控制在几十次成本可忽略但能有效避免买到「图片看着好、实物翻车」的商品。3.3 推送决策什么条件下才值得打扰你分析层的最终输出是一个决策推还是不推。我的经验是设置分级推送策略。第一级是「强推」价格低于历史均价 70%、卖家信用 4.9 以上、图片分析无异常这种直接发即时通知。第二级是「弱推」价格在 70%-85% 之间或者有一项指标不完美这种汇总成每小时一次的摘要推送。第三级是「不推」价格没优势或者风险项太多只记录到数据库供后续复盘。这样做的目的是控制信息噪音。监控系统最大的失败不是漏掉商品而是推送太多导致你开始忽略通知。一旦你对通知麻木了这套系统就白搭了。4. 通知层与任务调度怎么让消息在 10 秒内到达手机4.1 通知渠道选型别只盯着一个通道通知层的关键指标是到达速度和可靠性。常见方案有几种企业微信机器人、飞书机器人、钉钉机器人、Telegram Bot、邮件、短信。个人用最推荐企业微信或飞书机器人配置简单、到达快、支持 Markdown 格式手机上装个 App 就能收推送。如果对速度有极致要求可以同时接两个通道做冗余。比如主通道用飞书备用通道用邮件当飞书接口连续失败三次时自动切换。实际测试下来飞书机器人从触发到手机弹窗通常在 2-5 秒企业微信类似邮件则要看邮件服务商的推送策略可能延迟几十秒。import requests class Notifier: def __init__(self, webhook_url: str): self.webhook_url webhook_url def send(self, title: str, content: str, level: str strong): 发送通知level 控制消息样式 color red if level strong else blue payload { msg_type: interactive, card: { header: { title: {tag: plain_text, content: title}, template: color, }, elements: [ {tag: div, text: {tag: lark_md, content: content}}, ], }, } resp requests.post(self.webhook_url, jsonpayload, timeout5) if resp.status_code ! 200: print(f[通知失败] {resp.status_code}: {resp.text}) return resp.status_code 200参数说明webhook_url从机器人配置页获取注意不要泄露到公开仓库。level控制卡片颜色强推用红色、弱推用蓝色视觉上区分优先级。消息内容里建议带上商品标题、价格、卖家信用、直达链接让你不用打开 App 就能判断要不要点进去。4.2 任务调度用 APScheduler 还是自己写循环调度层决定采集任务什么时候跑、跑多快。简单场景用while True加time.sleep就够了但如果你要管理多个关键词、不同时间段的频率策略比如白天频率高、凌晨频率低建议用 APScheduler 做任务管理。from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.interval import IntervalTrigger scheduler BlockingScheduler() # 高频关键词每 30 秒跑一次 scheduler.add_job( monitor_task, IntervalTrigger(seconds30), args[索尼A7M3], idsony_a7m3, max_instances1, # 防止任务重叠 ) # 低频关键词每 5 分钟跑一次 scheduler.add_job( monitor_task, IntervalTrigger(minutes5), args[适马35art], idsigma_35, max_instances1, ) scheduler.start()关键参数是max_instances1防止上一次任务还没跑完下一次就启动了导致请求堆积触发风控。另外建议给每个任务加独立的异常捕获单个关键词失败不影响其他任务。5. 避坑与排查那些让我半夜爬起来修的问题5.1 现象Cookie 突然失效所有任务返回登录页原因闲鱼 Cookie 有效期不固定可能几天也可能几周触发风控时会提前失效。另外如果在同一账号上频繁切换 IP也会导致 Cookie 被强制下线。解决在采集层加一个响应检测如果返回内容里包含登录页特征比如「请登录」「扫码登录」立即停止该账号的所有任务并发送告警通知。同时准备备用账号人工介入更新 Cookie 后恢复。不要写自动登录逻辑验证码环节过不去反而增加封号风险。5.2 现象商品重复推送同一条消息收到好几次原因seen_ids集合只存在内存里程序重启后丢失导致已处理商品被重新推送。另外如果多个关键词命中同一商品也会重复推送。解决把seen_ids持久化到 SQLite 或 Redis设置合理的过期时间比如 7 天。对于多关键词命中在推送前做一次全局去重同一个商品 ID 只推一次取最高优先级的推送级别。import sqlite3 class SeenStore: def __init__(self, db_path: str seen.db): self.conn sqlite3.connect(db_path) self.conn.execute( CREATE TABLE IF NOT EXISTS seen (item_id TEXT PRIMARY KEY, ts INTEGER) ) def is_new(self, item_id: str) - bool: cur self.conn.execute(SELECT 1 FROM seen WHERE item_id?, (item_id,)) if cur.fetchone(): return False self.conn.execute(INSERT INTO seen VALUES (?, strftime(%s,now)), (item_id,)) self.conn.commit() return True def cleanup(self, days: int 7): self.conn.execute(DELETE FROM seen WHERE ts strftime(%s,now) - ?, (days*86400,)) self.conn.commit()5.3 现象请求返回 200 但解析不到任何商品原因页面结构变了或者返回的是验证码页面但状态码仍是 200。闲鱼前端改版频率不低解析规则需要定期维护。解决在解析函数里加一个「空结果检测」如果连续多次解析到 0 条商品但请求状态码正常就触发告警。同时把原始 HTML 存一份到本地方便排查是结构变化还是被反爬拦截。建议每周抽一次手动验证确认解析规则还有效。5.4 现象云服务器上跑几天后 IP 被限流请求全部超时原因云服务器 IP 段被闲鱼重点关照高频请求容易触发 IP 级别的限流。另外部分云厂商的 NAT 网关会导致多个用户共享出口 IP进一步增加被标记概率。解决降低单 IP 请求频率把间隔拉大到 10 秒以上。如果关键词多考虑分散到不同网络环境比如家里宽带跑一部分、云服务器跑一部分。不要用公开代理池质量不可控且可能引入更严重的风控。5.5 现象多模态分析接口偶尔超时导致整条流水线卡住原因外部模型接口有网络延迟和限流同步调用会阻塞主流程。解决把视觉分析改成异步任务采集和规则过滤走主流程通过规则的商品丢进队列由独立的 worker 消费队列做视觉分析分析完再决定推送级别。这样即使模型接口慢也不影响采集层的实时性。6. 进阶技巧用历史数据反推合理价格阈值监控系统跑了一周之后你会积累一批商品数据。这些数据不只是用来捡漏的还能反过来帮你优化阈值。具体做法是把每个关键词的历史成交价或已售出商品标价存下来算一个滚动 7 天的中位数和四分位距。当新商品价格低于中位数的 70% 时判定为「好价」低于 50% 时判定为「漏价」直接强推。import statistics def update_price_stats(db, keyword: str): 计算关键词的历史价格统计 rows db.execute( SELECT price FROM items WHERE keyword? AND ts strftime(%s,now) - 7*86400, (keyword,) ).fetchall() prices [r[0] for r in rows if r[0] 0] if len(prices) 10: return None # 样本太少不更新 return { median: statistics.median(prices), p25: statistics.quantiles(prices, n4)[0], p75: statistics.quantiles(prices, n4)[2], count: len(prices), }这个统计结果可以动态调整推送阈值而不是用固定数字。比如某款镜头平时成交价在 5000 左右突然出现一条 3200 的系统自动判定为漏价并强推。反过来如果某天大量商品集中上架导致价格整体下移中位数也会跟着变避免误判。另一个技巧是给卖家建画像。记录每个卖家的历史发布商品、成交速度、评价变化当某个卖家频繁发布低价商品且成交极快时可能是专业卖家在批量出货这类卖家的商品值得重点关注。反过来如果某个卖家历史评价里有「描述不符」的记录即使价格低也降级处理。我自己的习惯是每周花 10 分钟看一下推送记录和实际成交的对比看看哪些推对了、哪些漏了、哪些推了但其实是坑。这套系统不是搭完就一劳永逸的闲鱼在变、卖家在变、你的需求也在变定期校准才能保持好用。希望帮到你。本文还有配套的精品资源点击获取 
长河君 · 2026/10/10 13:43:07 · 阅读 1.2k 闲鱼智能监控机器人：从采集到通知的完整架构与避坑指南
简介这是一套面向二手交易场景的闲鱼多任务实时监控与智能分析工具基于 Playwright 与多模态大语言模型构建适合希望自动追踪商品、按需筛选好价的开发者与进阶用户。系统提供完整的 Web 管理界面支持任务可视化配置、AI 标准在线编辑、运行日志实时查看与结果筛选无需直接操作命令行用户只需用自然语言描述购买需求即可一键生成包含复杂筛选逻辑的监控任务并借助 GPT-4o 结合商品图文与卖家画像做深度分析再通过 ntfy.sh、企业微信机器人或 Bark 即时推送。资源包共 37 个文件约 12.31MB以 Python 源码为主体辅以 HTML、JS、CSS 前端页面、YAML 与 Dockerfile 部署配置、JSON 示例及说明文档覆盖爬虫、AI 处理、提示词生成与定时调度等模块。目前已有 442 人学习可帮助读者快速搭建可定制、可容器化部署的监控分析系统。1. 闲鱼智能监控机器人到底在监控什么从一条低价漏网商品说起你盯了三天的那款二手镜头标价 1800你觉得贵没下手。第四天早上刷到一条 950 的点进去已经显示「已售出」卖家说凌晨两点就被拍走了。这不是运气问题是信息差问题——闲鱼上真正的好价商品从发布到成交往往只有几分钟窗口靠人肉刷新根本抢不过脚本。闲鱼智能监控机器人要解决的就是这件事用程序代替你 7×24 小时盯着指定关键词的新发布商品自动过滤、自动分析、自动通知让你在别人还在睡觉的时候就知道有货上架了。这套「闲鱼任务监控分析系统」的核心不是爬虫本身而是三层结构采集层负责按关键词和筛选条件拉取新商品分析层负责判断这条商品值不值得推给你通知层负责在最短时间内把消息送到你手机上。适合谁用想捡漏的二手玩家、做无货源选品的卖家、需要监控竞品定价的运营以及任何愿意花一个下午把环境搭起来、之后长期受益的人。下面从架构选型一路讲到参数调优和踩坑记录能抄的代码我都贴出来。2. 采集层怎么搭关键词轮询、频率控制与反爬边界2.1 为什么不用官方 API 而走页面轮询闲鱼没有开放的商品搜索 API 给个人开发者市面上能用的方案基本是模拟搜索请求或解析搜索结果页。常见做法是构造搜索 URL带上关键词、排序方式、价格区间等参数然后解析返回的 HTML 或 JSON 数据。这里有个关键选择用无头浏览器Playwright、Selenium还是直接发 HTTP 请求。无头浏览器更接近真实用户行为能拿到动态渲染的内容但资源消耗大、速度慢直接请求快但需要自己处理签名和 Cookie。我的建议是混合策略首次登录和获取 Cookie 用无头浏览器后续轮询用轻量 HTTP 请求复用会话。这样单次轮询可以压到 1-2 秒一台 2 核 4G 的云服务器能同时跑几十个关键词任务。import requests import time import random from urllib.parse import quote class XianyuMonitor: def __init__(self, cookie: str): self.session requests.Session() self.session.headers.update({ User-Agent: Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.0 Mobile/15E148 Safari/604.1, Cookie: cookie, Referer: https://www.goofish.com/, }) self.seen_ids set() # 已处理商品 ID避免重复推送 def search(self, keyword: str, price_max: int None, page: int 1): 按关键词搜索返回商品列表 params { q: keyword, sort: newOnSale, # 按最新发布排序捡漏关键 page: page, } if price_max: params[priceMax] price_max # 实际接口路径以当前页面为准这里用搜索页做示例 url fhttps://www.goofish.com/search?{ .join(f{k}{quote(str(v))} for k,v in params.items()) } resp self.session.get(url, timeout10) resp.raise_for_status() return self._parse(resp.text) def _parse(self, html: str): 解析搜索结果提取商品 ID、标题、价格、链接 # 实际解析依赖页面结构建议用正则或 lxml 提取 # 这里返回模拟结构真实场景需根据页面调整 import re items [] # 示例匹配商品卡片中的关键字段 pattern re.compile(ritemId:(\d).*?title:(.*?).*?price:([\d.])) for m in pattern.finditer(html): item_id, title, price m.group(1), m.group(2), float(m.group(3)) if item_id not in self.seen_ids: items.append({id: item_id, title: title, price: price}) return items def run_once(self, keywords: list, interval: tuple (3, 8)): 轮询一轮所有关键词 for kw in keywords: try: items self.search(kw) for item in items: self.seen_ids.add(item[id]) # 交给分析层处理 yield item except Exception as e: print(f[采集失败] {kw}: {e}) time.sleep(random.uniform(*interval)) # 随机间隔降低被封风险这段代码的核心逻辑是用持久化 Session 复用 Cookie按newOnSale排序只拿最新商品用seen_ids集合去重。参数上interval控制请求间隔建议设在 3-8 秒之间随机浮动固定间隔反而容易被识别。price_max用来做粗筛把明显超出预算的商品在采集阶段就过滤掉减少后续分析压力。2.2 频率控制与账号安全的三条红线采集层最容易翻车的地方不是代码写错而是触发风控。血泪经验总结下来有三条红线不能碰第一单账号请求频率不要超过每分钟 20 次超过这个量级很容易被限流第二不要用同一个 IP 高频请求家用宽带和云服务器 IP 段的风控策略不同云服务器更容易被标记第三Cookie 过期后不要反复重试登录接口应该停下来人工介入更新。实际部署时我一般会准备 2-3 个账号轮换每个账号负责不同关键词组单账号日请求量控制在 5000 次以内。如果只是个人用一个账号监控 5-10 个关键词完全够用没必要上多账号。另外请求头里的User-Agent要跟 Cookie 来源设备一致用 iPhone 的 Cookie 就配 iPhone 的 UA混用会触发异常检测。提示Cookie 有效期通常几天到几周不等建议在系统里加一个失效检测连续返回登录页就发通知提醒更新。3. 分析层怎么做从关键词匹配到多模态判断值不值得推3.1 规则过滤价格、成色、卖家信用三要素采集回来的商品不能直接推给你否则一天几百条消息跟没监控一样。分析层第一道关卡是规则过滤用硬性条件把明显不合适的筛掉。核心看三个维度价格是否低于你设定的阈值、商品描述里有没有关键排除词比如「维修」「配件」「尸体」、卖家信用分和历史评价是否达标。class ItemAnalyzer: def __init__(self, config: dict): self.price_threshold config.get(price_threshold, {}) self.exclude_words config.get(exclude_words, [维修, 配件, 尸体, 缺件]) self.min_seller_score config.get(min_seller_score, 4.8) def rule_filter(self, item: dict, keyword: str) - bool: 返回 True 表示通过规则过滤 # 价格检查 max_price self.price_threshold.get(keyword) if max_price and item[price] max_price: return False # 排除词检查 title item.get(title, ) if any(w in title for w in self.exclude_words): return False # 卖家信用需在采集时补充 seller_score 字段 if item.get(seller_score, 5.0) self.min_seller_score: return False return True参数说明price_threshold按关键词分别设置比如「索尼 A7M3」设 8000「适马 35 1.4」设 2500不同品类价格差异大不能一刀切。exclude_words根据品类调整卖相机镜头和卖手机要排除的词完全不同。min_seller_score建议不低于 4.7低于这个分数的卖家纠纷率明显上升。3.2 多模态分析用视觉模型判断商品真实成色规则过滤只能处理结构化信息但闲鱼商品的核心价值往往藏在图片里。同样是「95 新」的描述有的镜头前组有明显划痕有的只是轻微使用痕迹。这时候可以接一个视觉分析模块用多模态模型对商品主图做初步判断。常见做法是调用支持图像理解的模型接口传入商品主图和一段提示词让模型输出成色评估和风险点。比如提示词可以写「这是一张二手相机镜头的商品图请判断镜片是否有划痕、霉斑、进灰镜身是否有明显磕碰用一句话总结成色等级优秀/良好/一般/较差。」模型返回结果后再跟卖家描述做交叉验证如果描述说「无划痕」但模型判断有划痕这条商品就标记为高风险不推送或者推送时附带警告。def analyze_image(image_url: str, model_client) - dict: 调用多模态模型分析商品图片 prompt ( 这是一张二手商品图片请判断\n 1. 商品外观是否有明显损伤划痕、磕碰、变形\n 2. 功能相关部件是否完整\n 3. 用一句话给出成色等级优秀/良好/一般/较差\n 输出格式{condition: 良好, issues: [轻微划痕], summary: ...} ) resp model_client.chat_with_image(image_url, prompt) # 解析模型返回实际需处理 JSON 解析异常 import json try: return json.loads(resp) except json.JSONDecodeError: return {condition: 未知, issues: [], summary: resp[:100]}这个环节的坑在于模型调用有延迟和成本不是每条商品都值得跑。我的策略是先用规则过滤筛掉 80% 的垃圾商品剩下的 20% 里再挑价格低于阈值 90% 的「高潜力商品」做视觉分析。这样每天实际调用量控制在几十次成本可忽略但能有效避免买到「图片看着好、实物翻车」的商品。3.3 推送决策什么条件下才值得打扰你分析层的最终输出是一个决策推还是不推。我的经验是设置分级推送策略。第一级是「强推」价格低于历史均价 70%、卖家信用 4.9 以上、图片分析无异常这种直接发即时通知。第二级是「弱推」价格在 70%-85% 之间或者有一项指标不完美这种汇总成每小时一次的摘要推送。第三级是「不推」价格没优势或者风险项太多只记录到数据库供后续复盘。这样做的目的是控制信息噪音。监控系统最大的失败不是漏掉商品而是推送太多导致你开始忽略通知。一旦你对通知麻木了这套系统就白搭了。4. 通知层与任务调度怎么让消息在 10 秒内到达手机4.1 通知渠道选型别只盯着一个通道通知层的关键指标是到达速度和可靠性。常见方案有几种企业微信机器人、飞书机器人、钉钉机器人、Telegram Bot、邮件、短信。个人用最推荐企业微信或飞书机器人配置简单、到达快、支持 Markdown 格式手机上装个 App 就能收推送。如果对速度有极致要求可以同时接两个通道做冗余。比如主通道用飞书备用通道用邮件当飞书接口连续失败三次时自动切换。实际测试下来飞书机器人从触发到手机弹窗通常在 2-5 秒企业微信类似邮件则要看邮件服务商的推送策略可能延迟几十秒。import requests class Notifier: def __init__(self, webhook_url: str): self.webhook_url webhook_url def send(self, title: str, content: str, level: str strong): 发送通知level 控制消息样式 color red if level strong else blue payload { msg_type: interactive, card: { header: { title: {tag: plain_text, content: title}, template: color, }, elements: [ {tag: div, text: {tag: lark_md, content: content}}, ], }, } resp requests.post(self.webhook_url, jsonpayload, timeout5) if resp.status_code ! 200: print(f[通知失败] {resp.status_code}: {resp.text}) return resp.status_code 200参数说明webhook_url从机器人配置页获取注意不要泄露到公开仓库。level控制卡片颜色强推用红色、弱推用蓝色视觉上区分优先级。消息内容里建议带上商品标题、价格、卖家信用、直达链接让你不用打开 App 就能判断要不要点进去。4.2 任务调度用 APScheduler 还是自己写循环调度层决定采集任务什么时候跑、跑多快。简单场景用while True加time.sleep就够了但如果你要管理多个关键词、不同时间段的频率策略比如白天频率高、凌晨频率低建议用 APScheduler 做任务管理。from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.interval import IntervalTrigger scheduler BlockingScheduler() # 高频关键词每 30 秒跑一次 scheduler.add_job( monitor_task, IntervalTrigger(seconds30), args[索尼A7M3], idsony_a7m3, max_instances1, # 防止任务重叠 ) # 低频关键词每 5 分钟跑一次 scheduler.add_job( monitor_task, IntervalTrigger(minutes5), args[适马35art], idsigma_35, max_instances1, ) scheduler.start()关键参数是max_instances1防止上一次任务还没跑完下一次就启动了导致请求堆积触发风控。另外建议给每个任务加独立的异常捕获单个关键词失败不影响其他任务。5. 避坑与排查那些让我半夜爬起来修的问题5.1 现象Cookie 突然失效所有任务返回登录页原因闲鱼 Cookie 有效期不固定可能几天也可能几周触发风控时会提前失效。另外如果在同一账号上频繁切换 IP也会导致 Cookie 被强制下线。解决在采集层加一个响应检测如果返回内容里包含登录页特征比如「请登录」「扫码登录」立即停止该账号的所有任务并发送告警通知。同时准备备用账号人工介入更新 Cookie 后恢复。不要写自动登录逻辑验证码环节过不去反而增加封号风险。5.2 现象商品重复推送同一条消息收到好几次原因seen_ids集合只存在内存里程序重启后丢失导致已处理商品被重新推送。另外如果多个关键词命中同一商品也会重复推送。解决把seen_ids持久化到 SQLite 或 Redis设置合理的过期时间比如 7 天。对于多关键词命中在推送前做一次全局去重同一个商品 ID 只推一次取最高优先级的推送级别。import sqlite3 class SeenStore: def __init__(self, db_path: str seen.db): self.conn sqlite3.connect(db_path) self.conn.execute( CREATE TABLE IF NOT EXISTS seen (item_id TEXT PRIMARY KEY, ts INTEGER) ) def is_new(self, item_id: str) - bool: cur self.conn.execute(SELECT 1 FROM seen WHERE item_id?, (item_id,)) if cur.fetchone(): return False self.conn.execute(INSERT INTO seen VALUES (?, strftime(%s,now)), (item_id,)) self.conn.commit() return True def cleanup(self, days: int 7): self.conn.execute(DELETE FROM seen WHERE ts strftime(%s,now) - ?, (days*86400,)) self.conn.commit()5.3 现象请求返回 200 但解析不到任何商品原因页面结构变了或者返回的是验证码页面但状态码仍是 200。闲鱼前端改版频率不低解析规则需要定期维护。解决在解析函数里加一个「空结果检测」如果连续多次解析到 0 条商品但请求状态码正常就触发告警。同时把原始 HTML 存一份到本地方便排查是结构变化还是被反爬拦截。建议每周抽一次手动验证确认解析规则还有效。5.4 现象云服务器上跑几天后 IP 被限流请求全部超时原因云服务器 IP 段被闲鱼重点关照高频请求容易触发 IP 级别的限流。另外部分云厂商的 NAT 网关会导致多个用户共享出口 IP进一步增加被标记概率。解决降低单 IP 请求频率把间隔拉大到 10 秒以上。如果关键词多考虑分散到不同网络环境比如家里宽带跑一部分、云服务器跑一部分。不要用公开代理池质量不可控且可能引入更严重的风控。5.5 现象多模态分析接口偶尔超时导致整条流水线卡住原因外部模型接口有网络延迟和限流同步调用会阻塞主流程。解决把视觉分析改成异步任务采集和规则过滤走主流程通过规则的商品丢进队列由独立的 worker 消费队列做视觉分析分析完再决定推送级别。这样即使模型接口慢也不影响采集层的实时性。6. 进阶技巧用历史数据反推合理价格阈值监控系统跑了一周之后你会积累一批商品数据。这些数据不只是用来捡漏的还能反过来帮你优化阈值。具体做法是把每个关键词的历史成交价或已售出商品标价存下来算一个滚动 7 天的中位数和四分位距。当新商品价格低于中位数的 70% 时判定为「好价」低于 50% 时判定为「漏价」直接强推。import statistics def update_price_stats(db, keyword: str): 计算关键词的历史价格统计 rows db.execute( SELECT price FROM items WHERE keyword? AND ts strftime(%s,now) - 7*86400, (keyword,) ).fetchall() prices [r[0] for r in rows if r[0] 0] if len(prices) 10: return None # 样本太少不更新 return { median: statistics.median(prices), p25: statistics.quantiles(prices, n4)[0], p75: statistics.quantiles(prices, n4)[2], count: len(prices), }这个统计结果可以动态调整推送阈值而不是用固定数字。比如某款镜头平时成交价在 5000 左右突然出现一条 3200 的系统自动判定为漏价并强推。反过来如果某天大量商品集中上架导致价格整体下移中位数也会跟着变避免误判。另一个技巧是给卖家建画像。记录每个卖家的历史发布商品、成交速度、评价变化当某个卖家频繁发布低价商品且成交极快时可能是专业卖家在批量出货这类卖家的商品值得重点关注。反过来如果某个卖家历史评价里有「描述不符」的记录即使价格低也降级处理。我自己的习惯是每周花 10 分钟看一下推送记录和实际成交的对比看看哪些推对了、哪些漏了、哪些推了但其实是坑。这套系统不是搭完就一劳永逸的闲鱼在变、卖家在变、你的需求也在变定期校准才能保持好用。希望帮到你。本文还有配套的精品资源点击获取