Item Pipeline（项目管道）

当一个项目被爬虫抓取后，它会被发送到 Item Pipeline，Item Pipeline 会通过几个按顺序执行的组件来处理它。

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

项目管道的典型用途是：

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

编写自己的项目管道

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

`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 对象）– 已关闭的爬虫

项目管道示例

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

让我们看看以下假设的管道，它调整那些不包含增值税（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):
        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 exports。

将项目写入 MongoDB

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

此示例的重点是展示如何获取爬虫以及如何正确清理资源。

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，但我们的爬虫返回了多个具有相同 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 Pipeline 组件

要激活 Item Pipeline 组件，您必须将其类添加到 ITEM_PIPELINES 设置中，如以下示例所示：

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

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

Scrapy shell

Feed exports (数据导出)