Scrapy 中的 Pipline管道

本小節(jié)中我們將詳細(xì)介紹 Scrapy 中的 Pipeline 及其多種用法和使用場(chǎng)景。Pipeline 是 Scrapy 框架的一個(gè)重要模塊，從前面的 Scrapy 架構(gòu)圖中我們可以看到它位于架構(gòu)圖的最左邊，用于連續(xù)處理從網(wǎng)頁(yè)中抓取到的每條記錄，就像一個(gè)流水線工廠加工食品那樣，完成食品最后的封裝、保存等操作。此外，我們還會(huì)介紹 Scrapy 內(nèi)置的圖片管道，可以自動(dòng)下載對(duì)應(yīng)地址的圖片。最后，我們會(huì)基于上述內(nèi)容完成一個(gè)小說(shuō)網(wǎng)站的爬取案例。

1. Scrapy 中的 Pipeline 介紹

Pipeline 的中文意思是管道，類似于工廠的流水線那樣。Scrapy 中的 Pipeline 通常是和 Items 聯(lián)系在一起的，其實(shí)就是對(duì) Items 數(shù)據(jù)的流水線處理。 一般而言，Pipeline 的典型應(yīng)用場(chǎng)景如下：

• 數(shù)據(jù)清洗、去重；
• 驗(yàn)證數(shù)據(jù)的有效性；
• 按照自定義格式保存數(shù)據(jù)；
• 存儲(chǔ)到合適的數(shù)據(jù)庫(kù)中 (如 MySQL、Redis 或者 MongoDB)；

通過(guò)前面的 Scrapy 架構(gòu)圖可知，Pipeline 位于 Scrapy 數(shù)據(jù)處理流程的最后一步，但是它也不是必須，Pipeline 默認(rèn)處于關(guān)閉狀態(tài)。如果需要的話，我們只需要在 settings.py 中設(shè)置 ITEM_PIPELINES 屬性值即可。它是一個(gè)數(shù)組值，我們可以定義多個(gè) Item Pipeline，并且在 ITEM_PIPELINES 中設(shè)置相應(yīng) Pipeline 的優(yōu)先級(jí)。這樣 Scrapy 會(huì)依次處理這些 Pipelines，最后達(dá)到我們想要的效果。

注意：上面的 pipeline 順序和功能都可以任意調(diào)整，保證邏輯性即可。比如有一個(gè)去重的 pipeline 和保存到數(shù)據(jù)庫(kù)的 pipeline，那么去重的 pipeline 一定要在保存數(shù)據(jù)庫(kù)之前，這樣保存的就是不重復(fù)的數(shù)據(jù)。

2. 如何編寫(xiě)自己的 Item Pipeline

編寫(xiě)自己的 Item Pipeline 非常簡(jiǎn)單，我們只需要編寫(xiě)一個(gè)簡(jiǎn)單的類，實(shí)現(xiàn)四個(gè)特定名稱的方法即可 (部分方法非必須)。我們來(lái)簡(jiǎn)單說(shuō)明下這三個(gè)方法：

• open_spider(spider)：非必需，參數(shù) spider 即被關(guān)閉的 Spider 對(duì)象。這個(gè)方法是 MiddlewareManager 類中的方法，在 Spider 開(kāi)啟時(shí)被調(diào)用，主要做一些初始化操作，如連接數(shù)據(jù)庫(kù)、打開(kāi)要保存的文件等；
• close_spider(spider)：非必需，參數(shù) spider 即被關(guān)閉的 Spider 對(duì)象。這個(gè)方法也是 MiddlewareManager 類中的方法，在 Spider 關(guān)閉時(shí)被調(diào)用，主要做一些如關(guān)閉數(shù)據(jù)庫(kù)連接、關(guān)閉打開(kāi)的文件等操作；
• from_crawler(cls, crawler)：非必需，在 Spider啟用時(shí)調(diào)用，且早于 open_spider() 方法。這個(gè)方法我們很少去重載，可以不用；
• process_item(item, spider)：必須實(shí)現(xiàn)。該函數(shù)有兩個(gè)參數(shù)，一個(gè)是表示被處理的 Item 對(duì)象，另一個(gè)是生成該 Item 的 Spider 對(duì)象。定義的 Item pipeline 會(huì)默認(rèn)調(diào)用該方法對(duì) Item 進(jìn)行處理，這也是 Pipeline 的工作核心；

