Scrapy Media Pipeline 完全指南：文件与图片的批量下载、存储与自定义

本文基于 Scrapy 官方文档 media-pipeline 及配套源码，系统讲解 Scrapy 媒体管道（Media Pipeline）的完整用法：从启用 FilesPipeline / ImagesPipeline、配置本地/S3/GCS/FTP 存储，到自定义文件命名、缩略图生成、尺寸过滤，直至基于 MediaPipeline 基类扩展自己的媒体管道。读完后你将掌握：如何把 item 中的媒体 URL 批量落地到各种存储、如何控制文件名与过期重下策略、以及源码层面“同一 URL 只下载一次”的去重机制是如何实现的。

一、什么是媒体管道：FilesPipeline 与 ImagesPipeline

Scrapy 提供可复用的 item pipeline 用于下载与 item 关联的文件（例如抓取商品时同时把商品图片下载到本地）。这些管道共享同一套功能与结构，统称为“媒体管道”，实际项目中通常二选一使用 Files Pipeline 或 Images Pipeline，核心实现位于 scrapy/pipelines。

两类管道共同实现的能力：

• 避免重复下载：对最近已下载过的媒体不再重复下载（由 FILES_EXPIRES / IMAGES_EXPIRES 控制保留期）；
• 多种存储后端：可把媒体存到文件系统目录、FTP 服务器、Amazon S3 桶、Google Cloud Storage 桶；
• 内部下载队列去重：管道维护一个“正在调度下载”的媒体 URL 队列，当多个 item 引用同一媒体时，只下载一次并把这些响应都连接到该队列上（从源码看，这一逻辑位于 MediaPipeline.SpiderInfo，其 downloading / downloaded / waiting 三个结构分别记录下载中、已完成的指纹缓存和等待者）。

Images Pipeline 在此基础上增加了图片专属处理：

• 把所有下载的图片统一转换为 JPG 格式、RGB 色彩模式；
• 自动生成缩略图（thumbnails）；
• 校验图片宽/高是否满足最小约束，过小的图片直接丢弃。

二、Files Pipeline 的典型工作流程

从 FilesPipeline 的文档描述与实际行为看，典型工作流分为四步：

1. 在 Spider 中抓取 item，把待下载文件 URL 放入 file_urls 字段；
2. item 从 Spider 返回后进入 item pipeline 链；
3. 当 item 到达 FilesPipeline 时，file_urls 中的 URL 会通过 标准 Scrapy downloader 下载——这意味着下载器中间件（downloader middlewares）生效，但 Spider 中间件不生效；item 在该管道阶段保持“锁定”状态，直到文件下载完成（或因故失败）；
4. 文件下载完成后，另一个字段 files 会被填充结果：一个字典列表，包含下载路径、原始抓取 URL（取自 file_urls）、文件 MD5 校验和、文件状态。files 中文件的顺序与原始 file_urls 保持一致；若某个文件下载失败，会记录一条错误日志，且该文件不会出现在 files 字段中。

从源码看，第 3 步的“锁定”对应 MediaPipeline.process_item：它先通过 get_media_requests() 拿到全部请求，然后并发处理（asyncio 可用时用 asyncio.gather，否则用 Twisted DeferredList），最后把结果交给 item_completed()。第 4 步的结果写入对应 FilesPipeline.item_completed：

ItemAdapter(item)[self.files_result_field] = [x for ok, x in results if ok]

只有 ok 为 True 的结果会写入结果字段，这解释了“失败文件不会出现”的行为。

三、Images Pipeline：不同的字段名与图片专属能力

使用 ImagesPipeline 与 FilesPipeline 几乎相同，只是默认字段名不同：图片 URL 放在 image_urls 字段，下载结果写入 images 字段。其价值在于可以配置缩略图生成、按尺寸过滤图片等额外能力。

需要注意：Images Pipeline 依赖 images 附加依赖（即 Pillow）。在 pyproject.toml 中该 extra 定义为 images = ["Pillow>=8.3.2"]；从 ImagesPipeline.init 看，若未安装 Pillow 会直接抛出 NotConfigured。

四、启用媒体管道

启用分两步，缺一不可。

第一步：加入 ITEM_PIPELINES 设置。

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

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

两者可以同时启用。

