使用文件管道

FilesPipeline 的典型工作流程如下：

1. 在 Spider 中抓取条目，将所需文件的 URL 放入 file_urls 字段。

2. Spider 返回条目，条目进入 item pipeline。

3. 条目到达 FilesPipeline 时，file_urls 中的 URL 会由 Scrapy 标准下载器下载，因此会经过下载器中间件，但不会经过 Spider 中间件。条目将一直停留在该管道阶段，直到文件下载完成或因某种原因失败。

4. 下载后，结果填入另一个字段 files。它是字典列表，每个字典包含下载路径、原始抓取 URL（来自 file_urls）、文件校验和与状态等信息。files 中的文件顺序与原 file_urls 相同。如果某个文件下载失败，会记录错误，该文件也不会出现在 files 中。

使用图片管道

ImagesPipeline 的用法与 FilesPipeline 很相似，但默认字段名不同：图片 URL 放在 image_urls 中，下载后图片的信息填入 images 字段。^[3]

用 ImagesPipeline 处理图片的优势是可以配置额外功能，例如生成缩略图、按尺寸过滤图片。^[3]

启用媒体管道

要启用媒体管道，首先将其加入项目的 ITEM_PIPELINES 设置。^[4]

图片管道配置如下：

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

文件管道配置如下：

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

随后，必须为目标存储设置配置有效值，用来存放下载的图片。否则，即使已经加入 ITEM_PIPELINES，管道仍然处于禁用状态。^[4]

文件管道需要设置 FILES_STORE：^[5]

FILES_STORE = "/path/to/valid/dir"

图片管道需要设置 IMAGES_STORE：^[6]

IMAGES_STORE = "/path/to/valid/dir"

文件命名

http://www.example.com/image.jpg

3afec3b4765f8f0a07b78f98c07b83f013567a0a

3afec3b4765f8f0a07b78f98c07b83f013567a0a.jpg

http://www.example.com/product/images/large/front/0000000004166

00b08510e4_front.jpg

import hashlib


def file_path(self, request, response=None, info=None, *, item=None):
    image_url_hash = hashlib.shake_256(request.url.encode()).hexdigest(5)
    image_perspective = request.url.split("/")[-2]
    image_filename = f"{image_url_hash}_{image_perspective}.jpg"

    return image_filename

import mimetypes

from scrapy.pipelines.files import FilesPipeline


class ContentTypeFilesPipeline(FilesPipeline):
    def file_path(self, request, response=None, info=None, *, item=None):
        path = super().file_path(request, response, info, item=item)
        if response is None:
            return path
        content_type = response.headers["Content-Type"].decode()
        return path + (mimetypes.guess_extension(content_type) or "")

from scrapy import Request, Spider
from scrapy.pipelines.files import FilesPipeline


class MySpider(Spider):
    def parse(self, response):
        yield {
            "file_urls": response.css("a.file::attr(href)").getall(),
            "referrer": response.url,
        }


class MyFilesPipeline(FilesPipeline):
    def get_media_requests(self, item, info):
        for file_url in item["file_urls"]:
            yield Request(file_url, headers={"Referer": item["referrer"]})

支持的存储方式

<IMAGES_STORE>/full/<FILE_NAME>

ftp://username:password@address:port/path
ftp://address:port/path

IMAGES_STORE = "s3://bucket/images"

IMAGES_STORE_S3_ACL = "public-read"

AWS_ENDPOINT_URL = "http://minio.example.com:9000"

AWS_USE_SSL = False  # or True (None by default)
AWS_VERIFY = False  # or True (None by default)

IMAGES_STORE = "gs://bucket/images/"
GCS_PROJECT_ID = "project_id"

IMAGES_STORE_GCS_ACL = "publicRead"

使用示例

使用媒体管道前，先启用它。^[30]

随后，Spider 返回的条目只要带有 URL 字段——文件管道为 file_urls，图片管道为 image_urls——管道就会将结果写入相应的 files 或 images 字段。^[31]

如果条目类型要求预先定义字段，必须同时定义 URL 字段和结果字段。例如图片管道的条目必须包含 image_urls 与 images。下面以 dataclass 为例：^[32]

from dataclasses import dataclass, field


@dataclass
class MyItem:
    # ... other item fields ...
    image_urls: list[str] = field(default_factory=list)
    images: list[dict] = field(default_factory=list)

URL 键和结果键也可以使用其他字段名，通过覆盖设置即可实现。

文件管道使用 FILES_URLS_FIELD 和／或 FILES_RESULT_FIELD：^[33][34]