完成這樣一個(gè) Item Pipeline 后，將該類的路徑地址添加到 settings.py 中的 ITEM_PIPELINES 中即可。下圖是我們一個(gè)簡(jiǎn)單項(xiàng)目完成的兩個(gè) pipelines。

3. 實(shí)戰(zhàn)演練

學(xué)習(xí)了上面的一些知識(shí)，我們來(lái)使用一個(gè)簡(jiǎn)單的網(wǎng)站進(jìn)行實(shí)戰(zhàn)演練，在該過(guò)程中介紹更多的和 Item Pipeline 相關(guān)的用法。

假設(shè)我們是一名小說(shuō)愛(ài)好者，我想到起點(diǎn)中文網(wǎng)上去找一些好的小說(shuō)看，我該怎么找呢？起點(diǎn)中文網(wǎng)的月票榜是一個(gè)不錯(cuò)的參考方式，如下圖所示：

其實(shí)簡(jiǎn)單看一看就知道月票榜的 url 組成：

• 主體 url：https://www.qidian.com/rank/yuepiao
• 參數(shù) month：02 表示 2 月份，03 表示 3 月份，目前為止最多到 7 月份；
• 參數(shù) chn：表示的是分類，-1 表示全部分類。21 表示玄幻，22表示仙俠；
• 參數(shù) page：表示第幾頁(yè)，一頁(yè)有20個(gè)作品。

目前我們只需要從 01 月份開(kāi)始到 07 月份的月票榜中，每次都取得第一頁(yè)的數(shù)據(jù)，也就是月票榜的前20 名。7 個(gè)月份的前 20 名加起來(lái)，然后再去重，就得到了曾經(jīng)的占據(jù)月票榜的作品，這中間大概率都是比較好看的書(shū)。完成這個(gè)簡(jiǎn)單的需求我們按照如下的步驟進(jìn)行：

創(chuàng)建初始項(xiàng)目 qidian_yuepiao：

[root@server ~]# pyenv activate scrapy-test
(scrapy-test) [root@server ~]# cd scrapy-test
(scrapy-test) [root@server scrapy-test]# scrapy startproject qidian_yuepia

(scrapy-test) [root@server qidian_yuepiao]# ls
 __init__.py  items.py  middlewares.py  pipelines.py settings.py  spider

接下來(lái)我們準(zhǔn)備獲取小說(shuō)作品的字段，大概會(huì)獲取如下幾個(gè)數(shù)據(jù)：

• 小說(shuō)名：name；
• 小說(shuō)作者：author；
• 小說(shuō)類型：fiction_type。比如玄幻、仙俠、科幻等；
• 小說(shuō)狀態(tài)：state。連載還是完結(jié)；
• 封面圖片地址：image_url；
• images：保存圖片數(shù)據(jù)；
• brief_introduction：作品簡(jiǎn)介；
• book_url：小說(shuō)的具體地址。

根據(jù)定義的這些字段，我們可以寫(xiě)出對(duì)應(yīng)的 Items 類，如下：

(scrapy-test) [root@server qidian_yuepiao]# cat items.py 
# Define here the models for your scraped items
#
# See documentation in:
# https://docs.scrapy.org/en/latest/topics/items.html

import scrapy


class QidianYuepiaoItem(scrapy.Item):
    # define the fields for your item here like:
    name = scrapy.Field()  
    author = scrapy.Field()
    fiction_type = scrapy.Field()
    state = scrapy.Field()
    image_url = scrapy.Field() 
    images = scrapy.Field()
    brief_introduction = scrapy.Field() 
    book_url = scrapy.Field()

到了最關(guān)鍵的地方，需要解析網(wǎng)頁(yè)數(shù)據(jù)，提取月票榜的作品信息。這個(gè)和前面一些，我們只需要完成相應(yīng)的 xpath 即可。此外，我們會(huì)從 01 月份的月票榜開(kāi)始，每次會(huì)新生成一個(gè) url，主要改動(dòng)的就是月份參數(shù)，每次將月份數(shù)加一；如果當(dāng)前月份大于07，則終止。

(scrapy-test) [root@server qidian_yuepiao]# touch spiders/qidian_yuepiao_parse.py

import re

from scrapy import Request
from scrapy.spiders import Spider
from  qidian_yuepiao.items import QidianYuepiaoItem


