简介这是一个基于Python开发的微信订阅机器人项目面向闲鱼二手商品监控场景适合希望学习微信接口自动化、爬虫与任务队列整合的Python开发者对于搭建私域订阅提醒或无人值守监控工具尤为实用。项目以Celery为调度核心完整搭建了四层功能模块爬虫端实时抓取闲鱼商品并写入MySQL交互式微信机器人可自动通过好友申请并将订阅需求存入MongoDB微信机器人Web接口为推送提供API推送端轮询商品数据库按MongoDB中的订阅分类请求接口将匹配结果推送至订阅好友。整体架构清晰从启动Web服务到启动Celery任务一次性联动爬虫、机器人、数据更新与推送四类任务目录按爬虫、机器人、接口、任务拆分Python脚本职责单一便于阅读与二次开发具备可运行、易扩展的实战参考价值。资源共78个文件其中75个为Python脚本另附README说明、License与gitignore配置压缩包仅99KB小巧但模块完整。目前已有504人学习下载适合用于课设、工具开发或二次改造。1. 闲鱼监控 微信推送这个微信订阅机器人到底帮你盯什么先说结论这不是一个「爬虫全家桶」而是一个结构很克制的 Python 微信订阅机器人核心场景就一句话——把闲鱼上的关键词搜索、新商品发布、价格变动变成推送到你微信里的订阅消息。你可能在蹲一块显卡、一台微单或者某个绝版配件手动刷闲鱼一天能刷几十次而这套机器人替你按固定间隔轮询命中关键词就把商品标题、价格、链接推过来。适合谁用两类人最合适。一类是二手捡漏党盯着特定型号、特定价格区间机器比人眼刷得勤另一类是倒卖或比价用户需要同时盯多个关键词、多个卖家手动切账号太累。整个项目用 Python 实现核心依赖是 requests 和定时调度不需要装浏览器内核也不用搞什么无头浏览器这对 Windows 和 Linux 服务器都友好。下文把资源包结构、监控逻辑、微信推送通道、部署踩坑一次讲透照着改就能用起来。2. 先看资源包结构目录划分与三个核心模块2.1 zip 里的文件都有什么拿到 zip 解压后目录大致分成源码、配置、依赖和说明四块。建议先按文件清单核对一遍缺了任何一个配置文件启动时都会翻车。文件/目录作用关键点main.py程序入口调度循环和异常兜底都在这monitor/闲鱼监控模块搜索接口封装、响应解析、关键词过滤pusher/微信推送模块pushplus / Server酱 通道封装config.yaml全部可调参数关键词、价格区间、轮询间隔、推送 tokenrequirements.txtPython 依赖清单requests、pyyaml、apscheduler 等README.md部署步骤环境要求、启动命令、常见报错这个拆分思路值得直接沿用监控、推送、配置三者分离。实际部署时你大概率要改的是config.yaml最多动monitor/里的匹配规则pusher/基本不用碰。2.2 监控模块搜索、解析、过滤三步拆开很多初学写法是把「发请求 → 解析 → 筛选 → 推送」全塞到一个函数里前两次能跑跑几天就崩——因为一旦某个环节报错整个循环就死了。这个资源包里把监控拆成三个独立步骤分别对应monitor/下的search.py、parser.py、filter.py。search.py只负责按关键词请求闲鱼搜索接口返回原始 JSON 文本parser.py从 JSON 里剥离出标题、价格、商品 ID、链接、发布时间、卖家昵称filter.py做关键词二次过滤和价格区间校验输出最终命中的商品列表。这样的好处是接口字段变了你只改parser.py过滤规则变了只改filter.py不用牵一发动全身。2.3 推送通道为什么默认走 pushplus 而不是微信个人号微信订阅机器人最容易卡住的环节就是把消息送进微信。直接用微信个人号接口属于灰产操作账号风险极高这个资源默认用的是 pushplus 的微信推送 API——扫码关注公众号后生成一个 token通过 HTTP POST 就能把消息推到你微信上。POST http://www.pushplus.plus/send这种方式没有封号风险token 就是你的身份标识丢给config.yaml里的push_token字段即可。如果你更习惯 Server酱把pusher/里的发送函数改一下 URL 和参数结构就能切换两者都是半小时内能跑通的方案。3. 把闲鱼关键词监控搭起来请求、解析与关键词匹配3.1 闲鱼搜索接口参数怎么填闲鱼的搜索接口走的是m.goofish.com的移动端 API返回 JSON 格式数据比渲染完再解析 HTML 稳得多。核心请求参数如下参数取值说明必填q搜索关键词需要 URL 编码是page页码从 1 开始是t时间戳防缓存建议填n每页条数一般 1020建议填具体请求代码写法如下import requests import time from urllib.parse import quote def search_goofish(keyword: str, page: int 1) - str: base_url https://m.goofish.com/api/h5/mtop.taobao.idle.search.api/1.0/ params { q: quote(keyword), page: page, t: int(time.time() * 1000), n: 20, } headers { User-Agent: Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X), Referer: https://m.goofish.com/, } resp requests.get(base_url, paramsparams, headersheaders, timeout5) resp.raise_for_status() return resp.text这段代码的关键在于quote(keyword)——关键词里有空格或中文时不编码会直接 400。t参数用毫秒级时间戳是为了避免闲鱼缓存旧结果实测加上之后新商品出现的延迟明显短一些。请求头里的User-Agent必须模拟手机浏览器闲鱼对 PC 端 UA 的请求限制更严这一步没做对后面全是滑块。3.2 解析响应从嵌套 JSON 里抠出商品字段闲鱼搜索接口返回的是一个嵌套很深的 JSON商品列表藏在data下的items数组里。直接下标索引很容易因为字段缺失报 KeyError所以解析函数要加.get()兜底。下面是核心解析逻辑import json def parse_items(raw_text: str) - list: data json.loads(raw_text) items data.get(data, {}).get(items, []) result [] for item in items: title item.get(title, ).strip() price item.get(price, 0) item_id item.get(itemId) or item.get(item_id) publish_time item.get(pubTime) or item.get(publishTime) seller item.get(seller, {}).get(nick, 未知卖家) link fhttps://m.goofish.com/item?id{item_id} result.append({ title: title, price: float(price) if price else 0.0, item_id: item_id, publish_time: publish_time, seller: seller, link: link, }) return result注意price的来源可能是字符串也可能是数字而且闲鱼有些商品是「面议」价格接口这时返回的可能是空串最好统一转成 float 并默认 0方便后面做价格区间过滤。itemId的字段名在不同端有细微差异老接口叫item_id新接口叫itemId所以这里用or做了双保险。链接拼接时id参数是必需的有些人只拿到标题没拿到 ID推送出来一堆无链接消息问题就出在这。3.3 关键词二次过滤全局匹配不如「标题 卖家」双维度很多新手以为搜索接口返回的商品一定都带关键词实际并非如此。闲鱼的模糊搜索会把「相近品类」也带进来比如你搜「索尼 A7M3」出来一堆「A7R3」「A7C」的配件。所以拿到parse_items的结果后还得再做一层本地过滤def filter_items(items: list, include_keywords: list, exclude_keywords: list, max_price: float) - list: matched [] for item in items: text item[title] item[seller] if exclude_keywords and any(k in text for k in exclude_keywords): continue if not all(k in text for k in include_keywords): continue if 0 max_price item[price]: continue matched.append(item) return matched这里include_keywords与exclude_keywords都是列表用any做排除、all做包含语义很直观。把卖家昵称拼进匹配文本是为了支持「指定卖家监控」——在include_keywords里填卖家昵称的一部分等于只看那个人的上新这个功能在蹲个人卖家时尤其好用。max_price默认 0 表示不限制这个判断逻辑是「先转 float 再比较」如果你直接在配置里写字符串会导致每次比较都报 TypeError。4. 对接微信订阅推送模板消息、轮询与去重4.1 推送通道配置pushplus token 和模板消息结构微信订阅机器人的最后一公里是 pushplus 推送。在config.yaml里要配三个字段push_token、pub_topic、template。push_token是从 pushplus 公众号绑定后拿到的令牌pub_topic是群组编号不填则推给个人template一般用html内容支持简单的标题加链接排版。配置示例pushplus: token: 你的token topic: template: html推送函数封装在pusher/里发送逻辑是一个标准的 POST 请求import requests def send_wechat(token: str, title: str, content: str, template: str html) - bool: url http://www.pushplus.plus/send payload { token: token, title: title, content: content, template: template, } resp requests.post(url, jsonpayload, timeout5) result resp.json() if result.get(code) ! 200: print(f推送失败: {result.get(msg)}) return False return True这里判断成功别只看 HTTP 状态码pushplus 接口响应体里有个code字段200 才是成功。很多人用resp.status_code判断返回 200 但实际推送失败就是因为code不是 200。如果你有多个监控关键词建议一个消息里打包多条商品而不是一条商品推一次——否则微信会刷屏而且 pushplus 免费版每天有推送条数上限打包发能省额度。4.2 轮询间隔与调度太短封号太长漏商品轮询间隔是一个典型的玄学参数。太短闲鱼的反爬策略分分钟给你上滑块验证太长热门关键词下的秒拍商品根本来不及推送。这个资源的默认配置是 60 秒一轮实测在单个关键词下比较安全。调度的核心循环可参考下面这段伪代码框架import time from monitor import search_goofish, parse_items, filter_items from pusher import send_wechat def run_once(config): keyword config[monitor][keyword] raw search_goofish(keyword) items parse_items(raw) hits filter_items(items, config[monitor][include], config[monitor][exclude], config[monitor][max_price]) if hits: send_wechat(config[pushplus][token], f闲鱼新上 {len(hits)} 件, format_content(hits)) return len(hits) while True: try: run_once(config) except Exception as e: print(f本轮异常: {e}) time.sleep(config[monitor][interval_seconds])while True加sleep是最直观的写法适合小规模监控如果你要同时盯几十个关键词建议换成 APScheduler 的BackgroundScheduler因为单个循环无法并发请求多个关键词一个关键词请求慢了会拖累后面所有关键词。异常兜底必须放在try-except里否则某次网络抖动导致请求超时整个程序就退出——这是部署后跑一段时间挂掉的头号原因。4.3 去重机制只推新链接不重复轰炸如果不做去重同一件商品每次轮询命中都会推一次一小时内你会被同一条链接轰炸几十遍。常见的做法是「已推送商品 ID 集合 固定窗口」seen_ids set() WINDOW_SIZE 200 def is_new(item_id: str) - bool: if item_id in seen_ids: return False seen_ids.add(item_id) if len(seen_ids) WINDOW_SIZE: seen_ids.pop() return True为什么不用数据库存储而是用一个 Python 集合因为监控商品量级在百级到千级内存集合足够程序重启后会重新学习和推送一次对你「看新商品」的目标影响不大。WINDOW_SIZE设为 200 的意思是保留最近 200 个商品 ID超过之后弹出最早那个——这保证热门商品下架重挂ID 不变时能被重新识别又不会无限膨胀内存。注意闲鱼商品重挂后item_id不变所以这个去重也能过滤掉「同链接改价再上架」的打扰。5. 部署避坑闲鱼反爬、推送丢失与 cookie 失效的常见排查5.1 频繁请求触发滑块验证码现象跑了一段时间后search_goofish返回值不再是正常 JSON而是html...验证码.../html或者直接抛 415 错误。原因请求频率超过闲鱼的阈值。闲鱼对同 IP 短时间高频请求非常敏感尤其是在多个关键词并发轮询时每个关键词都请求一遍单位时间请求数可能翻了几倍。解决把轮询间隔从 60 秒调大到 120180 秒并给每个关键词的请求加随机抖动避免整点齐射。比如time.sleep(interval random.uniform(0, 5))5.2 微信推送偶尔收不到返回 code 200 但无消息现象pushplus 接口返回{code:200}但微信里没收到推送或者延迟十几分钟。原因免费版 pushplus 对单日推送条数有上限超出后接口仍然返回 200实际会被静默丢弃。另外一条消息里 content 字段太长也会被截断。解决在推送函数里每次发送后记录当天累计条数超过 190 条就不再发送并输出日志。同时限制format_content里只保留前 5 条商品信息和链接超出的部分用「等 N 件更多商品」代替。5.3 闲鱼接口提示用户未登录或返回 4200 错误现象部署几天后parse_items解析出的items列表变成空数组抓包发现接口响应体里有subCode: 4200。原因闲鱼移动端接口要求登录态首次可以匿名访问但高频请求后服务端会要求补充 Cookie。没有 Cookie 的程序此时就失效了。解决从手机浏览器登录闲鱼网页版抓包获取Cookie字段填到config.yaml的request_headers中。注意 Cookie 有效期一般 37 天这属于正常运行周期不是 bug。request_headers: Cookie: 你的cookie User-Agent: Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X)5.4 关键词误报搜「A7M3」却推送「A7R3」的配件现象推送列表里出现大量不相关的商品标题里明明没有核心关键词但搜索结果把它带出来了。原因闲鱼搜索接口的推荐逻辑不是严格包含匹配它会把相关品类、同品牌商品混进来。本地filter_items里的include_keywords用all判断时如果只填一个词「A7M3」这个判断实际上拦不住不相关内容因为标题可能写「索尼A73」或包含其他子型号。解决用「双词约束」兜底把机型和必须出现的价格词分开填例如include_keywords: [A7M3, 索尼]。实测多加一个品牌词误报率能降一半以上同时把「A7R3」「A7C」放进exclude_keywords做排除。5.5 程序跑一段时间内存暴涨现象seen_ids集合无限增长运行一周后进程占用内存超过 1GB。原因我以为WINDOW_SIZE控制住了去重集合大小但如果filter_items返回的商品 ID 全部不重复seen_ids.pop()弹出的是任意元素不是最早插入的——Python 的 set 是无序的pop 弹出哪个元素是不可预测的。解决改用collections.deque辅助记录顺序同时保留 set 做查重from collections import deque seen_ids set() seen_order deque() def is_new(item_id: str) - bool: if item_id in seen_ids: return False seen_ids.add(item_id) seen_order.append(item_id) if len(seen_order) WINDOW_SIZE: old seen_order.popleft() seen_ids.discard(old) return True这样弹出的就是最早加入的那个 ID内存占用能稳定在几十 MB 以内。如果你已经用 set 裸跑了好几天修复后重启一次内存会立刻回落。6. 把监控调得更聪明发布时间过滤、白名单与改价收敛6.1 用发布时间戳过滤老链接闲鱼搜索默认按综合排序时稳定货源会一直排在前面每次轮询都命中同一个商品。如果想只看「新发布的」可以在filter_items里加一个min_publish_time参数——把轮询当天的零点时间戳传进去只保留publish_time在当天的商品。蹲秒拍商品时这个过滤能有效减少无意义推送。import datetime def is_today(publish_time_str: str) - bool: if not publish_time_str: return False dt datetime.datetime.strptime(publish_time_str, %Y-%m-%d %H:%M:%S) return dt.date() datetime.date.today()注意闲鱼接口的pubTime有两种格式老版本是2025-06-12 10:30:00新版本可能出现 ISO 格式带T和时区解析前做一次字符串归一化否则strptime会抛 ValueError。6.2 卖家白名单与黑名单如果你要盯的不是某个关键词下所有商品而是特定几个卖家的上新把卖家昵称拆出来单放一组配置。还有一种是黑名单专门屏蔽倒卖商——闲鱼上有些卖家是「上架瞬间改价」的类型你在exclude_keywords里填他们的昵称特征即可。whitelist_sellers: [数码老张, 二手小王子] blacklist_sellers: [批量回收商]过滤逻辑上白名单优先级高于关键词匹配只要命中了白名单卖家就算标题没完全匹配也推送命中黑名单则无条件丢弃。这个配置在config.yaml单独成段改起来不用动代码。6.3 高频改价场景下的重复提醒收敛闲鱼卖家经常会「微降价」刺激下单比如从 3500 改到 3450。如果你只按商品 ID 去重改价后的同 ID 商品被判定为重复不推送这其实错过了低价机会。但又不能每次都推否则就是骚扰。折中方案是「价格变化超过阈值才提醒」price_cache {} def should_notify_price(item_id: str, new_price: float, threshold: float 50.0) - bool: old_price price_cache.get(item_id) if old_price is None: price_cache[item_id] new_price return False if abs(new_price - old_price) threshold: price_cache[item_id] new_price return True return False首次发现不推送价格变化超过 50 元才提醒低于阈值的变化静默更新缓存。这比「每次都推」少很多噪音比「完全忽略」更能抓到真正的大幅降价。我会把threshold设为关键词价格的 2% 左右单价高的商品降价 50 元可能不算事但在单价 400 元的镜头盖场景50 元就是大事了。回头说说我这套习惯的来历最早跑这个项目时我只设了 30 秒的轮询间隔跑了不到一个小时就触发了滑块整个程序直接停摆。从那以后我每次部署微信订阅机器人都强制走一遍「请求频率 → 异常兜底 → 去重机制 → 推送额度」四步检查宁可慢一点也不要让监控程序在半夜悄悄死掉。这套闲鱼监控的资源包整体设计偏务实没有花哨的架构核心就是「轮询 解析 过滤 推送」四个环节你把上面的踩坑点对着检查一遍大概率能稳定跑几周不翻车。希望帮到你。本文还有配套的精品资源点击获取