Scrapy 中的 Pipline管道

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

1. Scrapy 中的 Pipeline 介紹

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

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

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

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

2. 如何編寫自己的 Item Pipeline

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

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

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

3. 實戰(zhàn)演練

學習了上面的一些知識，我們來使用一個簡單的網站進行實戰(zhàn)演練，在該過程中介紹更多的和 Item Pipeline 相關的用法。

假設我們是一名小說愛好者，我想到起點中文網上去找一些好的小說看，我該怎么找呢？起點中文網的月票榜是一個不錯的參考方式，如下圖所示：

其實簡單看一看就知道月票榜的 url 組成：

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

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

創(chuà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

接下來我們準備獲取小說作品的字段，大概會獲取如下幾個數(shù)據(jù)：

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

根據(jù)定義的這些字段，我們可以寫出對應的 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()

到了最關鍵的地方，需要解析網頁數(shù)據(jù)，提取月票榜的作品信息。這個和前面一些，我們只需要完成相應的 xpath 即可。此外，我們會從 01 月份的月票榜開始，每次會新生成一個 url，主要改動的就是月份參數(shù)，每次將月份數(shù)加一；如果當前月份大于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()
            # 注意一定要是列表，不然會報錯
            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ù)，同時也要提取請求的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
 
        # 一定要將月份轉成01, 02, s03這樣的形式，否則不能正確請求到數(shù)據(jù)
        next_month = num_to_str(month + 1)
        
        next_url = f"https://{prefix}?month={next_month}&{suffix}"
        yield Request(next_url)

最后到了我們本節(jié)課的重點。首先我想要將數(shù)據(jù)保存成 json 格式，存儲到文本文件中，但是在保存之前，需要對作品去重。因為有些作品會連續(xù)好幾個月出現(xiàn)在月票榜的前20位置上，會有比較多重復。我們通過作品的 url 地址來唯一確定該小說。因此需要定義兩個 Item Pipeline：

import json

from itemadapter import ItemAdapter
from scrapy.exceptions import DropItem


class QidianYuepiaoPipeline:
    """
    保存不重復的數(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:
    """
    去除重復的數(shù)據(jù)，重復數(shù)據(jù)直接拋出異常，不會進入下一個流水線處理
    """
    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

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

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

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

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

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

# 下載圖片存儲位置
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，我們可以從官網的介紹中知道該圖片下載功能的執(zhí)行流程如下：

• 在 spider 中爬取需要下載的圖片鏈接，將其放入 item 的 image_url 字段中；
• spider 將得到的 item 傳送到 pipeline 進行處理；
• 當 item 到達 Image Pipeline 處理時，它會檢測是否有 image_url 字段，如果存在的話，會將該 url 傳遞給 scrapy 調度器和下載器；
• 下載完成后會將結果寫入 item 的另一個字段 images，images 包含了圖片的本地路徑、圖片校驗、以及圖片的url；

完成了以上四步之后，我們的這樣一個簡單需求就算完成了。還等什么，快來運行看看！以下是視頻演示：

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

4. 小結

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