当前位置：首页 > news >正文

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

news 2025/11/13 16:05:09

一、理解挑战：为何要处理分页与增量爬取？

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数据：

单页响应示例：

{"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}
}

我们的目标是：爬取所有分页的文章，并且每次运行时只抓取新发布的文章。

步骤1：创建Scrapy项目与初始爬虫

步骤2：定义数据模型（Item）

在 items.py 中，定义我们要抓取的字段。

import scrapyclass NewsCrawlerItem(scrapy.Item):id = scrapy.Field()  # 用于去重的唯一标识title = scrapy.Field()content = scrapy.Field()publish_time = scrapy.Field()

步骤3：核心实现——分页逻辑

我们将分页逻辑放在爬虫的 parse 方法中。这里使用递归请求的方式处理分页。

在 example_news_api.py 中：

import scrapy
import json
from news_crawler.items import NewsCrawlerItemclass 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']):#     continueyield item# 处理分页：获取下一页current_page = pagination.get('current_page', 1)total_pages = pagination.get('total_pages', 1)if current_page < total_pages:next_page = current_page + 1next_page_url = f"https://api.example-news.com/v1/articles?page={next_page}"# 构造下一页的请求，并指定回调函数为self.parseyield 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 jsonclass SimpleDupeFilter:"""一个基于JSON文件的简单去重过滤器"""def __init__(self, file_path='./scraped_ids.json'):self.file_path = file_pathself.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_idsdef 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_iditem['title'] = article['title']item['content'] = article['content']item['publish_time'] = article['publish_time']# 在yield item之前，先将ID标记为已爬取（内存中）self.dupefilter.mark_as_scraped(item_id)yield itemcurrent_page = pagination.get('current_page', 1)total_pages = pagination.get('total_pages', 1)if current_page < total_pages:next_page = current_page + 1next_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)