简介:这是一套面向开发者与新媒体运营人员的微信公众号文章批量采集与归档工具源码,解决官方接口受限下对历史文章、阅读量、评论等数据的合规化存档需求。资源包共109个文件,含34个TypeScript核心逻辑文件、24个Vue前端组件、29张PNG图标与界面素材,以及配置类JSON、样式CSS、构建脚本JS等,完整覆盖前后端与部署能力,压缩后仅12.02MB,轻量易部署。已有717人学习下载,适用于内容合规审计、自媒体知识库建设、竞品分析及私有化数据中台搭建等场景。源码支持HTML格式100%还原原文排版(含内嵌图片与样式),提供Docker一键部署方案、订阅式自动抓取机制及开放API接口,同时集成合集下载、视频/图文消息解析、评论与转发量导出等功能,结构清晰、模块解耦,便于二次开发与定制化扩展。
1. 项目概述与核心价值
最近在整理一些行业资料时,发现很多有价值的深度分析都沉淀在各大微信公众号里。想系统性地保存下来离线阅读,或者做个本地知识库,手动一篇篇复制粘贴效率太低,而且格式容易乱。市面上的一些在线工具要么收费,要么限制多多,用起来总是不顺手。于是,我花时间研究并实现了一个基于Python的微信公众号文章批量下载工具,并决定把完整的源码和实现思路分享出来。这个工具的核心目标很明确:给你一个公众号的名称或ID,它能自动抓取该公众号的历史文章列表,并将文章正文、图片、视频等资源完整地下载到本地,生成结构清晰的HTML或Markdown文件,方便你永久保存和查阅。
这个工具特别适合内容创作者、市场分析师、学术研究者以及任何需要系统性归档网络信息的个人。比如,你可以用它来追踪竞品公众号的更新动态,批量下载某个垂直领域的教程合集建立个人知识库,或者单纯就是收藏自己喜欢的系列文章。整个过程完全自动化,解放双手。接下来,我会详细拆解这个工具的设计思路、关键技术点、具体的实现步骤,以及我在开发过程中踩过的那些坑和总结的实用技巧。你会发现,自己动手实现一个这样的工具,并没有想象中那么复杂。
2. 工具整体设计与核心思路拆解
2.1 为什么选择自己开发而非使用现成工具?
在决定动手之前,我调研过不少方案。有浏览器插件,有在线的“文章转PDF”网站,也有一些客户端软件。但它们普遍存在几个痛点:一是往往有数量限制,下载几十篇后就需要付费;二是对公众号文章的样式支持不完整,图片可能丢失,排版会错乱;三是无法实现真正的“历史文章”批量抓取,很多工具只能处理单篇文章链接。最重要的是,数据安全性和可控性。使用第三方在线服务,你的阅读列表和下载内容可能会经过别人的服务器,存在隐私泄露风险。自己写的工具,所有数据处理都在本地完成,源码在手,一切透明可控。
2.2 核心工作流程与模块划分
这个下载工具的工作流程可以抽象为四个核心步骤,对应四个功能模块:
- 公众号定位与列表抓取:输入一个公众号的名称或唯一标识(如biz参数),工具需要模拟微信的请求,获取到该公众号的“历史消息”页面列表。这是最核心也是最具挑战的一步,因为微信的反爬机制一直在升级。
- 文章链接提取与去重:从抓取到的列表页面HTML中,解析出每一篇文章的永久链接(通常以
https://mp.weixin.qq.com/s/...开头)。这里需要处理分页加载(微信列表是瀑布流)和链接去重,避免同一篇文章被多次下载。 - 文章内容解析与清洗:访问每一个文章链接,抓取完整的HTML页面。然后,需要从纷繁复杂的页面代码中,精准地提取出文章标题、作者、发布日期、正文内容(包含图文排版)、以及文章内的图片、视频、音频等资源的原始地址。
- 资源下载与本地化打包:将上一步解析出的正文HTML进行美化,并将文中引用的所有图片、视频等远程资源下载到本地文件夹中,同时修改HTML中的链接,使其指向本地文件。最后,将处理好的HTML文件,连同资源文件夹,按照日期和标题有序地保存起来。
整个工具的架构是模块化的,每个步骤相对独立。这样做的好处是,如果未来微信的页面结构发生变化,或者你想增强某个功能(比如增加导出为PDF),只需要修改对应的模块,而不会牵一发而动全身。
2.3 技术选型:Python生态的优势
我选择Python作为实现语言,主要基于以下几点考虑:
- 丰富的网络爬虫库:
requests用于发送HTTP请求简单高效,BeautifulSoup4和lxml用于解析HTML如鱼得水。 - 强大的异步支持:对于批量下载图片这种IO密集型任务,
aiohttp和asyncio可以大幅提升效率,实现并发下载。 - 成熟的HTML处理工具:
html2text可以方便地将HTML转为Markdown,readability之类的库可以帮助提取文章主体内容。 - 跨平台与易部署:Python脚本在Windows、macOS、Linux上都能运行,通过
pip安装依赖非常方便,也易于打包成可执行文件分享给不懂技术的朋友。
3. 关键技术细节与实操要点
3.1 如何获取公众号文章列表?—— 破解“历史消息”接口
这是整个项目的第一个技术难关。你不能直接通过一个公开的API拿到列表。经过抓包分析,我发现微信公众平台的文章列表是通过一个特殊的接口动态加载的,关键点在于以下几个参数:
__biz: 公众号的唯一身份标识,类似于ID。uin/key: 与用户会话相关的加密参数。offset: 控制分页的偏移量。count: 每页返回的文章数量(通常固定为10)。begin/end: 时间戳,用于按时间范围筛选。
实际操作中,最可靠的方法是模拟微信客户端或网页端的请求。你需要先通过搜索或已知文章,找到一个目标公众号的任意一篇文章。从这篇文章页面的URL或源代码中,可以提取出关键的__biz值。然后,构造一个携带正确Cookie和Header的请求,去访问一个固定格式的列表URL。Cookie尤为重要,它代表了你的登录状态,通常可以通过登录微信PC版或网页版后,从开发者工具的Network面板中复制出来。
重要提示:这里涉及模拟请求和抓取公开数据,务必遵守目标网站的
robots.txt协议,并将抓取频率控制在合理的、不对对方服务器造成压力的范围内(例如,在请求间添加随机延时)。本工具仅用于个人学习与研究目的的数据收集。
3.2 精准提取文章正文与资源
拿到文章链接后,下一步是获取纯净的正文。微信公众号文章的页面包含了大量无关元素:顶部关注引导、底部点赞评论、侧边栏广告、相关推荐等等。我们的目标是只保留文章本身。
这里有两种主流思路:
- CSS选择器精准定位:分析文章页面的HTML结构,找到包裹正文的那个
div标签(它的id或class通常包含js_content,rich_media_content等关键词)。用BeautifulSoup定位到这个节点,提取其内部HTML,就能得到相对干净的正文。这种方法速度快,但依赖微信前端的DOM结构,一旦微信改版,选择器可能失效。 - 使用内容提取库:比如
readability-lxml或goose3。这些库的算法会分析页面的标签密度、文本长度等因素,智能地判断哪一部分是核心文章内容。这种方法通用性更强,不依赖特定网站结构,但可能无法100%完美还原微信复杂的排版(如代码块、特殊格式等)。
我采用的是混合策略:优先尝试用预设的CSS选择器去定位,如果定位失败或内容过短,则回退到使用readability库进行提取。实测下来,这种方法在绝大多数情况下都能取得很好的效果。
对于图片、视频等资源,它们在HTML中通常以<img>requests>=2.25.1 beautifulsoup4>=4.9.3 lxml>=4.6.3 aiohttp>=3.8.1 readability-lxml>=0.8.1 html2text>=2020.1.16 tqdm>=4.62.3 # 用于显示进度条
在终端中,进入项目目录,运行以下命令一键安装:
pip install -r requirements.txt4.2 核心代码模块实现
我们将代码分成几个Python文件,便于管理。
1. 配置文件 (config.py)这里存放一些可配置的常量,比如请求头、超时时间、保存路径等。
import os # 请求头,模拟浏览器访问 HEADERS = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36', 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8', 'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8', } # 网络请求超时时间(秒) TIMEOUT = 10 # 文章保存的根目录 SAVE_ROOT = './wechat_articles' # 请求间隔延时,避免过快请求导致IP被限制(秒) REQUEST_DELAY = 22. 列表抓取模块 (crawler.py)这个模块负责获取公众号的文章列表。我们实现一个WeChatListCrawler类。
import requests import time import json from bs4 import BeautifulSoup from urllib.parse import urljoin, urlparse import re from config import HEADERS, TIMEOUT, REQUEST_DELAY class WeChatListCrawler: def __init__(self, cookie=None): self.session = requests.Session() if cookie: self.session.headers.update({'Cookie': cookie}) self.session.headers.update(HEADERS) def get_article_links_from_page(self, list_url): """从一个列表页中解析出文章链接""" links = [] try: resp = self.session.get(list_url, timeout=TIMEOUT) resp.raise_for_status() soup = BeautifulSoup(resp.text, 'lxml') # 微信公众号文章链接的常见模式 for link in soup.find_all('a', href=re.compile(r'^https?://mp\.weixin\.qq\.com/s')): href = link.get('href') if href and 'chksm' not in href: # 过滤掉一些带校验参数的临时链接 links.append(href) time.sleep(REQUEST_DELAY) # 礼貌性延时 except Exception as e: print(f"抓取列表页失败 {list_url}: {e}") return list(set(links)) # 去重 # 注意:这里简化了连续分页抓取的逻辑。实际中,你需要分析微信的Ajax接口或模拟滚动加载。 # 一个常见方法是不断改变URL中的`offset`参数来获取更多历史消息。由于微信历史列表的完整抓取涉及更复杂的接口分析和模拟,上述代码仅提供了基础框架。完整实现需要分析微信的Ajax请求,构造包含__biz,uin,key,offset等参数的POST请求,并解析返回的JSON数据。这部分代码因微信更新频繁而需要动态调整,是项目的核心难点之一。
3. 文章解析与下载模块 (article.py)这个模块负责处理单篇文章。
import os import aiohttp import asyncio from bs4 import BeautifulSoup from readability import Document import html2text from config import HEADERS, SAVE_ROOT import re class ArticleFetcher: def __init__(self, session=None): self.session = session or aiohttp.ClientSession(headers=HEADERS) async def fetch_article(self, url): """获取并解析单篇文章""" try: async with self.session.get(url) as resp: html = await resp.text() except Exception as e: print(f"下载文章失败 {url}: {e}") return None # 方法1: 尝试用CSS选择器定位微信正文 soup = BeautifulSoup(html, 'lxml') content_div = soup.find('div', id='js_content') or soup.find('div', class_=re.compile('rich_media_content')) if content_div and len(content_div.get_text(strip=True)) > 200: # 找到了正文区域 title = soup.find('meta', property='og:title') title = title['content'] if title else '未知标题' # 提取纯净的正文HTML content_html = str(content_div) else: # 方法2: 使用readability库智能提取 doc = Document(html) content_html = doc.summary() title = doc.title() # 提取文章发布日期 publish_time = None publish_meta = soup.find('meta', property='article:published_time') or soup.find('span', class_=re.compile('publish_time')) if publish_meta: publish_time = publish_meta.get('content') or publish_meta.get_text() return { 'url': url, 'title': title, 'publish_time': publish_time, 'raw_html': html, 'content_html': content_html } async def download_resource(self, url, save_path): """异步下载单个资源(图片/视频)到本地""" try: async with self.session.get(url) as resp: if resp.status == 200: content = await resp.read() os.makedirs(os.path.dirname(save_path), exist_ok=True) with open(save_path, 'wb') as f: f.write(content) return True except Exception as e: print(f"下载资源失败 {url}: {e}") return False async def localize_article(self, article_info, save_dir): """将文章资源本地化,并保存为HTML""" if not article_info: return False soup = BeautifulSoup(article_info['content_html'], 'lxml') resource_dir = os.path.join(save_dir, 'resources') os.makedirs(resource_dir, exist_ok=True) # 处理图片 img_tasks = [] for i, img in enumerate(soup.find_all('img')): src = img.get('data-src') or img.get('src') # 优先使用data-src if src and src.startswith('http'): local_filename = f'image_{i:03d}.jpg' local_path = os.path.join(resource_dir, local_filename) img_tasks.append(self.download_resource(src, local_path)) # 替换src属性为本地路径 img['src'] = f'./resources/{local_filename}' if 'data-src' in img.attrs: del img.attrs['data-src'] # 并发下载所有图片 if img_tasks: results = await asyncio.gather(*img_tasks) print(f"图片下载完成: {sum(results)} 成功, {len(results)-sum(results)} 失败") # 保存最终的HTML文件 final_html = f""" <!DOCTYPE html> <html> <head> <meta charset="utf-8"> <title>{article_info['title']}</title> <style> body {{ font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', 'Helvetica Neue', Arial, sans-serif; max-width: 800px; margin: 0 auto; padding: 20px; line-height: 1.6; color: #333; }} img {{ max-width: 100%; height: auto; display: block; margin: 1em auto; }} </style> </head> <body> <h1>{article_info['title']}</h1> <p><small>发布时间: {article_info['publish_time'] or '未知'} | 原文链接: <a href="{article_info['url']}">{article_info['url']}</a></small></p> <hr> {str(soup)} </body> </html> """ safe_title = re.sub(r'[\\/*?:"<>|]', "_", article_info['title'])[:50] # 清理非法文件名字符 filename = f"{article_info['publish_time'][:10] if article_info['publish_time'] else 'nodate'}_{safe_title}.html" filepath = os.path.join(save_dir, filename) with open(filepath, 'w', encoding='utf-8') as f: f.write(final_html) print(f"文章已保存: {filepath}") return True4. 主控与调度模块 (main.py)这是程序的入口,负责协调各个模块的工作。
import asyncio import os from crawler import WeChatListCrawler from article import ArticleFetcher from config import SAVE_ROOT async def main(): # 1. 初始化抓取器(需要填入有效的Cookie) cookie = "你的微信Cookie" # 请从浏览器开发者工具中复制 if not cookie or cookie == "你的微信Cookie": print("请先在config.py或此处配置有效的微信Cookie。") return list_crawler = WeChatListCrawler(cookie=cookie) # 2. 指定要抓取的公众号列表页URL(示例,需要替换) # 如何获取这个列表URL?通常可以从公众号资料页或历史消息页的地址栏复制。 # 例如:https://mp.weixin.qq.com/mp/profile_ext?action=home&__biz=MzA5NDk...==#wechat_redirect list_urls = [ 'https://mp.weixin.qq.com/mp/profile_ext?action=home&__biz=YOUR_BIZ_HERE&scene=124#wechat_redirect', ] all_article_urls = [] for url in list_urls: print(f"正在抓取列表: {url}") links = list_crawler.get_article_links_from_page(url) all_article_urls.extend(links) print(f"找到 {len(links)} 篇文章链接。") # 简单去重 all_article_urls = list(set(all_article_urls)) print(f"去重后,总计 {len(all_article_urls)} 篇待下载文章。") if not all_article_urls: print("未找到任何文章链接,请检查列表URL或Cookie。") return # 3. 创建文章保存目录 import datetime today_str = datetime.datetime.now().strftime('%Y%m%d_%H%M%S') save_dir = os.path.join(SAVE_ROOT, f'collection_{today_str}') os.makedirs(save_dir, exist_ok=True) # 4. 异步抓取并保存所有文章 fetcher = ArticleFetcher() tasks = [] for i, article_url in enumerate(all_article_urls[:10]): # 这里限制为前10篇作为演示 print(f"({i+1}/{len(all_article_urls[:10])}) 处理: {article_url}") task = asyncio.create_task(fetcher.fetch_and_save_article(article_url, save_dir)) tasks.append(task) # 控制并发度,避免过快 if len(tasks) >= 3: # 同时处理3篇文章 await asyncio.gather(*tasks) tasks = [] # 处理剩余任务 if tasks: await asyncio.gather(*tasks) await fetcher.session.close() print(f"\n所有文章处理完成!已保存至目录: {save_dir}") # 为ArticleFetcher添加一个便捷方法 async def fetch_and_save_article(self, url, save_dir): article_info = await self.fetch_article(url) if article_info: await self.localize_article(article_info, save_dir) ArticleFetcher.fetch_and_save_article = fetch_and_save_article if __name__ == '__main__': asyncio.run(main())4.3 如何获取关键的Cookie和列表URL?
这是工具能运行起来的前提。这里提供一个通用的手动获取方法:
获取Cookie:
- 在电脑上打开浏览器(推荐Chrome或Edge)。
- 访问
https://mp.weixin.qq.com并登录你的微信(需要能登录公众平台,普通用户扫码登录即可)。 - 按
F12打开开发者工具,切换到Network(网络) 面板。 - 刷新页面,在网络请求列表中,点击任意一个请求(如
profile_ext)。 - 在右侧
Headers(标头) 选项卡中,找到Request Headers部分,复制Cookie字段后面那一长串值。
获取公众号列表页URL(含__biz):
- 在微信PC客户端或网页版搜索找到目标公众号。
- 点击进入公众号主页,查看“历史消息”。
- 此时浏览器地址栏的URL,通常就包含了
__biz参数。将这个完整的URL复制下来,填入代码中的list_urls列表。
安全警告:Cookie是个人敏感信息,相当于你的登录凭证。切勿将包含真实Cookie的代码上传到GitHub等公开平台。建议将Cookie存储在环境变量或本地配置文件中,并通过
.gitignore忽略该配置文件。
5. 常见问题、排查技巧与优化建议
在实际使用和开发过程中,你肯定会遇到各种问题。下面是我总结的一些常见坑点和解决方案。
5.1 请求被拒绝或返回空数据
- 现象:代码运行后,抓取到的文章列表为空,或者请求返回403/404错误。
- 排查:
- Cookie失效:微信Cookie的有效期有限,可能已过期。重新登录并获取新的Cookie。
- 请求头不完整:模拟的
User-Agent不够像真实浏览器。可以尝试从浏览器开发者工具中直接复制一个完整的请求头信息,更新到config.py的HEADERS中。 - 频率过高:短时间内发送过多请求,触发反爬。务必在请求间添加随机延时(例如
time.sleep(random.uniform(1, 3)))。 - 列表URL或参数错误:公众号的
__biz可能不正确,或者列表接口已更新。需要重新抓包分析最新的请求格式。
5.2 文章内容提取不完整或格式错乱
- 现象:保存的HTML文件里缺少正文,或者排版混乱,图片显示不正常。
- 排查与解决:
- CSS选择器失效:微信前端改版了。打开一篇公众号文章,用开发者工具检查正文区域最新的HTML结构和CSS类名,更新
article.py中的content_div查找逻辑。 - 图片未成功本地化:检查图片链接是否成功替换。可能是图片的
style="width:16px;margin-left:4px;vertical-align:text-bottom;cursor:text;" />
- CSS选择器失效:微信前端改版了。打开一篇公众号文章,用开发者工具检查正文区域最新的HTML结构和CSS类名,更新