Scrapy Pipeline详解 - 数据处理管道完整指南

class MyPipeline:
    def process_item(self, item, spider):
        # 处理Item，必须返回Item或抛出DropItem
        return item

    def open_spider(self, spider):
        # 爬虫启动时调用，用于初始化资源（如打开文件、连接数据库）
        pass

    def close_spider(self, spider):
        # 爬虫关闭时调用，用于释放资源（如关闭文件、断开连接）
        pass

    @classmethod
    def from_crawler(cls, crawler):
        # 类方法，从Crawler对象创建Pipeline实例
        # 可以访问settings、signals等配置
        return cls()

ITEM_PIPELINES = {
    'myproject.pipelines.FilterPipeline': 100,   # 第一个执行：数据过滤
    'myproject.pipelines.MongoPipeline': 200,    # 第二个执行：数据库存储
    'myproject.pipelines.JsonPipeline': 300,     # 第三个执行：JSON文件写入
}

import scrapy
from itemadapter import ItemAdapter
from scrapy.exceptions import DropItem

class DuplicatesPipeline:
    """去重Pipeline - 基于URL字段过滤重复Item"""
    def __init__(self):
        self.ids_seen = set()

    def process_item(self, item, spider):
        adapter = ItemAdapter(item)
        if adapter.get('url') in self.ids_seen:
            raise DropItem(f"重复项: {adapter.get('url')}")
        self.ids_seen.add(adapter.get('url'))
        return item

class CleanDataPipeline:
    """数据清洗Pipeline - 去除空白字符、验证必填字段"""
    def process_item(self, item, spider):
        adapter = ItemAdapter(item)

        # 去除所有字符串字段的空白字符
        for field in adapter.field_names():
            value = adapter.get(field)
            if isinstance(value, str):
                adapter[field] = value.strip()

        # 验证必填字段
        if not adapter.get('title'):
            raise DropItem("缺少title字段，丢弃该Item")

        return item

import json
from itemadapter import ItemAdapter

class JsonWriterPipeline:
    """JSON文件写入Pipeline - 将Item逐条写入JSON数组"""

    def open_spider(self, spider):
        self.file = open('items.json', 'w', encoding='utf-8')
        self.file.write('[\n')
        self.first = True

    def close_spider(self, spider):
        self.file.write('\n]')
        self.file.close()

    def process_item(self, item, spider):
        adapter = ItemAdapter(item)
        line = json.dumps(dict(adapter), ensure_ascii=False)
        if not self.first:
            self.file.write(',\n')
        self.file.write('  ' + line)
        self.first = False
        return item

import sqlite3
from itemadapter import ItemAdapter

class SQLitePipeline:
    """SQLite存储Pipeline - 将Item写入SQLite数据库"""

    def open_spider(self, spider):
        self.conn = sqlite3.connect('scrapy_data.db')
        self.cursor = self.conn.cursor()
        self.cursor.execute('''
            CREATE TABLE IF NOT EXISTS articles (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                title TEXT,
                author TEXT,
                url TEXT UNIQUE,
                content TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')
        self.conn.commit()

    def close_spider(self, spider):
        self.conn.close()

    def process_item(self, item, spider):
        adapter = ItemAdapter(item)
        self.cursor.execute('''
            INSERT OR IGNORE INTO articles (title, author, url, content)
            VALUES (?, ?, ?, ?)
        ''', (
            adapter.get('title'),
            adapter.get('author'),
            adapter.get('url'),
            adapter.get('content'),
        ))
        self.conn.commit()
        return item

from itemadapter import ItemAdapter

class MongoPipeline:
    """MongoDB存储Pipeline - 通过settings配置连接信息"""

    def __init__(self, mongo_uri, mongo_db, collection_name):
        self.mongo_uri = mongo_uri
        self.mongo_db = mongo_db
        self.collection_name = collection_name

    @classmethod
    def from_crawler(cls, crawler):
        return cls(
            mongo_uri=crawler.settings.get('MONGO_URI', 'mongodb://localhost:27017'),
            mongo_db=crawler.settings.get('MONGO_DB', 'scrapy_data'),
            collection_name=crawler.settings.get('MONGO_COLLECTION', 'items'),
        )

    def open_spider(self, spider):
        import pymongo
        self.client = pymongo.MongoClient(self.mongo_uri)
        self.db = self.client[self.mongo_db]
        self.collection = self.db[self.collection_name]

    def close_spider(self, spider):
        self.client.close()

    def process_item(self, item, spider):
        adapter = ItemAdapter(item)
        self.collection.insert_one(dict(adapter))
        return item

Spider yield Item
    -> Pipeline1 (优先级100) - 数据清洗：去除空白字符
    -> Pipeline2 (优先级200) - 数据验证：检查必填字段，不合格则DropItem
    -> Pipeline3 (优先级300) - 数据存储：写入数据库
    -> Pipeline4 (优先级400) - 数据导出：写入JSON/CSV文件