當前位置：首頁 > 编程资源 > 编程问答 >内容正文

编程问答

第45讲：哪都能存，Item Pipeline 的用法

發布時間：2024/4/11 编程问答 51 豆豆

生活随笔收集整理的這篇文章主要介紹了第45讲：哪都能存，Item Pipeline 的用法小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

在前面的示例中我們已經了解了 Item Pipeline 項目管道的基本概念，本節課我們就深入詳細講解它的用法。

首先我們看看 Item Pipeline 在 Scrapy 中的架構，如圖所示。

圖中的最左側即為 Item Pipeline，它的調用發生在 Spider 產生 Item 之后。當 Spider 解析完 Response 之后，Item 就會傳遞到 Item Pipeline，被定義的 Item Pipeline 組件會順次調用，完成一連串的處理過程，比如數據清洗、存儲等。

它的主要功能有：

清洗 HTML 數據；
驗證爬取數據，檢查爬取字段；
查重并丟棄重復內容；
將爬取結果儲存到數據庫。

1. 核心方法

我們可以自定義 Item Pipeline，只需要實現指定的方法就可以，其中必須要實現的一個方法是：

process_item(item, spider)

另外還有幾個比較實用的方法，它們分別是：

open_spider(spider)
close_spider(spider)
from_crawler(cls, crawler)

下面我們對這幾個方法的用法做下詳細的介紹：

process_item(item, spider)

process_item() 是必須要實現的方法，被定義的 Item Pipeline 會默認調用這個方法對 Item 進行處理。比如，我們可以進行數據處理或者將數據寫入數據庫等操作。它必須返回 Item 類型的值或者拋出一個 DropItem 異常。

process_item() 方法的參數有如下兩個：

item，是 Item 對象，即被處理的 Item；
spider，是 Spider 對象，即生成該 Item 的 Spider。

下面對該方法的返回類型歸納如下：

如果返回的是 Item 對象，那么此 Item 會被低優先級的 Item Pipeline 的 process_item() 方法進行處理，直到所有的方法被調用完畢。
如果拋出的是 DropItem 異常，那么此 Item 就會被丟棄，不再進行處理。

open_spider(self, spider)

open_spider() 方法是在 Spider 開啟的時候被自動調用的，在這里我們可以做一些初始化操作，如開啟數據庫連接等。其中參數 spider 就是被開啟的 Spider 對象。

close_spider(spider)

close_spider() 方法是在 Spider 關閉的時候自動調用的，在這里我們可以做一些收尾工作，如關閉數據庫連接等，其中參數 spider 就是被關閉的 Spider 對象。

from_crawler(cls, crawler)

from_crawler() 方法是一個類方法，用 @classmethod 標識，是一種依賴注入的方式。它的參數是 crawler，通過 crawler 對象，我們可以拿到 Scrapy 的所有核心組件，如全局配置的每個信息，然后創建一個 Pipeline 實例。參數 cls 就是 Class，最后返回一個 Class 實例。

下面我們用一個實例來加深對 Item Pipeline 用法的理解。

2. 本節目標

我們以爬取 360 攝影美圖為例，來分別實現 MongoDB 存儲、MySQL 存儲、Image 圖片存儲的三個 Pipeline。

3. 準備工作

請確保已經安裝好 MongoDB 和 MySQL 數據庫，安裝好 Python 的 PyMongo、PyMySQL、Scrapy 框架，另外需要安裝 pillow 圖像處理庫，如果沒有安裝可以參考前文的安裝說明。

4. 抓取分析

我們這次爬取的目標網站為：https://image.so.com。打開此頁面，切換到攝影頁面，網頁中呈現了許許多多的攝影美圖。我們打開瀏覽器開發者工具，過濾器切換到 XHR 選項，然后下拉頁面，可以看到下面就會呈現許多 Ajax 請求，如圖所示。

我們查看一個請求的詳情，觀察返回的數據結構，如圖所示。

返回格式是 JSON。其中 list 字段就是一張張圖片的詳情信息，包含了 30 張圖片的 ID、名稱、鏈接、縮略圖等信息。另外觀察 Ajax 請求的參數信息，有一個參數 sn 一直在變化，這個參數很明顯就是偏移量。當 sn 為 30 時，返回的是前 30 張圖片，sn 為 60 時，返回的就是第 31~60 張圖片。另外，ch 參數是攝影類別，listtype 是排序方式，temp 參數可以忽略。

所以我們抓取時只需要改變 sn 的數值就好了。下面我們用 Scrapy 來實現圖片的抓取，將圖片的信息保存到 MongoDB、MySQL，同時將圖片存儲到本地。

5. 新建項目

首先新建一個項目，命令如下：

scrapy startproject images360

接下來新建一個 Spider，命令如下：

scrapy genspider images images.so.com

這樣我們就成功創建了一個 Spider。

6. 構造請求

