自从发布了 「原神」细节向初体验 这篇文章之后，粉丝朋友们纷纷感叹“原来你也（开始）玩原神”。

不过刚开始是因为等级不够，后来是找不到人，再后来是在做主线，目前我还没和别人联机过。

入坑一个多月一来，身边的朋友不停的向我安利米游社这个 App，终于，我下载下来看了看。

不得不说，如果是重度玩家的话，这个 App 确实能提升游戏体验，不过作为资深观景玩家，大地图是不可能用的。

于是我盯上了首页的信息流，想着身为技术人，不爬点数据下来有点对不起米哈游的 Slogan“技术宅拯救世界”，于是，开干。

一开始就选错了方向

米游社的内容主要集中在在手机端，而在 App 的动态数据获取方面，各大厂商的实现方案都大差不差，无非是请求接口、获取数据、展示界面。

于是我打开了 HttpCanary（一个安卓端网络抓包工具），一波操作之后，打开米游社 App，下滑加载内容。

然后跳出了网络连接失败的提示。

JRT 技术验证时，我对简书 App 抓包就遇到过类似的问题，很明显，这是 SSL 中间人攻击防护。

简单来说，App 会对服务端的 SSL 证书进行校验，如果不匹配，说明在这条连接中间，有节点在篡改数据。

这种情况并非无解，但需要对设备进行 Root，将抓包工具的证书添加到信任列表，或者对 App 进行反编译。前者费时费力，后者技术难度高且有法律风险，看来这条路行不通。

在我上网搜索相关资料的时候，无意间发现米游社有网页端，而且我要的信息流数据在网页端同样有展示。

技术难度一下子就下来了，只需要分析网络请求，然后针对性提取数据即可。

网络请求分析

打开米游社网页端（这里我们要爬的是原神区）：https://bbs.mihoyo.com/ys/

F12 调出开发者工具，然后...

进入了调试模式，数据根本没加载出来。

这是一种很常见的反爬措施，原理大概是这样：开发者工具打开的时候，遇到 JS 代码中的调试器（Debugger）语句就会暂停，否则跳过这段代码继续执行，只需要通过某种方式不断尝试打开调试器（比如死循环），就可以让我们打开开发者工具时无法正常获取数据。

解决方法也很简单，只需要点击这个按钮：

这个按钮会禁用掉断点调试功能，开启之后刷新网页，就可以正常获取到数据了。

切换到网络选项卡，筛选异步请求（Fetch/XHR）：

向下滚动页面，加载新内容，观察请求面板的变化：

很明显，红圈中的两个请求是加载时发起的。

查看请求参数，不难发现第二个请求的作用是根据帖子 ID 获取互动数据，我们暂且放在一边，主要关注第一个请求：

最近发现了一个很好用的网络请求工具：Hoppscotch，我们将请求信息复制进去，点击发送。

Bingo，响应数据出来了。看到这里，我不得不感叹一句，在游戏上米哈游算是同赛道顶尖，但在数据安全这方面，未免有些太过草率了。

响应数据结构分析

折叠具体数据，只查看结构部分：

我们来逐个分析。

retcode，猜一波是 return code（返回代码）的缩写，很明显是状态码，0 一般代表正常，类似 HTTP 状态码中的 200。

message，消息，是对状态码的描述，这里是 OK，印证了我们的猜测，一切正常。

data 里面就是我们要的数据了。

carousels，翻译一下是“旋转木马”，这个网页中什么东西是旋转的？答案是轮播图。

cover，遮罩，值是一个图片地址，访问一下试试：

猜对了，正是首页轮播图。

recommended_posts，对应帖子数据：

recommended_topics，对应推荐话题数据，位于网页的右侧边栏：

fixed_posts，可能是置顶帖子数据，空的，暂且不去理会。

selection_post_list，里面的帖子格式与 recommended_posts 不同，且没有规律，为简化数据获取流程，可以忽略。

