尧图网站设计 尧图网站设计YAOTU DESIGN
ARTICLE DETAIL

资讯详情

深耕网站设计与一线实操的经验洞察。

微信公众号爬虫实战:wechat_spider抓取历史文章与互动数据

微信公众号爬虫实战:wechat_spider抓取历史文章与互动数据 简介这是一份基于 Node.js 的微信爬虫项目采用中间人代理与 AnyProxy 解析微信 HTTPS 请求可批量获取公众号文章正文、阅读量、点赞量、在看数、评论以及历史文章链接等数据适合需要采集微信公众号数据的开发者、数据分析师或运营人员使用。项目支持本机运行也能通过 Docker 便捷部署到服务器降低了环境搭建门槛。资源包共 80 个文件体积仅 1.13MB以 JavaScript 源码40 个 js、9 个 jsx为主同时包含 JSON 配置、HTML 页面、CSS 样式、图片资源、HTTPS 证书crt/key及 Dockerfile、docker-compose.yml 等容器化部署文件目录结构清晰便于直接运行或二次开发。已有 2619 人学习下载代码内封装了客户端与服务端、代理规则、MongoDB/Redis 存储、文章历史链接抓取、评论与图片处理等完整模块拿来即可安装使用也可作为学习微信接口协议与爬虫代理原理的实战参考。 做公众号数据分析这行手里没有一套顺手的抓取工具基本等于瞎忙。我最早是手工复制文章链接、每天记录阅读量后来发现根本扛不住——几十个公众号、每天上百篇更新光整理数据就能耗掉大半天。后来我把wechat_spider这套微信爬虫思路落地专门用来获取公众号历史文章链接、文章正文、阅读量、点赞量和评论数据总算把这块时间压缩到了几分钟。如果你也在做竞品监测、内容复盘或者公众号矩阵运营这篇文章应该能帮你少走不少弯路。这套方案的核心思路并不复杂不碰网页版那套受限的登录流程直接从 PC 微信客户端的流量入口入手把请求参数抓下来再模拟接口去拉历史文章、统计数据。下面我按自己的实操过程把技术选型、关键细节、踩坑记录都拆开讲一遍。1. 项目概述与整体思路1.1 这个爬虫能做什么解决了什么痛点先说我为什么要折腾这么一套东西。当时的需求是同时跟进二十几个同行公众号每天记录三件事更新了哪些文章、阅读量多少、评论区有没有值得关注的反馈。如果全手工做等于每天多出一个小时的纯体力劳动而且很容易漏记。wechat_spider最终帮我把整个过程变成了三步拉历史文章链接、批量抓正文和指标、写入数据库。具体能拿到的数据包括公众号的历史文章链接列表按时间排序翻完整个人主页每篇文章的标题、摘要、正文内容、封面图、作者阅读量、点赞量部分情况下还能拿到在看数文章评论区的内容、点赞数、回复内容发布时间、图文类型、是否为多图文等基础信息。这套东西适合谁一个是做新媒体竞品分析的人每天要盯着对手动态另一个是公众号矩阵运营方需要统一汇总自己多个号的数据再一个就是想做垂直内容库的开发者比如做行业资讯聚合、文章存档检索。只要你不是拿它去恶意刷量、搬运抄袭自己研究数据是完全够用的。1.2 为什么我选择从 PC 微信进程入手市面上的思路主要有三条模拟网页版登录、模拟手机端协议、从 PC 客户端抓包。三条路我都试过说点真实感受。网页版微信的登录限制很严很多账号根本没有网页版入口而且网页版拿到的公众号文章数据非常有限历史列表接口基本被风控盯死。模拟手机端协议这条路需要处理大量加密逻辑和风险校验自己维护成本极高不亚于重新实现半个微信客户端太折腾了。最终我选了第三条路让 PC 微信自己干活我只需要在它和服务器之间加一个代理把公众号主页的 HTTP 请求照实记录下来。PC 微信在浏览公众号文章和打开公众号主页时会在本地发出完整的接口请求这些请求里带了当前登录用户的合法身份参数。我做的就是把这些参数“借”过来重放同样的请求拿到数据。这样做的好处很明显不需要破解协议、不需要碰加密数据身份校验由微信客户端自己完成我只需要模拟 HTTP 请求。缺点也明显参数有过期时间过几个小时或者切换账号后就要重新抓包。对个人研究来说这点麻烦完全能接受。2. 核心细节解析与关键点梳理2.1 历史文章列表背后的接口逻辑想抓全一个公众号的历史文章首先要搞清楚微信是用哪个接口产数据的。PC 端打开公众号主页、上滑加载更多历史消息时会触发一个类似这样的请求GET https://mp.weixin.qq.com/mp/profile_ext?actiongetmsg__biz...fjsonoffset0count10is_ok1scene124uin777key...pass_ticket...参数解释起来其实不复杂但每个都很关键参数作用注意事项__biz公众号的唯一标识相当于公众号的身份证号从公众号主页 URL 里就能提取到key当前登录态的临时凭证时效很短过期后接口会报 invalid key必须重新抓包更新pass_ticket另一个登录校验参数和 key 配套使用切换账号后立刻失效offset翻页偏移量第一次是 0之后每次递增每次返回的数量是 count 的值count每页数量通常填 10 比较稳填太大容易被风控scene场景值抓历史列表时固定为 124改错了可能拿不到数据返回内容是一个 JSON里面有一个general_msg_list字段它本身是一段转义后的 JSON 字符串展开后有一个list数组每一项对应一篇图文消息。每项内部又分为comm_msg_info基础信息和app_msg_ext_info文章扩展信息如果这天的推送是多图文app_msg_ext_info下面还会有一个multi_app_msg_item_list数组把副图文放在里面。拿到这个结构后我要做的事就是不断把offset往上加直到返回的列表为空就说明已经翻到最早的历史文章了。2.2 文章内容、阅读量、点赞量怎么拿历史列表接口只能拿到文章的标题、摘要、封面图、跳转链接拿不到正文和阅读数据。正文很好办从content_url直接请求文章页面再把 HTML 里的正文部分提取出来就行。真正麻烦的是阅读量、点赞量。如果你只想看一篇文章的阅读数据就不需要去解析页面上的小字。微信有一个独立的统计接口通过文章链接里的sn参数加上__biz等参数就能调出结构化数据GET https://mp.weixin.qq.com/mp/getappmsgext?__biz...appmsg_type9idx1sn...fjson这个接口返回的 JSON 里有个appmsgstat字段里面直接就是read_num、like_num、old_like_num这类数字。注意这个接口同样依赖 key 和 pass_ticket而且对请求频率异常敏感。我一开始就是一篇文章调一次接口结果连续请求一百多次后直接触发限制所有返回都变成ret: 200003。后来我改成每篇文章之间随机延迟 2 到 5 秒并且控制单批任务量情况才好转。文章正文和阅读数据之间的关联我用的是 URL 里的sn参数。每篇文章的content_url里都带一个唯一的sn它就是这篇文章在公众号内的“文章指纹”调用统计接口时填它基本不会错。2.3 评论数据的获取逻辑评论是单独一个接口很多资料里没有提我一开始也卡了很久。PC 微信打开一篇文章并拉取评论区时会请求GET https://mp.weixin.qq.com/mp/appmsg_comment?actiongetcomment__biz...appmsgid...idx1comment_id...key...offset0count20这里需要多说一句appmsgid不是comm_msg_info.id而是文章 URL 里那个mid参数comment_id也藏在页面源码的某个变量里。第一次抓评论时我怎么都找不到comment_id后来在文章 HTML 的window.__commentData或者是页面里的一段 JSON 初始化数据里看到了它。提取方式有点绕先请求文章页面从 HTML 中解析出comment_id再拿去调评论接口。评论接口返回的结构也很标准comment数组里是每一条评论包含昵称、头像、内容、点赞数每条评论下的回复会放在reply字段里。如果文章没有开启评论接口会返回空列表这属于正常情况不是代码写错了。3. 实操过程与核心环节实现3.1 环境准备与抓包接入实际操作前先把工具链备齐。我用的环境是Windows 10 系统安装 PC 版微信并保持登录一个长期使用的账号Python 3.9 以上装了requests、beautifulsoup4、lxml抓包工具用的是 mitmproxy开启后设置 Windows 系统代理让 PC 微信的流量走本地代理数据库先用 SQLite 起步数据结构稳定后可以迁到 MySQL。第一次配置时要注意给 mitmproxy 安装 CA 证书否则抓不到 HTTPS 明文流量。安装证书后重新启动 PC 微信然后随便打开一个公众号的主页向上翻几页历史消息。再回看 mitmproxy 的控制台就能看到刚才说的profile_ext请求。我的习惯是直接在 mitmproxy 的过滤栏里输入profile_ext把无关请求全部滤掉只看历史文章列表接口。把这条请求的完整 URL 拷出来尤其是key和pass_ticket这两个参数存到一个配置文件里。这是整个项目的数据源头后面所有请求都用它来带身份。3.2 核心代码逻辑演示整体代码结构不复杂核心就三块拉历史列表、解析文章信息、进详情页补全指标。直接看代码。先定义基础请求参数和拉取历史文章列表的函数import requests import json import time from urllib.parse import urlparse, parse_qs BIZ 你的__biz值 KEY 抓包得到的key PASS_TICKET 抓包得到的pass_ticket HEADERS { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36, Referer: https://mp.weixin.qq.com/, } def get_history(offset0, count10): url https://mp.weixin.qq.com/mp/profile_ext params { action: getmsg, __biz: BIZ, f: json, offset: offset, count: count, is_ok: 1, scene: 124, uin: 777, key: KEY, pass_ticket: PASS_TICKET, wxtoken: , x: 5, } resp requests.get(url, paramsparams, headersHEADERS, timeout15) data resp.json() if data.get(ret) ! 0: print(请求失败返回, data) return [] msg_json data.get(general_msg_list, {}) return json.loads(msg_json).get(list, [])这里有个容易踩的坑返回的general_msg_list是字符串不是对象必须先json.loads一次否则直接按字典操作会各种报错。拿到列表后需要把图文消息展开因为一篇文章可能包含多张副图文def parse_articles_from_msg(msg_item): base msg_item.get(comm_msg_info, {}) ext msg_item.get(app_msg_ext_info, {}) if not ext: return [] articles [] articles.append({ msg_id: base.get(id), publish_time: base.get(datetime), type: base.get(type), title: ext.get(title), digest: ext.get(digest), content_url: ext.get(content_url, ).replace(\\/, /), cover: ext.get(cover), author: ext.get(author), }) for sub in ext.get(multi_app_msg_item_list, []): articles.append({ msg_id: base.get(id), publish_time: base.get(datetime), type: base.get(type), title: sub.get(title), digest: sub.get(digest), content_url: sub.get(content_url, ).replace(\\/, /), cover: sub.get(cover), author: sub.get(author), }) return articles注意content_url里有很多转义斜杠\/不处理的话直接拼链接会出错。我统一做了一次replace(\\/, /)实测最省事。接下来是拿阅读量和点赞量。先解析文章链接里的sn再去调统计接口def get_article_stat(content_url): query urlparse(content_url).query params parse_qs(query) sn params.get(sn, [])[0] if not sn: return {} api https://mp.weixin.qq.com/mp/getappmsgext payload { __biz: BIZ, appmsg_type: 9, idx: 1, sn: sn, f: json, key: KEY, pass_ticket: PASS_TICKET, wxtoken: , } resp requests.get(api, paramspayload, headersHEADERS, timeout15) data resp.json() if data.get(appmsgstat): return data[appmsgstat] return {}appmsgstat里通常包含read_num、like_num、old_like_num、real_read_num等字段直接落到库里就行。如果返回错误的ret就把当前sn记录到一个文本文件里后续再重试。3.3 翻页、去重与断点续传历史文章列表的翻页其实就是一个offset累加的过程但直接无脑累加会有两个问题重复数据和中间中断。我的做法是维护一张article_list表用文章链接的唯一性做去重每篇文章入库前先查一次是否已存在同时把每个公众号的当前offset单独存一个字段每次任务启动时先读上次的offset这样即使中途断网或者超时也能接着跑不用从头翻。翻页终止条件也需要注意get_history返回空列表或者ret不等于 0就结束当前公众号的抓取。我个人还会加一个保险判断——如果连续三次返回空列表强制结束防止极端情况下接口异常导致的死循环。给一个完整的调度逻辑伪代码offset get_last_offset(BIZ) while True: msg_list get_history(offsetoffset) if not msg_list: break for msg in msg_list: articles parse_articles_from_msg(msg) for art in articles: save_to_db(art) time.sleep(random.uniform(1, 3)) offset len(msg_list) save_offset(BIZ, offset) time.sleep(random.uniform(2, 5))随机延迟我写进了循环里不是为了装模作样是真的能大幅降低风控命中概率。早期不加延迟几乎每次跑到两三百条就被限制。4. 常见问题与排查技巧实录4.1 高频报错对照表跑过这个项目的人基本都会遇到下面几个报错。我整理成一个速查表遇到问题直接对着排查报错信息出现原因解决办法invalid key/key expiredkey 或 pass_ticket 过期重新抓包获取最新的 key 和 pass_ticket替换后重试freq control/ret: 200003请求频率过高触发风控增加随机延迟降低并发暂停一段时间再跑链接内容不属于当前公众号content_url 里的 __biz 与当前配置不匹配核对文章 URL 中的 __biz确认抓取的是该公众号下的文章阅读量和点赞量返回 -1统计接口被风控或条件不满足清理本地代理缓存更换新 key降低请求频率评论接口返回空文章未开启评论或 comment_id 获取失败确认页面中能否看到评论重新从 HTML 解析 comment_id正文提取为空请求文章页面时被重定向或需要验证补全 headers 里的 Referer、User-Agent重新请求这里面最坑的是“链接内容不属于当前公众号”。这个报错我刚开始以为是代码问题排查半天最后发现是某篇文章的content_url里带的__biz是另一个公众号的值。出现这种情况多半是历史列表里混入了一篇转载文章或者广告投放文章。我的处理方式是在解析函数里加一道校验只有__biz和当前配置一致的文章才入库不一致的直接丢弃并记录日志不阻塞整个抓取流程。4.2 降频与合规经验风控这块是绕不过去的分享几个实测下来有效的降频手段。请求节奏上每篇文章详情页请求之间随机延迟 1 到 3 秒每翻一页延迟 2 到 5 秒单公众号单轮任务最多抓 500 篇就主动暂停等 15 分钟再继续。如果同时跑多个公众号我采用串行处理而不是开多个线程同时抓因为一旦并发PC 微信端会立刻检测到异常流量key 失效得极快。再就是账号管理。我建议用一个日常使用频率低、没有敏感操作的微信号来做这件事不要用主号。每次抓取前先手动打开几个公众号文章让微信客户端“热身”再开始跑任务实测能降低很多无谓的风控拦截。合规方面我的原则只有一条只抓取用于个人研究和学习的数据不对外传播、不洗稿、不用于任何商业变现。微信公众平台和各个公众号的资源都属于内容生产者的劳动成果爬虫只是数据获取工具最终怎么使用决定权在每个人自己手上。5. 从单号抓取到矩阵监控5.1 接入定时任务与异常通知跑通单号抓取后下一步自然是让它每天自动执行。我的方案是在服务器上用系统计划任务每天凌晨定时跑一次跑完把新增文章写入数据库再统计出前一天的阅读增量。定时任务最怕的就是静默失败所以我加了一个异常通知脚本运行结束时如果任务过程中有持续的错误日志或者是完全没有抓取到新数据就通过企业微信机器人推一条告警到群里。具体实现不复杂def send_alert(msg): webhook 你的企业微信机器人webhook payload {msgtype: text, text: {content: msg}} requests.post(webhook, jsonpayload, timeout10)这个动作看着小但非常实用。没有告警之前我曾经连续三天没发现问题数据库里全是旧数据等于白跑。5.2 多公众号统一管理单号稳定之后我把配置改成了列表驱动把二十几个公众号的__biz和对应的抓取状态统一放在一个 JSON 配置文件里循环遍历。每个公众号维护独立的offset和上次抓取时间互不干扰。这里会牵扯到一点小技巧多个公众号的 key 其实是同一个微信登录账号的 key所以只要在一个公共模块里统一管理 key 和 pass_ticket所有公众号共用即可。一旦 key 过期只需要重新抓一次包全局更新。如果某个公众号接口返回invalid key不影响其他公众号继续抓因为每个公众号都是独立循环报错时直接把当前公众号标记为“待重试”跳过去处理下一个。这样做到后面每天的数据自动汇总我只需要在周报里看一眼阅读曲线就能判断内容走向。整个项目从纯手工到全自动我个人的体感是省下的不是一点点时间而是把精力真正还给了内容分析本身。最后分享一个我踩过多次坑之后总结出来的经验这类抓取任务最重要的不是代码写得有多花哨而是节奏控制。一次抓多少、间隔多久、什么时候停这些细节决定了你能跑多久不出问题。先小批量调通流程再逐步加量远远好过一上来就开全速。我后期跑得最稳的配置反而是所有策略里最保守的那套。本文还有配套的精品资源点击获取
返回列表