运维学python之爬虫高级篇（四）Item Pipeline介绍(附爬取网站获取图片到本地代码) 推荐

1 Item Pipeline

当spider爬取到item后，它被发送到项目管道（Item Pipeline），通过几个组件按顺序进行处理。每一个Item Pipeline是一个实现了简单方法的Python类，它接收到一个item并对其执行一个操作，也要决定该item是否应该继续通过管道，或者被丢弃，不再进行处理。
Item Pipeline典型的用途是：
1.清理HTML数据
2.验证爬取的数据(检查items是否包含某些字段)
3.检查副本(并删除它们)
4.将item数据存储在数据库中

1.1 编写自己的Item Pipeline

每个Item Pipeline都是一个Python类，它必须实现以下方法:
process_item(self, item, spider)
这个方法可以被每个Item Pipeline调用，process_item()必须是:返回一个字典类型数据、返回一个条目(或任何子类)对象，返回一个 Twisted Deferred 或者DropItem异常，丢弃的item不再由进一步的Item Pipeline处理。
参数含义：
item： Item对象或字典，爬取的item
spider：spider对象，爬取了这个item的spider
此外，他们还可以实现以下方法:
open_spider(self, spider)
当spider打开时，函数就会被调用，spider参数含义：被打开的spider
close_spider(self, spider)
当spider关闭是，函数会被调用
from_crawler(cls, crawler)
如果存在，这个类方法被调用来从一个Crawler创建一个spider实例。它必须返回管道的一个新实例，Crawler对象提供对所有的scrapy核心组件的访问，比如设置和信号;这是管道访问它们并将其功能连接到scrapy的一种方式。

1.2 Pipeline示例

1.2.1 价格验证示例

from scrapy.exceptions import DropItem

class PricePipeline(object):

vat_factor = 1.15

def process_item(self, item, spider):
if item['price']:
if item['price_excludes_vat']:
item['price'] = item['price'] * self.vat_factor
return item
else:
raise DropItem("Missing price in %s" % item)

1.2.2 写入json文件

下面的Pipeline将所有经过的项目(从所有的spiders)存储到一个item.jl文件中，其中每行以JSON格式序列化:

import json

class JsonWriterPipeline(object):

def open_spider(self, spider):
self.file = open('items.jl', 'w')

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

def process_item(self, item, spider):
line = json.dumps(dict(item)) + "\n"
self.file.write(line)
return item

1.2.3 写入MongoDB

在本例中，我们将使用pymongo将items写入MongoDB。MongoDB地址和数据库名称在scrapy settings中指定;MongoDB集合以item类命名。本例的主要目的是展示如何使用from_crawler()方法以及如何正确地清理资源。

import pymongo

class MongoPipeline(object):

collection_name = 'scrapy_items'

def __init__(self, mongo_uri, mongo_db):
self.mongo_uri = mongo_uri
self.mongo_db = mongo_db

@classmethod
def from_crawler(cls, crawler):
return cls(
mongo_uri=crawler.settings.get('MONGO_URI'),
mongo_db=crawler.settings.get('MONGO_DATABASE', 'items')
)

def open_spider(self, spider):
self.client = pymongo.MongoClient(self.mongo_uri)
self.db = self.client[self.mongo_db]

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

def process_item(self, item, spider):
self.db[self.collection_name].insert_one(dict(item))
return item

1.2.4 重复过滤器

用于查找重复items的筛选器，并删除已处理的item,假设我们的items有一个惟一的id，但是我们的spider返回的是具有相同id的多个items:

from scrapy.exceptions import DropItem

class DuplicatesPipeline(object):

def __init__(self):
self.ids_seen = set()

def process_item(self, item, spider):
if item['id'] in self.ids_seen:
raise DropItem("Duplicate item found: %s" % item)
else:
self.ids_seen.add(item['id'])
return item

1.2.5 激活Item Pipeline

激活Item Pipeline 必须要在setting中设置ITEM_PIPELINES，示例如下：

ITEM_PIPELINES = {
'myproject.pipelines.PricePipeline': 300,
'myproject.pipelines.JsonWriterPipeline': 800,
}

在这个设置中分配给类的整数值决定了它们运行的顺序:item从低到高执行，整数值范围是0-1000。

2 Feed exports

执行scrapy时最常需要的特性之一就是能够正确地存储爬取出来的数据，scrapy提供了这个功能，允许使用多种序列化格式来生成一个Feed。