我们需要的是帖子数据，也就是 recommended_posts 中的内容。

帖子数据结构分析

随便选一条帖子数据，与网页上展示的内容进行比对：

可以看到，一条帖子的数据分为十几个部分，我们将对此一一说明。

post（帖子数据）

• game_id：游戏 ID，2 代表原神

• post_id：帖子 ID，唯一标识

• f_forum_id：论坛 ID，唯一标识

• uid：用户 ID，唯一标识

• subject：标题

• content：简介

• cover：题头图链接

• view_type：可能和访问方式有关，这里恒为 1

• created_at：创建时间，UNIX 时间戳格式，这里的时间为 2022 年 3 月 12 日 20:21:32

• images：图片数据，内部是图片链接

• post_status：帖子状态：

  • is_top：是否被置顶

  • is_good：是否被加精

  • is_official：是否为官方帖子

• topic_ids：所属的话题 ID

• view_status：可能和帖子可见性状态有关（正常、限流、被屏蔽等）

• max_floor：评论层数

• is_original：是否为原创

• republish_authorization：可能与转载授权类型有关

• reply_time：最后一次评论时间

• is_deleted：是否被删除

• is_interactive：是否允许互动

• score：可能和评分有关

forum（论坛数据）

• id：论坛 ID，唯一标识，与 post 中的 f_forum_id 相同

• name：论坛名称

topics（话题数据）

• id：话题 ID，唯一标识

• name：话题名称

• cover：话题题头图

• content_type：可能和话题的类型有关

user（用户数据）

• uid：用户 ID，唯一标识，与 post 中的 uid 相同

• nickname：用户昵称

• introduce：个人简介

• avatar：可能和用户头像有关（米游社不能自行上传头像，只能使用游戏中的角色图片作为头像，可选数量有限）

• gender：性别

• certification：认证称号

  • type：认证称号种类

  • label：认证称号名称

• level_exp：等级与经验

  • level：等级

  • exp：经验

• avatar_url：头像链接

• pendant：头像挂链接

stat（互动数据）

这几项数据不知道为什么均为 0，我们会在后面用另一个接口补全这几项数据。

• reply_num：评论量

• view_num：阅读量

• like_num：点赞量

• bookmark_num：收藏量

cover（题头图数据）

• url：题头图链接

• height：图片高度

• width： 图片宽度

• format：图片格式

• size：图片大小（字节）

• crop：裁剪数据

  • x：裁剪开始的横向坐标

  • y：裁剪开始的纵向坐标

  • w：裁剪宽度

  • h：裁剪高度

  • url：加入裁剪参数的题头图链接（实际上是阿里云对象存储的图片处理功能）

• is_user_set_cover：是否由用户设置题头图

• image_id：图片 ID，唯一标识

• entity_type：实体类型，含义未知

• entity_id：实体 ID，含义未知

image_list（图片数据）

各项数据含义与 cover 相同，不再重复说明。

其它数据

• self_operation：自营，含义未知

• is_official_master：是否为官方管理员

• is_user_master：是否为非官方管理员

• help_sys：帮助系统，含义未知

  • top_up：含义未知

• vote_count：票数，可能和论坛活动有关

• last_modify_time：最后一次更新时间（不正常数据，不应为 0）

• recommend_type：推荐类型

• collection：专题数据

构建数据库表结构

我们使用 Python 的 ORM 库 Peewee 与 SQLite 数据库进行交互，简化数据保存流程。

数据库定义相关代码存放在 db_config.py 文件中。

为了降低数据分析难度，我们将采集的内容分为几个部分，分别建立不同的表来存储：

• 帖子数据
• 论坛数据
• 用户数据
• 话题数据
• 图片数据
• 头像数据
• 认证称号数据

使用外键连接这些表，这在 SQL 中是一个稍显复杂的操作，但 Peewee 帮我们抽象了这个操作，我们只需指定字段名称、引用的表和这个字段在引用表中的名称即可。

