在Scrapy中如何处理API分页及增量爬取

原创

小白学大数据

发布于 2025-11-12 16:52:22

1410

一、理解挑战：为何要处理分页与增量爬取？1. API分页 API分页是一种将大量数据分割成多个较小、可管理块（即页面）的技术。常见的分页模式包括：页码分页：最直观的方式，通过 page 和 page_size 参数控制。游标分页：更现代、更稳定的方式，API返回一个指向下一组数据的 cursor 或 next_cursor 令牌，常用于实时流数据。偏移量分页：类似于页码分页，使用 offset 和 limit 参数。如果不能系统地处理分页，我们的爬虫将只能获取到第一页的数据，导致数据严重不完整。2. 增量爬取增量爬取指的是仅爬取自上次任务以来新增或发生变化的数据，而非每次都将目标数据全部重新抓取一遍。其核心价值在于：极大减少网络请求：提升爬取效率，降低带宽成本。减轻目标服务器负载：遵守良好的爬虫礼仪。近实时更新：对于监控类应用，可以快速感知数据变化。实现增量爬取的关键在于识别数据的“唯一性”和“变化性”，通常通过记录已爬取条目的ID、更新时间戳或哈希值来实现。二、实战演练：构建一个分页与增量爬取的Scrapy爬虫我们将以一个真实的示例来演示整个流程。假设我们需要从一个虚构的新闻网站API https://api.example-news.com/v1/articles 爬取文章列表。该API使用页码分页，并返回如下结构的JSON数据：单页响应示例：

我们的目标是：爬取所有分页的文章，并且每次运行时只抓取新发布的文章。步骤1：创建Scrapy项目与初始爬虫步骤2：定义数据模型（Item）在 items.py 中，定义我们要抓取的字段。

{
  "data": [
    {
      "id": 101,
      "title": "Python 3.12 发布，性能提升显著",
      "content": "文章内容...",
      "publish_time": "2023-10-01T10:00:00Z"
    },
    {
      "id": 100,
      "title": "Scrapy 2.8 新特性介绍",
      "content": "文章内容...",
      "publish_time": "2023-09-28T09:00:00Z"
    }
    // ... 更多文章
  ],
  "pagination": {
    "current_page": 1,
    "total_pages": 50,
    "total_items": 1000
  }
}

步骤3：核心实现——分页逻辑我们将分页逻辑放在爬虫的 parse 方法中。这里使用递归请求的方式处理分页。在 example_news_api.py 中：

import scrapy
import json
from news_crawler.items import NewsCrawlerItem

class ExampleNewsApiSpider(scrapy.Spider):
    name = 'example_news_api'
    allowed_domains = ['api.example-news.com']
    
    # 起始URL，第一页
    start_urls = ['https://api.example-news.com/v1/articles?page=1']
    
    def parse(self, response):
        # 解析API返回的JSON响应
        json_data = json.loads(response.text)
        articles = json_data.get('data', [])
        pagination = json_data.get('pagination', {})
        
        # 处理当前页的文章列表
        for article in articles:
            item = NewsCrawlerItem()
            item['id'] = article['id']
            item['title'] = article['title']
            item['content'] = article['content']
            item['publish_time'] = article['publish_time']
            
            # 在此处可以添加增量爬取判断（详见下一步）
            # if self.is_duplicate(item['id']):
            #     continue
            
            yield item
        
        # 处理分页：获取下一页
        current_page = pagination.get('current_page', 1)
        total_pages = pagination.get('total_pages', 1)
        
        if current_page < total_pages:
            next_page = current_page + 1
            next_page_url = f"https://api.example-news.com/v1/articles?page={next_page}"
            
            # 构造下一页的请求，并指定回调函数为self.parse
            yield scrapy.Request(
                url=next_page_url,
                callback=self.parse  # 递归调用自己处理下一页
            )

代码解释：爬虫从第一页开始。在 parse 方法中，首先解析JSON，提取 data 中的文章列表。遍历文章列表，生成 NewsCrawlerItem。检查分页信息，如果当前页不是最后一页，则构建下一页的URL并创建一个新的 Request 对象。这个新请求的回调函数仍然是 parse 自身，从而形成递归，直到处理完所有页面。步骤4：核心实现——增量爬取逻辑增量爬取的核心是“记忆”。我们需要一个持久化存储来记录已经处理过的文章ID。这里我们使用一个简单的文本文件（.txt 或 .json）来模拟，生产环境建议使用数据库（如SQLite, Redis, MongoDB）。a. 创建去重管理器在项目目录下创建一个新的文件 dupefilter.py：

import os
import json

class SimpleDupeFilter:
    """一个基于JSON文件的简单去重过滤器"""
    
    def __init__(self, file_path='./scraped_ids.json'):
        self.file_path = file_path
        self.scraped_ids = self._load_existing_ids()
    
    def _load_existing_ids(self):
        """从文件加载已爬取的ID集合"""
        if os.path.exists(self.file_path):
            with open(self.file_path, 'r', encoding='utf-8') as f:
                try:
                    return set(json.load(f))
                except json.JSONDecodeError:
                    return set()
        return set()
    
    def is_duplicate(self, item_id):
        """检查ID是否重复"""
        return item_id in self.scraped_ids
    
    def mark_as_scraped(self, item_id):
        """将ID标记为已爬取（内存中）"""
        self.scraped_ids.add(item_id)
    
    def save(self):
        """将已爬取的ID集合保存到文件"""
        with open(self.file_path, 'w', encoding='utf-8') as f:
            json.dump(list(self.scraped_ids), f, ensure_ascii=False)

b. 在爬虫中集成增量爬取修改 example_news_api.py：

import scrapy
import json
import base64
from news_crawler.items import NewsCrawlerItem
from news_crawler.dupefilter import SimpleDupeFilter  # 导入我们的去重器

