公众号爬虫实战：抓取、存储与互动数据

承接前文的原理与抓包基础，本文进入公众号爬虫的完整实战：先抓取公众号的第一篇文章，再翻页抓取全部历史文章，将抓取结果存储到 MongoDB，最后获取每篇文章的阅读数、点赞数等互动数据。

一、抓取第一篇文章

前文已经通过 Fiddler 抓到了公众号历史消息接口。网页版公众号的文章列表接口是 profile_ext，通过 action 参数区分不同操作：home 用于抓取第一页，getmsg 用于翻页抓取历史文章。

分析接口返回结构

访问公众号历史消息页后，浏览器会向 mp.weixin.qq.com/mp/profile_ext 发起请求。返回的 JSON 中关键字段是 general_msg_list（JSON 字符串），其中 list 数组包含每篇文章的信息：

json

{
  "general_msg_list": "{\"list\": [{\"comm_msg_info\": {...}, \"app_msg_ext_info\": {...}}]}"
}

每篇文章的 app_msg_ext_info 字段包含核心数据：

title：文章标题
content_url：文章链接（需进行 HTML 反转义）
digest：文章摘要
cover：封面图地址
author：作者
multi_app_msg_item_list：多图文推送时的副标题列表

抓取第一页的代码

python

import requests
import json

# 复用抓包得到的 Cookie 和参数
session = requests.Session()
session.headers.update({
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)",
    "Referer": "https://mp.weixin.qq.com/",
    "Cookie": "你的_cookie=xxx; ...",
})

url = "https://mp.weixin.qq.com/mp/profile_ext"
params = {
    "__biz": "你的__biz",
    "action": "home",          # home = 抓第一页
    "scene": "124",
    "f": "json",
    "ajax": "1",
    "begin": "0",
    "count": "10",
}
data = {
    "key": "你的_key",
    "pass_ticket": "你的_pass_ticket",
    "appmsg_token": "你的_appmsg_token",
    "x5": "0",
    "f": "json",
}

resp = session.post(url, params=params, data=data)
result = resp.json()

# 解析文章列表
if result.get("general_msg_list"):
    msg_list = json.loads(result["general_msg_list"])["list"]
    for item in msg_list:
        msg_info = item.get("app_msg_ext_info")  # 非图文消息没有此字段
        if not msg_info:
            continue
        print({
            "title": msg_info.get("title"),
            "content_url": msg_info.get("content_url", "").replace("&", "&"),
            "digest": msg_info.get("digest"),
        })

💡 现状说明：profile_ext 接口的 home 与 getmsg 参数在当前版本仍可正常返回文章列表，但 token、key、pass_ticket、appmsg_token 均有有效期（appmsg_token 约 2-4 小时），失效后需重新抓包获取。此外，也可以直接在浏览器中打开历史消息页，通过页面中的全局变量（如 window.appmsg_list）获取文章数据。

提取文章字段

接口返回的 content_url 等字段包含 HTML 转义字符（如 &），需要反转义后再使用。可以写一个工具函数 sub_dict 提取指定字段：

python

import html

def sub_dict(d: dict, keys: list) -> dict:
    return {k: html.unescape(d[k]) for k in d if k in keys}

d = {"title": "A&amp;B", "cover": "http://xxx.jpg"}
sub_dict(d, ["title", "cover"])
# {'title': 'A&B', 'cover': 'http://xxx.jpg'}

二、抓取所有历史文章

抓取第一页只是开始，公众号通常有几十上百篇历史文章，需要翻页抓取全部。翻页的核心是修改 action 为 getmsg，并通过 offset 参数控制抓取起点，直到接口返回 can_msg_continue == 0（表示没有更多文章）。

翻页抓取的完整代码

python

import requests
import json
import time

def get_articles(session, url, params, data, offset):
    """抓取从 offset 开始的 10 篇文章，返回 (文章列表, 是否还有下一页)"""
    params["offset"] = str(offset)
    resp = session.post(url, params=params, data=data)
    result = resp.json()

    msg_list = []
    if result.get("general_msg_list"):
        msg_list = json.loads(result["general_msg_list"])["list"]

    can_continue = result.get("can_msg_continue")  # 1=还有更多, 0=抓完
    return msg_list, bool(can_continue)


def crawl_all(session, url, params, data, start_offset=0):
    """抓取全部历史文章"""
    all_articles = []
    offset = start_offset
    while True:
        msg_list, can_continue = get_articles(session, url, params, data, offset)
        all_articles.extend(msg_list)
        print(f"已抓取 {len(all_articles)} 篇, 是否继续: {can_continue}")

        if not can_continue:
            break
        offset += len(msg_list)   # offset 累加，指向下一页
        time.sleep(1)             # 控制频率，避免风控
    return all_articles