从 peewee 库中导入我们使用的字段，并初始化一个名为 data.db 的 SQLite 数据库。

from peewee import (BooleanField, CharField, DateTimeField, IntegerField, Model, SqliteDatabase)

db = SqliteDatabase("data.db")

首先是帖子数据：

class Post(Model):
    id = IntegerField(primary_key=True)
    title = CharField()
    summary = CharField()
    content = CharField()
    created_time = DateTimeField()
    is_topped = BooleanField()
    is_best = BooleanField()
    is_official = BooleanField()
    is_original = BooleanField()
    is_deleted = BooleanField()
    is_interactive = BooleanField()
    visible_status = IntegerField()
    comments_count = IntegerField()
    republish_authorization = IntegerField()
    last_comment_time = DateTimeField()
    score = IntegerField()
    views_count = IntegerField()
    likes_count = IntegerField()
    comments_count = IntegerField()
    bookmarks_count = IntegerField()

    class Meta:
        database = db
        table_name = "posts"

论坛数据：

class Forum(Model):
    post = ForeignKeyField(Post, backref="forum")
    id = IntegerField()
    name = CharField()

    class Meta:
        database = db
        table_name = "forums"

用户数据：

class User(Model):
    post = ForeignKeyField(Post, backref="user")
    id = IntegerField()
    name = CharField()
    gender = IntegerField()
    introduction = CharField()
    pendant_url = CharField()
    level = IntegerField()
    exp = IntegerField()

    class Meta:
        database = db
        table_name = "users"

话题数据：

class Topic(Model):
    post = ForeignKeyField(Post, backref="topics")
    id = IntegerField(primary_key=True)
    name = CharField()
    cover_url = CharField()
    content_type = IntegerField()

    class Meta:
        database = db
        table_name = "topics"

图片数据：

class Image(Model):
    post = ForeignKeyField(Post, backref="images")
    id = IntegerField(primary_key=True)
    url = CharField()
    width = IntegerField()
    height = IntegerField()
    format = CharField()
    size = IntegerField()

    class Meta:
        database = db
        table_name = "images"

头像数据：

class Avatar(Model):
    user = ForeignKeyField(User, backref="avatar")
    id = IntegerField(primary_key=True)
    url = CharField()

    class Meta:
        database = db
        table_name = "avatars"

认证数据：

class Certification(Model):
    user = ForeignKeyField(User, backref="certification")
    id = IntegerField()
    name = CharField()

    class Meta:
        database = db
        table_name = "certifications"

编写数据库初始化函数并运行：

def InitDB():
    db.connect()
    db.create_tables([Post, Forum, User, Topic, Image, Avatar, Certification])

InitDB()

程序运行后，目录中会多出一个 data.db 文件，使用数据库管理工具打开，表结构正如我们所愿。

解析数据

新建 data_parse.py 文件，在其中编写我们的数据处理逻辑，以解析帖子数据为例：

def ParsePostData(json_data: Dict) -> Dict:
    return {
        "id": int(json_data["post"]["post_id"]),
        "title": json_data["post"]["subject"],
        "summary": json_data["post"]["content"],
        "content": json_data["post"]["full_content"],
        "created_time": datetime.fromtimestamp(json_data["post"]["created_at"]),
        "is_topped": json_data["post"]["post_status"]["is_top"],
        "is_best": json_data["post"]["post_status"]["is_good"],
        "is_official": json_data["post"]["post_status"]["is_official"],
        "is_original": json_data["post"]["is_original"],
        "is_deleted": bool(json_data["post"]["is_deleted"]),
        "is_interactive": json_data["post"]["is_interactive"],
        "visible_status": json_data["post"]["view_status"],
        "comments_count": json_data["post"]["max_floor"],
        "republish_authorization": json_data["post"]["republish_authorization"],
        "last_comment_time": datetime.fromisoformat(json_data["post"]["reply_time"]),
        "score": json_data["post"]["score"],
        "views_count": json_data["stat"]["view_num"],
        "likes_count": json_data["stat"]["like_num"],
        "comments_count": json_data["stat"]["reply_num"],
        "bookmarks_count": json_data["stat"]["bookmark_num"]
    }

