自從發(fā)布了 「原神」細(xì)節(jié)向初體驗(yàn) 這篇文章之后，粉絲朋友們紛紛感嘆“原來你也（開始）玩原神”。

不過剛開始是因?yàn)榈燃?jí)不夠，后來是找不到人，再后來是在做主線，目前我還沒和別人聯(lián)機(jī)過。

入坑一個(gè)多月一來，身邊的朋友不停的向我安利米游社這個(gè) App，終于，我下載下來看了看。

不得不說，如果是重度玩家的話，這個(gè) App 確實(shí)能提升游戲體驗(yàn)，不過作為資深觀景玩家，大地圖是不可能用的。

于是我盯上了首頁(yè)的信息流，想著身為技術(shù)人，不爬點(diǎn)數(shù)據(jù)下來有點(diǎn)對(duì)不起米哈游的 Slogan“技術(shù)宅拯救世界”，于是，開干。

一開始就選錯(cuò)了方向

米游社的內(nèi)容主要集中在在手機(jī)端，而在 App 的動(dòng)態(tài)數(shù)據(jù)獲取方面，各大廠商的實(shí)現(xiàn)方案都大差不差，無(wú)非是請(qǐng)求接口、獲取數(shù)據(jù)、展示界面。

于是我打開了 HttpCanary（一個(gè)安卓端網(wǎng)絡(luò)抓包工具），一波操作之后，打開米游社 App，下滑加載內(nèi)容。

然后跳出了網(wǎng)絡(luò)連接失敗的提示。

JRT 技術(shù)驗(yàn)證時(shí)，我對(duì)簡(jiǎn)書 App 抓包就遇到過類似的問題，很明顯，這是 SSL 中間人攻擊防護(hù)。

簡(jiǎn)單來說，App 會(huì)對(duì)服務(wù)端的 SSL 證書進(jìn)行校驗(yàn)，如果不匹配，說明在這條連接中間，有節(jié)點(diǎn)在篡改數(shù)據(jù)。

這種情況并非無(wú)解，但需要對(duì)設(shè)備進(jìn)行 Root，將抓包工具的證書添加到信任列表，或者對(duì) App 進(jìn)行反編譯。前者費(fèi)時(shí)費(fèi)力，后者技術(shù)難度高且有法律風(fēng)險(xiǎn)，看來這條路行不通。

在我上網(wǎng)搜索相關(guān)資料的時(shí)候，無(wú)意間發(fā)現(xiàn)米游社有網(wǎng)頁(yè)端，而且我要的信息流數(shù)據(jù)在網(wǎng)頁(yè)端同樣有展示。

技術(shù)難度一下子就下來了，只需要分析網(wǎng)絡(luò)請(qǐng)求，然后針對(duì)性提取數(shù)據(jù)即可。

網(wǎng)絡(luò)請(qǐng)求分析

打開米游社網(wǎng)頁(yè)端（這里我們要爬的是原神區(qū)）：https://bbs.mihoyo.com/ys/

F12 調(diào)出開發(fā)者工具，然后...

進(jìn)入了調(diào)試模式，數(shù)據(jù)根本沒加載出來。

這是一種很常見的反爬措施，原理大概是這樣：開發(fā)者工具打開的時(shí)候，遇到 JS 代碼中的調(diào)試器（Debugger）語(yǔ)句就會(huì)暫停，否則跳過這段代碼繼續(xù)執(zhí)行，只需要通過某種方式不斷嘗試打開調(diào)試器（比如死循環(huán)），就可以讓我們打開開發(fā)者工具時(shí)無(wú)法正常獲取數(shù)據(jù)。

解決方法也很簡(jiǎn)單，只需要點(diǎn)擊這個(gè)按鈕：

這個(gè)按鈕會(huì)禁用掉斷點(diǎn)調(diào)試功能，開啟之后刷新網(wǎng)頁(yè)，就可以正常獲取到數(shù)據(jù)了。

切換到網(wǎng)絡(luò)選項(xiàng)卡，篩選異步請(qǐng)求（Fetch/XHR）：

向下滾動(dòng)頁(yè)面，加載新內(nèi)容，觀察請(qǐng)求面板的變化：

很明顯，紅圈中的兩個(gè)請(qǐng)求是加載時(shí)發(fā)起的。

查看請(qǐng)求參數(shù)，不難發(fā)現(xiàn)第二個(gè)請(qǐng)求的作用是根據(jù)帖子 ID 獲取互動(dòng)數(shù)據(jù)，我們暫且放在一邊，主要關(guān)注第一個(gè)請(qǐng)求：