# 使用示例
session = requests.Session()
session.headers.update({
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)",
    "Cookie": "你的_cookie=xxx; ...",
})
url = "https://mp.weixin.qq.com/mp/profile_ext"
params = {
    "__biz": "你的__biz",
    "action": "getmsg",      # getmsg = 翻页抓取
    "count": "10",
    "f": "json",
    "ajax": "1",
}
data = {
    "key": "你的_key",
    "pass_ticket": "你的_pass_ticket",
    "appmsg_token": "你的_appmsg_token",
    "x5": "0",
    "f": "json",
}
articles = crawl_all(session, url, params, data)
print(f"共抓取 {len(articles)} 篇文章")

💡 要点：

翻页时 offset 表示已抓取的文章数量，接口从该位置返回下一批。

can_msg_continue 为 0 表示没有更多历史文章，停止翻页。

务必控制抓取频率（time.sleep），频繁请求会触发微信风控，导致接口返回异常或 Cookie 失效。

抓取大量文章时，appmsg_token 可能在过程中过期，建议每次请求后检查返回，若过期则重新抓包。

三、将数据存储到 MongoDB

抓取到的文章需要持久化存储。数据量小时可以直接存 CSV，但数据量大或需要结构化查询时，推荐使用 MongoDB 这类文档型数据库。

MongoDB 简介与安装

MongoDB 是目前最新版本是 8.0 的文档型数据库，无需预先定义表结构即可直接插入数据，非常适合爬虫数据的存储。

💡 安装方式：

macOS（Homebrew）：brew tap mongodb/brew && brew install mongodb-community

Windows：winget install MongoDB.Server

Docker（推荐）：docker run -d --name mongodb -p 27017:27017 mongo:8.0

启动 MongoDB：

bash

brew services start mongodb-community   # macOS
# 或
mongod --dbpath <path to data directory>

默认端口是 27017。可视化可选用 MongoDB Compass、Studio 3T 或命令行工具 mongosh。

MongoEngine 定义数据模型

MongoEngine 是 MongoDB 的 DOM（Document-Object Mapper）框架，类似关系型数据库的 ORM，可以更简洁地操作 MongoDB。

bash

pip install mongoengine

定义文章模型：

python

from datetime import datetime
from mongoengine import DateTimeField, Document, IntField, StringField, URLField, connect

# 连接 mongodb，无需事先创建数据库
connect('weixin', host='localhost', port=27017)


class Post(Document):
    """文章信息"""
    title = StringField()          # 文章标题
    content_url = StringField()    # 文章链接
    content = StringField()        # 文章内容
    source_url = StringField()     # 原文链接
    digest = StringField()         # 文章摘要
    cover = URLField(validation=None)   # 封面图
    p_date = DateTimeField()       # 推送时间

    read_num = IntField(default=0)       # 阅读数
    like_num = IntField(default=0)       # 点赞数
    comment_num = IntField(default=0)    # 评论数
    reward_num = IntField(default=0)     # 赞赏数
    author = StringField()         # 作者

    c_date = DateTimeField(default=datetime.now)  # 数据生成时间
    u_date = DateTimeField(default=datetime.now)  # 最后更新时间

将抓取结果保存到 MongoDB

抓取返回的 JSON 中有很多无用字段，通过 sub_dict 提取核心字段后，插入 MongoDB。多图文推送需要处理 multi_app_msg_item_list（副标题列表），因此把插入逻辑抽成私有方法 _insert 复用：

python

import html
import json
from datetime import datetime


def sub_dict(d: dict, keys: list) -> dict:
    return {k: html.unescape(d[k]) for k in d if k in keys}


def save(msg_list: str):
    """msg_list 为抓取接口返回的 general_msg_list JSON 字符串"""
    msg_list = msg_list.replace("\\/", "/")
    data = json.loads(msg_list)
    items = data.get("list")
    for msg in items:
        p_date = msg.get("comm_msg_info").get("datetime")
        msg_info = msg.get("app_msg_ext_info")   # 非图文消息没有此字段
        if msg_info:
            _insert(msg_info, p_date)
            # 多图文推送，把第二条第三条也保存
            multi_msg_info = msg_info.get("multi_app_msg_item_list")
            for msg_item in multi_msg_info:
                _insert(msg_item, p_date)
        else:
            print(f"此消息不是图文推送: {json.dumps(msg.get('comm_msg_info'))}")


def _insert(item: dict, p_date: int):
    keys = ('title', 'author', 'content_url', 'digest', 'cover', 'source_url')
    sub_data = sub_dict(item, keys)
    post = Post(**sub_data)
    post["p_date"] = datetime.fromtimestamp(p_date)
    try:
        post.save()
        print(f"保存成功: {post.title}")
    except Exception as e:
        print(f"保存失败: {post.title}", e)

💡 要点：

文字推送没有 app_msg_ext_info 字段，无需保存。

multi_app_msg_item_list 是多图文推送的副标题列表，字段与外层 app_msg_ext_info 一致（标题、封面、摘要、链接），因此共用 _insert 插入。

