项目管道¶

在爬虫抓取项目后，它会被发送到项目管道，项目管道会通过按顺序执行的多个组件来处理它。

每个项目管道组件（有时也简称为“项目管道”）都是一个 Python 类，它实现了一个简单的方法。它们接收一个项目并在其上执行操作，并决定项目是否应该继续通过管道或被丢弃不再处理。

项目管道的典型用途是

清理 HTML 数据
验证抓取的数据（检查项目是否包含某些字段）
检查重复项（并丢弃它们）
将抓取的项目存储到数据库中

编写自己的项目管道¶

每个项目管道组件都是一个 Python 类，必须实现以下方法

process_item(self, item, spider)¶

此方法对每个项目管道组件都会调用。

item 是一个项目对象，请参见支持所有项目类型。

process_item() 必须执行以下操作之一：返回一个项目对象，返回一个 Deferred 或引发 DropItem 异常。

丢弃的项目将不再由后续的管道组件处理。

参数：

item (项目对象) – 抓取的项目
spider (Spider 对象) – 抓取项目的爬虫

此外，它们还可以实现以下方法

open_spider(self, spider)¶

此方法在爬虫打开时调用。

参数：: spider (Spider 对象) – 已打开的爬虫

close_spider(self, spider)¶

此方法在爬虫关闭时调用。

参数：: spider (Spider 对象) – 已关闭的爬虫

classmethod from_crawler(cls, crawler)¶

如果存在，则会调用此类方法以从 Crawler 创建管道实例。它必须返回管道的新实例。Crawler 对象提供对所有 Scrapy 核心组件（如设置和信号）的访问；它是管道访问它们并将功能挂钩到 Scrapy 的一种方式。

参数：: crawler (Crawler 对象) – 使用此管道的爬虫

项目管道示例¶

价格验证和丢弃没有价格的项目¶

让我们看一下以下假设的管道，它会为那些不包含增值税（price_excludes_vat 属性）的项目调整 price 属性，并丢弃那些不包含价格的项目

from itemadapter import ItemAdapter
from scrapy.exceptions import DropItem


class PricePipeline:
    vat_factor = 1.15

    def process_item(self, item, spider):
        adapter = ItemAdapter(item)
        if adapter.get("price"):
            if adapter.get("price_excludes_vat"):
                adapter["price"] = adapter["price"] * self.vat_factor
            return item
        else:
            raise DropItem("Missing price")

将项目写入 JSON Lines 文件¶

以下管道将所有抓取的项目（来自所有爬虫）存储到单个 items.jsonl 文件中，每个项目一行，以 JSON 格式序列化

import json

from itemadapter import ItemAdapter


class JsonWriterPipeline:
    def open_spider(self, spider):
        self.file = open("items.jsonl", "w")

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

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

注意

JsonWriterPipeline 的目的只是为了介绍如何编写项目管道。如果您确实想要将所有抓取的项目存储到 JSON 文件中，则应使用 Feed 导出。

将项目写入 MongoDB¶

在此示例中，我们将使用 pymongo 将项目写入 MongoDB。MongoDB 地址和数据库名称在 Scrapy 设置中指定；MongoDB 集合以项目类的名称命名。

此示例的主要目的是展示如何使用 from_crawler() 方法以及如何正确清理资源。

import pymongo
from itemadapter import ItemAdapter


class MongoPipeline:
    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(ItemAdapter(item).asdict())
        return item

截取项目的屏幕截图¶

此示例演示了如何在 process_item() 方法中使用协程语法。

此项目管道向本地运行的 Splash 实例发送请求，以渲染项目 URL 的屏幕截图。下载请求响应后，项目管道将屏幕截图保存到文件并将文件名添加到项目中。

import hashlib
from pathlib import Path
from urllib.parse import quote

import scrapy
from itemadapter import ItemAdapter
from scrapy.http.request import NO_CALLBACK
from scrapy.utils.defer import maybe_deferred_to_future


class ScreenshotPipeline:
    """Pipeline that uses Splash to render screenshot of
    every Scrapy item."""

    SPLASH_URL = "https://127.0.0.1:8050/render.png?url={}"

    async def process_item(self, item, spider):
        adapter = ItemAdapter(item)
        encoded_item_url = quote(adapter["url"])
        screenshot_url = self.SPLASH_URL.format(encoded_item_url)
        request = scrapy.Request(screenshot_url, callback=NO_CALLBACK)
        response = await maybe_deferred_to_future(
            spider.crawler.engine.download(request)
        )

        if response.status != 200:
            # Error happened, return item.
            return item

        # Save screenshot to file, filename will be hash of url.
        url = adapter["url"]
        url_hash = hashlib.md5(url.encode("utf8")).hexdigest()
        filename = f"{url_hash}.png"
        Path(filename).write_bytes(response.body)

        # Store filename in item.
        adapter["screenshot_filename"] = filename
        return item

重复项过滤器¶

一个查找重复项目的过滤器，并丢弃已处理过的那些项目。假设我们的项目有唯一的 ID，但我们的爬虫返回了多个具有相同 ID 的项目

from itemadapter import ItemAdapter
from scrapy.exceptions import DropItem


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

    def process_item(self, item, spider):
        adapter = ItemAdapter(item)
        if adapter["id"] in self.ids_seen:
            raise DropItem(f"Item ID already seen: {adapter['id']}")
        else:
            self.ids_seen.add(adapter["id"])
            return item

激活项目管道组件¶

要激活项目管道组件，必须将其类添加到 ITEM_PIPELINES 设置中，如下例所示

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

在此设置中分配给类的整数值决定了它们运行的顺序：项目从较低的值到较高值的类依次通过。通常在 0-1000 范围内定义这些数字。