def num_to_str(num, size=2, padding='0'):
    """
    0 - > 00   1 -> 01   11 -> 11
    :param num:
    :param size:
    :param padding:
    :return:
    """
    str_num = str(num)
    while len(str_num) < size:
        str_num = padding + str_num
    return str_num


class QidianSpider(Spider):
    name = "qidian_yuepiao_spider"
    start_urls = [
        "https://www.qidian.com/rank/yuepiao?month=01&chn=-1&page=1"
    ]

    def parse(self, response):
        fictions = response.xpath('//div[@id="rank-view-list"]/div/ul/li')
        for fiction in fictions:
            name = fiction.xpath('div[@class="book-mid-info"]/h4/a/text()').extract_first()
            author = fiction.xpath('div[@class="book-mid-info"]/p[@class="author"]/a[1]/text()').extract_first()
            fiction_type = fiction.xpath('div[@class="book-mid-info"]/p[@class="author"]/a[1]/text()').extract_first()
            # 注意一定要是列表，不然會(huì)報(bào)錯(cuò)
            image_url = ['http:{}'.format(fiction.xpath('div[@class="book-img-box"]/a/img/@src').extract()[0])]
            brief_introduction = fiction.xpath('div[@class="book-mid-info"]/p[@class="intro"]/text()').extract_first()
            state = fiction.xpath('div[@class="book-mid-info"]/p[@class="author"]/a[2]/text()').extract()[0]
            book_url = fiction.xpath('div[@class="book-mid-info"]/h4/a/@href').extract()[0]  

            item = QidianYuepiaoItem()
            item['name'] = name
            item['author'] = author
            item['fiction_type'] = fiction_type
            item['brief_introduction'] = brief_introduction.strip()
            item['image_url'] = image_url
            item['state'] = state
            item['book_url'] = book_url

            yield item

        # 提取月份數(shù)，同時(shí)也要提取請(qǐng)求的url
        url = response.url
        regex = "https://(.*)\?month=(.*?)&(.*)"
        pattern = re.compile(regex)
        m = pattern.match(url)
        if not m:
            return []
        prefix = m.group(1)
        month = int(m.group(2))
        suffix = m.group(3)

        # 大于7月份則停止，目前是2020年7月20日
        if month > 7:
            return
 
        # 一定要將月份轉(zhuǎn)成01, 02, s03這樣的形式，否則不能正確請(qǐng)求到數(shù)據(jù)
        next_month = num_to_str(month + 1)
        
        next_url = f"https://{prefix}?month={next_month}&{suffix}"
        yield Request(next_url)

最后到了我們本節(jié)課的重點(diǎn)。首先我想要將數(shù)據(jù)保存成 json 格式，存儲(chǔ)到文本文件中，但是在保存之前，需要對(duì)作品去重。因?yàn)橛行┳髌窌?huì)連續(xù)好幾個(gè)月出現(xiàn)在月票榜的前20位置上，會(huì)有比較多重復(fù)。我們通過(guò)作品的 url 地址來(lái)唯一確定該小說(shuō)。因此需要定義兩個(gè) Item Pipeline：

import json

from itemadapter import ItemAdapter
from scrapy.exceptions import DropItem


class QidianYuepiaoPipeline:
    """
    保存不重復(fù)的數(shù)據(jù)到文本中
    """
    def open_spider(self, spider):
        self.file = open("yuepiao_top.json", 'w+')

    def close_spider(self, spider):
        self.file.close()

    def process_item(self, item, spider):
        data = json.dumps(dict(item), ensure_ascii=False)
        self.file.write(f"{data}\n")
        return item


class DuplicatePipeline:
    """
    去除重復(fù)的數(shù)據(jù)，重復(fù)數(shù)據(jù)直接拋出異常，不會(huì)進(jìn)入下一個(gè)流水線處理
    """
    def __init__(self):
        self.book_url_set = set() 

    def process_item(self, item, spider):
        if item['book_url'] in self.book_url_set:
            raise DropItem('duplicate fiction, drop it')
        self.book_url_set.add(item['book_url'])
        return item