最近發(fā)現(xiàn)了一個(gè)很好用的網(wǎng)絡(luò)請(qǐng)求工具：Hoppscotch，我們將請(qǐng)求信息復(fù)制進(jìn)去，點(diǎn)擊發(fā)送。

Bingo，響應(yīng)數(shù)據(jù)出來了?？吹竭@里，我不得不感嘆一句，在游戲上米哈游算是同賽道頂尖，但在數(shù)據(jù)安全這方面，未免有些太過草率了。

響應(yīng)數(shù)據(jù)結(jié)構(gòu)分析

折疊具體數(shù)據(jù)，只查看結(jié)構(gòu)部分：

我們來逐個(gè)分析。

retcode，猜一波是 return code（返回代碼）的縮寫，很明顯是狀態(tài)碼，0 一般代表正常，類似 HTTP 狀態(tài)碼中的 200。

message，消息，是對(duì)狀態(tài)碼的描述，這里是 OK，印證了我們的猜測(cè)，一切正常。

data 里面就是我們要的數(shù)據(jù)了。

carousels，翻譯一下是“旋轉(zhuǎn)木馬”，這個(gè)網(wǎng)頁(yè)中什么東西是旋轉(zhuǎn)的？答案是輪播圖。

cover，遮罩，值是一個(gè)圖片地址，訪問一下試試：

猜對(duì)了，正是首頁(yè)輪播圖。

recommended_posts，對(duì)應(yīng)帖子數(shù)據(jù)：

recommended_topics，對(duì)應(yīng)推薦話題數(shù)據(jù)，位于網(wǎng)頁(yè)的右側(cè)邊欄：

fixed_posts，可能是置頂帖子數(shù)據(jù)，空的，暫且不去理會(huì)。

selection_post_list，里面的帖子格式與 recommended_posts 不同，且沒有規(guī)律，為簡(jiǎn)化數(shù)據(jù)獲取流程，可以忽略。

我們需要的是帖子數(shù)據(jù)，也就是 recommended_posts 中的內(nèi)容。

帖子數(shù)據(jù)結(jié)構(gòu)分析

隨便選一條帖子數(shù)據(jù)，與網(wǎng)頁(yè)上展示的內(nèi)容進(jìn)行比對(duì)：

可以看到，一條帖子的數(shù)據(jù)分為十幾個(gè)部分，我們將對(duì)此一一說明。

post（帖子數(shù)據(jù)）

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

• post_id：帖子 ID，唯一標(biāo)識(shí)

• f_forum_id：論壇 ID，唯一標(biāo)識(shí)

• uid：用戶 ID，唯一標(biāo)識(shí)

• subject：標(biāo)題

• content：簡(jiǎn)介

• cover：題頭圖鏈接

• view_type：可能和訪問方式有關(guān)，這里恒為 1

• created_at：創(chuàng)建時(shí)間，UNIX 時(shí)間戳格式，這里的時(shí)間為 2022 年 3 月 12 日 20:21:32

• images：圖片數(shù)據(jù)，內(nèi)部是圖片鏈接

• post_status：帖子狀態(tài)：

  • is_top：是否被置頂

  • is_good：是否被加精

  • is_official：是否為官方帖子

• topic_ids：所屬的話題 ID

• view_status：可能和帖子可見性狀態(tài)有關(guān)（正常、限流、被屏蔽等）

• max_floor：評(píng)論層數(shù)

• is_original：是否為原創(chuàng)

• republish_authorization：可能與轉(zhuǎn)載授權(quán)類型有關(guān)

• reply_time：最后一次評(píng)論時(shí)間

• is_deleted：是否被刪除

• is_interactive：是否允許互動(dòng)

• score：可能和評(píng)分有關(guān)

forum（論壇數(shù)據(jù)）

• id：論壇 ID，唯一標(biāo)識(shí)，與 post 中的 f_forum_id 相同

• name：論壇名稱

topics（話題數(shù)據(jù)）

• id：話題 ID，唯一標(biāo)識(shí)

• name：話題名稱

• cover：話題題頭圖

• content_type：可能和話題的類型有關(guān)

user（用戶數(shù)據(jù)）

• uid：用戶 ID，唯一標(biāo)識(shí)，與 post 中的 uid 相同

• nickname：用戶昵稱

• introduce：個(gè)人簡(jiǎn)介