第二步：配置目标存储。 若未将存储设置配置为合法值，管道即使加入 ITEM_PIPELINES 也保持禁用状态。

# Files Pipeline
FILES_STORE = "/path/to/valid/dir"

# Images Pipeline
IMAGES_STORE = "/path/to/valid/dir"

从源码验证这一点：default_settings.py 中 FILES_STORE = None、IMAGES_STORE = None；而 FilesPipeline.init 在 store_uri 为空时会抛出 NotConfigured，错误信息明确指出 “FILES_STORE/IMAGES_STORE setting must be set to a valid path (not empty)”。

五、文件命名（File Naming）

5.1 默认命名：URL 的 SHA-1 哈希

默认情况下，文件名使用 URL 的 SHA-1 哈希值。例如：

http://www.example.com/image.jpg
SHA-1 哈希:
3afec3b4765f8f0a07b78f98c07b83f013567a0a
实际存储文件名:
3afec3b4765f8f0a07b78f0a07b78f98c07b83f013567a0a.jpg

对应源码是 FilesPipeline.file_path：先用 hashlib.sha1(request.url) 生成哈希，再从 URL path 中提取扩展名；若 path 没有扩展名则回退到原始 URL，仍不行则尝试用 mimetypes 猜测 MIME 类型再推扩展名，最后兜底为空字符串，最终返回 full/<哈希><扩展名>。

注意 ImagesPipeline 重写了该方法（见 images.py），固定返回 full/<哈希>.jpg——因为图片会被统一转码为 JPEG，所以扩展名恒定。

5.2 自定义命名：覆写 file_path

如果希望使用不同的文件命名规则（例如把分类元信息编进文件名），覆写管道类的 file_path 方法即可。例如对如下图片 URL：

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

生成带压缩哈希与视角 front 的文件名 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

警告：如果自定义命名规则依赖的元数据在多次抓取之间可能变化，会导致以新文件名意外地重复下载已有媒体。例如规则里用了商品标题，而站点两次抓取之间修改了标题，Scrapy 就会以更新后的文件名重新下载同一媒体。

5.3 基于 Response 命名

file_path 同时接收 response 参数，因此可以基于响应数据命名。典型用途：对不以文件名结尾的 URL，从 Content-Type 头推导扩展名：

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 "")

这需要把 FILES_EXPIRES 设为 0。原理（可从 FilesPipeline.media_to_download 与 _onsuccess 印证）：下载前 Scrapy 会以 response=None 先调用一次 file_path，再用返回路径去 stat 已有文件的修改时间；依赖于 response 的路径永远无法与这次检查匹配，因此依赖 response 命名的管道必须用 FILES_EXPIRES = 0 关闭过期检查——代价是每次运行都会重新下载全部文件。

六、支持的存储后端（Supported Storage）

FilesPipeline.STORE_SCHEMES 定义了 URI scheme 到存储类的映射：""/file → FSFilesStore，s3 → S3FilesStore，gs → GCSFilesStore，ftp → FTPFilesStore。FILES_STORE / IMAGES_STORE 按前缀自动选择后端。

6.1 文件系统存储

文件保存到如下路径：

<IMAGES_STORE>/full/<FILE_NAME>

• <IMAGES_STORE>：IMAGES_STORE 设置定义的目录；
• full：把完整图片与缩略图区分子目录（见缩略图一节）；
• <FILE_NAME>：文件名规则见上一节。

从源码看，FSFilesStore 初始化时会 mkdir 根目录，persist_file 时按需创建父目录（按 spider 缓存已创建目录避免重复系统调用），stat_file 则读取文件 mtime 并流式计算 MD5（每次读 8096 字节，避免把整个文件读进内存，见 _md5sum）。

6.2 FTP 服务器存储

FILES_STORE 和 IMAGES_STORE 可以直接指向 FTP 服务器，Scrapy 自动上传文件。合法格式：

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

若 URL 中未提供用户名/密码，则取自 FTP_USER 与 FTP_PASSWORD 设置（default_settings.py 中默认分别为 anonymous 与 guest）。

FTP 支持主动/被动两种连接模式，Scrapy 默认被动模式；要改用主动模式，设置：

FEED_STORAGE_FTP_ACTIVE = True

