(Python網絡爬蟲)；抓取B站404頁面小漫畫

一. 分析網頁

二. 準備工作

三. 實現爬蟲

1. 抓取工作

2. 分析工作

3. 拼接主函數&運行結果

四. 完整代碼清單

1.多線程版本spider.py：?

2.異步版本async_spider.py：

經常逛B站的同志們可能知道，B站的404頁面做得別具匠心：

可以通過https://www.bilibili.com/404來訪問這個頁面，我們還能換一換：

不過，能刷出哪個小漫畫都是隨機的，究竟有多少張我確實不知道。所以我突發奇想，想把這些東西爬下來，正好作為爬蟲案例練手也不錯呀！

一. 分析網頁

我們先“御駕親征”看看應該怎么下手，這時就需要用 F12 召喚開發者工具看網頁代碼了。

通過元素檢查，可以發現img對象的src屬性包含了一個鏈接，這就是圖片的來源。

每次換頁，這個鏈接就會變更。說明網頁資源是利用JavaScript動態渲染的，不應該像基礎爬蟲那樣直接從網頁源代碼抓取信息。那么首先，先看看是不是通過Ajax請求動態渲染的，如果是的話，直接模擬Ajax請求即可，不需要用selenium之類的工具。

經過篩查，有三個.xhr類型的請求：

其中中間的那個請求，它的響應是json數據，里面就是我們要找的信息——圖片id和鏈接：?

那現在問題就解決了，我們的思路很簡單——先模擬這個ajax請求，獲取響應的json數據；然后根據數據中的url，請求圖片并保存到計算機上。?我看了看一共只有20張圖片，所以不強要求異步爬蟲。正好我的異步編程水平很拉跨，沒個幾千條數據我是不會用異步的！

二. 準備工作

那么，用什么工具呢？首先是http庫，我們之前看到、這些請求的協議類型清一色都是"h2"：

那么我們就不能用requests或者標準庫urllib，它們不支持Http/2.0協議，但是可以用httpx這個第三方庫。httpx的API和requests很類似，在簡單的請求方面幾乎是完全一樣。不過它可以支持Http/2.0協議，還自帶異步框架，確實非常好用。

其次是解析庫，這里處理的是json數據不是html數據，什么靚湯呀正則呀都不需要了。不過json庫可能會用上，暫時把它考慮上吧。

最后是保存圖片，pathlib肯定會有用的，它太棒了。

至于其它的，我會用logging這個日志庫。合理記錄程序運行情況有助于調試爬蟲程序，沒個日志還要修爬蟲的BUG太折磨了；雖然不用異步，但可能會用concurrent庫來進行并發，提高爬取效率。

老B畢竟是大網站，沒個UA偽裝它還是會把我們當場擒獲的：

幸好，只加個UA就過了，畢竟404頁面沒必要下太多功夫防御吧：

好，有了這些分析，下面我們來實現這個程序。?

三. 實現爬蟲

新建一個文件夾，在其中創建spider.py文件，現在開始實現它。

1. 抓取工作

首先是導入工作和必要的準備。為了便于閱讀，我直接在代碼片段中寫(部分1)之類的提示，盡管這是不合法的：

(部分1)
import logging
from pathlib import Path
from concurrent import futuresimport httpxfrom my_modules.clock import clock
(部分2)
logging.basicConfig(level=logging.INFO,format='%(asctime)s-%(levelname)s:%(message)s'
)
logging.getLogger('httpx').setLevel(logging.WARNING)
(部分3)
headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36'
}
def get_client(): return httpx.Client(http2=True, headers=headers, timeout=10)
BASE_URL = 'https://api.bilibili.com/x/activity/operation/list?source_id=630edcfddbd0b39ca7371ad2'
SAVED_DIR = Path('images')

(1)：這一部分導入要用的庫。根據PEP8的建議，在Python中導入模塊時，標準庫、第三方庫、和自己編寫的模塊要分開寫，并用空行隔開。這個clock是個裝飾器，用來計時，看看這個爬蟲程序的效率如何。