class ExampleNewsApiSpider(scrapy.Spider):
    name = 'example_news_api'
    allowed_domains = ['api.example-news.com']
    start_urls = ['https://api.example-news.com/v1/articles?page=1']
    
    # 代理配置信息
    proxyHost = "www.16yun.cn"
    proxyPort = "5445"
    proxyUser = "16QMSOML"
    proxyPass = "280651"
    
    def __init__(self, *args, **kwargs):
        super(ExampleNewsApiSpider, self).__init__(*args, **kwargs)
        # 初始化去重过滤器
        self.dupefilter = SimpleDupeFilter()
        # 构建代理认证信息
        self.proxy_auth = self._get_proxy_auth()
    
    def _get_proxy_auth(self):
        """生成代理认证头信息"""
        proxy_auth = base64.b64encode(f"{self.proxyUser}:{self.proxyPass}".encode()).decode()
        return f"Basic {proxy_auth}"
    
    def start_requests(self):
        """重写start_requests方法，为初始请求添加代理"""
        for url in self.start_urls:
            yield scrapy.Request(
                url=url,
                callback=self.parse,
                meta={
                    'proxy': f"http://{self.proxyHost}:{self.proxyPort}",
                    'proxy_authorization': self.proxy_auth
                }
            )
    
    def parse(self, response):
        json_data = json.loads(response.text)
        articles = json_data.get('data', [])
        pagination = json_data.get('pagination', {})
        
        for article in articles:
            item_id = article['id']
            
            # !!! 增量爬取核心：检查是否重复 !!!
            if self.dupefilter.is_duplicate(item_id):
                self.logger.info(f"跳过重复文章 ID: {item_id}")
                # 一个重要优化：如果遇到重复ID，假设后续都是旧的，可以中断爬取。
                # 这适用于按发布时间倒序排列的API。
                # yield None  # 取消注释此行来启用此优化
                continue  # 跳过本条记录
            # !!! 核心逻辑结束 !!!
            
            item = NewsCrawlerItem()
            item['id'] = item_id
            item['title'] = article['title']
            item['content'] = article['content']
            item['publish_time'] = article['publish_time']
            
            # 在yield item之前，先将ID标记为已爬取（内存中）
            self.dupefilter.mark_as_scraped(item_id)
            
            yield item
        
        current_page = pagination.get('current_page', 1)
        total_pages = pagination.get('total_pages', 1)
        
        if current_page < total_pages:
            next_page = current_page + 1
            next_page_url = f"https://api.example-news.com/v1/articles?page={next_page}"
            yield scrapy.Request(
                url=next_page_url,
                callback=self.parse,
                meta={
                    'proxy': f"http://{self.proxyHost}:{self.proxyPort}",
                    'proxy_authorization': self.proxy_auth
                }
            )
    
    def closed(self, reason):
        """爬虫关闭时自动调用，用于保存去重记录"""
        self.dupefilter.save()
        self.logger.info("已保存去重记录文件。")

代码解释：在爬虫的 __init__ 方法中初始化了我们的 SimpleDupeFilter。在 parse 方法中，对于每篇文章，首先检查其 id 是否存在于已爬取集合中。如果存在，则记录日志并跳过 (continue)。如果不存在，则处理该文章，并将其 id 立刻加入到内存中的已爬取集合 (mark_as_scraped)。我们添加了一个重要的优化：由于新闻API通常按发布时间倒序排列，当我们遇到一个重复的ID时，意味着这一页及之后的所有文章都是我们已经爬取过的。此时，我们可以直接 break 或 return 来终止整个爬取任务，极大提升效率。代码中提供了相关注释。重写了 closed 方法，当爬虫正常或异常结束时，它会自动将内存中的已爬取ID集合持久化到文件中，供下次运行使用。三、进阶优化与生产环境建议上述示例提供了清晰的实现路径，但在生产环境中，你还可以考虑以下优化：使用Scrapy内置的去重：Scrapy自带 DUPEFILTER_CLASS，但其默认基于URL指纹去重，不适用于API分页（URL可能不变或只有页码变化）。你可以自定义一个基于响应内容ID的去重过滤器。数据库集成：对于海量数据，使用文件存储ID集合会变得缓慢。将其迁移到Redis或SQLite数据库是更好的选择。Redis的Set数据结构非常适合此场景。基于时间的增量爬取：如果API支持按时间过滤，可以记录上次爬取的最晚时间，然后请求 publish_time 大于该时间的文章。这比基于ID的去重更精确，能捕捉到文章的更新。请求URL可改为：f"https://api.example-news.com/v1/articles?page=1&after={last_crawl_time}"处理速率限制：在 settings.py 中配置 DOWNLOAD_DELAY 和 AUTOTHROTTLE_ENABLED，礼貌地爬取。游标分页的实现：如果API使用游标分页，逻辑更简洁。你只需要在 parse 方法中提取出 next_cursor，并将其作为参数加入到下一个请求中，直到 next_cursor 为 null 或空。python

# 游标分页伪代码
def parse(self, response):
    data = json.loads(response.text)
    for item in data['items']:
        yield process_item(item)
    next_cursor = data['pagination']['next_cursor']
    if next_cursor:
        yield scrapy.Request(
            f"https://api.example.com/v1/items?cursor={next_cursor}",
            callback=self.parse
        )

结论通过结合Scrapy的请求调度能力和一个外部的持久化去重机制，我们可以高效、稳健地实现API的分页爬取与增量抓取。关键在于：分页：通过分析API响应结构，递归或循环地生成后续页面的请求。增量：通过记录已爬取数据的唯一标识（如ID、时间戳），在数据生成端（Item Pipeline）或请求发起端（Spider）进行过滤。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

scrapy