2.1 序列化格式

用于序列化scrapy的数据格式主要有以下几种类型：

JSON

JSON lines

CSV

XML

你也可以通过setting中的FEED_EXPORTERS字段来扩展支持的格式。
JSON
FEED_FORMAT: json
使用的类: JsonItemExporter
JSON lines
FEED_FORMAT: jsonlines
使用的类: JsonLinesItemExporter
CSV
FEED_FORMAT: csv
使用的类: CsvItemExporter
XML
FEED_FORMAT: xml
使用的类: XmlItemExporter
Pickle
FEED_FORMAT: pickle
使用的类: PickleItemExporter
Marshal
FEED_FORMAT: marshal
使用的类: MarshalItemExporter

2.2 使用方法

进入项目目录，执行命令：

scrapy crawl tushu -o tushu.json

通过-o参数后面接要输出的格式即可。

3 下载和处理文件和图像

scrapy提供了可重用的 item pipelines，用于下载与特定item 相关的文件（例如，当你爬取了产品并想要在本地下载它们的图像时），这些pipelines共享一些功能和结构(我们将它们称为media pipelines)，但是通常要么使用Files Pipeline 要么使用 Images Pipeline。
这两个Pipeline都实现了这些特性:

避免重新下载最近下载的媒体

指定存储介质的位置（文件系统目录等）

Image Pipeline有一些额外的功能用于处理图像:

将所有下载的图像转换为通用格式(JPG)和模式(RGB)

生成缩略图

检查图像宽度/高度以确保它们满足最小约束条件

Pipeline为正准备下载的media url的保留了内部队列，将包含相同媒体的response连接到该队列，这样可以避免在多个items共享的情况下下载相同的媒体。

3.1 使用Files Pipeline

使用Files Pipeline典型的工作流程如下：
1.在一个spider中，你将一个item提取并且将所需的urls放入file_urls字段中；
2.item将从spider返回并进入item pipeline；
3.当item到达FilePipeline，在file_urls字段中的urls会使用标准scrapy调度器和下载器下载（这意味着调度程序和下装程序中间件被重用），如果优先级更高，会在其他页面被抓取前处理。item会在这个特定的pipline中保持“locker”状态，知道完成下载（或由于某些原因未完成下载）。
4.当下载文件时，将使用结果填充另一个字段(files)，这个字段将包含一个关于下载文件的信息的字典，例如下载路径、原始url（来自file_urls字段）和文件校验。文件字段列表中的files将保留原来的file_urls字段的相同顺序，如果有下载失败的文件，错误将会被记录，而file不会被记录到files字段中。

3.2 使用Images Pipeline

Images Pipeline的使用方式与File Pipeline方式类似，只是默认的字段名称不同，使用image_urls作为一个item的图片urls，它将填充一个图像image字段，以获取关于下载的图像的信息。
使用ImagesPipeline对于处理image files的优点是，您可以配置一些额外的功能，比如生成缩略图和根据它们的大小过滤图像。
Images Pipeline程序使用Pillow模块格式化图片为JPEG/RGB格式，所以你还需要安装Pillow模块，大多数情况下我们使用PIL，但众所周知，在某些情况下会引起麻烦，所以我们建议用Pillow。

3.3 使用Media Pipeline

如果要使用Media Pipeline你必须要在项目的setting中增加ITEM_PIPELINES设置，对于Images Pipeline，使用：

ITEM_PIPELINES = {'scrapy.pipelines.images.ImagesPipeline': 1}

Files Pipeline，使用：

ITEM_PIPELINES = {'scrapy.pipelines.files.FilesPipeline': 1}

注意：Images Pipeline和Files Pipeline可以同时使用。

然后，将目标存储设置配置为一个有效值，该值将用于存储下载的图像。否则即使你配置了ITEM_PIPELINES,也是被禁用的。
如果是File Pipeline，在setting中增加FILES_STORE:

FILES_STORE = '/path/to/valid/dir'

如果是Image Pipeline，在setting中增加IMAGES_STORE:

IMAGES_STORE = '/path/to/valid/dir'

3.4 支持的存储

目前官方唯一支持的是文件系统，但是也支持类似的Amazon S3 and和 Google Cloud Storage.

3.5 示例

1.首先使用media pipline首先要启用它，在setting中配置：

