简介:这是一份基于 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 | 用户标识 | 登录后的 cookie |
key | 本次会话的密钥 | 登录后从 cookie 或接口响应取 |
pass_ticket | 票据 | 登录后 cookie |
appmsg_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落库(也可以用 sqlite),lxml解析 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", encoding="utf-8") as f: f.write(json.dumps(data, ensure_ascii=False) + "\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", encoding="utf-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}, upsert=True) # 多图文里的子文章 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}, upsert=True) if __name__ == "__main__": parse_file("history.json") print(f"入库完成,总数:{col.count_documents({})}")逻辑说明:type == 49是图文消息的类型码,其他类型(文字、图片、视频)跳过。app_msg_ext_info是主文章,multi_app_msg_item_list是同一推送里的第二篇及以后。用appmsgid做唯一键去重,upsert=True保证重复跑不会产生脏数据。参数上,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, data=payload, headers=headers, timeout=10) 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", data=payload, timeout=10) 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一定要写进配置文件并加过期提醒,别硬编码在脚本里,不然换一次改一次,烦得很。希望帮到你。
本文还有配套的精品资源,点击获取