从源码看，该值由 FilesPipeline._update_stores 读入 FTPFilesStore.USE_ACTIVE_MODE，上传（persist_file）与状态检查（MDTM 取修改时间 + RETR 流式算 MD5）都遵循该模式。

6.3 Amazon S3 存储

需安装 s3 附加依赖（pyproject.toml 中为 boto3>=1.20.0；从源码看实际调用的是 botocore 客户端，缺库时 S3FilesStore 抛 NotConfigured）。FILES_STORE / IMAGES_STORE 可表示 S3 桶：

IMAGES_STORE = "s3://bucket/images"

可用 FILES_STORE_S3_ACL / IMAGES_STORE_S3_ACL 修改存储文件的 ACL 策略，默认 private（见 default_settings.py 中 FILES_STORE_S3_ACL = "private"）。公开访问用 public-read：

IMAGES_STORE_S3_ACL = "public-read"

S3 兼容存储：Minio 等自托管 S3 兼容服务同样适用，只需配置 endpoint：

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

# 自托管场景下可能还需要关闭 SSL / 关闭证书校验
AWS_USE_SSL = False  # 或 True（默认 None）
AWS_VERIFY = False   # 或 True（默认 None）

若要复用一个文件并行检查/上传过程中的连接，可相应调大 AWS_MAX_POOL_CONNECTIONS。这些设置由 _update_stores 统一注入 S3FilesStore，最终体现在 botocore 客户端创建 的 endpoint_url、use_ssl、verify、max_pool_connections 参数上。

6.4 Google Cloud Storage 存储

需安装 gcs 附加依赖（google-cloud-storage>=1.29.0）。FILES_STORE / IMAGES_STORE 可表示 GCS 桶：

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

认证使用 Google Cloud 标准的 Application Default Credentials。ACL 由 FILES_STORE_GCS_ACL / IMAGES_STORE_GCS_ACL 定义，默认空字符串（表示桶的默认对象 ACL 生效）；公开访问用 publicRead：

IMAGES_STORE_GCS_ACL = "publicRead"

从源码看，GCSFilesStore 初始化时会校验 storage.objects.get / storage.objects.create 两个 IAM 权限：缺少前者只告警（意味着无法做“是否过期”检查、每次都会重新下载），缺少后者则报错（无法保存文件）；上传时统一带 Cache-Control: max-age=172800 头并写入 predefined_acl（persist_file）。

七、完整使用示例

7.1 item 字段与结果字段

启用管道（见第四节）后，只要 Spider 返回的 item 带有 URL 字段（Files 管道为 file_urls，Images 管道为 image_urls），管道就会把结果写入对应字段（files 或 images）。

若使用预定义字段的 item 类型，必须同时定义 URL 字段与结果字段。以 dataclass 为例：

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)

7.2 自定义字段名

如需更换 URL 键或结果键名，可用相应设置（默认值 file_urls/files、image_urls/images 见 FilesPipeline 与 ImagesPipeline 的类属性）：

# Files Pipeline
FILES_URLS_FIELD = "field_name_for_your_files_urls"
FILES_RESULT_FIELD = "field_name_for_your_processed_files"

# Images Pipeline
IMAGES_URLS_FIELD = "field_name_for_your_images_urls"
IMAGES_RESULT_FIELD = "field_name_for_your_processed_images"

7.3 按管道类名隔离设置

如果你有多个继承自 ImagesPipeline（或 FilesPipeline）的管道，希望它们各自使用不同设置，可以把管道类名（大写）作为前缀。例如类名 MyPipeline，则：

MYPIPELINE_IMAGES_URLS_FIELD = "..."
MYPIPELINE_FILES_EXPIRES = 180

从源码看，这一机制由 MediaPipeline._key_for_pipe 实现：它先尝试读取 <类名大写>_<设置名>，未配置时回退到无前缀的通用设置；且只有当“直接父类恰好是 FilesPipeline / ImagesPipeline”时才生效，因此同名类名的不同子管道可以互相隔离设置。

八、进阶功能

8.1 文件过期策略（FILES_EXPIRES / IMAGES_EXPIRES）

