ARTICLE DETAIL

资讯详情

深耕商务建站与企业官网运营的一线实战洞察。

Python爬虫实战:公众号历史文章批量采集与解析

Python爬虫实战:公众号历史文章批量采集与解析 简介本资源是一套面向 Python 爬虫学习者与数据分析人员的微信公众号文章采集工具包聚焦于从微信公众平台批量获取公众号所有文章的 URL 地址适合具备一定 Python 基础、希望练习真实站点抓取与接口调试的开发者。压缩包共 30 个文件约 17.22MB以 13 个 py 脚本为核心辅以 5 个 md 说明文档、若干 jpg/png 截图以及 chromedriver 驱动文件覆盖文章链接抓取、文章信息解析、结果读取输出等模块并配有 requirements.txt 便于快速搭建运行环境。文档部分整理了获取 appmsg_token、cookie_token 等关键参数的思路以及微信公众号接口使用与自动获取微信参数的说明能帮助读者理解抓取流程中的鉴权与请求构造环节。目前已有 583 人学习下载可作为微信公众号爬虫入门与接口调试的实践参考。1. 公众号内容采集一个被低估的爬虫练手项目很多人第一次接触爬虫都是从公众号文章开始的。原因很直接页面结构相对稳定数据量适中反爬强度不算极端非常适合拿来练手。但真上手之后你会发现事情没那么简单——文章链接藏在历史消息里正文被各种内联样式包裹图片还做了防盗链处理。我见过不少开发者在这个项目上翻车不是卡在请求上就是卡在解析上。这个项目要解决的问题很明确给定一个公众号批量抓取它的历史文章列表再逐篇提取标题、作者、发布时间、正文内容和封面图。适合谁刚学完 requests 和 BeautifulSoup、想找一个完整项目练手的 Python 开发者也适合需要做内容归档、数据分析的运营或研究人员。技术栈不复杂核心就是 requests BeautifulSoup 正则存储用 CSV 或 SQLite 都行。2. 项目架构拆解从入口请求到数据落盘的完整链路2.1 整体模块划分与数据流向这个项目的架构不复杂但该有的分层不能少。我一般会把它拆成四个模块请求层、解析层、存储层、调度层。请求层负责模拟客户端发送 HTTP 请求拿到原始 HTML解析层从 HTML 里提取结构化数据存储层把数据写到本地文件或数据库调度层控制翻页逻辑和请求频率。数据流向是这样的调度层发起第一页请求 → 请求层返回 HTML → 解析层提取文章列表 → 存储层保存列表数据 → 调度层根据分页信息发起下一页请求 → 重复直到没有更多数据 → 对每篇文章详情页发起请求 → 解析正文 → 存储。为什么这么分因为公众号的列表页和详情页结构完全不同解析逻辑必须分开写。而且请求频率控制、Cookie 管理这些通用逻辑放在请求层后面改起来方便。2.2 请求层Cookie 与请求头的处理公众号的很多接口需要携带有效的 Cookie 才能返回数据尤其是历史文章列表。常见做法是手动从浏览器开发者工具里复制 Cookie 和 User-Agent写进配置里。import requests import time import random # 配置请求头Cookie 需要从浏览器手动获取 HEADERS { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36, Cookie: 你的Cookie字符串, # 从浏览器开发者工具复制 Referer: https://mp.weixin.qq.com/ } def fetch_page(url, retry3): 发送 GET 请求带重试机制 for i in range(retry): try: resp requests.get(url, headersHEADERS, timeout10) if resp.status_code 200: return resp.text else: print(f状态码 {resp.status_code}第 {i1} 次重试) except requests.RequestException as e: print(f请求异常{e}第 {i1} 次重试) time.sleep(random.uniform(2, 5)) # 随机延时降低被封风险 return None这段代码的关键点有三个。第一HEADERS里的Cookie是必须的没有它很多接口直接返回空数据或者跳转到验证页。第二retry参数控制重试次数网络抖动时自动重试比直接报错友好得多。第三time.sleep(random.uniform(2, 5))是保命操作固定间隔的请求很容易被识别为机器行为随机延时能有效降低风险。参数怎么改timeout根据你的网络情况调整一般 10 秒够用。retry建议不超过 5 次太多会浪费时间。Referer字段有些接口会校验带上更稳妥。2.3 解析层列表页与详情页的提取逻辑列表页返回的通常是 JSON 或者嵌在 HTML 里的 JSON 片段详情页则是完整的 HTML。两种解析方式不一样。from bs4 import BeautifulSoup import re import json def parse_article_list(html): 从列表页 HTML 中提取文章条目 articles [] soup BeautifulSoup(html, html.parser) # 文章列表通常在特定的 div 或 script 标签中 items soup.select(div.article-item) # 根据实际页面结构调整 for item in items: title item.select_one(h3).get_text(stripTrue) link item.select_one(a)[href] # 补全相对链接 if link.startswith(/): link https://mp.weixin.qq.com link articles.append({title: title, url: link}) return articles def parse_article_detail(html): 从详情页提取正文、作者、发布时间 soup BeautifulSoup(html, html.parser) # 标题 title_tag soup.select_one(h1.rich_media_title) title title_tag.get_text(stripTrue) if title_tag else # 作者 author_tag soup.select_one(span.rich_media_meta_text) author author_tag.get_text(stripTrue) if author_tag else # 发布时间 time_tag soup.select_one(em#publish_time) publish_time time_tag.get_text(stripTrue) if time_tag else # 正文内容 content_tag soup.select_one(div#js_content) content content_tag.get_text(stripTrue) if content_tag else # 封面图 cover_tag soup.select_one(img.rich_media_thumb) cover cover_tag[src] if cover_tag else return { title: title, author: author, publish_time: publish_time, content: content, cover: cover }解析层的核心是选择器要写对。公众号页面的 class 名经常变所以选择器不能写死最好用多个备选方案。比如正文容器可能是div#js_content也可能是div.rich_media_content可以两个都试一下。get_text(stripTrue)会把标签内的文本提取出来并去掉首尾空白。正文里如果有换行和段落get_text默认会丢掉格式如果需要保留段落结构可以用get_text(separator\n)。2.4 存储层CSV 与 SQLite 的选择数据量小的时候 CSV 够用数据量大或者需要频繁查询就用 SQLite。我一般两个都写方便切换。import csv import sqlite3 def save_to_csv(articles, filenamearticles.csv): 将文章列表写入 CSV 文件 if not articles: return keys articles[0].keys() with open(filename, a, newline, encodingutf-8-sig) as f: writer csv.DictWriter(f, fieldnameskeys) # 文件为空时写入表头 if f.tell() 0: writer.writeheader() writer.writerows(articles) def save_to_sqlite(articles, db_patharticles.db): 将文章写入 SQLite 数据库 conn sqlite3.connect(db_path) cursor conn.cursor() cursor.execute( CREATE TABLE IF NOT EXISTS articles ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT, author TEXT, publish_time TEXT, content TEXT, cover TEXT, url TEXT UNIQUE ) ) for art in articles: try: cursor.execute( INSERT OR IGNORE INTO articles (title, author, publish_time, content, cover, url) VALUES (?, ?, ?, ?, ?, ?), (art.get(title), art.get(author), art.get(publish_time), art.get(content), art.get(cover), art.get(url)) ) except sqlite3.Error as e: print(f写入失败{e}) conn.commit() conn.close()CSV 用utf-8-sig编码是为了 Excel 打开时不乱码这个坑我踩过好几次。SQLite 的url字段加了UNIQUE约束配合INSERT OR IGNORE可以自动去重避免重复抓取同一篇文章。3. 运行环境与依赖安装从零把项目跑起来3.1 Python 版本与依赖清单这个项目对 Python 版本要求不高3.8 以上都能跑。依赖就三个核心库pip install requests beautifulsoup4 lxmlrequests负责发请求beautifulsoup4负责解析 HTMLlxml是解析器后端比 Python 自带的html.parser快不少。如果要用 SQLitePython 标准库自带sqlite3不用额外装。我一般会再装一个fake-useragent用来随机生成 User-Agent降低被封的概率pip install fake-useragent3.2 配置文件与参数说明把 Cookie、目标公众号标识、请求间隔这些可变参数抽到一个config.py里改起来不用翻主逻辑代码。# config.py COOKIE 你的Cookie字符串 TARGET_BIZ 目标公众号的biz参数 # 从文章链接里提取 MAX_PAGES 50 # 最大翻页数防止无限循环 REQUEST_DELAY (2, 5) # 随机延时范围单位秒 OUTPUT_FORMAT sqlite # 可选 csv 或 sqlite DB_PATH articles.db CSV_PATH articles.csvTARGET_BIZ是公众号的唯一标识通常藏在文章链接的__biz参数里。MAX_PAGES是保险丝万一翻页逻辑出问题不会无限跑下去。REQUEST_DELAY控制请求频率数值越大越安全但越慢。3.3 主流程串联与运行把前面几个模块串起来主流程大概长这样from config import MAX_PAGES, OUTPUT_FORMAT from fetcher import fetch_page from parser import parse_article_list, parse_article_detail from storage import save_to_csv, save_to_sqlite import time import random def main(): all_articles [] for page in range(MAX_PAGES): # 构造列表页 URL具体参数根据实际情况调整 list_url fhttps://mp.weixin.qq.com/mp/profile_ext?actiongetmsgpage{page} html fetch_page(list_url) if not html: print(f第 {page} 页请求失败停止) break articles parse_article_list(html) if not articles: print(没有更多文章了) break # 逐篇抓取详情 for art in articles: detail_html fetch_page(art[url]) if detail_html: detail parse_article_detail(detail_html) detail[url] art[url] all_articles.append(detail) time.sleep(random.uniform(2, 5)) print(f第 {page} 页完成累计 {len(all_articles)} 篇) # 落盘 if OUTPUT_FORMAT csv: save_to_csv(all_articles) else: save_to_sqlite(all_articles) print(f抓取完成共 {len(all_articles)} 篇) if __name__ __main__: main()运行方式就是python main.py。第一次跑建议把MAX_PAGES设成 2 或者 3先验证流程通不通再放开跑全量。4. 避坑与常见问题那些让我熬夜的报错4.1 请求返回空数据或跳转验证页现象请求列表页返回的 HTML 里没有文章条目或者直接跳到一个验证页面。原因Cookie 过期了或者请求头里的 User-Agent 和 Cookie 不匹配。公众号对 Cookie 的有效期管得比较严一般几小时到一天就会失效。解决重新从浏览器复制 Cookie确保 User-Agent 和复制 Cookie 时用的浏览器一致。如果还是不行检查Referer字段是否缺失。4.2 正文提取为空或只有部分内容现象parse_article_detail返回的content字段是空字符串或者只有开头一小段。原因公众号正文可能是懒加载的初始 HTML 里只有占位符。另外有些文章正文在 iframe 里直接解析外层 HTML 拿不到。解决先确认页面是否完整加载。如果是懒加载需要找到实际的数据接口。如果是 iframe用requests请求 iframe 的src地址再解析。我一般会先把 HTML 存到本地用浏览器打开看看结构再决定怎么解析。4.3 图片链接失效或防盗链现象抓下来的封面图链接在浏览器里打不开返回 403。原因公众号图片有防盗链机制直接请求图片 URL 会被拒绝。解决下载图片时带上Referer头值设为https://mp.weixin.qq.com/。如果还不行可能需要把图片 URL 里的wx_fmt参数去掉再试。4.4 翻页逻辑死循环现象程序一直跑但抓到的文章列表始终是同一批。原因翻页参数没变或者接口返回的数据结构在最后一页时没有正确判断。解决在循环里加一个判断如果本次解析出的文章列表和上一次完全一样就break。另外MAX_PAGES一定要设这是最后的保险。4.5 写入 CSV 时中文乱码现象用 Excel 打开 CSV 文件中文全是乱码。原因Excel 默认用 GBK 编码打开 CSV而文件是 UTF-8 编码。解决写入时用utf-8-sig编码这个编码会在文件开头加 BOM 标记Excel 能正确识别。5. 二次开发与改造思路从能跑到好用5.1 增量抓取与去重策略第一次跑全量之后后面再跑就应该只抓新文章。最简单的做法是在存储层加一个url唯一索引每次插入前先查一下是否已存在。SQLite 的INSERT OR IGNORE配合UNIQUE约束就能做到。如果数据量大可以再加一个last_fetch_time字段记录每篇文章的抓取时间增量抓取时只请求比这个时间新的文章。不过公众号列表页一般按时间倒序排列所以更简单的做法是每次从第一页开始翻遇到已存在的 URL 就停止翻页。def is_url_exists(url, db_patharticles.db): 检查 URL 是否已存在于数据库 conn sqlite3.connect(db_path) cursor conn.cursor() cursor.execute(SELECT 1 FROM articles WHERE url ? LIMIT 1, (url,)) result cursor.fetchone() conn.close() return result is not None在翻页循环里每解析出一篇文章就先调这个函数判断如果已存在就跳过连续跳过 N 篇就终止翻页。5.2 正文清洗与结构化输出原始正文里通常夹杂着大量无关内容广告、推荐阅读、二维码提示等。我一般会做一轮清洗把正文按段落切分过滤掉长度过短的段落和包含特定关键词的段落。def clean_content(raw_text): 清洗正文去掉广告和无关段落 lines raw_text.split(\n) cleaned [] skip_keywords [点击上方, 关注我们, 长按识别, 推荐阅读] for line in lines: line line.strip() if len(line) 10: continue if any(kw in line for kw in skip_keywords): continue cleaned.append(line) return \n.join(cleaned)清洗规则可以根据实际抓取结果不断调整。我习惯先把原始正文存一份清洗后的另存一份方便对比和回滚。5.3 多公众号批量采集的调度改造单个公众号跑通之后很自然会想扩展到多个公众号。这时候需要把TARGET_BIZ改成列表外层再加一层循环。但要注意不同公众号的请求间隔要独立控制不能共用一个延时否则总请求频率会翻倍。TARGET_LIST [ {biz: 公众号A的biz, name: 账号A}, {biz: 公众号B的biz, name: 账号B}, ] for target in TARGET_LIST: print(f开始抓取 {target[name]}) crawl_one(target[biz]) time.sleep(random.uniform(10, 20)) # 账号之间多等一会儿账号之间的间隔建议比单账号内的间隔更长给服务器一个缓冲时间。我一般设 10 到 20 秒。5.4 验证改造效果的方法改完之后怎么确认没改坏我的习惯是准备一个小型测试集选一个文章数量少的公众号手动记录前 5 篇文章的标题和发布时间然后跑一遍程序对比抓取结果是否一致。如果一致再放开跑全量。另外每次改完解析逻辑先把 HTML 存到本地用本地文件跑解析函数这样不用反复发请求调试效率高很多。# 调试用从本地文件读取 HTML 并解析 with open(debug_page.html, r, encodingutf-8) as f: html f.read() result parse_article_detail(html) print(result)从那以后我每次改解析规则都强制先走一遍本地 HTML 调试确认没问题再发真实请求。这个习惯帮我省了大量等待时间也避免了因为调试请求太频繁被限制。希望帮到你。本文还有配套的精品资源点击获取
返回列表
PREV
查看更多资讯
NEXT
返回资讯列表