(2)：這一部分配置logging。首先，level=logging.INFO指出會報告'info'及以上級別的日志；format規定日志輸出的形式；最后那個getLogger('httpx')……是為了阻止httpx庫報告warning以下級別的日志，不然它每次請求成功都會自動給我發一個info日志，直接就刷屏了，煩人。

(3)：上來就是請求頭headers，只做了一下UA偽裝。這個get_client()函數返回一個可以執行請求的client對象，為什么我不直接寫client = httpx.Client(……)然后在全局使用它呢？

一來，這樣的話client需要用close()關閉，用with語句更好。然而，成品client對象是不能在with中使用的，所以我選擇用這個函數動態創建client；
二來，我害怕client不是線程安全的，共用一個可能導致意外BUG。為了避免麻煩還是為每個請求動態分配一個吧。不過我沒有考證，沒準是線程安全的？

當然還有其他策略，還可以參考最后我給出的異步版本的那種方案。那兩個常量BASE_URL和SAVED_DIR分別是 Ajax請求的url 和圖片保存的路徑。

然后，我們定義一個通用爬取函數，返回的是響應。?這樣后面不管是獲取text,json()還是content都能復用它：

def scrape_url(url: str) -> httpx.Response|None:"""抓取url的頁面,返回響應;超時則返回None"""logging.info('開始抓取:%s', url)try:with get_client() as client:response = client.get(url)except httpx.TimeoutException:logging.warning('警告!請求超時:%s', url)return Noneexcept httpx.RequestError:logging.critical('重大錯誤!抓取時發生異常:%s', url, exc_info=True)raiseelse:logging.info('抓取成功:%s', url)return response

這里關鍵點只有try-else語句，僅在try語句完美地執行下來沒有出錯時，才會執行else子句的內容。logging.info()這樣的函數以不同的等級進行日志報告，比如warning是警告，critical是嚴重錯誤。

然后，復用它來定義兩個具體抓取函數，一個模擬ajax請求、一個用來抓取圖片。這里的注解用到了Any，需要從typing導入它：

def scrape_ajax(url: str) -> Any|None:"""抓取ajax請求返回的響應,返回其中的json數據"""response = scrape_url(url)if response is not None:return response.json()else:return None
def scrape_img(url: str) -> bytes|None:"""抓取圖片url的響應,獲取其二進制數據"""response = scrape_url(url)if response is not None:logging.info('獲取圖片數據成功:%s', url)return response.contentelse:return None

下一步我們來看看怎么從scrape_ajax返回的數據中獲取我們想要的信息。

2. 分析工作

通過源碼觀察ajax請求的響應，返回的東西是這樣的：

這是個json數據，首先我們應該提取“data”，然后提取里面的“list”，這里面放著的是各個圖片的數據。對于每張圖片，我想要“id”和“object_id”兩個數據，前者拿來在保存時命名，后者供scrape_img使用來獲取二進制數據。那么我可以先把“Image”封裝為一個數據類，每個Image有id和url兩個屬性。我喜歡用typing的NamedTuple，它能很方便地創造一個輕量級數據類：

class Image(NamedTuple):"""封裝圖片數據的類"""id: strurl: str

?將這段代碼寫在準備工作那里，現在就可以開始寫分析數據的函數了。

def parse_ajax(data: Any) -> list[Image]:"""分析ajax請求的響應數據,封裝為Image對象列表返回"""imgs = []img_lst = data.get('data').get('list')for img in img_lst:id = img.get('id')url = img.get('object_id')new_img = Image(id, url)imgs.append(new_img)return imgs

這個函數分析ajax響應的json數據，從中提取信息封裝為Image對象，并打包為列表返回。?

然后，是一個保存圖片用的函數，負責消費這些Images對象：

def save_img(image: Image) -> None:"""保存單張圖片"""name, url = imagedata = scrape_img(url)if data is not None:(SAVED_DIR / f'{name}.png').write_bytes(data)logging.info('圖片保存成功:%s', name)else:logging.error('圖片獲取失敗:%s', name)

