ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

公众号历史文章爬虫实战:接口分析、参数解析与Python实现全指南

公众号历史文章爬虫实战:接口分析、参数解析与Python实现全指南 很多人一开始以为“公众号的文章抓不到”因为微信的生态比较封闭没有公开的API网页端也没有直接的列表页。但真做起来你会发现核心就一句话公众号的历史文章是通过一个内部接口分页拉取的只要拿到链接里的几个关键参数就能用Python把所有历史文章一页一页抓下来。这篇文章我就把我自己从零到一实现这个爬虫的完整过程写出来包括接口分析、参数解析、代码实现、并发设计、防封策略以及我真实踩过的坑。内容基于我多年写爬虫的实际经验每一步都会讲清楚为什么这么做适合刚入门Python爬虫的读者也适合想快速实现公众号数据归档的老手。1. 一篇文章是怎么从链接到数据的历史文章抓取链路在动手写代码之前必须先搞清楚一件事浏览器里打开一篇公众号文章和爬虫去抓取一篇公众号文章走的路径完全不同。1.1 从一条文章链接说起随便打开一篇公众号文章链接长这样https://mp.weixin.qq.com/s/xxxxxxxxxxxxxxxx短链接形式重定向后会变成类似这样的长链接https://mp.weixin.qq.com/s?__bizMzA3MzI4MjgzMwmid2650821234idx1sn9c2f9d8e7b6a5c4d3e2f1a0b9c8d7e6f注意这里的__biz和mid两个参数。__biz是公众号的唯一标识相当于公众号的身份证号所有的文章链接里都带它。mid是文章消息的ID可以理解为这一批推送消息的编号。但靠单篇文章链接是拿不到历史列表的。你想爬“所有历史文章”需要一个能按页返回文章列表的接口而不是一篇一篇去搜集链接。1.2 历史文章列表接口在哪里公众号网页版有个隐藏功能点击公众号主页的“历史消息”按钮会加载出一个列表这个列表背后实际上调用了这样一个接口https://mp.weixin.qq.com/mp/profile_ext?actiongetmsg__bizxxxxxxxxxxfjsonoffset0count10is_ok1scene126wxtokenappmsg_tokenxxxxxxxx这个接口返回的是JSON数据里面有一个app_msg_list数组就是当前页的文章列表还有can_msg_continue是否还有下一页和next_offset下一页的偏移量这两个字段。所以整个抓取逻辑就清晰了得到目标公众号的__biz。带着必要的Cookie和appmsg_token请求getmsg接口。解析JSON里的app_msg_list拿到本页文章。用next_offset更新偏移量继续请求下一页直到can_msg_continue为0。这个流程理解透了后面所有代码都是在围绕这四步做文章。1.3 为什么不能直接用requests打开文章链接有的朋友刚上手就会写import requests url https://mp.weixin.qq.com/s/xxxxxxxx resp requests.get(url) print(resp.text)这样确实能拿到HTML但拿到的是一个不完整的框架真正的正文内容是在JS渲染之后才出现的直接抓取你会得到一堆空壳div。更关键的是微信对单篇文章抓取有频率限制连续请求几十次就可能弹出环境异常。所以我们抓取的主路径应该是接口而不是HTML页面接口返回的是结构化JSON省去了解析HTML的麻烦频率控制上也更好把握。2. 从零开始的完整实现参数获取与核心代码搞清楚了链路下一步就是把每个环节落地。我先说参数怎么拿这部分很多人卡了很久。2.1 参数获取__biz 和 appmsg_token__biz获取方式很简单在微信里打开任意一篇目标公众号的文章点击右上角“...”选择“复制链接”粘贴出来后就能看到__biz后面的值。或者直接在微信内打开公众号主页再点历史消息从浏览器开发者工具里找到getmsg请求URL里就有。appmsg_token就没那么直白了。这个参数是微信服务器下发的临时令牌在打开“历史消息”页面时自动生成并绑定当前登录的微信身份。所以只拿到__biz不够还必须拿到一个合法的appmsg_token和对应的Cookie。我的做法是在PC端微信里打开公众号的历史消息按F12打开开发者工具切到Network面板刷新历史消息页面找到getmsg这个请求把请求头里的Cookie和请求URL里的appmsg_token复制出来。这里有个细节Cookie里的slave_sid和slave_user两个值最关键它们代表的是你的微信身份凭证。整个抓取过程需要一直带着这组Cookie一旦失效通常几小时到一天接口就会返回错误。2.2 核心抓取函数拿到参数之后核心代码其实很简短。我直接把可用的版本贴出来基于requests库import requests import time import json from urllib.parse import urlencode class WeChatSpider: def __init__(self, biz, cookie, token): self.biz biz self.cookie cookie self.token token self.base_url https://mp.weixin.qq.com/mp/profile_ext self.headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36, Cookie: self.cookie, Referer: fhttps://mp.weixin.qq.com/mp/profile_ext?actionhome__biz{self.biz}scene124, } def get_articles(self, offset0, count10): params { action: getmsg, __biz: self.biz, f: json, offset: offset, count: count, is_ok: 1, scene: 126, wxtoken: , appmsg_token: self.token, } url self.base_url ? urlencode(params) resp requests.get(url, headersself.headers, timeout10) data resp.json() if data.get(base_resp, {}).get(ret) ! 0: raise Exception(f接口返回错误: {data.get(base_resp)}) return data def fetch_all(self, max_pagesNone): offset 0 page 0 all_articles [] while True: data self.get_articles(offsetoffset) msg_list data.get(app_msg_list, []) if not msg_list: break all_articles.extend(msg_list) page 1 offset data.get(next_offset, offset) can_continue data.get(can_msg_continue, 0) print(f第 {page} 页累计 {len(all_articles)} 篇) if not can_continue or (max_pages and page max_pages): break time.sleep(1) return all_articles这段代码做了一件事循环调用getmsg接口每次用上一次返回的next_offset偏移量翻页直到接口告诉你“没有下一页了”。2.3 关于 count 参数的选择count是每次请求返回的条数理论上最大可以调到10。我试过调大比如count20发现接口并不会返回那么多最多就是10条。所以不要在这个参数上花心思老老实实count10通过降低翻页次数来减少请求频率反而更稳妥。另外接口返回的JSON里每篇文章包含这些核心字段字段含义aid文章唯一ID用于详情页跳转appmsgid公众号后台文章IDtitle文章标题digest摘要link文章链接update_time发布时间Unix时间戳cover封面图URLis_pay是否付费文章其中link就是我们可以用来抓全文的链接update_time可以用于排序和增量更新。3. 文章正文抓取与清洗拿到列表只是第一步列表拿到了不等于“所有文章”都拿到了。你还得进入每一篇文章把正文抓下来存成结构化数据。这一步的坑比接口还多。3.1 正文页面的真实结构前面说过直接用requests.get(link)拿到的HTML不是最终渲染后的结果。文章正文其实藏在idjs_content的div里而且里面的图片是懒加载的——src是一个灰色占位图真正的图片地址在>import re import requests def clean_html(html_content): # 去掉script和style html_content re.sub(rscript.*?/script, , html_content, flagsre.S) html_content re.sub(rstyle.*?/style, , html_content, flagsre.S) # 把data-src替换成src html_content re.sub(rdata-src([^]), rsrc\1, html_content) # 去掉空标签 html_content re.sub(r([a-zA-Z])[^]*\s*/\1, , html_content, flagsre.S) return html_content def get_article_content(link): headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36, } resp requests.get(link, headersheaders, timeout10) match re.search(rdiv[^]*idjs_content[^]*(.*?)/div, resp.text, flagsre.S) if not match: return return clean_html(match.group(1))注意这里的.*?非贪婪匹配有时候会因为js_contentdiv嵌套太多而匹配失败。更稳妥的方案是用BeautifulSoup直接找idjs_content的节点取innerHTML。我后面改进版本里就是这么做的。3.2 第一个容易踩的坑正文被截断微信文章正文很长的时候HTML里的js_contentdiv内会出现一个qqmusic占位、投票组件、小程序卡片等额外内容。如果你只按div标签切分可能把正文内部嵌套的div也一起截断了。我推荐用Python的html.parser或lxml来定位节点而不是纯正则。这里给一个用BeautifulSoup的稳定版本from bs4 import BeautifulSoup def get_content_stable(link): resp requests.get(link, headersheaders, timeout10) soup BeautifulSoup(resp.text, lxml) content_div soup.find(div, idjs_content) if not content_div: return # 把内容里的data-src替换成src for img in content_div.find_all(img): if img.get(data-src): img[src] img[data-src] return str(content_div)3.3 文章链接里的坑有的链接需要二次跳转部分公众号文章是“图文消息”里的分享卡片link字段可能是短链接也可能是http://mp.weixin.qq.com/s?__biz...mid...idx...的长链接。建议在抓正文前加一个allow_redirectsTruerequests默认就是不需要特殊处理。但有一种特殊情况视频消息。如果文章内容是视频号视频link可能指向finder.video.qq.com抓下来正文就是空的。这种情况下我们需要在列表阶段就过滤掉appmsgid异常的记录或者判断正文为空的就跳过。4. 数据存储与增量更新爬全不算完还要能续爬文章列表和正文都拿到手之后一定不要直接存成一个大JSON完事那样后续做检索、去重、增量更新都会很痛苦。4.1 为什么选择SQLite单机爬虫数据量在几万篇以内我最推荐SQLite。理由很简单零配置不用装数据库服务。单文件方便备份和迁移。支持SQL查询后续做筛选、统计很顺手。你需要建两张表articles存文章列表信息article_contents存正文内容或者把正文和列表合在一张表里也行我习惯分开。CREATE TABLE IF NOT EXISTS articles ( appmsgid INTEGER PRIMARY KEY, title TEXT, digest TEXT, link TEXT, update_time INTEGER, cover TEXT, is_pay INTEGER DEFAULT 0, crawled INTEGER DEFAULT 0 ); CREATE TABLE IF NOT EXISTS article_contents ( appmsgid INTEGER PRIMARY KEY, content TEXT, clean_text TEXT, crawl_time INTEGER );4.2 增量更新的思路所谓增量更新就是不要让爬虫每次都从第一页开始重新抓。因为接口返回的文章列表是按时间倒序排列的第一页永远是最新的10篇。如果跑了50页到了第51页发现next_offset对应的文章在数据库里已经存在说明后面的都已经抓过了直接停掉即可。我在代码里加的判断逻辑是这样的def fetch_all_incremental(spider, conn): offset 0 page 0 while True: data spider.get_articles(offsetoffset) msg_list data.get(app_msg_list, []) if not msg_list: break newest_appmsgid msg_list[0][appmsgid] # 如果最新的appmsgid已经在库里说明此前已经爬过了 # 但仍要检查这一页有没有新的 need_break True for item in msg_list: if not article_exists(conn, item[appmsgid]): save_article(conn, item) need_break False if need_break: break page 1 offset data.get(next_offset, offset) can_continue data.get(can_msg_continue, 0) if not can_continue: break time.sleep(0.8)这样设计之后以后每次定时运行爬虫只需要几分钟就能完成一次增量更新不必重新跑全量。5. 并发与频率控制爬得快不是本事爬得稳才算公众号接口对频率非常敏感。很多人写完代码一跑前50页好好的第60页突然返回错误码200013然后被限制访问。这就是频率控制没做好。5.1 翻页接口的限速策略getmsg接口的限速策略我实测下来大致是这样的具体数值可能随时间变化但规律不变单账号连续请求如果间隔小于0.5秒几十次之后就会触发频率限制。如果间隔在1~2秒之间可以连续抓几百页不出问题。如果对方公众号文章特别多比如几千篇一次性抓完可能会触发“操作频繁”的校验。所以我的建议是翻页请求间隔不低于1秒最保守是2秒。一次抓取会话最多500~800页就休息一会儿。抓几千篇文章用2秒间隔大概需要40~60分钟这个时间成本是必须承担的。5.2 并发请求正文的取舍有了文章列表之后抓正文是一个天然的并发场景——文章之间没有依赖关系。很多教程直接上ThreadPoolExecutor开50个线程实际上很容易被微信的风控系统盯上。我踩过的坑是一开始用30个线程并发抓正文结果抓了200篇之后IP直接被限制了所有请求都返回“环境异常”。后来我把并发降到5间隔0.3秒反而稳定跑完了3000多篇。这里的关键是正文页面和接口的限速阈值不一样。接口频率限制很严格但单篇文章页面的限制相对宽松5个线程完全够用。如果你要爬非常多的文章可以自己实现一个简单的“自适应限速器”——当连续出现超时或异常时自动扩大间隔当一切正常时逐步缩小间隔。下面是我改进后的并发抓取正文代码import threading import time import requests from concurrent.futures import ThreadPoolExecutor, as_completed def crawl_content(article): link article[link] for retry in range(3): try: content get_content_stable(link) return article[appmsgid], content except Exception as e: time.sleep(2 * (retry 1)) return article[appmsgid], None def crawl_all_contents(articles, max_workers5): results {} with ThreadPoolExecutor(max_workersmax_workers) as executor: futures {executor.submit(crawl_content, a): a[appmsgid] for a in articles} for future in as_completed(futures): appmsgid, content future.result() results[appmsgid] content return results我用的是ThreadPoolExecutor因为这是标准库不需要额外安装。asynciohttpx异步库其实性能更好但对新手不友好而且微信公众号服务器对异步并发请求的响应不太稳定线程池足够用了。5.3 代理池到底有没有必要说实话对于公众号爬虫来说大多数情况下不需要代理池。因为限制的核心是“微信身份”而非IP换IP不能解决Token失效问题。但有一种情况需要代理你的请求触发了IP层面的临时封禁。这时挂一个干净的住宅代理等IP解封再切换回来比干等效率高。如果你只是个人学习用途建议先用本机IP跑出现限制就等几个小时再跑完全没必要一上来就引入代理那会让复杂度倍增。6. 高频报错排查我遇到的五种典型问题与解决记录这部分是从实际操作中总结的网上很多教程不会写但对新手来说非常关键。6.1 错误码base_resp.ret ! 0接口返回的JSON里base_resp.ret非0最常见的是ret含义解决办法0成功-200013频率限制降低请求频率等待几分钟恢复200003Token无效或过期重新获取appmsg_token0但msg为空Cookie异常检查Cookie里的slave_sid、slave_user-3参数不完整检查__biz是否带了URL编码是否被转义我遇到最多的是200013通常出现在连续请求超过几十次之后。解决办法很简单sleep时间从1秒调到2.5秒等10分钟再继续。6.2appmsg_token莫名其妙就失效了appmsg_token的有效期和你的微信登录状态强绑定。如果你中途用同一台电脑的微信客户端点击了其他操作或者重新登录了微信旧Token立即失效。我一开始很困惑明明什么都没变怎么Token说失效就失效。后来发现是因为我在测试时频繁切换微信账号。建议一个爬虫任务对应一个微信登录状态不要在抓取过程中切换账号。6.3 翻页翻到一半next_offset不涨了接口正常返回can_msg_continue1但next_offset和上一次一样导致死循环。这种情况我遇到过一次是公众号删除了部分文章导致服务端返回的偏移量异常。解决办法是在代码里加一个保护判断如果连续3次next_offset相同且app_msg_list非空就强制终止。6.4 正文里的图片全都加载不出来前面提到>resp.encoding resp.apparent_encoding7. 进阶扩展从“能跑”到“好用”的几个优化方向抓完所有历史文章之后你会发现原始数据离“好用”还差很远。下面这几个优化是我在实际项目中都会做的你可以根据自己的需求取舍。7.1 提取纯文本而不是HTML如果只是做全文检索或数据分析HTML里的标签全是噪音。用BeautifulSoup提取正文div之后再调用.get_text()把纯文本拿出来存到article_contents.clean_text字段里。这一步能让后续搜索、词频统计的代码简单非常多。7.2 按日期归档与统计公众号文章列表里有update_time可以按年/月分组统计发布频率找出这个号的更新规律。这块用SQL直接搞定SELECT strftime(%Y-%m, datetime(update_time, unixepoch, 8 hours)) AS month, COUNT(*) AS cnt FROM articles GROUP BY month ORDER BY month DESC;如果后续有做数据分析的需求这个表结构非常方便。7.3 配合定时任务做长期监控增量更新的代码跑通之后配合系统的cron或Windows任务计划程序每天凌晨2点自动运行一次爬虫就能实现“公众号更新后自动抓取”。这是很多“公众号监控”类工具的实现基础。我自己的一个实际项目就是定时抓取几个行业头部公众号的更新然后自动汇总成当天的行业早报邮件。整个系统中爬虫部分占用的代码量很少核心价值全在增量更新逻辑和稳定的频率控制上。7.4 文章去重与更新检测有些公众号会删除旧文章或者修改已发布文章。增量更新时你可以把当前的title和数据库里的title对比如果标题变了说明文章被修改过了可以重新抓取正文。这个功能对于“追踪文章变化”的场景很有用实现也不复杂SELECT appmsgid, title FROM articles WHERE title ! ?;写在最后的一点个人经验公众号历史文章爬虫这个需求我前后做了好几个版本最大的体会是这套系统的难点从来不是Python代码本身而是对接口机制的理解和对频率控制的敬畏。很多人一上来就追求并发拉满、效率最高结果没跑几百篇就触发了限制反而耽误整体时间。我自己的稳定配置是翻页接口间隔1.5秒正文抓取5线程每300篇暂停30秒。这样跑一个几千篇文章的公众号通常几个小时能完成全程不需要人工干预。如果你只是自己用比如备份某个技术公众号的学习资料、做某领域的内容归档这个配置足够稳定了。最后再多说一句很多朋友问我要不要配置代理IP池。我的答案很直接个人学习场景别碰代理纯属给自己添乱企业级的批量采集场景代理是标配但那时候你需要的也不只是爬虫了而是一整套采集调度系统。从这篇文章的定位出发把单机版跑稳、跑透远比追求那些花哨的高级功能有价值。
返回列表