简介:这是一份基于 Node.js 的微信爬虫项目,面向需要批量采集微信公众号文章数据(正文、阅读量、点赞量、在看数、评论及历史文章链接)的开发者。项目以 AnyProxy 中间人代理原理解析微信 HTTPS 请求,支持本地电脑运行或 Docker 部署,并借助 MongoDB 与 Redis 完成数据存储和任务调度。压缩包共 80 个文件,以 js/jsx 源码为主,辅以 Dockerfile、docker-compose.yml、package.json、证书文件及少量静态资源,整体仅 1.13MB,便于快速下载与部署。内部目录按 client、server、plugins、rule、api 等模块清晰划分,前端界面、代理规则、数据存储逻辑彼此独立;同时附带 Docker 编排配置和证书生成文件,可省去繁琐的环境搭建步骤。已有 2625 人学习下载,适合对爬虫、HTTP 代理及微信数据挖掘感兴趣的 Node.js 中高级开发者参考实践。
1. 微信文章采集需求:核心不是爬页面,而是把后台数据接口串成一条管线
接到“帮我把某公众号所有历史文章连同阅读量、点赞、评论一起导出来”这种需求时,很多人的第一反应是去抓文章页的 HTML。真正动手会发现死路一条:微信文章页源码里没有阅读量和点赞量,评论也是异步加载,模拟点击翻页既慢又容易触发风控。搜到 wechat_spider 这类方案时,你找到的思路是利用公众平台后台自己的文章数据接口,按“历史列表 → 文章详情 → 互动数据”三步把数据取回来。“所有历史文章链接”对应列表接口的翻页游标,阅读量、点赞和评论分别对应另外两个独立接口。这篇文章把这三层接口的调用方式、参数含义和踩坑点一次讲透,适合自己运营公众号做内容分析,或者盯竞品公开文章的开发者。
2. 先过登录态这道坎:token、cookie 与 fakeid 的取法与封装
2.1 为什么选后台接口而不是抓文章页 HTML
文章页拿不到完整数据是最直接的原因。公开文章页渲染出来的 HTML 只有标题、作者、正文和发布时间,阅读量、点赞量这类互动数据在页面源码里根本不存在,它们是后端单独下发的。评论虽然页面里有入口,但也是滚动到位置后才异步请求,抓取时要模拟浏览器行为,效率低且容易被识别。
公众平台后台的情况不一样。后台的图文管理、超链接插入、数据统计模块本身就有查询文章列表和文章数据的接口,运营者登录后访问这些接口是产品设计的正常路径,返回的是干净 JSON,不需要渲染 JavaScript。你拿到的不是“爬”出来的页面,而是后台组件自己用的数据源。办法的核心价值就是把这些接口找出来、把参数补齐、把数据存下来。
我一般会先确认一下需求边界:如果只需要正文和链接,一个列表接口就能搞定;如果需要阅读量、点赞、评论,那必须接受“列表一次拿全量元数据、详细数据逐篇请求”这种两段式设计。认清这一点,后面写代码时就不会觉得接口太多是负担。
2.2 三个关键参数的来源与有效期
开始写代码前需要准备三个配置项:token、cookie、fakeid。它们来源不同,有效期差异也很大,混用或抄错是新手最容易翻车的地方。
| 配置项 | 获取方式 | 有效期特征 |
|---|---|---|
| token | 登录公众平台后台后,打开任意一个后台页面,地址栏?token=后面那一长串 | 跟随登录态,几小时到几天,后台掉线即失效 |
| cookie | 后台页面按 F12 打开开发者工具,复制当前请求头里的完整 Cookie | 与 token 同步,两者最好一起更新 |
| fakeid | 在浏览器打开目标公众号的任意一篇文章,查看页面源码,搜索fakeid变量,复制引号内的值 | 长期有效,与登录态无关,一次获取可反复用 |
fakeid 看起来是一段很长的编码字符串,很多人以为它是公众号微信号,其实不是。它是由公开页面传输的公众号身份标识,同一个公众号的 fakeid 是固定的,换浏览器、换登录人都不变。token 和 cookie 则是你的登录凭证,过期后所有请求都会返回错误,这是后面排查时第一个要怀疑的对象。
三个参数都准备好之后,把它们放进一个配置文件里集中管理。不要散落在脚本各处,因为后期增量更新时你大概率要频繁替换 token 和 cookie,集中管理能少改很多地方。
2.3 用 requests 固定登录态的最小封装
这个方案用 requests 就够,不需要上 scrapy 这类重框架。接口是 HTTP + JSON,没有渲染需求,requests 的会话管理足够应对。下面是基础封装,也是后面所有请求的公共入口。
import requests import time # 从浏览器复制,不要手动修改 CONFIG = { "token": "从后台地址栏?token=后面复制", "cookie": "从浏览器开发者工具复制完整Cookie头", "fakeid": "从公众号主页源码中搜索fakeid变量得到", } # API_BASE 取自后台地址栏的域名前缀,不要写死 API_BASE = "后台域名" HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/120.0 Safari/537.36", "Cookie": CONFIG["cookie"], "Referer": f"https://{API_BASE}/", # 与后台页面保持一致 } def get_json(path, params): """统一请求入口:带重试的GET,失败时抛异常""" url = f"https://{API_BASE}{path}" for attempt in range(2): try: resp = requests.get(url, params=params, headers=HEADERS, timeout=10) j = resp.json() if j.get("ret") == 0: return j print("ret非0:", j.get("ret"), j.get("errmsg")) except Exception as e: print(f"请求异常,第{attempt + 1}次:{e}") time.sleep(1) raise RuntimeError("请求失败")这个封装的逻辑很简单:每个接口请求都走get_json,先判断返回的ret是否为 0,非 0 说明业务层报错,直接打印出来;网络层异常重试两次后放弃。重试次数不要设太多,接口报错通常是参数错误或登录态失效,重试再多次也解决不了,反而会掩盖真正的问题。
HEADERS里最重要是 Cookie 和 Referer。Referer 必须保留,后台接口对来源有校验,Referer 缺失或域名对不上时会返回异常。User-Agent 不要用默认的python-requests,容易被风控系统标记,随便填一个常见浏览器的 UA 就行。
3. 历史文章全量链接:时间戳翻页模型与去重逻辑
3.1 appmsgpublish 返回结构里必须留下的字段
标题里“所有历史文章链接”的实现,核心接口是/cgi-bin/appmsgpublish。这个接口一次返回一页文章列表,每一篇包含字段:appmsgid、idx、title、link、update_time、digest、cover、author、content。看到content字段别惊讶,列表接口确实直接带正文 HTML,这意味着拉历史文章的同时正文已经拿到了,不需要再逐篇请求详情页。
这里有两个字段必须特别留意:appmsgid和idx。appmsgid是文章的唯一标识,但同一篇文章在发布、修改后appmsgid可能不变;idx是文章在发布当天的序号,阅读量接口和评论接口都要求同时传这两个值。我的做法是把appmsgid + idx拼成联合主键,入库时按这个键去重。
link字段多数情况下是可直接访问的文章地址,个别场景下返回的是相对路径,落地时要加一层判断,遇到以/s?开头的值就手动拼上标准前缀。这个不算 bug,是后台接口在不同版本下的返回差异,留个心眼就行。
3.2 begin 不是页码,是时间游标
第一次写翻页时很容易想当然:第一页begin=0,第二页begin=1,第三页begin=2——这是错的。这个接口的翻页维度是时间,不是页码。
具体机制是:第一页传begin=0,返回最新发布的 5 篇文章;第二页的begin必须传第一页最后一条的update_time(10 位秒级时间戳),接口会返回从这个时间点往前推的文章。之后每页都用上一页最后一条的update_time作为游标,直到拿到最早那篇。这个机制的好处是不会因为新文章发布导致重复或漏拉,坏处是如果你把游标存错,翻页会直接卡住或死循环。
count参数默认是 5,我试过调大,但稳定性和返回速度都一般,接口对这个参数有上限限制,超过会直接报错。建议保持默认值 5,不要为了减少请求次数去调大它,翻页次数多一些没关系,数据完整比速度重要。
3.3 全量拉取的最小实现
把上面两个要点落到代码里,就是一个带时间游标的循环。这个函数可以直接跑,参数只需要 fakeid、token 和停止时间。
import time def fetch_all_articles(stop_ts=0, max_pages=2000): """ 全量拉取历史文章,返回文章列表。 stop_ts: 遇到 update_time 小于等于该值的文章就停止,0 表示拉到最早。 """ articles = [] cursor = 0 # 第一页传0,表示从最新开始 page = 0 fakeid = CONFIG["fakeid"] token = CONFIG["token"] while page < max_pages: params = { "sub": "list", "search_field": "null", "begin": cursor, "count": 5, "query": "", "fakeid": fakeid, "type": "101_1", "free_publish_id": "", "publish_time": "", "token": token, "lang": "zh_CN", } data = get_json("/cgi-bin/appmsgpublish", params).get("data", {}) batch = data.get("list", []) if not batch: break for item in batch: info = { "appmsgid": item["appmsgid"], "idx": item["idx"], "title": item["title"], "link": item["link"], "update_time": item["update_time"], "content": item.get("content", ""), } articles.append(info) # 增量场景:时间戳已经低于停止线,直接返回 if stop_ts and item["update_time"] <= stop_ts: return articles # 下一页游标 = 本页最后一条的发布时间 cursor = batch[-1]["update_time"] page += 1 time.sleep(1) # 列表接口也要控速 return articles这段代码的关键逻辑在两处。第一,游标更新必须取batch[-1]["update_time"],而不是你自己记录的时间,因为接口返回的顺序可能和你想象的不完全一致;第二,stop_ts参数承担增量职责——第一次全量传 0 拉到底,后面每天跑传上次的最大update_time,接口拉到跨过这条线就停。
max_pages是个保护项,防止游标异常时无限循环。2000 页对应 1 万篇文章,对绝大多数公众号都够用;如果目标号发文量极大,把这个值调大即可。跳出 while 循环后,说明已经拉穿了这个公众号的全部历史文章,返回的articles就是标题里“所有历史文章链接”的原始数据。
3.4 增量场景里的“所有历史”怎么定义
“所有历史文章”这个词在不同需求下边界完全不同。第一种是真全量:把一个公众号从注册到现在的所有文章都拉下来,这时stop_ts=0,循环跑到接口返回空为止。第二种是增量:你只需要从某一天之后的新文章,此时把上次同步的最大update_time传进去,接口会自动停在时间线边界。
我用增量时会在 SQLite 里记录每个 fakeid 对应的max_update_time,每次跑完任务后更新。这样做有两个好处:一是重启脚本不用从头拉,二是如果当天新增文章较多导致脚本中断,下次再跑时只要看游标位置,不会重复请求已经处理过的文章。增量模式唯一的坑点是同一秒发布多篇文章时,时间戳相同会导致边界判定不准,这种情况我在第 5 章的避坑清单里单独说。
4. 正文、阅读量、点赞与评论:三个接口拼出完整数据
4.1 正文直接取自列表接口的 content 字段
很多人以为正文要打开文章页再抓一次,实际上是多余的。appmsgpublish返回的每条记录里就有content字段,内容是完整的正文 HTML,包含段落标签、图片标签和排版样式。你只需要把它原样存下来,后面做全文检索或转纯文本时再清洗。
这块最需要注意的不是接口,而是图片。正文 HTML 里的图片地址有两种形态:一种是src属性直接可访问的完整 URL,另一种是>def fetch_stat(appmsgid, idx): """单篇阅读量/点赞量""" params = { "appmsgid": appmsgid, "idx": idx, "fakeid": CONFIG["fakeid"], "token": CONFIG["token"], "lang": "zh_CN", } data = get_json("/cgi-bin/appmsgstat", params).get("data", {}) return { "read_num": data.get("read_num", 0), "like_num": data.get("like_num", 0), }
返回结构里的read_num是阅读量,like_num是点赞量,都是整数。需要留意的是大号场景:阅读量超过 10 万的公众号,页面显示是“10万+”,但接口返回的往往还是具体数字,这两个口径在数据分析时要统一,别混着比较。
这个接口要逐篇请求,意味着全量拉取 1000 篇文章就要调 1000 次接口。我对它的限速策略是间隔 2 到 3 秒,不做并发。并发确实能快很多,但触发风控的概率会指数级上升,被限制了反而拖慢整体进度。如果数据量确实大,建议分批次跑,一天跑一部分,比一次性冲完更稳。
4.3 评论分页:comment/list 与一个黑匣子字段
评论接口/cgi-bin/comment/list同样依赖appmsgid + idx,支持分页。返回结构里的elected_comment数组就是已展示的精选评论,每条包含content、like_num、comment_id和reply_list。
import base64 def fetch_comments(appmsgid, idx): """分页拉精选评论,返回评论列表""" comments = [] begin = 0 while True: params = { "action": "list", "appmsgid": appmsgid, "idx": idx, "fakeid": CONFIG["fakeid"], "token": CONFIG["token"], "lang": "zh_CN", "begin": begin, "count": 20, } data = get_json("/cgi-bin/comment/list", params).get("data", {}) batch = data.get("elected_comment", []) for c in batch: content = c.get("content", "") # 部分返回的 content 是编码串,识别异常字符时尝试解码 if content and not content.isprintable(): try: content = base64.b64decode(content).decode("utf-8", "ignore") except Exception: pass comments.append({ "comment_id": c.get("comment_id"), "content": content, "like_num": c.get("like_num", 0), "reply_list": c.get("reply_list", []), }) if len(batch) < 20: break begin += 20 time.sleep(0.5) return comments这段代码里藏了两个边界说明。第一,接口只返回精选评论,不是全部评论,这是微信产品设计决定的,接口层面无法改变;第二,content字段在不同版本接口下有可能是编码串,这是我实际遇到过的情况——某天代码跑得好好的,第二天突然返回一堆乱码,后来才发现是内容做了编码处理。代码里的isprintable()判断就是为此预留的兜底,识别到异常就把内容按 base64 解回来,解失败就保留原样,保证下游不崩。
评论点赞数在like_num字段,每条评论的回复在reply_list里,存储时建议直接以 JSON 字符串形式落库,因为回复层级和结构不稳定,强拆成关系表反而麻烦。
4.4 数据落库:文章主表与互动数据表拆分
数据怎么落库直接决定后面分析的效率。我用的方案是拆两张表:文章主表存静态信息,互动数据表存每天变化的动态指标。
| 表名 | 字段 | 说明 |
|---|---|---|
| articles | appmsgid, idx | 联合主键 |
| articles | title, link, update_time | 标题、链接、发布时间 |
| articles | digest, author, content | 摘要、作者、正文 HTML |
| article_stats | appmsgid, idx | 联合主键,关联 articles |
| article_stats | read_num, like_num | 阅读量、点赞量 |
| article_stats | comment_count, fetch_date | 评论数、抓取日期 |
文中提到的comment_count来自评论接口返回里的总数,不是elected_comment的长度,这个区分很重要。拆分的逻辑很直接:正文和链接不会随时间变化,阅读量和点赞量会变。每天新增抓取时,文章表用联合主键去重,互动表按fetch_date记录新一行,这样就能画出每篇文章阅读量随时间的增长曲线,而不只是当天快照。
5. 高频避坑清单:登录态、翻页游标与接口风控的五个现场
5.1 返回 ret=-3 但 token 看起来没过期
现象:昨天还能正常跑的脚本,今天所有请求都返回ret=-3,错误信息指向 token 无效。你去后台页面看,页面能正常打开,地址栏里的 token 和配置文件里一对比,也确实一样。 原因:token 和 cookie 是配套的。后台页面能打开不代表接口能通,真正下发请求时校验的是 cookie 里的完整登录态,token 只是其中一部分。很多时候后台页面有缓存,页面显示正常,但 cookie 里的关键字段已经过期。 解决:不要手动刷新 token,直接重新登录一次后台,然后用新的 cookie 和 token 同时替换配置文件。我还会在get_json里对ret != 0做提前返回,让主循环及时停下来,而不是带着无效登录态白白重试几百次。
5.2 翻页翻到一半列表为空,总数却对不上
现象:全量拉取时,前几页正常,之后某一页接口返回的list是空数组,程序因此退出,但最终统计的文章数和后台显示的总数差了一截。 原因:最容易犯的错是把“这一批不足 5 条”当成“已经到底了”。公众号如果只发了 3 篇文章,第一页返回 3 条,按not batch判断会正常结束;但如果某段时间发文频率低,翻页时正好遇到不足 5 条的一批,程序就可能提前退出。 解决:终止条件改成“batch为空才停”,不足 5 条只能说明这一页不满,不代表后面没有更早的文章。另外,同秒发布多篇文章时时间戳相同会导致游标不动,需要在进入下一页前判断游标是否和上一轮相同,相同就说明遇到边界情况,把最后一条的idx作为二级游标或者直接结束本轮。
5.3 fakeid 复制少一位导致全部请求落空
现象:配置都填好了,但所有列表请求返回空数据或者直接报错,检查参数格式也看不出问题。 原因:fakeid 是一段很长的编码字符串,手动从源码里复制时很容易漏掉末尾一两位。它又不像 token 有明显的格式特征,少了字符肉眼完全看不出来。 解决:复制后和源码里的原始值做一次全长比对,或者在脚本里加一个断言,长度少于预期值就直接报警。fakeid 本身可以长期使用,一次拿对后面都不用动,这个调试成本值得花。
5.4 评论字段是乱码,不是内容坏了而是编码问题
现象:评论接口调用成功,elected_comment里也有数据,但content字段显示为一串无法阅读的字符。 原因:接口在某个版本里对部分字段做了编码处理,返回的不再是原始文本。这个行为没有文档说明,只能通过返回数据特征判断,属于接口里的黑匣子参数。 解决:加入isprintable()检测,异常时尝试 base64 解码。如果解出来还是乱码,就把原始报文落盘,等接口恢复正常后再重新拉取。评论数据不是高频变化内容,晚几天补齐影响不大。
5.5 拉了几百篇开始返回验证页
现象:连续请求一段时间后,resp.json()解析失败,打印出来的响应内容是一段 HTML 验证页面。 原因:请求频率触发了风控策略,接口不再返回 JSON 数据。这跟前面几种错误性质不同,属于数据量设计问题,不是参数配置错误。 解决:列表接口请求间隔保持 1 秒以上,阅读量、评论等详情接口间隔 2 到 3 秒;单批任务控制在 500 篇以内,跑完暂停一段时间再继续。如果目标号数据量特别大,把任务拆成按天增量执行,比一次性全量更容易被容忍。
6. 增量更新与校验三连:把采集任务变成每天能自动跑的管线
6.1 用 SQLite 记录游标与已见文章
增量跑批的关键是把游标和已见文章落盘,不能每次重启都从头拉。我用 SQLite 存状态,表结构很简单:一张表记录每个 fakeid 的max_update_time,一张表存文章主数据。这样即使脚本中断,下次启动也能从断点继续,不用重新拉已经处理过的历史数据。
import sqlite3 conn = sqlite3.connect("spider_state.db") conn.execute("""CREATE TABLE IF NOT EXISTS state ( fakeid TEXT PRIMARY KEY, max_update_time INTEGER, updated_at TEXT )""") conn.execute("""CREATE TABLE IF NOT EXISTS articles ( appmsgid TEXT, idx INTEGER, title TEXT, link TEXT, update_time INTEGER, content TEXT, PRIMARY KEY (appmsgid, idx) )""")每天定时任务执行时,先从state表读出max_update_time,传入fetch_all_articles作为stop_ts,跑完后用新列表里的最大值覆盖旧值。失败的任务不需要回滚,下次跑批时按照时间游标自然跳过已同步部分。
6.2 跑完任务后的三项校验
数据入库后不能直接信,我每次跑完都做三个校验。第一是总数核对:后台“发表记录”或“图文素材”页面有文章总数,拿库里去重后的记录数和它对比,差太多说明翻页提前终止了;第二是链接抽查:随机抽 5 到 10 篇文章的link直接浏览器打开,能正常显示说明链接字段没问题;第三是空值检测:统计title、content为空的比例,超过 5% 基本可以断定列表接口返回值发生了变化,需要重新检查字段名。
6.3 一个值得养成的排错习惯
最后说一个我自己的习惯:每次接口请求都把原始返回 JSON 落一份到本地raw/目录,文件名带时间戳。这个习惯救过我一次——某次接口字段悄悄变了,我的解析代码直接报错,但报错信息根本看不出是字段名问题,后来翻原始报文才定位到是返回结构里多了一层嵌套。如果没有这份原始报文,排查时间至少翻倍。跑批脚本的排错和功能开发不一样,你不是在写代码,是在维护一条数据管线,原始报文就是管线的日志,留一份永远不吃亏。希望帮到你。
本文还有配套的精品资源,点击获取