图片/文件管道会避免重复下载近期已下载的文件。保留期（单位：天）由 FILES_EXPIRES 或 IMAGES_EXPIRES 控制，两者默认均为 90 天（对应 FilesPipeline.EXPIRES 与 ImagesPipeline.EXPIRES 的类属性）：

# 文件 120 天过期
FILES_EXPIRES = 120

# 图片 30 天过期
IMAGES_EXPIRES = 30

判断逻辑在 _onsuccess：用文件“最后修改时间”计算年龄天数，age_days > self.expires 才判定过期、触发重新下载，否则返回 status: "uptodate" 的 FileInfo 且不重新下载；IMAGES_STORE 为空目录时该检查失败则强制下载。

8.2 缩略图生成（IMAGES_THUMBS）

Images Pipeline 可为下载的图片自动生成缩略图。将 IMAGES_THUMBS 设为字典，键为缩略图名、值为尺寸：

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

启用后，每种尺寸都会生成一个缩略图，路径格式：

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

其中 <size_name> 是 IMAGES_THUMBS 的键，<image_id> 是图片 URL 的 SHA-1 哈希。完整示例：

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

第一个是原图。源码实现见 ImagesPipeline.get_images：先产出全尺寸图片，再遍历 self.thumbs 对每个 thumb_id 调用 thumb_path()（默认 thumbs/<size name>/<哈希>.jpg，见 thumb_path）并 convert_image 缩放。缩放的细节值得注意：RGBA 的 PNG/WEBP 及调色板模式（P 模式）图片会先合成到白色背景再转 RGB；缩放使用 LANCZOS 重采样（Pillow < 9.1.0 时回退到 ANTIALIAS）；输出统一保存为 JPEG。

8.3 过滤过小的图片（IMAGES_MIN_HEIGHT / IMAGES_MIN_WIDTH）

Images Pipeline 可以丢弃过小的图片，最小尺寸由 IMAGES_MIN_HEIGHT 和 IMAGES_MIN_WIDTH 指定（默认 0，即不过滤）：

IMAGES_MIN_HEIGHT = 110
IMAGES_MIN_WIDTH = 110

注意：尺寸约束不影响缩略图生成。可以只设其中一个约束，也可以两个都设——都设时只有同时满足两个最小值的图片才会保存。以上例为例，(105×105)、(105×200)、(200×105) 都会被丢弃，因为至少有一个维度小于约束。

从源码看，检查发生在 get_images，且先做 EXIF 方向修正（ImageOps.exif_transpose）再测量宽高，避免手机拍摄图片因 EXIF 旋转标记导致的宽高误判；不满足时抛出 ImageException("Image too small ...")。

8.4 允许重定向（MEDIA_ALLOW_REDIRECTS）

媒体管道默认忽略重定向：请求媒体文件 URL 时若发生 HTTP 重定向，该次媒体下载即视为失败。要处理媒体重定向：

MEDIA_ALLOW_REDIRECTS = True

从源码看，MediaPipeline.init 读取该设置后，在 _modify_media_request 中给媒体请求注入 handle_httpstatus_list（排除 300–399，即正常跟随 3xx）或 handle_httpstatus_all（把 3xx 当普通状态码交给管道、从而判定下载失败）。

九、扩展媒体管道：可覆写的方法

基类 MediaPipeline 定义了完整生命周期钩子：media_to_download（下载前检查，可短路返回 FileInfo）、get_media_requests、media_downloaded、media_failed、item_completed、file_path。FilesPipeline 与 ImagesPipeline 在其上提供了文档化的覆写点。

9.1 FilesPipeline 可覆写方法

file_path(self, request, response=None, info=None, *, item=None) 每个已下载 item 调用一次，返回该文件（来自指定 response）的存储路径。除 response 外，还接收原始 request、info（MediaPipeline.SpiderInfo）与 item。例如若文件 URL 以常规路径结尾（如 https://example.com/a/b/c/foo.png），可把文件按原始文件名存入 files 目录：

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 决定路径（见 5.3 节）。默认实现返回 full/<request URL 哈希>.<扩展名>。

get_media_requests(item, info) 管道通过该方法从 item 取得待下载请求，可覆写以改变返回内容：

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)

这些请求由管道处理，下载完成后结果以两元素元组列表传给 item_completed：(success, file_info_or_error)，其中 success 为布尔；file_info_or_error 成功时是包含以下键的字典，失败时是 Twisted 的 Failure：