我來(lái)簡(jiǎn)單介紹下上面實(shí)現(xiàn)的兩個(gè) pipelines 的代碼。首先爬蟲(chóng)抓取的 item 結(jié)果經(jīng)過(guò)的是 DuplicatePipeline 這個(gè)管道 (我們通過(guò)管道的優(yōu)先級(jí)控制)，我們?cè)?DuplicatePipeline 中定義了一個(gè)全局的集合 (set)，在 管道的核心方法process_item() 中，我們先判斷傳過(guò)來(lái)的 item 中 book_url 的值是否存在，如果存在則判定重復(fù)，然后拋出異常，這樣下一個(gè)管道 (即 QidianYuepiaoPipeline) 就不會(huì)去處理；

在經(jīng)過(guò)的第二個(gè)管道 (QidianYuepiaoPipeline) 中，我們主要是將不重復(fù) item 保存到本地文件中，因此我們會(huì)在 open_spider() 方法中打開(kāi)文件句柄，在 close_spider() 方法中關(guān)閉文件句柄，而在 process_item() 中將 item 數(shù)據(jù)保存到指定的文件中。

接著就是將這兩個(gè) Pipelines 加到 settings.py 中：

ITEM_PIPELINES = {
    'qidian_yuepiao.pipelines.DuplicatePipeline': 200,
    'qidian_yuepiao.pipelines.QidianYuepiaoPipeline': 300,
}

最后，我們來(lái)介紹一個(gè) Scrapy 內(nèi)置的圖片管道，其實(shí)現(xiàn)的 Pipeline 代碼位置為：scrapy/pipelines/images.py，對(duì)應(yīng)的還有一個(gè)內(nèi)置的文件管道。我們不需要編寫(xiě)任何代碼，只需要在 settings.py 中指定下載的圖片字段即可：

# 下載圖片存儲(chǔ)位置
IMAGES_STORE = '/root/scrapy-test/qidian_yuepiao/qidian_yuepiao/images'
# 保存下載圖片url地址的字段
IMAGES_URLS_FIELD = 'image_url'
# 圖片保存地址字段
IMAGES_RESULT_FIELD = 'images'
IMAGES_THUMBS = {
  'small': (102, 136),
  'big': (150, 200)
}

# ...

ITEM_PIPELINES = {
    'scrapy.pipelines.images.ImagesPipeline': 1,
    'qidian_yuepiao.pipelines.DuplicatePipeline': 200,
    'qidian_yuepiao.pipelines.QidianYuepiaoPipeline': 300,
}

由于 ImagesPipeline 繼承自 FilesPipeline，我們可以從官網(wǎng)的介紹中知道該圖片下載功能的執(zhí)行流程如下：

• 在 spider 中爬取需要下載的圖片鏈接，將其放入 item 的 image_url 字段中；
• spider 將得到的 item 傳送到 pipeline 進(jìn)行處理；
• 當(dāng) item 到達(dá) Image Pipeline 處理時(shí)，它會(huì)檢測(cè)是否有 image_url 字段，如果存在的話，會(huì)將該 url 傳遞給 scrapy 調(diào)度器和下載器；
• 下載完成后會(huì)將結(jié)果寫(xiě)入 item 的另一個(gè)字段 images，images 包含了圖片的本地路徑、圖片校驗(yàn)、以及圖片的url；

完成了以上四步之后，我們的這樣一個(gè)簡(jiǎn)單需求就算完成了。還等什么，快來(lái)運(yùn)行看看！以下是視頻演示：

這樣爬取數(shù)據(jù)是不是非常有趣？使用了 Scrapy 框架后，我們的爬取流程會(huì)變得比較固定化以及流水線化。但我們不僅僅要學(xué)會(huì)怎么使用 Scrapy 框架，還要能夠基于 Scrapy 框架在特定場(chǎng)景下做些改造，這樣才能達(dá)到完全駕馭 Scrapy 框架的目的。

4. 小結(jié)

本小節(jié)中，我們介紹了 Scrapy 中 Pipeline 相關(guān)的知識(shí)并在起點(diǎn)中文網(wǎng)上進(jìn)行了簡(jiǎn)單的演示。在我們的爬蟲(chóng)項(xiàng)目中使用了兩個(gè)自定義管道，分別用于去除重復(fù)小說(shuō)以及將非重復(fù)的小說(shuō)數(shù)據(jù)保存到本地文件中；另外我們還啟用了 Scrapy 內(nèi)置的圖片下載管道，幫助我們自動(dòng)處理圖片 URL 并下載。