?第一行直接拆包Image對象拿到id和url，然后嘗試用scrape_img抓取二進制數據。如果不是None，則保存到SAVED_DIR中并給出一條報告。否則，說明請求超時，也給出一條報告。

最后，我們把這些“積木”拼起來寫成主函數，這個程序就完成啦。

3. 拼接主函數&運行結果

下面是主函數，它“奮六世之余烈”，把我們一直以來做的工作協調起來：

@clock(report_upon_exit=True)
def main():"""啟動!"""SAVED_DIR.mkdir(exist_ok=True)ajax_json = scrape_ajax(BASE_URL)if ajax_json is None:raise RuntimeError('第一個請求都超時了,還想做爬蟲?')images = parse_ajax(ajax_json)with futures.ProcessPoolExecutor() as executor:executor.map(save_img, images)if __name__ == '__main__':main()

這里有很多老面孔打贏復活賽了，尤其是concurrent.futures。

首先，用了我自己的裝飾器來計時。
然后，SAVED_DIR.mkdir()函數是讓程序創建這個目錄。但是如果目錄已經存在這一句就會報錯，指定exist_ok=True可以避免這種行為。
獲取ajax請求的響應，如果是None的話說明超時了。要是這么重要的數據都得不到，那我們也沒有繼續運行程序的理由了，直接報錯吧。
用parse_ajax來獲取封裝的圖片數據。
用concurrent.futures.ProcessPoolExecutor來多進程地執行save_img任務，可以提高效率。

很好，下面我們來運行一下看看，終端的結果如下：

11秒！這個效率我覺得還是不行。我試了試完全順序執行，用時15秒。這里多進程的提升確實有限。感覺get_client也有鍋，在異步爬蟲里必須更改這部分邏輯，否則它會阻塞協程。

把上面的ProcessPoolExecutor改為ThreadPoolExecutor就能改為多線程版本，它的運行時間如下：

用時9.4s，看來多線程在這里強于多進程。即使因為GIL，多線程常被人詬病，但是它輕量，在這里就超過了開銷更大的多進程版本。

我還實現了一版異步爬蟲，它的用時如下：

用時7.9s接近8s，可以看到，盡管中間存在阻塞步驟，異步的效率還是更強。它是真適合干這個呀，異步拿了MVP！異步的腳本跟多進程的相比改動有些大、更改了一些邏輯、我就不詳細說了，最后我會給出異步的代碼清單。

不管怎樣，成功把圖片下載下來了，這就是成功！還要啥自行車啊？

?好了，完事兒(*￣▽￣)~*。下面我來放一下多線程和異步版本的完整代碼清單。

四. 完整代碼清單

1.多線程版本spider.py：?

import logging
from pathlib import Path
from concurrent import futures
from typing import Any, NamedTupleimport httpxfrom my_modules.clock import clocklogging.basicConfig(level=logging.INFO,format='%(asctime)s-%(levelname)s:%(message)s'
)
logging.getLogger('httpx').setLevel(logging.WARNING)headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36'
}
def get_client(): return httpx.Client(http2=True, headers=headers, timeout=10)
BASE_URL = 'https://api.bilibili.com/x/activity/operation/list?source_id=630edcfddbd0b39ca7371ad2'
SAVED_DIR = Path('images')class Image(NamedTuple):"""封裝圖片數據的類"""id: strurl: strdef scrape_url(url: str) -> httpx.Response|None:"""抓取url的頁面,返回響應;超時則返回None"""logging.info('開始抓取:%s', url)try:with get_client() as client:response = client.get(url)except httpx.TimeoutException:logging.warning('警告!請求超時:%s', url)return Noneexcept httpx.RequestError:logging.critical('重大錯誤!抓取時發生異常:%s', url, exc_info=True)raiseelse:logging.info('抓取成功:%s', url)return response
def scrape_ajax(url: str) -> Any|None:"""抓取ajax請求返回的響應,返回其中的json數據"""response = scrape_url(url)if response is not None:return response.json()else:return None
def scrape_img(url: str) -> bytes|None:"""抓取圖片url的響應,獲取其二進制數據"""response = scrape_url(url)if response is not None:logging.info('獲取圖片數據成功:%s', url)return response.contentelse:return None
def parse_ajax(data: Any) -> list[Image]:"""分析ajax請求的響應數據,封裝為Image對象列表返回"""imgs = []img_lst = data.get('data').get('list')for img in img_lst:id = img.get('id')url = img.get('object_id')new_img = Image(id, url)imgs.append(new_img)return imgs
def save_img(image: Image) -> None:"""保存單張圖片"""name, url = imagedata = scrape_img(url)if data is not None:(SAVED_DIR / f'{name}.png').write_bytes(data)logging.info('圖片保存成功:%s', name)else:logging.error('圖片獲取失敗:%s', name)@clock(report_upon_exit=True)
def main():"""啟動!"""SAVED_DIR.mkdir(exist_ok=True)ajax_json = scrape_ajax(BASE_URL)if ajax_json is None:raise RuntimeError('第一個請求都超時了,還想做爬蟲?')images = parse_ajax(ajax_json)with futures.ThreadPoolExecutor() as executor:executor.map(save_img, images)if __name__ == '__main__':main()