• url — 文件来源 URL（即 get_media_requests 返回的请求 URL）；
• path — 文件保存路径（相对 FILES_STORE）；
• checksum — 文件内容的 MD5 哈希；
• status — 文件状态指示，取值为：
  • downloaded：文件已下载；
  • uptodate：文件未下载，因为按过期策略它近期已下载过；
  • cached：文件取自缓存（响应带 "cached" 标志，例如来自 HttpCacheMiddleware）。

item_completed 收到的元组列表保证与 get_media_requests 返回请求的顺序一致（从 process_item 按 requests 顺序收集 results 可以印证）。典型 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) 当单个 item 的全部文件请求完成（无论成功还是失败）时调用。该方法必须返回（或丢弃）要传递给后续管道阶段的 item。示例：把下载路径存到 file_paths 字段，若没有任何文件则丢弃 item：

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 直接返回 item（见 FilesPipeline.item_completed）；注意基类 MediaPipeline.item_completed 还会按 LOG_FAILED_RESULTS 标志为每个失败结果记录错误日志。

9.2 ImagesPipeline 额外可覆写方法

ImagesPipeline 是 FilesPipeline 的扩展（见 scrapy/pipelines/images.py），自定义字段名之外还增加了图片行为：

• file_path(...)：与 Files Pipeline 相同，默认返回 full/<request URL 哈希>.jpg；
• thumb_path(self, request, thumb_id, response=None, info=None, *, item=None)：每个 item 的每个 IMAGES_THUMBS 项各调用一次，返回缩略图存储路径，可基于 item 定制；默认返回 thumbs/<size name>/<request URL 哈希>.<扩展名>；
• get_media_requests(item, info)：与 Files Pipeline 相同，但使用图片 URL 字段名；
• item_completed(results, item, info)：与 Files Pipeline 相同，但使用图片结果字段名存储下载结果。

此外还有一个文档未单列但源码可见的扩展点：get_images 以生成器方式产出 (path, image, buf) 三元组（全图 + 各缩略图），覆写它即可插入自己的图片处理逻辑。

十、完整示例：自定义 Images Pipeline

一个把上述方法串起来的完整自定义图片管道：

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：

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

基于内容过滤的图片管道

覆写 get_images() 可以用分类器（如 TensorFlow 模型）过滤图片，只需把分类逻辑实现到 is_valid_image() 中：

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

抛出 ImageException（定义于，是 FileException 的子类）后，该图片会走 media_failed 路径被记为失败，从而不进入结果字段。

十一、源码速查与统计

• 管道基类与去重队列：scrapy/pipelines/media.py（MediaPipeline、FileInfo、SpiderInfo、_key_for_pipe）；
• Files Pipeline 与四种存储：scrapy/pipelines/files.py（FSFilesStore / S3FilesStore / GCSFilesStore / FTPFilesStore）；
• Images Pipeline 与图片处理：scrapy/pipelines/images.py（get_images、convert_image、thumb_path）;
• 默认配置值：scrapy/settings/default_settings.py（FILES_STORE/IMAGES_STORE 默认 None、FILES_STORE_S3_ACL/IMAGES_STORE_S3_ACL 默认 private、FTP_USER/FTP_PASSWORD、FEED_STORAGE_FTP_ACTIVE 默认 False）；
• 附加依赖声明：pyproject.toml（images、s3、gcs extras）；
• 相关测试：tests/test_pipeline_files.py、tests/test_pipeline_images.py、tests/test_pipeline_media.py。

运行过程中还可通过统计项核对行为：inc_stats 会累加 file_count 与 file_status_count/<status>（见 FilesPipeline.inc_stats），在 stats 中即可看到 downloaded / uptodate / cached 各状态的计数。

适用前提小结：Images Pipeline 需要安装 Pillow（images extra）；S3/GCS 存储分别需要 s3、gcs extra；FILES_EXPIRES/IMAGES_EXPIRES 默认 90 天；基于 response 的文件命名必须搭配 FILES_EXPIRES = 0；媒体重定向默认视为失败，需显式开启 MEDIA_ALLOW_REDIRECTS。以上取值与行为均可在当前仓库的源码与默认设置中直接验证。