FILES_URLS_FIELD = "field_name_for_your_files_urls"
FILES_RESULT_FIELD = "field_name_for_your_processed_files"

图片管道使用 IMAGES_URLS_FIELD 和／或 IMAGES_RESULT_FIELD：^[35][36]

IMAGES_URLS_FIELD = "field_name_for_your_images_urls"
IMAGES_RESULT_FIELD = "field_name_for_your_processed_images"

如果需要更复杂的功能并覆盖管道行为，参见“扩展媒体管道”。^[9]

如果有多个继承 ImagesPipeline 的管道，并希望各自采用不同设置，可以在设置键前加上管道类名的大写形式。例如管道名为 MyPipeline，要自定义 IMAGES_URLS_FIELD，就定义 MYPIPELINE_IMAGES_URLS_FIELD，该管道会使用此自定义值。^[3][35]

附加功能

# 120 days of delay for files expiration
FILES_EXPIRES = 120

# 30 days of delay for images expiration
IMAGES_EXPIRES = 30

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

<IMAGES_STORE>/thumbs/<size_name>/<image_id>.jpg

<IMAGES_STORE>/full/63bbfea82b8880ed33cdb762aa11fab722a90a24.jpg
<IMAGES_STORE>/thumbs/small/63bbfea82b8880ed33cdb762aa11fab722a90a24.jpg
<IMAGES_STORE>/thumbs/big/63bbfea82b8880ed33cdb762aa11fab722a90a24.jpg

IMAGES_MIN_HEIGHT = 110
IMAGES_MIN_WIDTH = 110

MEDIA_ALLOW_REDIRECTS = True

扩展媒体管道

自定义文件管道可以覆盖以下方法：

class scrapy.pipelines.files.FilesPipeline[source]

file_path(self, request, response=None, info=None, *, item=None)[source]

每个下载条目都会调用一次此方法。它返回指定 response 对应文件的下载路径。^[41]

除 response 外，此方法还接收原始 request、info 和 item。^[42][43]

可以覆盖此方法，自定义每个文件的下载路径。

例如，文件 URL 以普通路径结尾，如 https://example.com/a/b/c/foo.png，可以使用下面的方法，将所有文件按原文件名下载到 files 文件夹，例如 files/foo.png：

from pathlib import PurePosixPath
from scrapy.utils.httpobj import urlparse_cached

from scrapy.pipelines.files import FilesPipeline


class MyFilesPipeline(FilesPipeline):
    def file_path(self, request, response=None, info=None, *, item=None):
        return "files/" + PurePosixPath(urlparse_cached(request).path).name

同样，也可以根据 item 的某个属性确定路径，或者使用 response；后者参见“根据响应命名文件”。^[44]

默认情况下，file_path() 返回 full/<request URL hash>.<extension>。^[45]

get_media_requests(item, info)[source]

如工作流程所述，管道调用此方法，从条目中取得待下载文件的请求。可以覆盖此方法，改变返回的请求：

from itemadapter import ItemAdapter


def get_media_requests(self, item, info):
    adapter = ItemAdapter(item)
    for file_url in adapter["file_urls"]:
        yield scrapy.Request(file_url)

也可以用它设置请求头，参见“设置请求头”。^[46]

这些请求由管道处理，下载完成后，结果作为二元元组列表传给 item_completed()。每个元组为 (success, file_info_or_error)，其中：^[47]

• success 为布尔值：图片下载成功时为 True，因某种原因失败时为 False。

• 成功时 file_info_or_error 是包含以下键的字典；发生问题时则为 Failure。^[48]

  • url：下载文件的 URL，也就是 get_media_requests() 返回的请求 URL。^[49]

  • path：文件保存路径，相对于 FILES_STORE。^[5]

  • checksum：图片内容的 MD5 哈希。^[50]

  • status：文件状态。

    可能取以下值：

    • downloaded：文件已下载。

    • uptodate：依据过期策略，文件最近已经下载，因此本次未重新下载。

    • cached：文件来自缓存，响应带有 cached 标记，例如由 HttpCacheMiddleware 提供。^[51]

传入 item_completed() 的元组列表保证与 get_media_requests() 返回的请求顺序相同。^[47][49]

results 参数的典型值如下：

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

item_completed(results, item, info)[source]

单个条目的所有文件请求完成后，无论成功下载还是失败，都会调用 FilesPipeline.item_completed()。^[52]

item_completed() 必须返回传给后续条目管道阶段的输出。因此与其他管道一样，必须返回条目，或者丢弃条目。^[47]