• avatar：可能和用戶頭像有關(guān)（米游社不能自行上傳頭像，只能使用游戲中的角色圖片作為頭像，可選數(shù)量有限）

• gender：性別

• certification：認(rèn)證稱號(hào)

  • type：認(rèn)證稱號(hào)種類

  • label：認(rèn)證稱號(hào)名稱

• level_exp：等級(jí)與經(jīng)驗(yàn)

  • level：等級(jí)

  • exp：經(jīng)驗(yàn)

• avatar_url：頭像鏈接

• pendant：頭像掛鏈接

stat（互動(dòng)數(shù)據(jù)）

這幾項(xiàng)數(shù)據(jù)不知道為什么均為 0，我們會(huì)在后面用另一個(gè)接口補(bǔ)全這幾項(xiàng)數(shù)據(jù)。

• reply_num：評(píng)論量

• view_num：閱讀量

• like_num：點(diǎn)贊量

• bookmark_num：收藏量

cover（題頭圖數(shù)據(jù)）

• url：題頭圖鏈接

• height：圖片高度

• width： 圖片寬度

• format：圖片格式

• size：圖片大?。ㄗ止?jié)）

• crop：裁剪數(shù)據(jù)

  • x：裁剪開始的橫向坐標(biāo)

  • y：裁剪開始的縱向坐標(biāo)

  • w：裁剪寬度

  • h：裁剪高度

  • url：加入裁剪參數(shù)的題頭圖鏈接（實(shí)際上是阿里云對(duì)象存儲(chǔ)的圖片處理功能）

• is_user_set_cover：是否由用戶設(shè)置題頭圖

• image_id：圖片 ID，唯一標(biāo)識(shí)

• entity_type：實(shí)體類型，含義未知

• entity_id：實(shí)體 ID，含義未知

image_list（圖片數(shù)據(jù)）

各項(xiàng)數(shù)據(jù)含義與 cover 相同，不再重復(fù)說明。

其它數(shù)據(jù)

• self_operation：自營(yíng)，含義未知

• is_official_master：是否為官方管理員

• is_user_master：是否為非官方管理員

• help_sys：幫助系統(tǒng)，含義未知

  • top_up：含義未知

• vote_count：票數(shù)，可能和論壇活動(dòng)有關(guān)

• last_modify_time：最后一次更新時(shí)間（不正常數(shù)據(jù)，不應(yīng)為 0）

• recommend_type：推薦類型

• collection：專題數(shù)據(jù)

構(gòu)建數(shù)據(jù)庫(kù)表結(jié)構(gòu)

我們使用 Python 的 ORM 庫(kù) Peewee 與 SQLite 數(shù)據(jù)庫(kù)進(jìn)行交互，簡(jiǎn)化數(shù)據(jù)保存流程。

數(shù)據(jù)庫(kù)定義相關(guān)代碼存放在 db_config.py 文件中。

為了降低數(shù)據(jù)分析難度，我們將采集的內(nèi)容分為幾個(gè)部分，分別建立不同的表來存儲(chǔ)：

• 帖子數(shù)據(jù)
• 論壇數(shù)據(jù)
• 用戶數(shù)據(jù)
• 話題數(shù)據(jù)
• 圖片數(shù)據(jù)
• 頭像數(shù)據(jù)
• 認(rèn)證稱號(hào)數(shù)據(jù)

使用外鍵連接這些表，這在 SQL 中是一個(gè)稍顯復(fù)雜的操作，但 Peewee 幫我們抽象了這個(gè)操作，我們只需指定字段名稱、引用的表和這個(gè)字段在引用表中的名稱即可。

從 peewee 庫(kù)中導(dǎo)入我們使用的字段，并初始化一個(gè)名為 data.db 的 SQLite 數(shù)據(jù)庫(kù)。

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

db = SqliteDatabase("data.db")

首先是帖子數(shù)據(jù)：

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"

論壇數(shù)據(jù)：

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

    class Meta:
        database = db
        table_name = "forums"

用戶數(shù)據(jù)：

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"

話題數(shù)據(jù)：

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"

圖片數(shù)據(jù)：

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"

頭像數(shù)據(jù)：

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

    class Meta:
        database = db
        table_name = "avatars"

認(rèn)證數(shù)據(jù)：

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

    class Meta:
        database = db
        table_name = "certifications"

編寫數(shù)據(jù)庫(kù)初始化函數(shù)并運(yùn)行：

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

InitDB()

