scrapy框架之Pipeline管道類

Item Pipeline簡介

Item管道的主要責任是負責處理有蜘蛛從網頁中抽取的Item，他的主要任務是清洗、驗證和存儲數據。
當頁面被蜘蛛解析后，將被發送到Item管道，並經過幾個特定的次序處理數據。
每個Item管道的組件都是有一個簡單的方法組成的Python類。
他們獲取了Item並執行他們的方法，同時他們還需要確定的是是否需要在Item管道中繼續執行下一步或是直接丟棄掉不處理。

項目管道的執行過程

清理HTML數據
驗證解析到的數據（檢查Item是否包含必要的字段）
檢查是否是重復數據（如果重復就刪除）
將解析到的數據存儲到數據庫中

編寫自己的Item Pipeline

每個項目管道組件是一個Python類，必須實現以下方法：

process_item(self, item, spider)

對於每個項目管道組件調用此方法。process_item() 必須返回一個帶數據的dict，返回一個Item （或任何后代類）對象，返回一個Twisted Deferred或者raise DropItemexception。丟棄的項目不再由其他管道組件處理。

參數：

• item（Itemobject或dict） - 剪切的項目
• Spider（Spider對象） - 抓取物品的蜘蛛

另外，它們還可以實現以下方法：

# 當蜘蛛打開時調用此方法。
open_spider(self, spider)   # 參數spider打開的蜘蛛

# 當蜘蛛關閉時調用此方法。
close_spider(self, spider)   # 參數spider被關閉的蜘蛛

# 如果存在，則調用此類方法以從a創建流水線實例Crawler。它必須返回管道的新實例。Crawler對象提供對所有Scrapy核心組件（如設置和信號）的訪問; 它是管道訪問它們並將其功能掛鈎到Scrapy中的一種方式。 
from_crawler(cls, crawler)  # 參數crawler（Crawlerobject） - 使用此管道的crawler

將抓取的items以json格式保存到文件中

從spider抓取到的items將被序列化為json格式，並且以每行一個item的形式被寫入到items.jl文件中

import json 
  
class JsonWriterPipeline(object): 
  
  def __init__(self): 
    self.file = open('items.jl', 'wb') 
  
  def process_item(self, item, spider): 
    line = json.dumps(dict(item)) + "\n"
    self.file.write(line) 
    return item

刪除重復項

假設在spider中提取到的item有重復的id，那么我們就可以在process_item函數中進行過濾

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

激活ItemPipeline組件

在settings.py文件中，往ITEM_PIPELINES中添加項目管道的類名，就可以激活項目管道組件

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

圖像管道

items

定義過濾字段

import scrapy

class ImgpilelineproItem(scrapy.Item):
    # define the fields for your item here like:
    img_src = scrapy.Field()

spider

只是用來獲取圖片的下載地址並提交至itme

import scrapy
from imgPileLinePro.items import ImgpilelineproItem

class ImgSpider(scrapy.Spider):
    name = 'img'
    # allowed_domains = ['www.xxx.com']
    start_urls = ['http://pic.netbian.com/4kmeinv/']
    url = 'http://pic.netbian.com/4kmeinv/index_%d.html'
    page = 2

    def parse(self, response):
        li_list = response.xpath('//*[@id="main"]/div[3]/ul/li')
        for li in li_list:
            img_src = 'http://pic.netbian.com'+li.xpath('./a/img/@src').extract_first()
            item = ImgpilelineproItem()
            item['img_src'] = img_src

            yield item

        if self.page <= 2:  # 爬取前兩頁
            new_url = format(self.url%self.page)
            self.page += 1
            yield scrapy.Request(url=new_url,callback=self.parse)

pipelines

from scrapy.pipelines.images import ImagesPipeline
import scrapy

# 用來下載圖片的管道類
class ImgPileLine(ImagesPipeline):
    #接收item且將item中存儲的img_src進行請求發送
    def get_media_requests(self,item,info):
        yield scrapy.Request(url=item['img_src'])

    #指定數據存儲的路徑(文件夾【在配置文件中指定】+圖片名稱【該方法中返回】)
    def file_path(self,request,response=None,info=None):
        img_name = request.url.split('/')[-1]
        return img_name

    #就是將item傳遞給下一個即將被執行的管道類
    def item_completed(self,result,item,info):
        return item

settings中的配置

# 指定文件的下載路徑
IMAGES_STORE = './imgsLib'   # 該文件會自動創建
# 啟用管道
ITEM_PIPELINES = {
   'imgPileLinePro.pipelines.ImgPileLine': 300,
}

將item寫入到mongodb

import pymongo

class MongoPipeline(object):

    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')
        )
    # 爬蟲開始建立與mongodb的連接
    def open_spider(self, spider):
        self.client = pymongo.MongoClient(self.mongo_uri)
        self.db = self.client[self.mongo_db]
    # 爬蟲結束斷開與mongodb的連接
    def close_spider(self, spider):
        self.client.close()
    # 數據存儲
    def process_item(self, item, spider):
        # update 去重,以url_token為查找條件更新數據
        self.db["user"].update({"url_token":item["url_token"]},{"$set":item},True)
        return item

MONGO_URI = "localhost"
MONGO_DATABASE = "zhihu"