ITEM_PIPELINES = {'scrapy.pipelines.images.ImagesPipeline': 1}

2.然后设置字段images和image_urls：

import scrapy

class MyItem(scrapy.Item):

# ... other item fields ...
image_urls = scrapy.Field()
images = scrapy.Field()

3.在setting中添加下载路径和字段：

# 图片下载存储路径
ITEM_STORE = 'E:\\'

为了避免下载最近下载的文件，可以设置FILES_EXPIRES或IMAGES_EXPIRES来配置缓存时间：

# 120天后过期
FILES_EXPIRES = 120

# 30天后过期
IMAGES_EXPIRES = 30

Images Pipline可以自动创建下载图像的缩略图，在setting中增加IMAGES_THUMBS参数,参数为一个字典，其中的键是缩略图名称，而值是它们的维数：

IMAGES_THUMBS = {
'small': (50, 50),
'big': (270, 270),
}

如果想过滤掉小图片，通过设置IMAGES_MIN_HEIGHT和 IMAGES_MIN_WIDTH来指定图像大小：

IMAGES_MIN_HEIGHT = 110
IMAGES_MIN_WIDTH = 110

这个值的配置不会影响缩略图的生成。
通过上面的配置我们就可以为我们的爬虫添加下载图片功能了。

4 小爬虫

上面说了那么多，大家可能觉得已经一头雾水了，接下来我们就用一个小项目来具体说明一下，我们要爬取的网站是（搜房网二手房http://esf.fang.com/ 页面中的各个房源图片）如下图：







获取网页列表中，每个条目的详细页中的图片。

4.1 启用pipeline

setting.py中增加如下内容：

# 新增内容从这里开始#################################################################
# 启动pipline
ITEM_PIPELINES = {
# 注意，如果自定义图片名称时，此条内容要注释，不然自定义图片名不生效
'scrapy.pipelines.images.ImagesPipeline': 1,
# 自定义图片名称后，可以取消注释此条
# 'sp.pipelines.SpDownimagePipeline': 200,
}
# 图片保存地址
IMAGES_STORE = 'E:\\'
# 图片过期时间30天
IMAGES_EXPIRES = 30
# 设置缩略图
# IMAGES_THUMBS = {
#     'small': (50, 50),
#     'big': (270, 270),
# }
# 过滤小图片
# IMAGES_MIN_HEIGHT = 110
# IMAGES_MIN_WIDTH = 110
# 允许重定向
MEDIA_ALLOW_REDIRECTS = True
# 控制时间，下载等待时间3秒
DOWNLOAD_DELAY = 3
# 请求user-agent
USER_AGENT = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:57.0) Gecko/20100101 Firefox/57.0'
# 新增内容从这里结束#################################################################

4.2 配置items

设置要爬取的网页名字段image和爬取网页内的图片链接字段image_urls，items.py代码如下：

# -*- coding: utf-8 -*-

# Define here the models for your scraped items
#
# See documentation in:
# http://doc.scrapy.org/en/latest/topics/items.html 
import scrapy

class SpItem(scrapy.Item):
"""
定义item字段
"""
# 网页名称
image = scrapy.Field()
# 网页内图片链接
image_urls = scrapy.Field()

4.3 spider

我们的爬虫ftx.py代码如下：

# -*- coding: utf-8 -*-
import scrapy
from sp.items import SpItem

class MyBlog(scrapy.Spider):
name = 'ftx'
start_urls = ['http://esf.fang.com']

def parse(self, response):
"""
爬取初始start_urls列表项目的url(相对路径)，通过response.follow生成
request，作为参数传入回掉函数parse_item
"""
# 获取首页所有二手房的链接地址（相对地址）
page_urls = response.css("p.title a::attr(href)").extract()
for page_url in page_urls:
# 相对地址使用response.follow方式拼接url
request = response.follow(page_url, callback=self.parse_item)
# 如果获取的连接是绝对地址，用下面这条方法
# request = scrapy.Request(page_url, callback=self.parse_item)
yield request

def parse_item(self, response):
"""
处理item函数
:param response: 请求的网页内容
:return: item
"""
# 导入item类
item = SpItem()
# 每个个人页面中的图片
# image是每个详细页的标题， image_urls是每个详细页内图片的url
item['image'] = response.css("div.floatl::text").extract_first().strip()
item['image_urls'] = response.css("img.loadimg::attr(data-src)").extract()
yield item

