★ FEATURED ARTICLE 简介一套基于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以及从其他渠道拿到的合法链接都能直接喂进来。然后是去重与状态管理。文章链接是天然去重键但同一个链接可能因为搜索来源不同被带上了?fromtimeline之类的参数所以不仅要存原始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里这几个核心依赖值得逐个确认requests2.31.0 beautifulsoup44.12.3 lxml5.2.1 python-docx1.1.0 pymysql1.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( totalMAX_RETRY, backoff_factor0.5, status_forcelist[500, 502, 503, 504], allowed_methods[GET] ) adapter HTTPAdapter(max_retriesretry) session.mount(https://, adapter) session.headers.update({User-Agent: USER_AGENT}) return session def fetch_html(session, url): try: resp session.get(url, timeoutREQUEST_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_factor0.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(stripTrue) account_el soup.select_one(#js_name) account_name account_el.get_text(stripTrue) 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_modeWAL) 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 归一化与哈希去重抓历史文章时同一个链接往往从不同路径回来。比如文章在列表页里带?fromtimelineisappinstalled0在分享页里带?chksmxxxx。如果直接拿完整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(queryurlencode(query)).geturl()公众号文章的定位参数主要是biz、mid、idx、sn四件套其余都是统计或分享来源标记。归一化之后再用url_hash字段去重才能做到“无论从哪个入口进来同一条文章只抓一次”。4.2 三态状态字段驱动增量增量采集的关键是“状态驱动”不是“时间驱动”。每天定时任务启动后我只会捞status0的记录同时在内存里维护一个已处理URL集合双重拦截。具体流程是def get_pending_tasks(limit50): conn get_conn() rows conn.execute( SELECT id, url FROM article WHERE status0 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], level0) 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_errordeleted不要再重试。如果确认正文容器不同才去调整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 21这里的重点是定时任务和手动执行千万不要叠加。我见过最典型的事故是手动跑了一次全量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直接写满了系统盘。标准做法是使用RotatingFileHandlerimport logging from logging.handlers import RotatingFileHandler handler RotatingFileHandler( logs/spider.log, maxBytes5 * 1024 * 1024, backupCount3, encodingutf-8, ) logging.basicConfig(levellogging.INFO, handlers[handler])单个日志文件超过5MB后自动轮转保留最近3份。这样磁盘占用永远控制在20MB以内。我最开始觉得写日志越多越好直到日志文件占满了磁盘才明白保留最近几份有效信息比保留历史全部噪声更有价值。最后一个个人习惯每次修改解析逻辑后我都会先抓20篇文章做样本校验通过后再放全量。这个习惯让我在增量抓取时踩的坑大幅减少。希望这篇笔记里提到的结构、参数和避坑经验能帮你把这个爬虫系统源码更快跑成自己的稳定服务。本文还有配套的精品资源点击获取
阅读完成 · 觉得有帮助？
