简介:一套基于Python的微信公众号爬虫系统源码,面向具备一定爬虫基础、希望批量获取公众号文章的技术开发者。项目通过ADB与模拟器联动模拟手机微信客户端行为,结合抓包分页接口实现突破,支持多账号切换与多公众号并发采集,能有效规避传统方案中文章缺失、账号易被限制等问题。压缩包体积仅568KB,共52个文件,其中31个Python脚本构成爬虫执行、任务管理、代理配置等核心模块,6个JavaScript脚本负责请求拦截与规则处理,另有图片、JSON、MD文档等辅助内容,适合直接阅读与二次开发。项目采用模块化设计,包含任务生成、代理拦截、数据持久化等环节,并集成MongoDB存储和动态代理IP,实用性强。目前已有225人学习下载,如需快速搭建微信公众号采集链路,这套源码可提供完整的目录结构与基础逻辑供参考。
1. 一套能跑通的微信公众号爬虫系统,到底解决谁的什么问题
如果你手里攒了几十个知识类公众号,每周想集中归档几篇好文,靠人工复制粘贴撑不过一个月。基于Python的微信公众号爬虫系统,就是把这件重复劳动固化成一套可复用的源码工程:喂给它一批文章链接,它能自动提取标题、正文、发布日期和公众号信息,写进数据库,再按需导出Word或Markdown。它治的是内容运营、知识库建设和竞品情报收集这类“历史文章留存”的刚需。想找毕设项目的在校生、刚把Python语法学完准备练完整工程的从业者,都能从这个系统里直接抄到能跑通的答案。它不是破解接口的黑匣子,主要工作集中在“公开页面内容的下载与结构化”,接下来的每一章都围绕这个主链路展开。
2. 从公众号文章链接到数据落地:先把采集链路和模块边界画清楚
2.1 全链路:任务队列、请求层、解析层、存储层、导出层
拿到源码第一件事,不是打开主文件逐行读,而是先看它把哪几个环节拆开了。我见过的Python爬虫系统,绝大多数都长在下面这条链路上:任务来源(一批文章URL)→ 去重与状态管理 → HTTP请求层 → 正文解析层 → 结构化存储 → 文件导出。
先说任务来源。一个微信公众号文章的完整采集动作,其实分两段:第一段是“找到文章链接”,第二段是“把链接内容抓下来”。这套系统的核心往往在第二段。第一段常见的做法是用RPA采集微信公众号后台的历史消息列表,或者从搜狗微信搜索里拿搜索结果的链接,再把链接列表落成一个文本文件或数据表,交给爬虫继续跑。这类系统把任务来源抽象成“种子链接文件”,就是为了跟前端采集方式解耦。你手里积累的公众号文章URL,以及从其他渠道拿到的合法链接,都能直接喂进来。
然后是去重与状态管理。文章链接是天然去重键,但同一个链接可能因为搜索来源不同被带上了?from=timeline之类的参数,所以不仅要存原始URL,还要存一个去掉追踪参数后的“归一化URL”。这一层不做好,后面抓十遍的悲剧就每天都在发生。状态管理至少要有三态:待抓取、抓取成功、抓取失败。系统重启后,从失败状态里捞一批重跑,比全量重跑省太多时间。
请求层负责把HTML文本拿回来,解析层从HTML里抽出标题、正文、公众号名和发布时间,存储层把结构化结果写进SQLite或MySQL,导出层再按业务需要生成Word、Markdown或HTML快照。五层各管各的事,哪一环出问题都可以单独替换。这个边界感,比“一个大文件从头写到尾”的写法值得抄。
2.2 文章正文为什么用 requests 就够,而“列表页”才需要浏览器渲染
很多初学者拿到爬虫系统源码后,第一反应是:页面里有那么多JS,为什么还要用requests这种同步请求库?我的经验是,公众号文章链接多数长这样:https://mp.weixin.qq.com/s/xxxxxxx。这种文章落地页的服务端渲染做得比较彻底,正文就藏在返回的HTML里,不需要浏览器额外执行脚本。用requests拿到的那份HTML,已经包含完整正文、封面图地址、公众号名称、发布时间等关键字段。
真正需要模拟浏览器的是“列表页”。公众号的历史消息页面、搜狗微信的搜索结果页,这两类页面有较强的反爬逻辑和异步加载,requests直接拉会拿到一坨JS空壳。所以很多爬虫系统只在列表采集这一段上接入Selenium、Playwright或RPA操作浏览器,把拿到的文章链接再交给requests去抓正文。这样分工很合理:浏览器渲染成本高,只用在容易变化的列表页;正文页量大且格式固定,用轻量请求库抓效率高得多。
做一个简单的判断题:如果你的源码里所有页面都用了Selenium,那它多半是抓列表页或后台登录态页面用的,不一定代表这套系统写复杂了。判断标准是正文提取是否稳定,而不是用没用Selenium。“能少用浏览器就少用浏览器”这句话,在这个场景下能省一半的机器资源。
2.3 文章表怎么设计:字段、唯一约束和采集状态的三态设计
数据库表是整套系统真正的地基。我一般会建一张article表,字段设计成下面这样:
CREATE TABLE article ( id INTEGER PRIMARY KEY AUTOINCREMENT, url TEXT NOT NULL UNIQUE, url_hash TEXT NOT NULL UNIQUE, title VARCHAR(200), author VARCHAR(100), account_name VARCHAR(100), biz VARCHAR(100), publish_time DATETIME, content_text LONGTEXT, content_html LONGTEXT, word_path VARCHAR(255), status TINYINT DEFAULT 0, retry_count TINYINT DEFAULT 0, last_error TEXT, create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE INDEX idx_status ON article(status); CREATE INDEX idx_biz_time ON article(biz, publish_time);几个字段的取舍逻辑先交代清楚。url和url_hash都设唯一约束,是双保险。url本身就能唯一标识一篇文章,但URL可能长达两百多个字符,直接当索引键在数据量大时性能有压力。url_hash取URL的MD5值,固定32位,用来做快速查重。这个设计在百万级链接下仍然能保持毫秒级去重查询。
status字段用0、1、2表示待抓取、成功、失败。很多人喜欢用布尔值“是否抓过”,但实际跑起来你会发现,失败的需要重试,成功的可能需要定期回看,所以三态是最低要求。我还会加一个retry_count,因为一次失败就永远标记成失败太武断,网络抖动导致的失败完全可以重试两次再放弃。
biz是公众号的唯一标识,从文章页面里的var biz = "..."这类JS变量里可以取到。存它的价值在于后期能按公众号分组统计:哪个号更新最勤、哪些号的稿件平均质量高,这些分析都要靠biz字段支撑。publish_time建议直接存datetime类型,比存时间戳方便做跨天查询。
最后补一个关键提示:全文检索不要指望SQL里的LIKE,后面数据量到几十万条时,应该把content_text同步到ES里,或者用SQLite的FTS5扩展。
3. 把源码跑起来:环境准备到四个核心模块逐段读代码
3.1 创建虚拟环境并确认依赖
我建议直接用Python 3.10及以上版本,新版在类型注解和数据类上的支持更顺手,Win10/11和主流Linux发行版都能跑。先建立虚拟环境再装依赖,避免把系统Python环境搞乱。
python -m venv venv source venv/bin/activate # Windows 下改为 venv\Scripts\activate pip install -r requirements.txtrequirements.txt里这几个核心依赖值得逐个确认:
requests==2.31.0 beautifulsoup4==4.12.3 lxml==5.2.1 python-docx==1.1.0 pymysql==1.1.0选型时有个常见误区:上来就装Scrapy。这个项目的体量用Scrapy有点重,Scrapy的调度器、中间件、Twisted异步模型都是给大规模分布式采集准备的,你自己维护一个定时巡检脚本时反而会被框架约束。用requests手动管理Session和循环,逻辑最透明,出问题好排查。
如果你在VSCode里跑,记得在命令面板里执行“Python: Select Interpreter”,把解释器指到当前目录的venv目录下,否则会出现终端里装好的包,编辑器里import却报红的尴尬。之前在旧版本里跑过的老项目如果有兼容问题,优先检查是不是urllib3版本冲突。
3.2 config.py:只改这几个参数
源码包里一般会有一个集中放配置的文件,常见的名字是config.py。这里只挑核心参数说,别去动那些看起来很深奥的封装代码。
# config.py DB_PATH = "articles.db" SEED_FILE = "seed_links.txt" WORD_SAVE_DIR = "./export_word" DOWNLOAD_INTERVAL = 1.5 # 每次 HTTP 请求之间的间隔,单位秒 REQUEST_TIMEOUT = 10 # 单个请求超时时间 MAX_RETRY = 3 # 失败重试次数 USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ..." PROXY_POOL = [] # 可选代理列表,格式 ["http://ip:port"]DOWNLOAD_INTERVAL是最值得认真调的参数。公众号文章页面对单IP的请求频率比较敏感,间隔小于0.5秒时容易触发风控,1.5秒是一个兼顾效率和安全的起始值。单线程模式下,一万篇文章按平均2秒一篇算,大概需要5个多小时,这个速度对绝大多数归档场景都够用了。
USER_AGENT尽量用一个较新的Chrome或Edge的完整UA字符串。有些反爬逻辑会拦截不完整的UA,所以别偷懒写成python-requests/2.31。
PROXY_POOL默认留空,让代码走直连。只有当你发现部分文章出现区域性访问限制时,才考虑在池子里填代理地址。系统会在fetch_html里自动从池中轮换,但代理质量问题会导致大量超时,所以我对这个参数的态度是:能不用就不用。
3.3 请求层:带重试、超时、限速的 Session
请求层是整套系统的“腰”,腰如果软了,后面解析做得再好也白搭。最小可用实现长这样:
import requests import time from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry def build_session(): session = requests.Session() retry = Retry( total=MAX_RETRY, backoff_factor=0.5, status_forcelist=[500, 502, 503, 504], allowed_methods=["GET"] ) adapter = HTTPAdapter(max_retries=retry) session.mount("https://", adapter) session.headers.update({"User-Agent": USER_AGENT}) return session def fetch_html(session, url): try: resp = session.get(url, timeout=REQUEST_TIMEOUT) resp.raise_for_status() # 检查是否返回了验证码或风控页面 if "验证码" in resp.text[:2000] or "环境异常" in resp.text[:2000]: raise RuntimeError("risk control triggered") time.sleep(DOWNLOAD_INTERVAL) resp.encoding = resp.apparent_encoding if resp.encoding == "ISO-8859-1" else resp.encoding return resp.text except Exception as exc: print(f"[fetch fail] {url} -> {exc}") return None这里有两个设计点要注意。Retry里的backoff_factor=0.5,表示第一次重试等待0.5秒,第二次等待1秒,第三次等待2秒,而不是立即重试。这比连续快速重试对服务器的压力小得多,也能减少因为瞬时网络抖动导致的采集失败。
第二个点是风控页面的预检查。resp.text[:2000]这个裁剪是有讲究的,微信风控页的title和正文关键词都集中在前2KB里,判断一次只要几毫秒。如果命中了关键词,直接抛异常跳出,而不是把整个风控页当成正常文章送到解析层。这样做的好处是,解析层永远只处理真实文章HTML,不用到处写“如果不是文章就跳过”的判断。
还有一个隐藏参数值得说明:resp.encoding的处理。部分老文章或特定入口的页面会返回GBK编码,requests默认按ISO-8859-1猜,结果中文全乱码。上面的代码先把响应头判断排除掉ISO-8859-1,再用apparent_encoding去做检测,能覆盖绝大多数编码场景。
3.4 解析层 + 存储层:最小可用的 parse_article 与 save_article
解析层的核心任务是“从HTML里稳定抽出几个字段”。公众号文章的HTML结构这几年变化不算大,正文一般都在div#js_content里。下面这段代码是经过多次迭代后仍然稳定的最小版本:
from bs4 import BeautifulSoup import html as html_lib def parse_article(page_html, source_url): soup = BeautifulSoup(page_html, "lxml") # 标题优先取 og:title meta 字段,兼容没有 meta 的页面 title_meta = soup.find("meta", attrs={"property": "og:title"}) title = title_meta["content"].strip() if title_meta else soup.title.get_text(strip=True) account_el = soup.select_one("#js_name") account_name = account_el.get_text(strip=True) if account_el else "" content_div = soup.select_one("#js_content") if not content_div: return None # 微信正文图片地址放在>import sqlite3 def get_conn(): conn = sqlite3.connect(DB_PATH) conn.execute("PRAGMA journal_mode=WAL") return conn def save_article(article): conn = get_conn() try: conn.execute( """INSERT OR IGNORE INTO article (url, title, account_name, content_text, content_html, status) VALUES (:url, :title, :account_name, :content_text, :content_html, 1)""", article, ) conn.commit() finally: conn.close()INSERT OR IGNORE结合url的唯一索引,是防重复抓取的最后一道防线。状态字段置为1是“抓取成功”,这个动作在全部处理完成之后才做,避免文章内容还没解析完就误标成功。
4. 抓取微信公众号历史文章的工程化:去重、增量与 Word 导出
4.1 URL 归一化与哈希去重
抓历史文章时,同一个链接往往从不同路径回来。比如文章在列表页里带?from=timeline&isappinstalled=0,在分享页里带?chksm=xxxx。如果直接拿完整URL去重,同一条文章会被当成两条记录。我在系统里专门加了一个归一化函数:
from urllib.parse import urlparse, parse_qsl, urlencode def normalize_url(raw_url): parsed = urlparse(raw_url) # 只保留 biz、mid、idx、sn 这几个公众号文章定位参数 keep_keys = {"biz", "mid", "idx", "sn"} query = [(k, v) for k, v in parse_qsl(parsed.query) if k in keep_keys] query.sort() return parsed._replace(query=urlencode(query)).geturl()公众号文章的定位参数主要是biz、mid、idx、sn四件套,其余都是统计或分享来源标记。归一化之后再用url_hash字段去重,才能做到“无论从哪个入口进来,同一条文章只抓一次”。
4.2 三态状态字段驱动增量
增量采集的关键是“状态驱动”,不是“时间驱动”。每天定时任务启动后,我只会捞status=0的记录,同时在内存里维护一个已处理URL集合,双重拦截。具体流程是:
def get_pending_tasks(limit=50): conn = get_conn() rows = conn.execute( "SELECT id, url FROM article WHERE status=0 LIMIT ?", (limit,) ).fetchall() conn.close() return rows拉取时用LIMIT限定每批次条数,是因为如果积累的待抓取链接有几万条,一次性导入内存会导致任务中断后全部重来。每批50条,处理完一批再拉下一批,天然支持断点续跑。失败的任务会把status置为2,同时retry_count加一,当retry_count超过3时不再自动重试,等着人工排查。我在实际项目中见过一次坑:把所有失败任务无限重试,结果对同一批坏链接反复请求了几个小时,风险页面越积越多。加了重试上限后,这种“死循环式请求”才被根治。
4.3 用 python-docx 把文章存成 Word 文档
把文章保存为Word是这个项目里使用频率最高的导出功能。python-docx本身是个不错的库,但它有个明显边界:它不认识HTML标签。把<p>、<section>直接塞给docx会变成一段纯文本,样式全部丢失。所以我的导出策略分两种:
第一种是“纯文本归档型”,适合内部知识库:
from docx import Document from docx.shared import Pt def export_text_to_word(article, save_path): doc = Document() doc.add_heading(article["title"], level=0) meta = doc.add_paragraph() meta.add_run(f"公众号:{article['account_name']}\n") meta.add_run(f"原文链接:{article['url']}") for para in article["content_text"].split("\n"): para = para.strip() if not para: continue p = doc.add_paragraph() run = p.add_run(para) run.font.size = Pt(12) doc.save(save_path)第二种是“带图片的完整导出型”。这种方案要把正文里的图片全部下载到本地,再插入Word,流程是:解析HTML → 提取所有img→ 下载图片到assets目录 → 用docx.add_picture插入。下载图片时要注意带上Referer: https://mp.weixin.qq.com/,否则会被防盗链拒绝。图片插入Word后会挤压排版,我的处理是把每张图片单独放一个段落,居中显示,下方加图片原始链接,方便追溯。
两种方案我都会做,但默认导出用第一种。原因很现实:公众号文章带宽限流,一次导出几百篇文章时,图片下载会拖垮整个采集速度。先保证文字内容100%归档,图片作为二期增强功能按需开启。
4.4 Word 导出完成后,要在库里记录导出路径
很多系统第一次跑完Word导出就结束了,再跑一次时又把所有文章重新生成一遍Word。我在article表里留了word_path字段,每次导出成功后把路径写进去。生成前先判断word_path是否为空,为空才执行docx处理。这个小小的“导出状态”,能避免重复劳动。
5. 这五处避坑记录:403、空正文、防盗链、重复抓取、乱码
5.1 请求返回403或者“验证码”页面
现象:日志里频繁出现resp.status_code == 403,或者抓回来的HTML标题是“请输入验证码”。 原因:单IP在短时间内请求次数过高,触发微信风控;或者是User-Agent太单一,容易被识别为脚本。 解决:把DOWNLOAD_INTERVAL从1.5秒调到3秒,同一批任务里对同一个公众号连续不超过20篇。如果风控页已经出现,这时候再加间隔意义不大,更好的做法是让整个队列进入“暂停期”,停30分钟到1小时再继续。我见过有人在风控触发后马上无限重试,结果风控时间被越拉越长。暂停不是示弱,是用时间换稳定。
5.2 正文解析出来全是空字符串
现象:parse_article返回None,或者content_text长度只有几十个字符。 原因:第一种是请求回来的是被拦截的HTML空壳页面,页面里根本没有#js_content;第二种是某些政策类或营销类文章使用了不同的正文容器;第三种是文章已被删除,微信返回的是“该内容已被发布者删除”的提示页。 解决:遇到None时,先打印page_html[:500],人工看一眼是哪种情况。如果是风控空壳,回到上一个坑的处理流程。如果是删除页,直接把status置为2并记录last_error="deleted",不要再重试。如果确认正文容器不同,才去调整soup.select_one的选择器。调整后必须拿一批真实文章样本验证,别只测一篇。
5.3 图片全部裂开:防盗链与懒加载
现象:导出的Word里文字正常,但所有图片显示为“无法加载”,HTML快照里图片也全是裂图。 原因:微信图片服务器对Referer做了校验,非微信域名的请求会被拒绝;另外图片懒加载导致src属性里是占位符,真实地址在>SELECT COUNT(*) AS total, SUM(CASE WHEN title IS NULL OR LENGTH(title) < 5 THEN 1 ELSE 0 END) AS bad_title, SUM(CASE WHEN content_text IS NULL OR LENGTH(content_text) < 200 THEN 1 ELSE 0 END) AS bad_content, SUM(CASE WHEN publish_time IS NULL OR publish_time < '2000-01-01' THEN 1 ELSE 0 END) AS bad_time FROM article WHERE status = 1;
当一个bad_*字段占比超过5%时,我就知道解析规则可能过期了,需要去抽样看页面结构。这个脚本每天跑一次,配合定时任务自动执行,采集系统才算有“仪表盘”。后期如果要做成可视化界面,把这段查出来的结果转成JSON,用Flask挂一个/health接口,前端页面就能实时显示采集健康度。
6.2 用 Cron 做增量巡检,但别让两个进程同时跑
增量巡检用系统级定时器最省事。Linux下crontab配置如下:
# 每天早上 8 点执行一次增量采集 0 8 * * * cd /opt/wechat_spider && /opt/wechat_spider/venv/bin/python run_daily.py >> logs/daily.log 2>&1这里的重点是,定时任务和手动执行千万不要叠加。我见过最典型的事故是:手动跑了一次全量,Cron到点又启动了一个进程,两个爬虫同时写SQLite,结果数据库被锁死,数据出现重复。解决方案是加一个文件锁:
import fcntl, sys def acquire_lock(): lock_file = open("spider.lock", "w") try: fcntl.flock(lock_file, fcntl.LOCK_EX | fcntl.LOCK_NB) return lock_file except BlockingIOError: print("另一个爬虫进程已在运行,本次退出") sys.exit(0)每次主程序启动时先获取锁,拿不到锁就退出。这样无论Cron还是手动任务都不会撞车。
6.3 日志自动瘦身,别等磁盘满了才发现
爬虫系统的日志增长比想象中快得多。我遇到过一次,跑了三周,日志文件撑到40GB,直接写满了系统盘。标准做法是使用RotatingFileHandler:
import logging from logging.handlers import RotatingFileHandler handler = RotatingFileHandler( "logs/spider.log", maxBytes=5 * 1024 * 1024, backupCount=3, encoding="utf-8", ) logging.basicConfig(level=logging.INFO, handlers=[handler])单个日志文件超过5MB后自动轮转,保留最近3份。这样磁盘占用永远控制在20MB以内。我最开始觉得写日志越多越好,直到日志文件占满了磁盘才明白,保留最近几份有效信息,比保留历史全部噪声更有价值。
最后一个个人习惯:每次修改解析逻辑后,我都会先抓20篇文章做样本校验,通过后再放全量。这个习惯让我在增量抓取时踩的坑大幅减少。希望这篇笔记里提到的结构、参数和避坑经验,能帮你把这个爬虫系统源码更快跑成自己的稳定服务。
本文还有配套的精品资源,点击获取