4.4 自定义image pipeline

直接上代码pipelines.py：

# -*- coding: utf-8 -*-

# Define your item pipelines here
#
# Don't forget to add your pipeline to the ITEM_PIPELINES setting
# See: http://doc.scrapy.org/en/latest/topics/item-pipeline.html import scrapy
from scrapy.pipelines.images import ImagesPipeline

class SpDownimagePipeline(ImagesPipeline):
"""
自定义图片下载类
"""
def get_media_requests(self, item, info):
"""
ImagesPipeline类的方法,必须返回每个图像URL的Request
:param item:获取的item
"""
# 从item中获取图片url并发送请求，image_urls就是items.py中定义的字段
for image_url in item['image_urls']:
# meta作用就是可以将item的值传给下一个函数使用，类似于先缓存起来
yield scrapy.Request(image_url, meta={'item': item})

def item_completed(self, results, item, info):
"""
此处没有做修改，只是把ImagesPipeline的方法拿过来用，必须返回item
"""
if isinstance(item, dict) or self.images_result_field in item.fields:
item[self.images_result_field] = [x for ok, x in results if ok]
return item

def file_path(self, request, response=None, info=None):
"""
file_path为ImagePipeline自带的方法，这里我们重写这个方法，
为了能够自定义图片的名称，如果不重写，SHA1 hash格式，类似full/63bbfea82b8880ed33cdb762aa11fab722a90a24.jpg
"""
# 获取item，从get_media_requests的Request中获取
item = request.meta['item']
# 图片名称，一版用split（‘/’）分割后取最后一个值也就是-1，这里没用-1是因为图片最后一个字段不是随机数
# 是长乘以宽如：452x340c.jpg，容易重名，所以用的-2，倒数第二个字段
image_guid = request.url.split('/')[-2] + '.jpg'
# 全名，包括路径和图片名
fullname = "full/%s/%s" % (item['image'], image_guid)
return fullname

对于ImagesPipeline的两个方法get_media_requests和item_completed这里解释一下：
get_media_requests(item, info)
pipeline会获取image的urls从item下载它，因此我们可以重写get_media_requests方法并且返回每一个url的request：

def get_media_requests(self, item, info):
for file_url in item['file_urls']:
yield scrapy.Request(file_url)

这些请求将由pipeline处理,当完成下载时结果将会以2-元素的元组形式被发送到item_completed方法，每个元组将包含（success， file_info_or_error）类似下面这种形式：

[(True,
{'checksum': '2b00042f7481c7b056c4b410d28f33cf',
'path': 'full/0a79c461a4062ac383dc4fade7bc09f1384a3910.jpg',
'url': 'http://www.example.com/files/product1.pdf'}),
(False,
Failure(...))]

success：布尔值，如果下载图片成功，返回True,如果下载图片失败，返回False。file_info_or_error：返回的是一个字典，其中包括，url、path和checksum,如果出现问题返回Twisted Failure。

url代表文件从哪里下载的，这是从get_media_requests返回的request的url

path代表文件存储路径

checksum代表图像内容的MD5 hash

item_completed(results, item, info)
当一个单独项目中所有图片请求完成时（下载完成或者下载失败），此方法将会被调用，其中results参数为get_media_requests下载完成后返回的结果，item_completed必须返回输出发送到下一个阶段的pipeline。所以你必须返回或删除item，和之前其它pipeline操作一样。
下面的一个示例，我们将下载的文件路径(在results中传递)存储在file_path item字段中，如果不包含任何文件，则删除该项目。

from scrapy.exceptions import DropItem

def item_completed(self, results, item, info):
file_paths = [x['path'] for ok, x in results if ok]
if not file_paths:
raise DropItem("Item contains no files")
item['file_paths'] = file_paths
return item

下面是一个完整的自定义Image pipeline示例：

import scrapy
from scrapy.pipelines.images import ImagesPipeline
from scrapy.exceptions import DropItem

class MyImagesPipeline(ImagesPipeline):

def get_media_requests(self, item, info):
for image_url in item['image_urls']:
yield scrapy.Request(image_url)

def item_completed(self, results, item, info):
image_paths = [x['path'] for ok, x in results if ok]
if not image_paths:
raise DropItem("Item contains no images")
item['image_paths'] = image_paths
return item