如何以Scrapy风格实现Item的跨站数据补全与数据库更新？

我使用Python Scrapy框架实现了如下自定义爬虫：

class MyCustomSpider(scrapy.Spider):
    def __init__(self, name=None, **kwargs):
        super().__init__(name, **kwargs)
        self.days = getattr(self, 'days', None)

    def start_requests(self):
        start_url = f'https://some.url?days={self.days}&format=json'
        yield scrapy.Request(url=start_url, callback=self.parse)

    def parse(self, response):
        json_data = response.json() if response and response.status == 200 else None
        if json_data:
            for entry in json_data['entries']:
                yield self.parse_json_entry(entry)
        
            if 'next' in json_data and json_data['next'] != "":
                yield response.follow(f"https://some.url?days={self.days}&time={self.time}&format=json", self.parse)

    def parse_json_entry(self, entry):
        ...
        item = loader.load_item()
        return item

我通过Pipeline将解析后的Item Upsert到数据库中，现需添加以下功能：

1. 在Upsert前从数据库读取Item的当前状态；
2. 若Item不存在于数据库，或存在但部分字段为空，则基于Item内容生成另一网站的URL，通过Scrapy发起请求（利用其缓存等特性），爬取内容补全Item后再存入数据库；
3. 若Item已存在且字段齐全，则仅更新Item的状态。

目前我在Pipeline中直接调用外部网站进行补全，但未借助Scrapy框架。请问如何以Scrapy风格实现上述第2点功能？是通过Pipeline实现，还是应将数据库检查、回调逻辑全部放在Spider中？

解决方案 #

核心思路：Spider控流程，Pipeline管持久化 #

Scrapy的设计逻辑是Spider负责请求调度和数据解析的流程控制，Pipeline专注数据的清洗、验证和持久化。不建议在Pipeline里发起新请求（会绕开Scrapy的调度、缓存、并发控制体系），正确做法是把数据库检查和补全请求的逻辑放在Spider中，让整个流程符合Scrapy的异步请求流。

具体实现步骤 #

1. 给Spider添加数据库查询能力

在Spider的__init__方法中初始化数据库连接，实现查询Item状态的方法：

class MyCustomSpider(scrapy.Spider):
    name = "my_custom_spider"

    def __init__(self, name=None, **kwargs):
        super().__init__(name, **kwargs)
        self.days = getattr(self, 'days', None)
        # 初始化数据库连接（示例用SQLAlchemy，也可使用原生DBAPI）
        self.db_engine = create_engine(settings.get('DATABASE_URL'))

    def check_item_status(self, item):
        """查询数据库中Item的当前状态，返回是否存在、字段是否齐全"""
        with self.db_engine.connect() as conn:
            result = conn.execute(
                text("SELECT * FROM items WHERE id = :item_id"),
                {"item_id": item['id']}
            ).fetchone()
            if not result:
                return {"exists": False, "fields_complete": False}
            # 根据你的Item字段调整检查逻辑
            fields_to_check = ['field1', 'field2', 'field3']
            all_complete = all(result[field] is not None for field in fields_to_check)
            return {"exists": True, "fields_complete": all_complete}

2. 修改parse_json_entry，加入状态检查和补全请求

原本直接返回Item，现在改为根据数据库状态决定流程走向：

def parse_json_entry(self, entry):
        # 解析基础Item字段
        loader = ItemLoader(item=MyItem(), response=None)
        loader.add_value('id', entry['id'])
        loader.add_value('basic_field', entry['basic_field'])
        # ... 其他基础字段解析
        item = loader.load_item()

        # 检查数据库中Item状态
        status = self.check_item_status(item)
        if not status['exists'] or not status['fields_complete']:
            # 生成补全URL，将未补全的Item存入meta传递给回调
            completion_url = f"https://another.site/detail/{item['id']}"
            yield scrapy.Request(
                url=completion_url,
                callback=self.parse_completion_data,
                meta={'item': item}
            )
        else:
            # 字段齐全，标记状态更新
            item['status'] = 'updated'
            yield item

3. 实现补全请求的回调函数

解析补全页面内容，填充Item缺失字段后返回给Pipeline：

def parse_completion_data(self, response):
        # 从meta中取出未补全的Item
        item = response.meta['item']
        # 解析补全页面内容
        loader = ItemLoader(item=item, response=response)
        loader.add_xpath('field1', '//div[@class="field1"]/text()')
        loader.add_css('field2', '.field2::text')
        # ... 其他需要补全的字段
        # 标记补全状态
        item['status'] = 'completed'
        yield loader.load_item()

4. Pipeline专注Upsert逻辑

Pipeline仅负责接收最终Item，执行Upsert操作：

class MyPipeline:
    def __init__(self, db_url):
        self.db_url = db_url
        self.engine = create_engine(db_url)

    @classmethod
    def from_crawler(cls, crawler):
        return cls(
            db_url=crawler.settings.get('DATABASE_URL')
        )

    def process_item(self, item, spider):
        with self.engine.connect() as conn:
            # 执行Upsert逻辑，根据id判断插入或更新
            conn.execute(
                text("""
                    INSERT INTO items (id, basic_field, field1, field2, status)
                    VALUES (:id, :basic_field, :field1, :field2, :status)
                    ON CONFLICT (id) DO UPDATE SET
                        basic_field = EXCLUDED.basic_field,
                        field1 = COALESCE(EXCLUDED.field1, items.field1),
                        field2 = COALESCE(EXCLUDED.field2, items.field2),
                        status = EXCLUDED.status
                """),
                item
            )
            conn.commit()
        return item

为什么不建议在Pipeline中处理请求？ #

• 破坏异步调度：Pipeline是同步处理Item的，直接发起请求会阻塞流程，无法利用Scrapy的并发、缓存、重试机制。
• 违背单一职责：请求逻辑和数据持久化逻辑耦合，后续修改补全规则时需改动Pipeline，维护成本高。

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