Scrapy技术问题：自定义Item Pipeline后Feed Exports无数据输出？

相关代码 #

Spider代码 #

class ExampleSpider(scrapy.Spider):
    name = 'example'

    custom_settings = {
        'ITEM_PIPELINES': {'img_clear.pipelines.DuplicatesPipeline': 100,},
        'FEEDS': {
            'feeds/example/tags.csv': {
                'format': 'csv',
                'fields': ["tag_id", "url", "title"],
                'item_export_kwargs': {
                    'include_headers_line': False,
                },
                'item_classes': [ExampleTagItem],
                'overwrite': False
            },
            'feeds/example/galleries.csv': {
                'format': 'csv',
                'fields': ["id", "url", "tag_ids"],
                'item_export_kwargs': {
                    'include_headers_line': False,
                },
                'item_classes': [ExampleGalleryItem],
                'overwrite': False,
            }
        }
    }

去重Pipeline代码 #

class DuplicatesPipeline():
    def open_spider(self, spider):
        if spider.name == "example":
            with open("feeds/example/galleries.csv", "r") as rf:
                csv = rf.readlines()
            self.ids_seen = set([str(line.split(",")[0]) for line in csv])
            
            with open("feeds/example/tags.csv", "r") as rf:
                tags_csv = rf.readlines()
            self.tag_ids_seen = set([str(line.split(",")[0]) for line in tags_csv])

    def process_item(self, item, spider):
        if isinstance(item, ExampleTagItem):
            self.process_example_tag_item(item, spider)    
        elif isinstance(item, ExampleGalleryItem):
            self.process_example_gallery_item(item, spider)

    def process_example_tag_item(self, item, spider):
        adapter = ItemAdapter(item)
        if adapter['tag_id'] in self.tag_ids_seen:
            raise DropItem(f"Duplicate item found: {item!r}")
        else:
            self.tag_ids_seen.add(adapter['tag_id'])
            return item

    def process_example_gallery_item(self, item, spider):
        adapter = ItemAdapter(item)
        if adapter['id'] in self.ids_seen:
            raise DropItem(f"Duplicate item found: {item!r}")
        else:
            self.ids_seen.add(adapter['id'])
            return item

现象与排查信息 #

• 启用Pipeline后，日志显示已丢弃重复项，也能抓取到有效项，但Feed文件始终无数据写入；
• 注释掉ITEM_PIPELINES配置后，数据可正常写入CSV；
• 启用Pipeline时执行scrapy crawl example -o test.csv，生成的CSV为空；
• 在Pipeline返回Item前打印内容，显示正常。

核心原因 #

DuplicatesPipeline的process_item方法没有返回处理后的Item。Scrapy的Pipeline是链式执行的，每个process_item必须返回Item对象（或抛出DropItem），否则后续环节（包括Feed Exports扩展）接收不到数据。你的代码只是调用了子处理方法，但没把子方法的返回值传递下去，相当于默认返回None，导致Feed扩展拿不到有效数据。

修复代码 #

修改process_item方法，添加返回逻辑：

def process_item(self, item, spider):
    if isinstance(item, ExampleTagItem):
        return self.process_example_tag_item(item, spider)    
    elif isinstance(item, ExampleGalleryItem):
        return self.process_example_gallery_item(item, spider)
    # 兜底返回，避免漏处理的情况
    return item

验证说明 #

修改后，有效Item会被正常返回给后续环节，Feed Exports就能接收到数据并写入文件；重复项依然会被DropItem抛出，日志也会正常显示丢弃信息。

内容的提问来源于stack exchange，提问作者Pa7rickStar