程序運(yùn)行后，目錄中會(huì)多出一個(gè) data.db 文件，使用數(shù)據(jù)庫(kù)管理工具打開，表結(jié)構(gòu)正如我們所愿。

解析數(shù)據(jù)

新建 data_parse.py 文件，在其中編寫我們的數(shù)據(jù)處理邏輯，以解析帖子數(shù)據(jù)為例：

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"]
    }

這個(gè)函數(shù)接收數(shù)據(jù)字典，并以字典形式返回提取后的數(shù)據(jù)。

類似的，我們可以編寫出論壇、用戶、話題、圖片、用戶頭像、用戶認(rèn)證這幾類數(shù)據(jù)的解析函數(shù)。

使用一個(gè)函數(shù)對(duì)完整的數(shù)據(jù)字典進(jìn)行解析：

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)
    }

由于發(fā)起一次網(wǎng)絡(luò)請(qǐng)求時(shí)，會(huì)獲取到一組數(shù)據(jù)，為了簡(jiǎn)化主邏輯，我們編寫一個(gè)解析數(shù)據(jù)列表的函數(shù)：

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

到現(xiàn)在，我們的程序已經(jīng)可以實(shí)現(xiàn)數(shù)據(jù)的獲取與解析了。

數(shù)據(jù)預(yù)處理

由于信息流接口獲取到的數(shù)據(jù)存在一些問題，比如互動(dòng)數(shù)據(jù)異常，缺少帖子完整內(nèi)容等，我們需要對(duì)請(qǐng)求到的數(shù)據(jù)進(jìn)行一些處理，替換掉錯(cuò)誤的數(shù)據(jù)，加入我們希望獲取的數(shù)據(jù)。

替換互動(dòng)數(shù)據(jù)

首先，在 data_fetch.py 中增加一個(gè)數(shù)據(jù)獲取函數(shù)：

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("解析互動(dòng)數(shù)據(jù) Json 時(shí)出現(xiàn)異常")

    if data["retcode"] != 0:
        raise ValueError(f"獲取互動(dòng)數(shù)據(jù)時(shí)出現(xiàn)異常，錯(cuò)誤碼：{data['retcode']}，錯(cuò)誤信息：{data['message']}")

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

這個(gè)接口支持一次獲取一組動(dòng)態(tài)數(shù)據(jù)，只需要將對(duì)應(yīng)帖子的 post_id 以英文逗號(hào)分隔的格式作為接口的 post_ids 參數(shù)即可。

為了確保函數(shù)封裝良好，我們將函數(shù)的參數(shù)指定為由字符串形式的 post_id 組成的列表。（因?yàn)樾畔⒘鹘涌诜祷氐?post_id 是字符串類型）

接下來，在 data_process.py 文件中編寫一個(gè)函數(shù)，將傳入的 data 列表中的所有互動(dòng)數(shù)據(jù)替換成對(duì)應(yīng)的正確數(shù)據(jù)：

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

加入帖子完整內(nèi)容數(shù)據(jù)

同樣的，在 data_fetch.py 中定義一個(gè)函數(shù)：

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 時(shí)出現(xiàn)異常")

    if data["retcode"] != 0:
        raise ValueError(f"獲取帖子全文時(shí)出現(xiàn)異常，錯(cuò)誤碼：{data['retcode']}，錯(cuò)誤信息：{data['message']}")

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

這里我們遇到了一些問題，調(diào)試這個(gè)接口的時(shí)候會(huì)出現(xiàn) 403 錯(cuò)誤，我們將常見反爬措施會(huì)校驗(yàn)的內(nèi)容（Cookie / UA / Referer 等）全部復(fù)制到請(qǐng)求工具中，再次請(qǐng)求發(fā)現(xiàn)數(shù)據(jù)正常返回。

之后我們逐一刪除這些內(nèi)容，最后發(fā)現(xiàn)，添加 Referer 即可規(guī)避這一反爬措施。

同樣的，編寫一個(gè)函數(shù)對(duì)原先的 data 列表進(jìn)行替換：

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

由于這個(gè)接口一次只能獲取一條數(shù)據(jù)，我們需要使用多線程來提高程序的運(yùn)行效率。

過早的優(yōu)化是萬(wàn)惡之源，如果不能確定這個(gè)函數(shù)會(huì)拖慢程序的運(yùn)行速度，可以先使用單線程請(qǐng)求，后期通過性能分析找到瓶頸，再針對(duì)性優(yōu)化。