接下來定義爬取的頁數。比如爬取 50 頁、每頁 30 張，也就是 1500 張圖片，我們可以先在 settings.py 里面定義一個變量 MAX_PAGE，添加如下定義：

MAX_PAGE = 50

定義 start_requests() 方法，用來生成 50 次請求，如下所示：

def start_requests(self):data = {'ch': 'photography', 'listtype': 'new'}base_url = 'https://image.so.com/zjl?'for page in range(1, self.settings.get('MAX_PAGE') + 1):data['sn'] = page * 30params = urlencode(data)url = base_url + paramsyield Request(url, self.parse)

在這里我們首先定義了初始的兩個參數，sn 參數是遍歷循環生成的。然后利用 urlencode 方法將字典轉化為 URL 的 GET 參數，構造出完整的 URL，構造并生成 Request。

還需要引入 scrapy.Request 和 urllib.parse 模塊，如下所示：

from scrapy import Spider, Request from urllib.parse import urlencode

再修改 settings.py 中的 ROBOTSTXT_OBEY 變量，將其設置為 False，否則無法抓取，如下所示：

ROBOTSTXT_OBEY = False

運行爬蟲，即可以看到鏈接都請求成功，執行命令如下所示：

scrapy crawl images

運行示例結果如圖所示。

所有請求的狀態碼都是 200，這就證明圖片信息爬取成功了。

7. 提取信息

首先定義一個叫作 ImageItem 的 Item，如下所示：

from scrapy import Item, Field class ImageItem(Item):collection = table = 'images'id = Field()url = Field()title = Field()thumb = Field()

在這里我們定義了 4 個字段，包括圖片的 ID、鏈接、標題、縮略圖。另外還有兩個屬性 collection 和 table，都定義為 images 字符串，分別代表 MongoDB 存儲的 Collection 名稱和 MySQL 存儲的表名稱。

接下來我們提取 Spider 里有關信息，將 parse 方法改寫為如下所示：

def parse(self, response):result = json.loads(response.text)for image in result.get('list'):item = ImageItem()item['id'] = image.get('id')item['url'] = image.get('qhimg_url')item['title'] = image.get('title')item['thumb'] = image.get('qhimg_thumb')yield item

首先解析 JSON，遍歷其 list 字段，取出一個個圖片信息，然后再對 ImageItem 進行賦值，生成 Item 對象。
這樣我們就完成了信息的提取。

8. 存儲信息

接下來我們需要將圖片的信息保存到 MongoDB、MySQL 中，同時將圖片保存到本地。

MongoDB

首先確保 MongoDB 已經正常安裝并且能夠正常運行。

我們用一個 MongoPipeline 將信息保存到 MongoDB 中，在 pipelines.py 里添加如下類的實現：

import pymongo class MongoPipeline(object):def __init__(self, mongo_uri, mongo_db):self.mongo_uri = mongo_uriself.mongo_db = mongo_db@classmethoddef from_crawler(cls, crawler):return cls(mongo_uri=crawler.settings.get('MONGO_URI'),mongo_db=crawler.settings.get('MONGO_DB'))def open_spider(self, spider):self.client = pymongo.MongoClient(self.mongo_uri)self.db = self.client[self.mongo_db]def process_item(self, item, spider):self.db[item.collection].insert(dict(item))return itemdef close_spider(self, spider):self.client.close()

這里需要用到兩個變量，MONGO_URI 和 MONGO_DB，即存儲到 MongoDB 的鏈接地址和數據庫名稱。我們在 settings.py 里添加這兩個變量，如下所示：

MONGO_URI = 'localhost' MONGO_DB = 'images360'

這樣一個保存到 MongoDB 的 Pipeline 的就創建好了。這里最主要的方法是 process_item()，直接調用 Collection 對象的 insert 方法即可完成數據的插入，最后返回 Item 對象。

MySQL

首先需要確保 MySQL 已經正確安裝并且正常運行。

新建一個數據庫，名字還是 images360，SQL 語句如下所示：

CREATE DATABASE images360 DEFAULT CHARACTER SET utf8 COLLATE utf8_general_ci

新建一個數據表，包含 id、url、title、thumb 四個字段，SQL 語句如下所示：

CREATE TABLE images (id VARCHAR(255) NULL PRIMARY KEY, url VARCHAR(255) NULL , title VARCHAR(255) NULL , thumb VARCHAR(255) NULL)

執行完 SQL 語句之后，我們就成功創建好了數據表。接下來就可以往表里存儲數據了。

接下來我們實現一個 MySQLPipeline，代碼如下所示：