建议使用 html.unescape 对 content_url 等字段反转义后再入库。

四、获取文章互动数据

存储了文章基础信息后，往往还需要文章的阅读数、点赞数、评论数、赞赏数等互动数据。这些数据通过 getappmsgext 接口获取，且必须依赖微信登录态（未登录时这些字段不返回）。

分析 getappmsgext 接口

从抓包可以看到，getappmsgext 是一个 POST 接口，请求参数包含文章标识（mid、sn）、__biz，以及登录凭证（key、pass_ticket、appmsg_token）。

💡 现状说明（2024-2026）：

互动数据（阅读数、点赞数）必须在微信登录态下才能返回，未登录时 read_num、like_num 不返回，wxtoken 为 777。

新增了 appmsg_like_type、req_id 等参数，点赞机制有更新。

appmsg_token 有效期缩短到 2-4 小时，pass_ticket 更短，批量抓取建议每次不超过 50 篇，超时需重新抓包。

获取单篇文章互动数据的最小示例

python

import requests
import time

# 从 Fiddler / Charles 抓包 getappmsgext 请求，复制 Cookie、token、pass_ticket 等参数
session = requests.Session()
session.headers.update({
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)",
    "Referer": "https://mp.weixin.qq.com/",
    "Cookie": "你的_cookie=xxx; wap_sid2=xxx; ...",   # ⚠️ 必须为登录态 Cookie
})

url = "https://mp.weixin.qq.com/mp/getappmsgext"
params = {
    "__biz": "你的__biz",          # 公众号标识
    "appmsg_type": "9",            # 固定值
    "mid": "你的_mid",             # 文章标识
    "sn": "你的_sn",               # 文章标识
    "idx": "1",                    # 在公众号中的序号
    "scene": "27",
    "title": "文章标题(URL编码)",
    "ct": str(int(time.time())),
    "is_need_ad": "0",
    "is_need_reward": "1",
    "is_only_read": "1",           # 新版参数：仅获取阅读数
    "appmsg_like_type": "2",       # 新版点赞机制
    "comment_id": "",
    "f": "json",
    "r": "0.8",                    # 随机数，防重放
}
data = {
    "key": "你的_key",             # 从抓包获取
    "pass_ticket": "你的_pass_ticket",
    "appmsg_token": "你的_appmsg_token",   # 有效期 2-4 小时，失效需重新抓包
    "x5": "0",
    "f": "json",
    "req_id": "你的_req_id",       # 防重放随机 ID
}

resp = session.post(url, params=params, data=data)
result = resp.json()

# 解析互动数据（以抓包返回的实际字段为准）
if result.get("appmsgstat", {}).get("is_login"):
    stat = result["appmsgstat"]
    print("阅读数:", stat.get("read_num"))
    print("点赞数:", stat.get("like_num"))
    print("真实阅读数:", stat.get("real_read_num"))
else:
    print("未获取到登录态数据，请检查 Cookie 是否过期：", result.get("base_resp"))

说明：

互动数据（阅读数、点赞数）必须在微信登录态下才能返回，未登录时 read_num、like_num 不返回，wxtoken 为 777。

appmsg_token 有效期仅 2-4 小时，pass_ticket 更短，批量抓取建议每次不超过 50 篇，超时后重新抓包。

单篇文章循环调用此请求即可批量更新互动数据。

批量更新互动数据

实际项目中，可以从 MongoDB 中读出已存储的文章，循环调用 getappmsgext 更新每篇文章的互动数据。核心逻辑与上例一致，只需遍历文章列表，对每篇构造对应的 mid、sn 请求即可。注意控制频率并在 Cookie 失效时及时刷新。

小结

本文完成了公众号爬虫的完整闭环：

抓取第一篇文章：定位 profile_ext 接口，解析 general_msg_list 提取文章字段。
抓取全部历史文章：通过 action=getmsg + offset 翻页，can_msg_continue 判断结束。
存储到 MongoDB：用 MongoEngine 定义模型，sub_dict 提取字段并入库。
获取互动数据：通过 getappmsgext 接口在登录态下获取阅读数、点赞数等。

整个流程的关键在于实时抓包获取最新参数，并在参数失效后及时刷新。下一节将对抓取到的数据进行分析与可视化。

版本差异（爬虫技术栈 → 当前版本）

库	本文编写时	当前稳定版
`requests`	2.28/2.31	2.32.x
`Scrapy`	1.x/2.0	2.11.x（API 稳定）
`httpx`	0.24	0.28.x
`Playwright`	1.3x	1.6x（Python 版）
`lxml`/`BeautifulSoup`	旧版	保持稳定
Python	3.8-3.12	3.14（推荐）

本文讲解的爬虫原理（HTTP、解析、反爬、存储）与核心 API 在最新版本中成立；注意 Python 3.9 及以下已 EOL，新项目使用 3.13/3.14。