這里使用的是 Python 的內(nèi)置庫(kù) joblib，上述代碼將使用 10 個(gè)線程對(duì) GetPostContent 函數(shù)發(fā)起調(diào)用，并將結(jié)果以正確的順序存入 contents_list 中。

之后我們通過 for 循環(huán)，將每條帖子對(duì)應(yīng)的 content 插入到其數(shù)據(jù)的 post 項(xiàng)中。

數(shù)據(jù)存儲(chǔ)

由于我們?cè)跀?shù)據(jù)解析過程中，就將解析結(jié)果字典的鍵與數(shù)據(jù)庫(kù)的字段進(jìn)行了一一對(duì)應(yīng)，所以我們可以直接使用字典解包，將鍵值對(duì)變?yōu)閰?shù)，傳入 Peewee 的 create 函數(shù)中，從而快速實(shí)現(xiàn)數(shù)據(jù)的存儲(chǔ)。

示例代碼如下：

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

但在對(duì)論壇數(shù)據(jù)進(jìn)行保存時(shí)，我們遇到了一個(gè)問題：將要被保存的數(shù)據(jù)可能已經(jīng)存在于數(shù)據(jù)庫(kù)中，這樣會(huì)因?yàn)橹麈I重復(fù)而產(chǎn)生異常。

因此，我們需要對(duì)主鍵重復(fù)產(chǎn)生的 IntegrityError 進(jìn)行捕獲，并將其忽略：

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

類似的，我們可以編寫出其它類型數(shù)據(jù)的存儲(chǔ)函數(shù)。

最后，用一個(gè)函數(shù)將它們聚合起來：

def SaveData(data: Dict):
    with db.atomic():  # 開啟事務(wù)
        post_obj = SavePostData(data["post_data"])
        if data["forum_data"]:  # 如果板塊數(shù)據(jù)不為空
            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)

這里我們使用了數(shù)據(jù)庫(kù)的事務(wù)功能，在事務(wù)中的數(shù)據(jù)庫(kù)操作，只可能全部成功或者全部失敗。

這樣做由兩個(gè)原因，其一，可以防止程序出錯(cuò)時(shí)對(duì)數(shù)據(jù)庫(kù)造成污染；其二，通過減少提交操作（Commit），可以提高數(shù)據(jù)存儲(chǔ)的性能。

和 data_parse.py 中一樣，我們可以對(duì)一組數(shù)據(jù)進(jìn)行保存：

def SaveDataList(data_list: List[Dict]):
    for data in data_list:
        try:
            SaveData(data)
        except IntegrityError:  # 主鍵重復(fù)
            print(f"帖子 {data['post_data']['id']} 出現(xiàn)重復(fù)，已自動(dòng)跳過")

這個(gè)函數(shù)考慮到了數(shù)據(jù)在獲取過程中產(chǎn)生變動(dòng)，從而導(dǎo)致主鍵重復(fù)時(shí)的處理。

主邏輯

導(dǎo)入庫(kù)部分省略。

我們先來定義一些常量：

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

這樣做可以幫助我們快速修改運(yùn)行參數(shù)，而不需要對(duì)使用到這些參數(shù)的每一個(gè)位置進(jìn)行改動(dòng)。

如果可調(diào)整的參數(shù)超過 10 個(gè)，或者這個(gè)爬蟲需要定期運(yùn)行，最好使用配置文件（比如 YAML）進(jìn)行管理。

由于數(shù)據(jù)保存需要消耗較長(zhǎng)時(shí)間，我們將其抽離成獨(dú)立的一個(gè)線程，因此，需要定義一個(gè)列表，用來存放待保存的數(shù)據(jù)：

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

接下來是數(shù)據(jù)保存線程：

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)} 條數(shù)據(jù)，耗時(shí) {round(time() - start_time, 2)} 秒")
                data_to_save_list.clear()
        sleep(DATA_SAVE_INTERVAL)

由于數(shù)據(jù)的獲取、保存到待保存列表的清空需要一定時(shí)間，在此期間，如果有新數(shù)據(jù)加入列表，就會(huì)導(dǎo)致未保存數(shù)據(jù)的丟失，因此，我們需要在數(shù)據(jù)保存期間對(duì)數(shù)據(jù)進(jìn)行加鎖。

接下來，我們對(duì)需要采集的頁(yè)數(shù)進(jìn)行計(jì)算，并對(duì)必要的數(shù)據(jù)進(jìn)行校驗(yàn)，最后初始化數(shù)據(jù)庫(kù)。