2.異步版本async_spider.py：

import asyncio
import logging
from pathlib import Path
from typing import Any, NamedTupleimport httpxfrom my_modules.clock import clocklogging.basicConfig(level=logging.INFO,format='%(asctime)s-%(levelname)s:%(message)s'
)
logging.getLogger('httpx').setLevel(logging.WARNING)headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36'
}
BASE_URL = 'https://api.bilibili.com/x/activity/operation/list?source_id=630edcfddbd0b39ca7371ad2'
SAVED_DIR = Path('async_images')class Image(NamedTuple):"""封裝圖片數據的類"""id: strurl: strasync def scrape_url(client: httpx.AsyncClient, url: str) -> httpx.Response|None:"""異步抓取url的頁面,返回響應;超時則返回None"""logging.info('開始抓取:%s', url)try:response = await client.get(url)except httpx.TimeoutException:logging.warning('警告!請求超時:%s', url)return Noneexcept httpx.RequestError:logging.critical('重大錯誤!抓取時發生異常:%s', url, exc_info=True)raiseelse:logging.info('抓取成功:%s', url)return response
async def scrape_ajax(client: httpx.AsyncClient, url: str) -> Any|None:"""異步抓取ajax請求返回的響應,返回其中的json數據"""response = await scrape_url(client, url)if response is not None:return response.json()return None
async def scrape_img(client: httpx.AsyncClient, url: str) -> bytes|None:"""異步抓取圖片url的響應,獲取其二進制數據"""response = await scrape_url(client, url)if response is not None:logging.info('獲取圖片數據成功:%s', url)return response.contentreturn None
def parse_ajax(data: Any) -> list[Image]:"""分析ajax請求的響應數據,封裝為Image對象列表返回"""imgs = []img_lst = data.get('data', {}).get('list', [])for img in img_lst:id = img.get('id')url = img.get('object_id')if id and url:imgs.append(Image(id, url))return imgs
async def save_img(client: httpx.AsyncClient, image: Image) -> None:"""異步保存單張圖片"""name, url = imagedata = await scrape_img(client, url)if data is not None:(SAVED_DIR / f'{name}.png').write_bytes(data)logging.info('圖片保存成功:%s', name)else:logging.error('圖片獲取失敗:%s', name)@clock(report_upon_exit=True)
async def async_main():"""啟動!異步爬蟲"""SAVED_DIR.mkdir(exist_ok=True)async with httpx.AsyncClient(http2=True, headers=headers, timeout=10) as client:ajax_json = await scrape_ajax(client, BASE_URL)if ajax_json is None:raise RuntimeError('第一個請求都超時了,還想做爬蟲?')images = parse_ajax(ajax_json)tasks = [save_img(client, image) for image in images]await asyncio.gather(*tasks)if __name__ == '__main__':asyncio.run(async_main())