数据项管道

数据项被 spider 抓取后，会被发送到 Item Pipeline，后者会依次通过多个组件对其进行处理。

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

数据项管道的典型用途包括：

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

编写自己的数据项管道

每个数据项管道都是一个组件，必须实现以下方法

process_item(self, item, spider)

此方法针对每个数据项管道组件调用。

item 是一个数据项对象，详见支持所有数据项类型。

process_item() 必须：返回一个数据项对象，返回一个Deferred 或引发一个 DropItem 异常。

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

参数:

item (数据项对象) – 抓取的数据项
spider (Spider 对象) – 抓取该数据项的 spider

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

open_spider(self, spider)

此方法在 spider 启动时调用。

参数:: spider (Spider 对象) – 已启动的 spider

close_spider(self, spider)

此方法在 spider 关闭时调用。

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

数据项管道示例

价格验证和丢弃没有价格的数据项

让我们看看下面这个假设的管道，它会调整不包含增值税（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 文件

以下管道将所有抓取的数据项（来自所有 spider）存储到单个 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 集合以数据项类命名。

此示例的主要目的是展示如何获取 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 = "http://localhost: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，但我们的 spider 返回了多个具有相同 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 范围内。