ARTICLE DETAIL

资讯详情

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

基于Python的微信公众号爬虫系统:从Token获取到FTS5全文检索

基于Python的微信公众号爬虫系统:从Token获取到FTS5全文检索 简介一份基于Python的微信公众号爬虫系统源码面向需要批量采集公众号文章内容、搭建内容库或进行舆情监控的开发者和爬虫工程师适用于内容管理、行业分析、学术研究等场景。项目以Scrapy爬虫框架为基础通过ADB与Android模拟器实现微信客户端自动化操作利用抓包获取历史文章分页接口规避传统方式中文章不全、账号易封的痛点并支持多账号切换和动态代理显著提升采集稳定性。资源包含52个文件以31个Python脚本为核心覆盖任务生成、代理拦截、数据解析、爬虫执行等环节另有6个JavaScript代理拦截脚本、配置文件、依赖清单、结构示意图和说明文档压缩包仅568KB体量精炼、模块边界清晰。学习者可从中掌握ADB操控微信、Scrapy中间件编写、MongoDB数据持久化等完整实现思路也可直接部署运行或按需二次开发用于多公众号的高效抓取。已有225人学习下载适合具备一定爬虫基础、希望构建高可用微信采集系统的开发者参考。1. 一份能跑的 Python 微信公众号爬虫先分清抓取对象和链路拿到(源码)基于Python的微信公众号爬虫系统.zip第一件事不是急着pip install而是先厘清这套系统究竟在抓什么。常规的公众号采集对象是某公众号的已发布文章列表再展开到每篇文章的标题、摘要、作者、正文、封面图、发布时间和阅读量。爬虫系统要解决的是把“账号”和“文章”两层数据拉平再在增量维度上做去重和归档。常见做法就是先翻列表分页再对列表里未入库的appmsgid请求正文最后统一落库。这套流程不依赖单集浏览器核心是一个可控的 requests 会话配上 cookie、token 和必要的签名参数。对有 5 年经验的后端开发来说难点已经不在解析 HTML而在限频调度、缓存策略和故障兜底。适用人群是一线爬虫工程师、数据团队以及要在企业内部做公众号内容聚合的后端小组。这篇文章直接按一个可复现的最小系统来讲把 token 获取、分页去重、正文抽取和检索归档都落到代码层。2. 微信公众号文章采集的三条门路官方接口、搜狗搜索、后台导出2.1 三个来源的采集路径和可用性对比市面能落到代码的公众号采集方案绕不开三条路微信公众平台后台的接口、搜狗微信搜索、第三方内容聚合 API。先看横向对比后面选型才有依据。数据源访问方式权限门槛可获取字段反爬强度公众平台后台接口登录 Cookie Token 调/cgi-bin/appmsg公众号运营者扫码授权标题、摘要、作者、正文、封面、阅读量中频率敏感搜狗微信搜索搜索页 列表页 HTML匿名访问多数时候只有摘要正文需二次解析极高常常要求 JS 验证第三方聚合 APIHTTPS 调用商业授权全字段由服务商承担我一般优先选第一路。搜狗微信搜索对非浏览器访问的 JS 校验很重返回的正文是转义过的 HTML还要再过一层解码可用性很差。第三方 API 适合赶工期但单价、字段稳定性都不可控。自己写爬虫系统最通用的是抓公众号后台的“图文素材”和“发表记录”因为这个入口返回的是结构化 JSON翻页和抽字段都稳定。2.2 走官方接口方向Token 与 Cookie 的来源访问/cgi-bin/appmsg之前要保证登录态里有slave_sid和slave_user这两个字段藏在 Cookie 里。token 可以从首页 HTML 里正则抽取。下面这段是经常出现在各类公众号爬虫源码里的取 token 姿势import re import requests session requests.Session() # 读者需替换为自己的登录 Cookie session.headers.update({ User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64), Cookie: slave_sid***; slave_user***;, }) resp session.get(https://mp.weixin.qq.com/cgi-bin/home?thome/indexlangzh_CN) m re.search(rtoken(\d), resp.text) token m.group(1) if m else print(token:, token)这段代码只做两件事注入登录态再从首页脚本里抽取 token。token 有时出现在window.token或var token赋值里抽不到时优先看resp.text里token前后 20 个字符的上下文确认正则是没匹配到还是页面结构已经改版。token 是有时效的通常以小时计所以爬虫系统里要把它做成可刷新参数而不是硬编码。2.3 文章列表接口的参数组合拿到 token 后请求文章列表。接口路径是/cgi-bin/appmsg用 GET 请求关键参数如下import requests url https://mp.weixin.qq.com/cgi-bin/appmsg params { action: list_ex, begin: 0, count: 5, fakeid: MjM5MTAxMDAwMA, type: 9, token: token, lang: zh_CN, } resp requests.get(url, paramsparams, cookiessession.cookies, headerssession.headers) data resp.json() for item in data.get(app_msg_list, []): print(item[title], item[link], item[update_time]) print(总文章数:, data[app_msg_cnt])参数说明begin从 0 开始count表示每页条数这里写 5 是为了测试时不触发限频跑稳了再往上调到 10 或 20。type9表示已发表内容fakeid是公众号的 base64 标识可以通过个人主页接口拿到。app_msg_cnt是账号的文章总数用来判断翻页是否结束。需要注意app_msg_list里的link是临时链接过期很快入库时最好只保存appmsgid后续再次访问详情页时要用新的列表数据补全。3. 用 Python 搭一个可增量抓取的公众号文章采集架构3.1 requests 会话与会话级重试源码里最值得学习的不一定是某个解析函数而是请求层是否统一收敛。所有请求共用同一个requests.Session()把超时、重试、Header 都挂上才能在上千次请求中不散架。下面是常见做法from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry def build_session(retries: int 3, backoff: float 0.8) - requests.Session: s requests.Session() retry_cfg Retry( totalretries, connectretries, readretries, backoff_factorbackoff, status_forcelist[502, 503, 504], ) adapter HTTPAdapter(max_retriesretry_cfg) s.mount(https://, adapter) s.mount(http://, adapter) return s逻辑说明Retry里的total是总重试次数backoff_factor0.8表示第一次重试等待 0.8 秒第二次 1.6 秒指数递增。status_forcelist只对 5xx 状态码触发不要把 403 加进去因为 403 是业务态重复请求只会加强风控标记。connect和read分别控制连接阶段和读取阶段的超时重试分开设是为了应对连接被重置、但读响应正常的场景。3.2 列表分页与增量去重分页不能无脑begin count。公众号后台对同一会话的请求频率有限制连续翻页几十次后接口会返回ok: false, errmsg: freq control。增量采集的做法是先查本地已有appmsgid集合遇到已存在的直接跳过当前剩余分页。import sqlite3 def fetch_page_and_dedupe(session, params, seen_ids, db_conn): resp session.get(https://mp.weixin.qq.com/cgi-bin/appmsg, paramsparams) data resp.json() if not data.get(app_msg_list): return 0, data.get(app_msg_cnt, 0) new_count 0 for item in data[app_msg_list]: appmsgid item[appmsgid] if appmsgid in seen_ids: continue seen_ids.add(appmsgid) _save_meta(db_conn, item) new_count 1 return new_count, data.get(app_msg_cnt, 0)这里逐条把begin加 1而不是按页大小偏移是为了配合去重时跳过已存在的记录。假如一页里 5 条只新增了两条剩下 3 条跳过那偏移量要精确到条而不是页否则会漏掉下一页开头的数据。seen_ids用内存 set 加速首次运行时从 SQLite 表里灌入SELECT appmsgid FROM article_meta WHERE appmsgid IS NOT NULL;配合这套逻辑爬虫中断后重启已入库的appmsgid会被跳过实现断点续爬。如果你在源码里看到的是begin count建议改成逐条偏移这是高频踩坑点。3.3 正文抽取与字段清洗公众号正文有几种形态编辑器的富文本 HTML、图片加文字的扫描文档、外链跳转。正文抽取建议用trafilatura比BeautifulSoup手写选择器更扛结构变化安装与调用如下pip install trafilaturaimport trafilatura def extract_article(link: str, session: requests.Session) - dict: downloaded trafilatura.fetch_url(link, sessionsession) text trafilatura.extract( downloaded, include_commentsFalse, include_tablesFalse ) return {content: text or , content_len: len(text or )}参数含义include_commentsFalse是去掉读者的评论区块include_tablesFalse是过滤表格内容因为公众号表格常混排影响文本纯净度。trafilatura.extract返回清洗后的纯文本不是 HTML对后续全文检索友好。需要保留封面图和正文里的图片时得换readability-lxml入库字段也要从纯文本改成 HTML。两者取舍是文本完整度优先用trafilatura结构性展示优先用 readability。4. 采集运行期的常见坑限频、403、zip 伪加密与调试4.1 拿到 zip 源码包以后先做解压校验这个 zip 包在传播中有几种典型损坏方式截断、伪加密、目录缺文件。Linux 上建议先跑测试unzip -t wechat_crawler.zip输出No errors detected in compressed data才说明文件结构完整。如果出现bad CRC大概率是传输丢字节这时候重新下载比任何修复都省事。伪加密是另一个常见情况unzip -l能看到文件列表一解压就要求密码但包说明里没标密码。快速检测方式是看加密标志位zipinfo -v wechat_crawler.zip | grep -i encryption如果输出里encryption是None却还要密码那就是伪加密。可以在 Python 里清掉文件头标志位再解import zipfile with zipfile.ZipFile(wechat_crawler.zip) as zf: for info in zf.infolist(): if info.flag_bits 0x1: info.flag_bits ^ 0x1这段代码把每个文件头的加密标志位强制清 0。它只对伪加密有效真加密必须用正确密码不要尝试对不明来源的 zip 暴力破解既浪费时间也有安全风险。解压后第一件事是检查目录结构是否有requirements.txt和入口文件没有requirements.txt的源码包依赖版本通常要对齐 Python 3.8 到 3.10 之间的版本。4.2 403 和 freq control 出现时先降速再换 cookie403 在公众号接口场景里通常有三层含义Cookie 过期、IP 被风控、签名参数缺失。处理顺序建议固定先确认 Cookie 是否过期再降请求频率最后才怀疑参数问题。import time import random def rate_limited_get(session, url, params, cookie_expire_ts): if time.time() cookie_expire_ts: raise RuntimeError(cookie expired, need manual scan) time.sleep(random.uniform(1.2, 2.8)) resp session.get(url, paramsparams, timeout10) if resp.status_code 403: raise RuntimeError(forbidden, check fakeid and token validity) return resp这里把time.sleep放在请求前让每次请求前都有一段随机等待。随机区间的上下限直接影响采集效率和触发风控的概率区间越大越安全但总量下降明显。cookie_expire_ts可以维护成全局变量每次成功响应后从Set-Cookie里重新解析过期时间。注意不要在调用层再包一层 time.sleep否则双重等待会让采集速度慢到不可用。4.3 在 VSCode 里配置 Python 调试环境没有调试器的爬虫容易在翻页循环里栽跟头。VSCode 配 Python 环境时不要直接在全局解释器装依赖先给项目建虚拟环境python -m venv .venv source .venv/bin/activate pip install -r requirements.txt然后创建.vscode/launch.json设置justMyCode为true只进自己的函数不跳site-packages。翻页循环经常在fetch_page_and_dedupe里出问题打断点后观察begin与seen_ids的关系能快速判断是偏移写错还是去重误删。日志级别建议调到DEBUGrequests 库会输出每个请求的 URL 和状态码配合 grep 就能定位卡在哪个fakeid。5. 采集后的核心校验用 SQLite FTS5 做正文全文检索和重复率复核正文入库后能搜到才是硬指标。SQLite 自带 FTS5 全文检索不需要起 Elasticsearch。关键是 tokenizer 选trigram对中文不用分词插件直接按 3 字符滚动切分百万级数据内响应速度都可接受。建表语句如下CREATE VIRTUAL TABLE IF NOT EXISTS fts_articles USING fts5( title, content, contentarticle_meta, content_rowidid, tokenizetrigram );contentarticle_meta表示外部内容表content_rowidid指定主键。用外部内容表的好处是正文更新后不用重建整条 FTS 记录同步即可。增量同步用 Python 侧执行import sqlite3 conn sqlite3.connect(wechat_articles.db) conn.execute( INSERT INTO fts_articles(rowid, title, content) SELECT id, title, content FROM article_meta WHERE id ? , (last_sync_id,)) conn.commit()这里的last_sync_id每次跑完要更新到本地 marker 表否则重启后会重复同步。查询时用 FTS5 的 match 语法SELECT title, snippet(fts_articles, 1, [, ], …, 12) FROM fts_articles WHERE fts_articles MATCH 微信 OR 爬虫 ORDER BY rank LIMIT 20;match里的短语用双引号括起避免被解析成单词项。snippet返回高亮片段参数分别对应FTS 表名、列下标、左边界符、右边界符、省略号、上下文长度。这套检索方案适合采集完成后的运营查询按“公众号爬虫系统”这类长尾词就能直接拉出历史文章不用等第三方搜索收录。最后顺手复核去重质量。不同appmsgid的文章可能正文完全相同多出现于转载场景。统计正文的md5指纹即可量化重复率SELECT COUNT(*) FROM ( SELECT md5(content) AS h FROM article_meta GROUP BY h HAVING COUNT(*) 1 );md5 用于内部去重指纹没有安全问题不需要换 SHA256。这个 SQL 输出的是重复组数不是重复记录数要算重复率就把它除以SELECT COUNT(*) FROM article_meta。到这一步这个基于 Python 的公众号采集系统就具备了从分页抓取、正文清洗、断点续爬到全文检索的完整闭环。后续想扩展点赞数或评论数只需在appmsg接口返回值里多保留一个键再在_save_meta里加对应字段。本文还有配套的精品资源点击获取
返回列表