这个函数接收数据字典，并以字典形式返回提取后的数据。

类似的，我们可以编写出论坛、用户、话题、图片、用户头像、用户认证这几类数据的解析函数。

使用一个函数对完整的数据字典进行解析：

def ParseData(json_data: Dict) -> Dict:
    return {
        "post_data": ParsePostData(json_data),
        "forum_data": ParseForumData(json_data),
        "user_data": ParseUserData(json_data),
        "topics_data": ParseTopicsData(json_data),
        "images_data": ParseImagesData(json_data),
        "user_avatar_data": ParseUserAvatarData(json_data),
        "user_certification_data": ParseUserCertificationData(json_data)
    }

由于发起一次网络请求时，会获取到一组数据，为了简化主逻辑，我们编写一个解析数据列表的函数：

def ParseDataList(data_list: List[Dict]) -> List[Dict]:
    return [ParseData(data) for data in data_list]

到现在，我们的程序已经可以实现数据的获取与解析了。

数据预处理

由于信息流接口获取到的数据存在一些问题，比如互动数据异常，缺少帖子完整内容等，我们需要对请求到的数据进行一些处理，替换掉错误的数据，加入我们希望获取的数据。

替换互动数据

首先，在 data_fetch.py 中增加一个数据获取函数：

def GetInteractiveData(post_ids: List[str]) -> Dict[int, Dict[str, int]]:
    url = "https://bbs-api.mihoyo.com/post/wapi/getDynamicData"
    params = {
        "gids": 2,
        "post_ids": ",".join(post_ids)
    }
    response = httpx_get(url, params=params)

    response.raise_for_status()
    data = response.text
    try:
        data = ujson_loads(data)
    except ValueError:
        raise ValueError("解析互动数据 Json 时出现异常")

    if data["retcode"] != 0:
        raise ValueError(f"获取互动数据时出现异常，错误码：{data['retcode']}，错误信息：{data['message']}")

    result = {}
    for item in data["data"]["list"]:
        result[item["post_id"]] = item["stat"]
    return result

这个接口支持一次获取一组动态数据，只需要将对应帖子的 post_id 以英文逗号分隔的格式作为接口的 post_ids 参数即可。

为了确保函数封装良好，我们将函数的参数指定为由字符串形式的 post_id 组成的列表。（因为信息流接口返回的 post_id 是字符串类型）

接下来，在 data_process.py 文件中编写一个函数，将传入的 data 列表中的所有互动数据替换成对应的正确数据：

def ReplaceInteractiveData(data: List[Dict]) -> List[Dict]:
    post_ids = (item["post"]["post_id"] for item in data)
    interactive_data = GetInteractiveData(post_ids)

    result = []
    for item in data:
        item["stat"] = interactive_data[item["post"]["post_id"]]
        result.append(item)
    return result

加入帖子完整内容数据

同样的，在 data_fetch.py 中定义一个函数：

def GetPostContent(post_id: int) -> str:
    url = "https://bbs-api.mihoyo.com/post/wapi/getPostFull"
    params = {
        "gids": 2,
        "post_id": post_id
    }
    headers = {
        "Referer": f"https://bbs.mihoyo.com/ys/article/{post_id}"
    }
    response = httpx_get(url, params=params, headers=headers)

    response.raise_for_status()
    data = response.text
    try:
        data = ujson_loads(data)
    except ValueError:
        raise ValueError("解析帖子全文 Json 时出现异常")

    if data["retcode"] != 0:
        raise ValueError(f"获取帖子全文时出现异常，错误码：{data['retcode']}，错误信息：{data['message']}")

    return data["data"]["post"]["post"]["content"]

