一、理解挑战：为何要处理分页与增量爬取？
- API分页
API分页是一种将大量数据分割成多个较小、可管理块（即页面）的技术。常见的分页模式包括：
页码分页：最直观的方式，通过 page 和 page_size 参数控制。
游标分页：更现代、更稳定的方式，API返回一个指向下一组数据的 cursor 或 next_cursor 令牌，常用于实时流数据。
偏移量分页：类似于页码分页，使用 offset 和 limit 参数。
如果不能系统地处理分页，我们的爬虫将只能获取到第一页的数据，导致数据严重不完整。 - 增量爬取
增量爬取指的是仅爬取自上次任务以来新增或发生变化的数据，而非每次都将目标数据全部重新抓取一遍。其核心价值在于：
极大减少网络请求：提升爬取效率，降低带宽成本。
减轻目标服务器负载：遵守良好的爬虫礼仪。
近实时更新：对于监控类应用，可以快速感知数据变化。
实现增量爬取的关键在于识别数据的“唯一性”和“变化性”，通常通过记录已爬取条目的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 scrapy
class 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 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）进行过滤。