然后啟動(dòng)數(shù)據(jù)保存線程：

data_save_thread = Thread(target=DataSaveJob, daemon=True)  # 設(shè)置為守護(hù)線程，避免影響主線程退出
data_save_thread.start()
print("數(shù)據(jù)獲取線程啟動(dòng)成功...")

核心采集邏輯如下：

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

這里省略了關(guān)于出錯(cuò)重試的邏輯。

然后將數(shù)據(jù)加入待保存列表：

with data_to_save_list_lock:
    data_to_save_list.extend(data)

在數(shù)據(jù)全部采集完畢后，我們需要等待保存線程將它們?nèi)看嫒霐?shù)據(jù)庫(kù)：

print("等待數(shù)據(jù)存儲(chǔ)完成...")
while True:
    if not data_to_save_list:
        with data_to_save_list_lock:  # 獲取到鎖則證明全部數(shù)據(jù)已保存完畢
            print("數(shù)據(jù)存儲(chǔ)完成...")
            exit()
    else:
        sleep(1)

運(yùn)行程序，輸出如下：

總數(shù)據(jù)量：30000   單頁(yè)數(shù)據(jù)個(gè)數(shù)：30   總頁(yè)數(shù)：1000
等待時(shí)間：0s   錯(cuò)誤重試間隔：20s   數(shù)據(jù)保存間隔：20s
初始化數(shù)據(jù)庫(kù)成功...
數(shù)據(jù)獲取線程啟動(dòng)成功...
開始采集數(shù)據(jù)...
開始采集第 1 頁(yè)
第 1 頁(yè)采集成功，耗時(shí) 4.39 秒
開始采集第 2 頁(yè)
第 2 頁(yè)采集成功，耗時(shí) 2.67 秒
開始采集第 3 頁(yè)
第 3 頁(yè)采集成功，耗時(shí) 2.57 秒
開始采集第 4 頁(yè)
第 4 頁(yè)采集成功，耗時(shí) 2.42 秒
開始采集第 5 頁(yè)
第 5 頁(yè)采集成功，耗時(shí) 2.48 秒
開始采集第 6 頁(yè)
第 6 頁(yè)采集成功，耗時(shí) 2.07 秒
開始采集第 7 頁(yè)
第 7 頁(yè)采集成功，耗時(shí) 2.16 秒
開始采集第 8 頁(yè)
已成功保存 210 條數(shù)據(jù)，耗時(shí) 3.91 秒
第 8 頁(yè)采集成功，耗時(shí) 5.24 秒
（以下省略）

采集結(jié)果

本以為數(shù)據(jù)量至少在十萬(wàn)量級(jí)，沒想到爬到五百多頁(yè)就開始連續(xù)報(bào)錯(cuò)，手動(dòng)請(qǐng)求接口發(fā)現(xiàn)已經(jīng)沒有新的數(shù)據(jù)返回，排除掉反爬原因后，可以確定是數(shù)據(jù)已經(jīng)被爬取完成。

我采集數(shù)據(jù)的時(shí)間是 2022 年 3 月 27 日，data.db 文件的大小為 71.6MB，總數(shù)據(jù)量 14997 條。

后記

本文中，我們對(duì)米游社的動(dòng)態(tài)加載請(qǐng)求進(jìn)行了分析，設(shè)計(jì)了保存這些數(shù)據(jù)的數(shù)據(jù)庫(kù)結(jié)構(gòu)，通過自動(dòng)化請(qǐng)求接口實(shí)現(xiàn)了數(shù)據(jù)的獲取、解析與保存。

感興趣的小伙伴可以自行探索以下內(nèi)容：

• 將本文中的數(shù)據(jù)獲取相關(guān)代碼改寫為異步形式，提升網(wǎng)絡(luò)請(qǐng)求性能
• 優(yōu)化數(shù)據(jù)庫(kù)操作性能
• 使用裝飾器實(shí)現(xiàn)數(shù)據(jù)預(yù)處理
• 使用 addict 庫(kù)改寫數(shù)據(jù)解析邏輯，增強(qiáng)可讀性
• 使用 rich 庫(kù)輸出不同顏色的終端信息

文中的程序會(huì)在本系列完結(jié)后開源，屆時(shí)倉(cāng)庫(kù)地址將更新在此處。

在接下來的文章中，我們將對(duì)這些數(shù)據(jù)進(jìn)行進(jìn)一步的分析。