这里我们遇到了一些问题，调试这个接口的时候会出现 403 错误，我们将常见反爬措施会校验的内容（Cookie / UA / Referer 等）全部复制到请求工具中，再次请求发现数据正常返回。

之后我们逐一删除这些内容，最后发现，添加 Referer 即可规避这一反爬措施。

同样的，编写一个函数对原先的 data 列表进行替换：

def AddFullContentData(data: List[Dict]) -> List[Dict]:
    post_ids = (item["post"]["post_id"] for item in data)

    contents_list = Parallel(n_jobs=10)(delayed(GetPostContent)(post_id) for post_id in post_ids)
    result = []
    for index, item in enumerate(data):
        item["post"]["full_content"] = contents_list[index]
        result.append(item)
    return result

由于这个接口一次只能获取一条数据，我们需要使用多线程来提高程序的运行效率。

过早的优化是万恶之源，如果不能确定这个函数会拖慢程序的运行速度，可以先使用单线程请求，后期通过性能分析找到瓶颈，再针对性优化。

这里使用的是 Python 的内置库 joblib，上述代码将使用 10 个线程对 GetPostContent 函数发起调用，并将结果以正确的顺序存入 contents_list 中。

之后我们通过 for 循环，将每条帖子对应的 content 插入到其数据的 post 项中。

数据存储

由于我们在数据解析过程中，就将解析结果字典的键与数据库的字段进行了一一对应，所以我们可以直接使用字典解包，将键值对变为参数，传入 Peewee 的 create 函数中，从而快速实现数据的存储。

示例代码如下：

def SavePostData(post_data: Dict) -> Post:
    return Post.create(**post_data)

但在对论坛数据进行保存时，我们遇到了一个问题：将要被保存的数据可能已经存在于数据库中，这样会因为主键重复而产生异常。

因此，我们需要对主键重复产生的 IntegrityError 进行捕获，并将其忽略：

def SaveForumData(forum_data: Dict, post_obj: Post) -> None:
    try:
        Forum.create(post=post_obj, **forum_data)
    except IntegrityError:
        pass

类似的，我们可以编写出其它类型数据的存储函数。

最后，用一个函数将它们聚合起来：

def SaveData(data: Dict):
    with db.atomic():  # 开启事务
        post_obj = SavePostData(data["post_data"])
        if data["forum_data"]:  # 如果板块数据不为空
            SaveForumData(data["forum_data"], post_obj)
        user_obj = SaveUserData(data["user_data"], post_obj)
        SaveTopicsData(data["topics_data"], post_obj)
        SaveImagesData(data["images_data"], post_obj)
        SaveAvatarData(data["user_avatar_data"], user_obj)
        SaveCertificationData(data["user_certification_data"], user_obj)

这里我们使用了数据库的事务功能，在事务中的数据库操作，只可能全部成功或者全部失败。

这样做由两个原因，其一，可以防止程序出错时对数据库造成污染；其二，通过减少提交操作（Commit），可以提高数据存储的性能。

和 data_parse.py 中一样，我们可以对一组数据进行保存：

def SaveDataList(data_list: List[Dict]):
    for data in data_list:
        try:
            SaveData(data)
        except IntegrityError:  # 主键重复
            print(f"帖子 {data['post_data']['id']} 出现重复，已自动跳过")

这个函数考虑到了数据在获取过程中产生变动，从而导致主键重复时的处理。

主逻辑

导入库部分省略。

我们先来定义一些常量：

TOTAL_DATA_COUNT = 30000
DATA_COUNT_PER_PAGE = 30
SLEEP_TIME = 0
ERROR_SLEEP_TIME = 20
DATA_SAVE_INTERVAL = 20

这样做可以帮助我们快速修改运行参数，而不需要对使用到这些参数的每一个位置进行改动。