下面的示例把 results 中的文件路径保存到条目的 file_paths 字段；如果没有任何文件，则丢弃该条目：^[47]

from itemadapter import ItemAdapter
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")
    adapter = ItemAdapter(item)
    adapter["file_paths"] = file_paths
    return item

默认情况下，item_completed() 返回条目。^[47]

自定义图片管道可以覆盖以下方法：

class scrapy.pipelines.images.ImagesPipeline[source]

ImagesPipeline 扩展了 FilesPipeline，定制字段名，并添加针对图片的行为。^[3]

file_path(self, request, response=None, info=None, *, item=None)[source]

每个下载条目都会调用一次此方法，返回指定 response 对应文件的下载路径。^[41]

除 response 外，此方法还接收原始 request、info 和 item。^[42][43]

可以覆盖此方法来自定义每个文件的下载路径。

例如，文件 URL 以普通路径结尾，如 https://example.com/a/b/c/foo.png，可以用下面的方法按原文件名下载到 files 文件夹，例如 files/foo.png：

from pathlib import PurePosixPath
from scrapy.utils.httpobj import urlparse_cached

from scrapy.pipelines.images import ImagesPipeline


class MyImagesPipeline(ImagesPipeline):
    def file_path(self, request, response=None, info=None, *, item=None):
        return "files/" + PurePosixPath(urlparse_cached(request).path).name

同样，可以根据 item 的属性或 response 确定路径；后者见“根据响应命名文件”。^[44]

默认 file_path() 返回 full/<request URL hash>.<extension>。^[53]

thumb_path(self, request, thumb_id, response=None, info=None, *, item=None)[source]

每个下载条目会针对 IMAGES_THUMBS 中的每一项调用此方法，返回指定 response 对应图片的缩略图下载路径。^[38][41]

除 response 外，此方法还接收原始 request、thumb_id、info 和 item。^[42][43]

可以覆盖此方法自定义每张图片的缩略图路径，也可以根据 item 的某个属性确定路径。

默认 thumb_path() 返回 thumbs/<size name>/<request URL hash>.<extension>。^[54]

get_media_requests(item, info)[source]

工作方式与 FilesPipeline.get_media_requests() 相同，但图片 URL 使用不同的字段名。

必须为每个图片 URL 返回一个 Request。

item_completed(results, item, info)[source]

单个条目的所有图片请求完成后，无论成功还是失败，都会调用 ImagesPipeline.item_completed()。^[55]

工作方式与 FilesPipeline.item_completed() 相同，但使用不同字段名来保存图片下载结果。

默认 item_completed() 返回条目。^[56]

访问下载失败时的响应

下载失败时，item_completed() 接收到一个 Failure，其 value 是 media_downloaded() 抛出的异常。对于非 2xx 响应，该异常是只用消息构造的 FileException，不保留响应引用。若要访问响应，例如把 status 记录到条目中，可以覆盖 media_downloaded()，将响应附加到自定义异常：^[48]

from scrapy.pipelines.files import FilesPipeline
from scrapy.pipelines.media import FileException


class ResponseFileException(FileException):
    def __init__(self, response):
        super().__init__(f"download-error ({response.status})")
        self.response = response


class MyFilesPipeline(FilesPipeline):
    async def media_downloaded(self, response, request, info, *, item=None):
        try:
            return await super().media_downloaded(response, request, info, item=item)
        except FileException:
            raise ResponseFileException(response)

    def item_completed(self, results, item, info):
        errors = [
            value.value.response.status
            for ok, value in results
            if not ok and value.check(ResponseFileException)
        ]
        if errors:
            item["download_errors"] = errors
        return super().item_completed(results, item, info)

自定义图片管道示例

下面是前述图片管道方法的完整示例：

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


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")
        adapter = ItemAdapter(item)
        adapter["image_paths"] = image_paths
        return item

要启用自定义媒体管道组件，必须将其类导入路径加入 ITEM_PIPELINES，例如：^[4]

ITEM_PIPELINES = {"myproject.pipelines.MyImagesPipeline": 300}

from scrapy.pipelines.images import ImagesPipeline, ImageException


class ImageClassifierPipeline(ImagesPipeline):
    def is_valid_image(self, image):
        raise NotImplementedError

    def get_images(self, response, request, info, *, item=None):
        for path, image, buf in super().get_images(response, request, info, item=item):
            if not self.is_valid_image(image):
                raise ImageException("Image does not match criteria")
            yield path, image, buf