简介这是一份基于 Node.js 的微信爬虫项目源码面向需要批量采集公众号数据的开发者与数据分析人员通过 AnyProxy 中间人代理原理抓取微信文章内容、阅读量、点赞量、在看数及评论等数据支持 Docker 部署可运行于个人电脑或服务器。资源包共 80 个文件以 40 个 js 与 9 个 jsx 脚本为核心涵盖代理规则、数据模型、内容处理与导出逻辑另含 json 配置、html 页面、png 截图及证书文件等整体约 1.13MB结构清晰便于二次开发。项目已适配 AnyProxy 4并配套 MongoDB、Redis 存储方案代码中可见文章链接抓取、历史页面处理、数据合并与导出等模块。目前已有 2623 人学习下载适合想研究微信数据采集、代理抓包或 Node.js 爬虫实战的读者参考借鉴。1. 公众号历史文章抓取从阅读量到评论一套能跑通的采集链路做内容分析、竞品监测或者私域运营的人迟早会碰到同一个需求把某个公众号的历史文章链接全部拉下来顺带拿到阅读量、点赞量、评论这些互动数据。手动翻历史消息不现实几百篇翻到手酸还容易漏。wechat_spider这类工具解决的正是这件事——输入公众号名称或 biz输出一份结构化的文章清单包含标题、链接、发布时间、阅读数、点赞数、评论内容。它适合三类人做选题分析的内容运营、需要监控竞品发文节奏的市场人员、以及想把自己账号历史数据沉淀下来的号主。核心难点不在写代码而在微信的反爬机制和登录态维护。下面按「先搞懂数据从哪来 → 再动手跑通最小链路 → 最后处理翻车场景」的顺序展开每一步都给可复现的命令和参数。2. 数据从哪来公众号文章的四个数据入口与登录态原理2.1 历史文章列表的三个来源公众号文章链接不是随便就能拿到的微信没有开放「列出某号所有文章」的公开接口。实际能用的入口有三个第一个是公众号主页的图文列表。在微信客户端里点进一个号能看到历史消息但这个列表是分页加载的且客户端做了加密直接抓包拿到的是密文。常见做法是通过「微信公众平台」的素材管理接口但那只对你自己运营的号有效。第二个是搜狗微信搜索。搜狗曾经提供公众号文章搜索能按公众号名筛文章但这些年收录越来越不全很多号的文章搜不到且翻页有验证码。它适合做补充不适合做全量。第三个是文章内的超链接和「阅读原文」。很多号会在文末放往期文章链接顺着爬能拿到一部分但覆盖不全属于辅助手段。真正能拿到全量历史链接的是通过中间人代理抓取微信客户端与服务器之间的通信。微信客户端在加载历史消息时会向服务器请求一个列表接口返回的数据里包含每篇文章的appmsgid、title、url、datetime。这个接口需要携带登录态的 cookie 和 token。wechat_spider这类项目的核心思路就是在本机起一个代理让微信客户端的流量经过它从中提取这些接口响应并落库。注意这里说的代理是本地抓包代理用于分析自己账号的通信不涉及任何绕过网络限制的操作。请确保只采集你有权限查看的公开数据。2.2 登录态和关键参数biz、uin、key、pass_ticket要让请求合法几个参数缺一不可参数含义从哪来biz公众号唯一标识形如MzA5...文章链接里的__biz字段uin用户标识登录后的 cookiekey本次会话的密钥登录后从 cookie 或接口响应取pass_ticket票据登录后 cookieappmsg_token文章列表令牌历史消息接口响应biz是定位公众号的关键同一个号的所有文章__biz都一样。你可以在任意一篇该号文章的链接里找到它。key和pass_ticket会过期通常几小时到一天过期后需要重新登录获取。2.3 阅读量和点赞量为什么难拿阅读量不在文章列表接口里而在文章详情页的js_相关接口中。早期微信返回的阅读数是明文后来改成通过https://mp.weixin.qq.com/mp/getappmsgext接口返回且需要key和pass_ticket同时有效。点赞在看数同理。评论则需要调appmsg_comment接口分页拉取。这三个接口都有频率限制短时间大量请求会触发验证或直接返回空。所以采集节奏要控制通常建议每篇文章间隔 2 到 5 秒单账号单日采集量控制在几百篇以内。3. 跑通最小链路代理配置、抓包落库、导出文章清单3.1 环境准备与依赖安装先准备 Python 环境建议 3.8 以上。核心依赖是抓包和 HTTP 请求库# 创建虚拟环境 python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate # 安装依赖 pip install mitmproxy requests pymongo lxmlmitmproxy负责中间人抓包requests发请求pymongo落库也可以用 sqlitelxml解析 HTML。如果你不想用 mitmproxy也可以用fiddler或charles手动导出但自动化程度低。3.2 配置抓包代理并提取接口响应启动 mitmproxy写一个脚本把历史消息接口的响应存下来# save_history.py from mitmproxy import http import json def response(flow: http.HTTPFlow): # 匹配历史消息列表接口 if mp.weixin.qq.com/mp/profile_ext in flow.request.pretty_url: # 只处理返回 JSON 的响应 if application/json in flow.response.headers.get(content-type, ): try: data json.loads(flow.response.text) # 提取文章列表 if general_msg_list in data: with open(history.json, a, encodingutf-8) as f: f.write(json.dumps(data, ensure_asciiFalse) \n) print(f已保存一页文章数{len(data[general_msg_list])}) except Exception as e: print(f解析失败{e})启动命令mitmproxy -s save_history.py -p 8080然后在微信客户端设置代理指向本机 8080 端口手机和电脑在同一局域网。打开目标公众号的历史消息页往下翻几页脚本会把每页响应追加到history.json。逻辑说明profile_ext是历史消息接口的路径特征general_msg_list字段里是文章数组。每翻一页触发一次请求脚本就存一条。参数上-p 8080是监听端口可以改成别的-s指定脚本。3.3 解析 JSON 并落库拿到history.json后解析出每篇文章的字段# parse_history.py import json from pymongo import MongoClient client MongoClient(mongodb://localhost:27017/) db client[wechat_spider] col db[articles] def parse_file(path): with open(path, r, encodingutf-8) as f: for line in f: data json.loads(line) msg_list json.loads(data[general_msg_list]) for item in msg_list[list]: # 只处理图文消息 if item.get(comm_msg_info, {}).get(type) ! 49: continue info item[comm_msg_info] appmsg item.get(app_msg_ext_info, {}) # 主文章 doc { appmsgid: info[id], title: appmsg.get(title), url: appmsg.get(content_url), datetime: info[datetime], biz: appmsg.get(content_url, ).split(__biz)[-1].split()[0] if __biz in appmsg.get(content_url, ) else None, } col.update_one({appmsgid: doc[appmsgid]}, {$set: doc}, upsertTrue) # 多图文里的子文章 for sub in appmsg.get(multi_app_msg_item_list, []): sub_doc { appmsgid: sub.get(itemidx), title: sub.get(title), url: sub.get(content_url), datetime: info[datetime], biz: doc[biz], } col.update_one({appmsgid: sub_doc[appmsgid]}, {$set: sub_doc}, upsertTrue) if __name__ __main__: parse_file(history.json) print(f入库完成总数{col.count_documents({})})逻辑说明type 49是图文消息的类型码其他类型文字、图片、视频跳过。app_msg_ext_info是主文章multi_app_msg_item_list是同一推送里的第二篇及以后。用appmsgid做唯一键去重upsertTrue保证重复跑不会产生脏数据。参数上MongoDB 连接串按实际改没有 MongoDB 可以换成 sqlite3。3.4 补采阅读量、点赞量和评论文章链接拿到后逐篇请求详情接口# fetch_stats.py import requests import time from pymongo import MongoClient client MongoClient(mongodb://localhost:27017/) col client[wechat_spider][articles] # 从抓包或登录后获取会过期 KEY 你的key PASS_TICKET 你的pass_ticket UIN 你的uin def get_stats(url, biz): # 从 url 提取 mid 和 idx params dict(p.split() for p in url.split(?)[1].split()) api https://mp.weixin.qq.com/mp/getappmsgext payload { is_only_read: 1, is_temp_url: 0, appmsg_type: 9, mid: params.get(mid), idx: params.get(idx), sn: params.get(sn), __biz: biz, key: KEY, pass_ticket: PASS_TICKET, uin: UIN, } headers {User-Agent: Mozilla/5.0 (iPhone; CPU iPhone OS 14_0 like Mac OS X) AppleWebKit/605.1.15} r requests.post(api, datapayload, headersheaders, timeout10) if r.status_code 200: data r.json() return { read_num: data.get(appmsgstat, {}).get(read_num), like_num: data.get(appmsgstat, {}).get(like_num), old_like_num: data.get(appmsgstat, {}).get(old_like_num), } return {} for doc in col.find({read_num: {$exists: False}}).limit(50): stats get_stats(doc[url], doc[biz]) if stats: col.update_one({_id: doc[_id]}, {$set: stats}) print(f{doc[title]} 阅读{stats.get(read_num)}) time.sleep(3) # 控制频率避免触发限制逻辑说明getappmsgext是阅读量接口mid、idx、sn从文章 URL 里解析。key和pass_ticket过期后会返回错误码需要重新登录抓包更新。time.sleep(3)是硬性节流别去掉。参数上limit(50)控制单次跑的量建议分批。评论接口类似路径是appmsg_comment参数多一个comment_id从详情页 HTML 里提取。这里不展开思路一致。4. 避坑与排查登录态失效、频率限制、数据缺失的五个翻车现场4.1 现象请求返回ret: -3或空数据原因key或pass_ticket过期。微信的登录态通常几小时失效尤其是切换网络或长时间不操作后。解决重新在微信客户端打开一次公众号历史页用抓包脚本重新捕获最新的key和pass_ticket。建议把这两个值写到配置文件方便替换。如果频繁失效检查是不是同一账号在多设备登录互踢会导致会话失效。4.2 现象翻页到第三页后返回空列表原因微信对历史消息接口有翻页深度限制通常只能翻到最近几个月或前几十页更早的文章需要其他方式。解决常见做法是结合「文章内链」补采——从已抓到的文章正文里提取往期链接递归抓取。另外可以按时间分段先抓最近半年再通过搜狗微信搜索补早期文章。没有百分百全量的方案接受一定缺失率。4.3 现象阅读量字段存在但数值为 0原因部分文章在采集时阅读量确实为 0刚发布或者接口返回的是缓存值。另外某些账号关闭了阅读量展示。解决对阅读量为 0 的文章做二次采集间隔几小时后再跑一次。如果始终为 0检查该号是否在后台关闭了「显示阅读量」。这个没法绕过只能标记为不可用。4.4 现象评论拉取时comment_id找不到原因评论接口需要从文章详情页的 HTML 里提取comment_id但部分文章没有开启评论或者页面结构变了。解决先请求文章详情页用正则匹配comment_id匹配不到就跳过该篇。正则写法comment_id\s*[:]\s*[](\d)[]。如果大面积匹配失败说明微信改了页面结构需要更新正则。4.5 现象跑了几百篇后 IP 被限所有请求超时原因请求频率过高触发微信的风控。表现是接口返回 403 或直接连接超时。解决立即停止采集等待几小时。后续把间隔调到 5 到 10 秒单日单账号控制在 200 篇以内。如果有多账号可以轮换uin和key但注意别用同一 IP 高频请求。另外请求头里的User-Agent要模拟真实客户端别用 Python 默认的。5. 进阶增量采集、评论分页与数据校验的三个实用技巧5.1 用appmsgid做增量避免重复跑全量每次采集前先查库里已有的最大appmsgid只抓比它新的文章。历史消息接口返回的列表是按时间倒序的翻到appmsgid小于等于已有最大值就可以停。这样日常增量采集只需翻一两页速度快且不容易触发限制。# 增量采集的停止条件 max_id col.find_one(sort[(appmsgid, -1)])[appmsgid] # 在解析时判断 if int(info[id]) int(max_id): print(已到已有数据边界停止) break5.2 评论分页comment_id加offset循环拉取评论接口每次返回一页需要循环。关键参数是comment_id和offset每页通常 10 条。循环直到返回的comment_list为空或has_more为 0。def fetch_comments(comment_id, biz, mid, idx, sn): offset 0 all_comments [] while True: payload { action: getcomment, comment_id: comment_id, __biz: biz, mid: mid, idx: idx, sn: sn, offset: offset, limit: 10, key: KEY, pass_ticket: PASS_TICKET, } r requests.post(https://mp.weixin.qq.com/mp/appmsg_comment, datapayload, timeout10) data r.json() comments data.get(comment_list, []) if not comments: break all_comments.extend(comments) offset 10 time.sleep(2) return all_comments逻辑说明offset每次加 10直到返回空列表。limit固定 10改大了也没用微信服务端写死。每条评论里有nick_name、content、like_num按需存。5.3 数据校验用阅读量和点赞量的比值做异常检测采集完一批数据后快速判断有没有采错。正常文章的点赞率点赞/阅读通常在 0.5% 到 5% 之间。如果某篇比值超过 20%大概率是阅读量没采到为 0 或极小或者点赞数串了。写个校验脚本for doc in col.find({read_num: {$gt: 0}, like_num: {$gt: 0}}): ratio doc[like_num] / doc[read_num] if ratio 0.2: print(f异常{doc[title]} 阅读{doc[read_num]} 点赞{doc[like_num]})发现异常就回查该篇的接口响应看是不是read_num字段缺失或为 0。这个习惯帮我省了很多事后返工。我自己跑这套链路两年多最大的教训是别贪快。一开始为了赶分析报告把sleep调到 0.5 秒结果跑了 300 篇后整个账号的key直接失效等了半天才恢复。后来固定 3 到 5 秒间隔单日不超过 300 篇再没翻过车。另外key和pass_ticket一定要写进配置文件并加过期提醒别硬编码在脚本里不然换一次改一次烦得很。希望帮到你。本文还有配套的精品资源点击获取