1. 项目缘起为什么我们需要一个公众号数据抓取脚本做内容运营、市场分析或者竞品研究的朋友十有八九都动过这个念头要是能把某个公众号的历史文章一股脑儿全扒下来就好了。无论是想分析某个大号的爆款规律还是想给自己的内容库做个备份手动一篇篇点开、复制、粘贴效率低到令人发指而且一旦遇到文章被删除或者账号迁移数据就彻底丢了。市面上虽然有一些现成的工具但要么收费不菲要么功能受限要么就是操作复杂对非技术背景的同事不够友好。我自己就经常遇到这种需求。团队需要分析行业头部账号的内容策略老板一句话“把这个号近三年的文章都拉出来看看”如果手动操作这工作量简直不敢想。于是我就琢磨着自己写一个脚本。我的核心诉求很简单一键、全自动、稳定、数据全。所谓“一键”就是输入目标公众号的名称或者ID脚本就能自动运行把文章列表、标题、发布时间、阅读数如果能获取到、点赞数、摘要、原文链接甚至正文内容都给我抓取下来并规整地保存成结构化的数据比如Excel或者数据库。经过一段时间的折腾和优化这个脚本终于能比较稳定地跑起来了今天就来和大家分享一下我的实现思路、踩过的坑以及一些实用的技巧。2. 核心思路与方案选型不走寻常路的抓取策略在动手之前我们必须先搞清楚微信公众号数据的获取途径。直接去爬取微信公众平台的网页是条死路因为它的反爬机制非常严格页面结构复杂且动态加载登录态也难以维持。经过一番调研和测试我总结出几条可行的路径并最终选择了其中一条相对稳定和高效的路。2.1 主流抓取路径分析目前获取公众号历史文章数据主要有以下几种方式各有优劣通过搜狗微信搜索这是最古老的方法。搜狗有一个专门的微信搜索频道可以搜索到公众号和文章。但问题很多首先搜狗的数据本身就不全很多新号或者文章可能搜不到其次搜狗页面也有反爬需要处理验证码和频率限制最重要的是它无法获取到阅读数、点赞数等关键互动数据数据维度不全。通过微信公众号手机端接口逆向这是目前最主流、数据最全的方法。微信手机客户端App在查看公众号历史消息列表和文章详情时会调用后台的API接口。如果我们能模拟手机端的请求就能拿到最原始、最完整的数据包括阅读量、点赞量、评论等。这需要抓包分析App的请求找到关键的接口和参数。这种方法技术门槛较高但一旦跑通效果最好。通过第三方聚合平台或工具一些数据公司提供了付费的API服务可以直接调用获取公众号数据。这对于企业级应用且预算充足的情况是很好的选择但对于个人或小团队来说成本是首要考虑因素。通过微信公众号后台仅限自己管理的号如果你本身就是公众号的管理员那么后台提供了最全面的数据导出功能。但这显然不适用于抓取别人的公众号。综合来看逆向手机端API接口是平衡了数据完整性、稳定性和自主可控性的最佳选择。我的脚本也正是基于这个思路构建的。2.2 技术栈选择Python生态是首选确定了抓取路径接下来是技术选型。对于这类自动化抓取任务Python几乎是唯一的选择因为它有极其丰富的网络请求和HTML解析库。请求库requests和httpx。requests简单易用是标准选择。但在处理大量异步请求时我选择了httpx因为它同时支持同步和异步客户端性能更好能显著提升抓取大量文章详情页的速度。解析库BeautifulSoup4和lxml。BeautifulSoup4解析HTML非常友好写起来快。但lxml的解析速度更快在需要处理成千上万页面时优势明显。我的脚本里混合使用列表页用BeautifulSoup快速开发详情页解析用lxml提升效率。数据存储pandasSQLite/MySQL。pandas是数据处理的瑞士军刀抓下来的数据先用它清洗、整理然后可以轻松导出为Excel.xlsx或CSV文件。对于需要长期存储或关系查询的数据我会选择存入SQLite轻量级单文件或MySQL。任务调度与异步asyncioaiohttp。为了实现“一键”和高效异步IO是关键。使用asyncio和aiohttp可以同时发起数十上百个网络请求去获取文章详情将原本需要数小时的串行抓取任务缩短到几分钟内完成。模拟请求关键fake-useragent和 代理IP池。为了避免被目标服务器封禁IP随机生成User-Agent和使用代理IP是必须的。fake-useragent库可以方便地生成各种浏览器的随机UA。对于代理IP可以根据实际情况使用付费服务或维护一个免费IP池稳定性较差。注意逆向和抓取他人公开数据需注意法律风险与平台规则。本脚本及分享仅用于技术学习与个人已授权数据的备份请勿用于侵犯他人权益、商业爬取或对目标服务器造成压力的场景。务必设置合理的请求间隔如每次请求间休眠1-3秒体现技术人的操守。3. 实操拆解从零构建抓取脚本的关键步骤下面我将分步拆解整个脚本的核心模块。为了清晰我会用伪代码和关键代码片段来说明思路你完全可以基于此骨架填充实现。3.1 第一步获取目标公众号的唯一标识biz或__biz这是整个流程的起点也是最棘手的一步之一。微信公众号没有一个像微博ID那样显而易见的、固定的公开标识。我们通常需要找到一个包含__biz参数的URL。这个__biz值就是公众号在微信系统中的唯一ID。如何获取从历史文章链接中提取在微信里打开任意一篇该公众号的历史文章点击右上角“...”选择“复制链接”。你会得到一个长链接其中包含__bizXXXXXX这样的参数。这个XXXXXX就是我们要的。通过公众号名片页在微信搜索公众号进入其主页分享名片到文件传输助手然后在电脑浏览器中打开这个名片页查看页面源代码搜索__biz也能找到。通过第三方工具查询有些网站提供了通过公众号名称查询__biz的服务但可靠性需要自行甄别。拿到__biz后我们就有了叩开数据大门的钥匙。import re def extract_biz_from_url(url): 从公众号文章URL中提取__biz参数 pattern r‘__biz([A-Za-z0-9_-])’ match re.search(pattern, url) if match: return match.group(1) else: raise ValueError(f“无法从URL中提取__biz参数: {url}”) # 示例用法 sample_url “https://mp.weixin.qq.com/s?__bizMzA5NDk4MDA1MAmid2641234567idx1snxxxxx” biz extract_biz_from_url(sample_url) print(f“提取的biz为: {biz}”)3.2 第二步模拟请求获取历史文章列表有了__biz我们就可以构造请求获取公众号的历史消息列表。这里我们模拟的是微信手机端翻看历史消息的动作。关键接口是https://mp.weixin.qq.com/mp/profile_ext。这是一个GET请求需要携带一系列参数其中最重要的除了__biz还有action: 操作类型getmsg表示获取消息。offset: 偏移量用于分页。微信一次返回10条文章信息offset控制从第几条开始获取。count: 每次获取的数量通常是10。uin/key/pass_ticket: 这些是登录态或令牌参数是难点所在。它们通常包含在公众号主页的页面源代码中或者需要通过更底层的接口获取。一种常见做法是先请求一次公众号主页https://mp.weixin.qq.com/mp/profile_ext?actionhome__biz...从返回的HTML中解析出这些动态生成的令牌。import requests from bs4 import BeautifulSoup import json import time def get_history_list(biz, offset0, count10): 获取一页历史文章列表 base_url “https://mp.weixin.qq.com/mp/profile_ext” # 注意以下参数需要从主页HTML中动态获取此处为示例 params { ‘action’: ‘getmsg’, ‘__biz’: biz, ‘offset’: offset, ‘count’: count, ‘uin’: ‘MTIzNDU2Nzg’, # 示例需动态获取 ‘key’: ‘abcdef123456’, # 示例需动态获取 ‘pass_ticket’: ‘xyz789’, # 示例需动态获取 ‘appmsg_token’: ‘token_here’, # 可能需要的令牌 ‘f’: ‘json’, ‘scene’: ‘124’, } headers { ‘User-Agent’: ‘Mozilla/5.0 (iPhone; CPU iPhone OS 14_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148 MicroMessenger/8.0.10(0x18000a2a) NetType/WIFI Language/zh_CN’, ‘Referer’: f‘https://mp.weixin.qq.com/mp/profile_ext?actionhome__biz{biz}’, } try: resp requests.get(base_url, paramsparams, headersheaders, timeout10) resp.raise_for_status() data resp.json() # 解析返回的JSON提取文章基本信息 if data.get(‘ret’) 0: article_list data.get(‘general_msg_list’, {}).get(‘list’, []) return parse_article_list(article_list) else: print(f“请求失败返回码: {data.get(‘ret’)}”) return [] except Exception as e: print(f“获取历史列表出错: {e}”) return [] def parse_article_list(msg_list): 解析JSON中的文章列表信息 articles [] for item in msg_list: app_msg item.get(‘app_msg_ext_info’) if app_msg: # 主文章 articles.append({ ‘title’: app_msg.get(‘title’), ‘link’: app_msg.get(‘content_url’), ‘digest’: app_msg.get(‘digest’), ‘cover’: app_msg.get(‘cover’), ‘create_time’: app_msg.get(‘create_time’), ‘article_id’: app_msg.get(‘aid’), }) # 同一推送下的多篇文章副条 multi_app_msg app_msg.get(‘multi_app_msg_item_list’, []) for sub in multi_app_msg: articles.append({ ‘title’: sub.get(‘title’), ‘link’: sub.get(‘content_url’), ‘digest’: sub.get(‘digest’), ‘cover’: sub.get(‘cover’), ‘create_time’: app_msg.get(‘create_time’), # 通常与主文相同 ‘article_id’: sub.get(‘aid’), }) return articles实操心得一令牌token的获取与维持脚本稳定性的核心在于如何稳定获取uin、key、pass_ticket等参数。我的经验是不能硬编码。必须在每次启动抓取任务前先模拟访问一次公众号主页actionhome用BeautifulSoup或正则表达式从返回的HTML中提取这些动态生成的参数。有时这些参数的有效期很短如果抓取过程中中断可能需要重新获取。可以考虑将获取令牌的逻辑封装成一个独立函数并在检测到请求失败如返回ret非0时自动重新获取令牌并重试。3.3 第三步异步抓取文章详情与关键数据获取到文章列表后我们得到了每篇文章的content_url内容链接。但这个链接是经过加密的需要解密才能得到真实的文章地址。解密算法是固定的通常是Base64解码后与一个固定字符串进行异或运算。网上有公开的解密函数这里不展开。得到真实URL后就可以抓取文章详情页了。这里我强烈推荐使用异步IO因为文章数量可能成百上千串行抓取太慢。import asyncio import aiohttp from lxml import etree async def fetch_article_detail(session, article_info, semaphore): 异步获取单篇文章详情 async with semaphore: # 使用信号量控制并发数避免被封 url decrypt_content_url(article_info[‘link’]) # 先解密URL if not url.startswith(‘http’): url ‘https://mp.weixin.qq.com’ url try: async with session.get(url, timeout10) as response: html await response.text() # 使用lxml解析效率高 tree etree.HTML(html) # 提取阅读数、点赞数、评论数等 # 这些数据通常藏在特定的js变量或标签属性里 read_num extract_read_count(tree) like_num extract_like_count(tree) comment_data extract_comment_info(tree) # 提取正文 content_div tree.xpath(‘//div[id“js_content”]’)[0] # 清理正文中的样式、脚本等无关标签 cleaned_content clean_html_content(content_div) article_info[‘read_num’] read_num article_info[‘like_num’] like_num article_info[‘comment_info’] comment_data article_info[‘content’] cleaned_content article_info[‘detail_url’] url print(f“已抓取: {article_info[‘title’]}”) return article_info except Exception as e: print(f“抓取文章失败 {article_info[‘title’]}: {e}”) article_info[‘error’] str(e) return article_info async def batch_fetch_details(article_list): 批量异步抓取文章详情 connector aiohttp.TCPConnector(limit20, sslFalse) # 限制连接数 timeout aiohttp.ClientTimeout(total30) semaphore asyncio.Semaphore(10) # 控制最大并发数为10 async with aiohttp.ClientSession(connectorconnector, timeouttimeout) as session: tasks [] for article in article_list: task asyncio.create_task(fetch_article_detail(session, article, semaphore)) tasks.append(task) results await asyncio.gather(*tasks, return_exceptionsTrue) # 过滤掉异常结果 valid_results [r for r in results if isinstance(r, dict)] return valid_results实操心得二解析细节与反爬应对文章详情页的数据提取是另一个难点。阅读数、点赞数并非直接写在HTML里而是通过JavaScript动态加载的。你需要分析页面源代码找到包含这些数据的JS变量或特定的接口。例如阅读数可能在一个名为var msg_read_num 10000;的变量里或者在一个window.__stat的对象里。使用正则表达式或简单的字符串查找可以提取出来。此外微信页面结构可能变动所以解析规则要有一定的容错性最好能定期检查。对于反爬除了控制并发、使用代理IP和随机UA外在请求头中正确设置Referer通常为公众号主页或历史列表页和Cookie如果维持了会话也非常重要。3.4 第四步数据清洗、存储与导出所有数据抓取完毕后我们得到的是一个字典列表。接下来就是用pandas进行清洗和整理。import pandas as pd from datetime import datetime def save_to_excel(articles_data, filename“公众号数据.xlsx”): 将抓取的数据保存到Excel df pd.DataFrame(articles_data) # 数据清洗 # 1. 处理时间戳 if ‘create_time’ in df.columns: df[‘publish_date’] df[‘create_time’].apply(lambda x: datetime.fromtimestamp(x) if x else None) # 2. 确保阅读数、点赞数为数值类型 for col in [‘read_num’, ‘like_num’]: if col in df.columns: df[col] pd.to_numeric(df[col], errors‘coerce’).fillna(0).astype(int) # 3. 选择要保存的列并排序 columns_to_save [‘title’, ‘publish_date’, ‘read_num’, ‘like_num’, ‘digest’, ‘detail_url’] # 如果抓取了正文可以单独存一个文件因为正文可能很长 df_text df[[‘title’, ‘publish_date’, ‘content’]].copy() if ‘content’ in df.columns else None df[columns_to_save].to_excel(filename, indexFalse) print(f“数据已保存到 {filename}”) if df_text is not None: text_filename filename.replace(‘.xlsx’, ‘_正文.xlsx’) df_text.to_excel(text_filename, indexFalse) print(f“正文内容已保存到 {text_filename}”) return df除了Excel你也可以选择存入数据库便于后续的复杂查询和分析。使用sqlalchemy库可以轻松地将DataFrame写入SQLite或MySQL。4. 一键整合与调度让脚本真正“一键”运行将上述模块组合起来并添加一些用户交互和错误处理就构成了我们的主脚本。import asyncio import sys def main(): print(“ 公众号历史数据一键抓取脚本 ”) # 1. 用户输入 biz_input input(“请输入公众号的__biz参数或包含该参数的文章链接: “).strip() biz extract_biz_from_url(biz_input) if ‘http’ in biz_input else biz_input max_pages input(“请输入要抓取的页数每页约10篇文章: “).strip() try: max_pages int(max_pages) except: max_pages 10 print(f“输入无效默认抓取{max_pages}页”) # 2. 获取令牌这里简化实际需从主页获取 token_params get_token_params(biz) if not token_params: print(“获取令牌失败请检查网络或__biz参数是否正确。”) return all_articles [] offset 0 # 3. 分页获取文章列表 for page in range(max_pages): print(f“正在获取第 {page1} 页列表...”) articles get_history_list(biz, offsetoffset, count10, **token_params) if not articles: print(“没有更多文章或获取失败。”) break all_articles.extend(articles) offset len(articles) time.sleep(2) # 列表页请求间隔 print(f“共获取到 {len(all_articles)} 篇文章列表。”) if not all_articles: print(“未获取到任何文章程序结束。”) return # 4. 异步抓取文章详情 print(“开始异步抓取文章详情请稍候...”) loop asyncio.new_event_loop() asyncio.set_event_loop(loop) detailed_articles loop.run_until_complete(batch_fetch_details(all_articles)) loop.close() # 5. 保存数据 df save_to_excel(detailed_articles, f“{biz[:10]}_data.xlsx”) # 6. 简单统计 print(“\n抓取完成简单统计”) print(f“文章总数: {len(df)}”) if ‘read_num’ in df.columns: print(f“总阅读量: {df[‘read_num’].sum():,}”) print(f“平均阅读量: {df[‘read_num’].mean():,.0f}”) print(f“数据文件: {biz[:10]}_data.xlsx”) if __name__ “__main__”: main()这个主函数流程清晰输入参数 - 获取令牌 - 循环翻页抓列表 - 异步批量抓详情 - 保存数据。你可以将其打包成命令行工具或者加上简单的GUI比如用tkinter使其对非技术人员更友好。5. 常见问题与避坑指南实录在实际开发和运行过程中我遇到了不少坑。这里总结一下希望能帮你节省时间。5.1 请求频繁被限制或封禁这是最常见的问题。微信的反爬策略会检测异常流量。症状返回ret值为非0如 -1或请求直接超时或返回验证页面。解决方案降低请求频率在每次请求无论是列表页还是详情页之间加入随机延时例如time.sleep(random.uniform(2, 5))。列表页请求间隔可以更长一些。使用高质量代理IP这是应对IP封禁最有效的方法。建议使用付费的代理IP服务它们通常提供高匿、稳定的IP池并支持自动切换。在requests或aiohttp的请求中设置proxies参数即可。模拟真实用户行为除了随机UA还可以随机化请求间隔模拟人的阅读速度。在抓取详情页时不要一次性发起几百个并发用asyncio.Semaphore将并发数控制在10-20以内。维护Cookie池如果可能尝试获取并维护有效的登录态Cookie但这部分难度较大。5.2 无法获取阅读数、点赞数等数据症状抓取到的文章详情页HTML中找不到阅读数等元素。原因与解决页面结构变化微信前端可能改版。需要重新分析页面找到新的数据存放位置。通常数据在script标签的JavaScript变量中。用浏览器的开发者工具在“网络”选项卡中筛选XHR/Fetch请求有时这些数据是通过额外接口异步加载的你需要找到并模拟那个接口。数据需要授权某些公众号的详细数据如阅读数来源分布可能只有管理员可见。公开页面只能看到基础阅读数和点赞数。如果连基础数据都没有可能是该公众号选择了不公开或者你的请求缺少了必要的参数如appmsg_token。5.3 抓取到的文章链接是加密的content_url症状从列表接口获取的content_url是一串看似乱码的字符串无法直接访问。解决这是微信的固定加密方式。你需要一个固定的解密函数。这个函数通常是Base64解码后与一个固定的密钥如‘https://mp.weixin.qq.com/s?‘的某部分进行逐字节异或操作。这个解密算法在网上是公开的搜索“微信公众号 content_url 解密”就能找到多种语言的实现。务必确保你使用的解密函数是正确的否则会得到错误的链接。5.4 异步抓取时出现大量错误或数据丢失症状使用asyncio并发抓取详情时部分任务失败返回None或异常。解决方案做好异常捕获就像示例代码中那样在每个异步任务内部用try...except包裹即使出错也不影响其他任务并将错误信息记录到结果中。使用return_exceptionsTrue在asyncio.gather中设置此参数让异常作为结果返回而不是直接抛出中断整个程序。实现重试机制对于失败的请求可以将其放入一个重试队列稍后重新尝试。可以定义一个简单的装饰器或函数来实现带指数退避的重试逻辑。控制并发量过高的并发会导致本地端口耗尽或服务器拒绝。通过asyncio.Semaphore严格控制并发数并根据网络情况和目标服务器响应动态调整。5.5 数据存储混乱或格式错误症状导出的Excel文件乱码、日期格式不对、数字被当成字符串。解决编码问题确保在读写文件时指定编码为utf-8-sigExcel兼容的UTF-8带BOM格式。数据类型使用pandas的pd.to_numeric、pd.to_datetime函数进行强制类型转换并处理转换错误errors‘coerce’。清理HTML正文内容包含大量HTML标签和样式直接存入Excel会很乱。可以使用BeautifulSoup的get_text()方法提取纯文本或者用lxml的xpath(‘string(.)’)。如果希望保留部分格式如加粗可以编写自定义的清理函数。最后这个脚本是一个强大的工具但能力越大责任越大。请务必遵守robots.txt虽然微信可能没有明确的以合理的频率抓取不要对服务器造成负担。将抓取的数据用于个人学习、分析或已授权的内容备份远离任何可能侵权的商业用途。技术本身无罪关键在于使用它的人。希望这个分享能帮你打开一扇窗更高效地处理信息而不是带来麻烦。