如果可调整的参数超过 10 个，或者这个爬虫需要定期运行，最好使用配置文件（比如 YAML）进行管理。

由于数据保存需要消耗较长时间，我们将其抽离成独立的一个线程，因此，需要定义一个列表，用来存放待保存的数据：

data_to_save_list: List[Dict] = []
data_to_save_list_lock = Lock()

接下来是数据保存线程：

def DataSaveJob():
    while True:
        if data_to_save_list:
            with data_to_save_list_lock:
                start_time = time()
                SaveDataList(data_to_save_list)
                print(f"已成功保存 {len(data_to_save_list)} 条数据，耗时 {round(time() - start_time, 2)} 秒")
                data_to_save_list.clear()
        sleep(DATA_SAVE_INTERVAL)

由于数据的获取、保存到待保存列表的清空需要一定时间，在此期间，如果有新数据加入列表，就会导致未保存数据的丢失，因此，我们需要在数据保存期间对数据进行加锁。

接下来，我们对需要采集的页数进行计算，并对必要的数据进行校验，最后初始化数据库。

然后启动数据保存线程：

data_save_thread = Thread(target=DataSaveJob, daemon=True)  # 设置为守护线程，避免影响主线程退出
data_save_thread.start()
print("数据获取线程启动成功...")

核心采集逻辑如下：

data = GetMainData(page, DATA_COUNT_PER_PAGE)
data = ReplaceInteractiveData(data)
data = AddFullContentData(data)
data = ParseDataList(data)

这里省略了关于出错重试的逻辑。

然后将数据加入待保存列表：

with data_to_save_list_lock:
    data_to_save_list.extend(data)

在数据全部采集完毕后，我们需要等待保存线程将它们全部存入数据库：

print("等待数据存储完成...")
while True:
    if not data_to_save_list:
        with data_to_save_list_lock:  # 获取到锁则证明全部数据已保存完毕
            print("数据存储完成...")
            exit()
    else:
        sleep(1)

运行程序，输出如下：

总数据量：30000   单页数据个数：30   总页数：1000
等待时间：0s   错误重试间隔：20s   数据保存间隔：20s
初始化数据库成功...
数据获取线程启动成功...
开始采集数据...
开始采集第 1 页
第 1 页采集成功，耗时 4.39 秒
开始采集第 2 页
第 2 页采集成功，耗时 2.67 秒
开始采集第 3 页
第 3 页采集成功，耗时 2.57 秒
开始采集第 4 页
第 4 页采集成功，耗时 2.42 秒
开始采集第 5 页
第 5 页采集成功，耗时 2.48 秒
开始采集第 6 页
第 6 页采集成功，耗时 2.07 秒
开始采集第 7 页
第 7 页采集成功，耗时 2.16 秒
开始采集第 8 页
已成功保存 210 条数据，耗时 3.91 秒
第 8 页采集成功，耗时 5.24 秒
（以下省略）

采集结果

本以为数据量至少在十万量级，没想到爬到五百多页就开始连续报错，手动请求接口发现已经没有新的数据返回，排除掉反爬原因后，可以确定是数据已经被爬取完成。

我采集数据的时间是 2022 年 3 月 27 日，data.db 文件的大小为 71.6MB，总数据量 14997 条。

后记

本文中，我们对米游社的动态加载请求进行了分析，设计了保存这些数据的数据库结构，通过自动化请求接口实现了数据的获取、解析与保存。

感兴趣的小伙伴可以自行探索以下内容：

• 将本文中的数据获取相关代码改写为异步形式，提升网络请求性能
• 优化数据库操作性能
• 使用装饰器实现数据预处理
• 使用 addict 库改写数据解析逻辑，增强可读性
• 使用 rich 库输出不同颜色的终端信息

文中的程序会在本系列完结后开源，届时仓库地址将更新在此处。

在接下来的文章中，我们将对这些数据进行进一步的分析。