import pymysql class MysqlPipeline():def __init__(self, host, database, user, password, port):self.host = hostself.database = databaseself.user = userself.password = passwordself.port = port@classmethoddef from_crawler(cls, crawler):return cls(host=crawler.settings.get('MYSQL_HOST'),database=crawler.settings.get('MYSQL_DATABASE'),user=crawler.settings.get('MYSQL_USER'),password=crawler.settings.get('MYSQL_PASSWORD'),port=crawler.settings.get('MYSQL_PORT'),)def open_spider(self, spider):self.db = pymysql.connect(self.host, self.user, self.password, self.database, charset='utf8', port=self.port)self.cursor = self.db.cursor()def close_spider(self, spider):self.db.close()def process_item(self, item, spider):data = dict(item)keys = ', '.join(data.keys())values = ', '.join(['% s'] * len(data))sql = 'insert into % s (% s) values (% s)' % (item.table, keys, values)self.cursor.execute(sql, tuple(data.values()))self.db.commit()return item

如前所述，這里用到的數據插入方法是一個動態構造 SQL 語句的方法。

這里還需要幾個 MySQL 的配置，我們在 settings.py 里添加幾個變量，如下所示：

MYSQL_HOST = 'localhost' MYSQL_DATABASE = 'images360' MYSQL_PORT = 3306 MYSQL_USER = 'root' MYSQL_PASSWORD = '123456'

這里分別定義了 MySQL 的地址、數據庫名稱、端口、用戶名、密碼。這樣，MySQL Pipeline 就完成了。

Image Pipeline

Scrapy 提供了專門處理下載的 Pipeline，包括文件下載和圖片下載。下載文件和圖片的原理與抓取頁面的原理一樣，因此下載過程支持異步和多線程，十分高效。下面我們來看看具體的實現過程。

官方文檔地址為：https://doc.scrapy.org/en/latest/topics/media-pipeline.html。

首先定義存儲文件的路徑，需要定義一個 IMAGES_STORE 變量，在 settings.py 中添加如下代碼：

IMAGES_STORE = './images'

在這里我們將路徑定義為當前路徑下的 images 子文件夾，即下載的圖片都會保存到本項目的 images 文件夾中。

內置的 ImagesPipeline 會默認讀取 Item 的 image_urls 字段，并認為該字段是一個列表形式，它會遍歷 Item 的 image_urls 字段，然后取出每個 URL 進行圖片下載。

但是現在生成的 Item 的圖片鏈接字段并不是 image_urls 字段表示的，也不是列表形式，而是單個的 URL。所以為了實現下載，我們需要重新定義下載的部分邏輯，即需要自定義 ImagePipeline，繼承內置的 ImagesPipeline，重寫方法。

我們定義 ImagePipeline，如下所示：

from scrapy import Request from scrapy.exceptions import DropItem from scrapy.pipelines.images import ImagesPipeline class ImagePipeline(ImagesPipeline):def file_path(self, request, response=None, info=None):url = request.urlfile_name = url.split('/')[-1]return file_namedef item_completed(self, results, item, info):image_paths = [x['path'] for ok, x in results if ok]if not image_paths:raise DropItem('Image Downloaded Failed')return itemdef get_media_requests(self, item, info):yield Request(item['url'])

在這里我們實現了 ImagePipeline，繼承 Scrapy 內置的 ImagesPipeline，重寫了下面幾個方法。

get_media_requests()。它的第一個參數 item 是爬取生成的 Item 對象。我們將它的 url 字段取出來，然后直接生成 Request 對象。此 Request 加入調度隊列，等待被調度，執行下載。
file_path()。它的第一個參數 request 就是當前下載對應的 Request 對象。這個方法用來返回保存的文件名，直接將圖片鏈接的最后一部分當作文件名即可。它利用 split() 函數分割鏈接并提取最后一部分，返回結果。這樣此圖片下載之后保存的名稱就是該函數返回的文件名。
item_completed()，它是當單個 Item 完成下載時的處理方法。因為并不是每張圖片都會下載成功，所以我們需要分析下載結果并剔除下載失敗的圖片。如果某張圖片下載失敗，那么我們就不需保存此 Item 到數據庫。該方法的第一個參數 results 就是該 Item 對應的下載結果，它是一個列表形式，列表每一個元素是一個元組，其中包含了下載成功或失敗的信息。這里我們遍歷下載結果找出所有成功的下載列表。如果列表為空，那么該 Item 對應的圖片下載失敗，隨即拋出異常 DropItem，該 Item 忽略。否則返回該 Item，說明此 Item 有效。

現在為止，三個 Item Pipeline 的定義就完成了。最后只需要啟用就可以了，修改 settings.py，設置 ITEM_PIPELINES，如下所示：

ITEM_PIPELINES = {'images360.pipelines.ImagePipeline': 300,'images360.pipelines.MongoPipeline': 301,'images360.pipelines.MysqlPipeline': 302, }

這里注意調用的順序。我們需要優先調用 ImagePipeline 對 Item 做下載后的篩選，下載失敗的 Item 就直接忽略，它們就不會保存到 MongoDB 和 MySQL 里。隨后再調用其他兩個存儲的 Pipeline，這樣就能確保存入數據庫的圖片都是下載成功的。
接下來運行程序，執行爬取，如下所示：