如何使用Scrapy管道将爬取的图片直接保存到MySQL数据库

问题根源 #

默认scrapy.pipelines.images.ImagesPipeline处理后，item['['images']仅存储图片本地路径、校验码、URL等元数据，不包含图片二进制内容，因此无法直接用于数据库存储。 #

修改步骤 #

1. 新增自定义图片管道（替换默认管道） #

在pipelines.py中新增继承自ImagesPipeline的自定义类，直接抓取图片二进制内容，无需写入本地磁盘：

import scrapy
from scrapy.pipelines.images import ImagesPipeline
from scrapy.exceptions import DropItem
import io
from PIL import Image
import mysql.connector
from decouple import config

class CustomImagesPipeline(ImagesPipeline):
    def get_images(self, response, request, info, *, item=None):
        # 直接获取响应中的图片二进制内容
        image_bytes = response.body
        # 校验图片有效性
        try:
            Image.open(io.BytesIO(image_bytes))
        except Exception:
            raise DropItem(f"无效图片链接：{request.url}")
        # 绑定图片URL和二进制内容存入item
        if not item.get("image_bytes_list"):
            item["image_bytes_list"] = []
        item["image_bytes_list"].append({
            "url": request.url,
            "content": image_bytes
        })
        # 跳过默认的本地文件写入逻辑
        return []

    def item_completed(self, results, item, info):
        # 校验所有图片下载成功
        for ok, err in results:
            if not ok:
                raise DropItem(f"图片下载失败：{err.value}")
        return item

2. 修改items.py新增二进制存储字段 #

import scrapy

class MyItem(scrapy.Item):
    image_urls = scrapy.Field()
    image_bytes_list = scrapy.Field() # 存储图片URL和对应二进制

3. 修改settings.py调整管道配置 #

替换默认图片管道，调整执行优先级，无需配置本地存储路径IMAGES_STORE：

ITEM_PIPELINES = {
    # 替换为你的实际项目名，自定义图片管道先执行
    "你的项目名.pipelines.CustomImagesPipeline": 1,
    # MySQL存储管道后执行
    "你的项目名.pipelines.SavetoMySQL": 300
}

4. 修正SavetoMySQL存储逻辑 #

修复原代码中的表名不统一、缺少事务提交、未指定操作数据库的问题：

class SavetoMySQL:
    def __init__(self):
        self.user = config("DB_USER")
        self.password = config("DB_PASSWORD")
        self.host = config("DB_HOST")
        self.port = config("DB_PORT")
        self.db_name = config("DB_NAME") # 新增配置项，填你的数据库名称
    
    def connect_database(self):
        self.conn = mysql.connector.connect(
            host=self.host,
            user=self.user,
            password=self.password,
            port=self.port,
            database=self.db_name # 指定操作的数据库
        )
    def create_table(self):
        self.curr = self.conn.cursor()
        self.curr.execute("""CREATE TABLE IF NOT EXISTS myitem_db(
                                image_urls VARCHAR(300),
                                images LONGBLOB);
                            """)

    def store_db(self, item):
        # 表名和建表语句保持一致
        insert_stmt = "INSERT INTO myitem_db(image_urls, images) VALUES (%s,%s)"
        for img_info in item["image_bytes_list"]:
            data = (img_info["url"], img_info["content"])
            self.curr.execute(insert_stmt, data)
        # 提交事务，否则数据不会写入数据库
        self.conn.commit()

    def open_spider(self, spider):
        self.connect_database()
        self.create_table()

    def process_item(self, item, spider):
        self.store_db(item)
        return item
    
    def close_spider(self, spider):
        self.curr.close()
        self.conn.close()

依赖安装 #

需要安装Pillow库用于图片有效性校验：

pip install pillow

内容的提问来源于stack exchange，提问作者TAN YONG SHENG