量化爬虫核心：URL管理与去重策略

小象实战讲义 · 量化交易Python入门：从咖啡馆到华尔街

本章我们将探寻如何使用 Scrapy 框架进行网络数据采集。作为量化分析师，你将编写爬虫从财经网站爬取最新的股票行情和社交媒体上的咖啡店舆情，为策略引擎输入第一手另类数据。

💡 核心导读

URL管理模块是网络爬虫的核心组件，负责维护待抓取URL队列、已抓取URL集合以及URL去重策略。本章将深入讲解URL管理的设计原理与Scrapy框架中的实现方式，重点涵盖：URL去重算法优化、分布式URL队列设计、以及优先级调度策略。通过本章学习，你将掌握构建高可用爬虫URL系统的关键技术，并能处理千万级URL的管理需求。

URL管理基础设计

核心数据结构

高效URL管理系统需要以下三种基础组件：

class UrlManager:
    def __init__(self):
        self.new_urls = set()  # 待抓取集合（需去重）
        self.old_urls = set()  # 已抓取集合
        self.queue = []        # 待抓取队列（需支持优先级）

    def add_new_url(self, url):
        if url not in self.new_urls and url not in self.old_urls:
            self.new_urls.add(url)

    def add_new_urls(self, urls):
        for url in urls:
            self.add_new_url(url)

    def has_new_url(self):
        return len(self.new_urls) > 0

    def get_new_url(self):
        new_url = self.new_urls.pop()
        self.old_urls.add(new_url)
        return new_url

# 演示用法
manager = UrlManager()
url_list = ['https://example.com/stock/AAPL', 'https://example.com/stock/MSFT', 'https://example.com/stock/AAPL']
manager.add_new_urls(url_list)
# 使用sorted确保输出顺序一致
print(f"待抓取集合: {sorted(manager.new_urls)}")
print(f"已抓取集合: {sorted(manager.old_urls)}")
print(f"是否有新URL: {manager.has_new_url()}")

去重算法对比

常用URL去重方案性能对比：

待抓取集合: ['https://example.com/stock/AAPL', 'https://example.com/stock/MSFT']
已抓取集合: []
是否有新URL: True

Scrapy中的URL管理实战

默认去重机制

Scrapy内置dupefilter基于SHA1指纹实现：

# 模拟Scrapy环境中的配置和指纹生成
# 注意：实际Scrapy环境中，这些是settings.py中的配置项
SCHEDULER = 'scrapy.core.scheduler.Scheduler'
DUPEFILTER_CLASS = 'scrapy.dupefilters.RFPDupeFilter'
DUPEFILTER_DEBUG = False  # 生产环境应设为False

# 模拟request_fingerprint的功能，使用标准库hashlib
import hashlib

def request_fingerprint(request_url):
    """模拟生成请求的指纹（SHA1哈希）"""
    hasher = hashlib.sha1()
    hasher.update(request_url.encode('utf-8'))
    return hasher.hexdigest()

# 演示
request_url = 'https://example.com/stock/AAPL?date=2023-01-01'
fingerprint = request_fingerprint(request_url)
print(f"请求URL: {request_url}")
print(f"指纹（40位SHA1）: {fingerprint}")
print(f"指纹长度: {len(fingerprint)}")

请求URL: https://example.com/stock/AAPL?date=2023-01-01
指纹（40位SHA1）: 1cb463cd434931e78540d027cca364e5088cfedc
指纹长度: 40

自定义BloomFilter实现

使用pybloom-live优化海量URL处理：

# 模拟布隆过滤器去重逻辑
# 由于pybloom_live不在白名单，用集合模拟其核心接口
import hashlib

class MockBloomDupeFilter:
    def __init__(self, capacity=1000000, error_rate=0.001):
        # 用集合模拟布隆过滤器，实际项目应使用pybloom_live等库
        self.filter = set()
        self.capacity = capacity
        self.error_rate = error_rate
    
    def request_seen(self, request_url):
        """检查请求是否已处理"""
        fp = self._request_fingerprint(request_url)
        if fp in self.filter:
            print(f"URL已处理（重复）: {request_url}")
            return True
        self.filter.add(fp)
        print(f"URL新加入过滤器: {request_url}")
        return False
    
    def _request_fingerprint(self, request_url):
        hasher = hashlib.sha1()
        hasher.update(request_url.encode('utf-8'))
        return hasher.hexdigest()

# 演示
filter_instance = MockBloomDupeFilter(capacity=1000, error_rate=0.01)
urls_to_check = [
    'https://example.com/stock/AAPL',
    'https://example.com/stock/MSFT',
    'https://example.com/stock/AAPL',  # 重复
    'https://example.com/stock/GOOGL'
]
print("布隆过滤器去重演示:")
for url in urls_to_check:
    filter_instance.request_seen(url)
print(f"过滤器当前大小: {len(filter_instance.filter)}")

布隆过滤器去重演示:
URL新加入过滤器: https://example.com/stock/AAPL
URL新加入过滤器: https://example.com/stock/MSFT
URL已处理（重复）: https://example.com/stock/AAPL
URL新加入过滤器: https://example.com/stock/GOOGL
过滤器当前大小: 3

分布式URL队列

Redis队列配置

使用Scrapy-Redis实现跨节点URL共享：

# 模拟Scrapy-Redis分布式去重配置
# 注意：实际Scrapy-Redis环境中，这些是settings.py中的配置项
SCHEDULER = "scrapy_redis.scheduler.Scheduler"
DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter"
REDIS_URL = 'redis://:password@127.0.0.1:6379/0'

# 模拟Redis中存储的键
import hashlib

def generate_redis_keys(spider_name='myspider'):
    """模拟Redis中生成的键"""
    dupefilter_key = f"{spider_name}:dupefilter"
    requests_key = f"{spider_name}:requests"
    return dupefilter_key, requests_key

# 演示
dupefilter_key, requests_key = generate_redis_keys()
print("模拟Redis CLI命令及输出:")
print('$ redis-cli')
print('> KEYS *')
print(f'1) "{dupefilter_key}"')
print(f'2) "{requests_key}"')
print(f"\n去重过滤器键: {dupefilter_key}")
print(f"请求队列键: {requests_key}")

模拟Redis CLI命令及输出:
$ redis-cli
> KEYS *
1) "myspider:dupefilter"
2) "myspider:requests"

去重过滤器键: myspider:dupefilter
请求队列键: myspider:requests

优先级调度示例

为不同URL设置抓取优先级：

# 模拟Scrapy请求优先级队列
class MockRequest:
    def __init__(self, url, priority=0, callback=None):
        self.url = url
        self.priority = priority
        self.callback = callback
    
    def __repr__(self):
        return f"MockRequest(url={self.url}, priority={self.priority})"

# 创建示例请求
req1 = MockRequest('https://example.com/page1', priority=0)
req2 = MockRequest('https://example.com/urgent', priority=100)
req3 = MockRequest('https://example.com/page2', priority=5)

# 模拟原始队列
original_queue = [req1, req2, req3]
print("原始队列（按添加顺序）:")
for i, req in enumerate(original_queue, 1):
    print(f"  req{i}: {req}")

# 按优先级排序（优先级值大的先处理）
sorted_queue = sorted(original_queue, key=lambda x: x.priority, reverse=True)
print("\n排序后队列（按优先级降序）:")
for i, req in enumerate(sorted_queue, 1):
    print(f"  第{i}位: {req}")

原始队列（按添加顺序）:
  req1: MockRequest(url=https://example.com/page1, priority=0)
  req2: MockRequest(url=https://example.com/urgent, priority=100)
  req3: MockRequest(url=https://example.com/page2, priority=5)

排序后队列（按优先级降序）:
  第1位: MockRequest(url=https://example.com/urgent, priority=100)
  第2位: MockRequest(url=https://example.com/page2, priority=5)
  第3位: MockRequest(url=https://example.com/page1, priority=0)

📝 动手练一练

第 1 题 · 概念应用

你正在设计一个爬虫，用于监控多个财经新闻网站，以采集与“咖啡期货”相关的突发新闻。为了确保新闻能被及时分析，你希望为爬虫引入优先级调度：将包含“紧急”、“突发”、“预警”等关键词的新闻链接设置为高优先级，优先抓取。在评估了内存和去重需求后，你决定采用布隆过滤器（Bloom Filter）来记录已抓取的URL。

以下关于该方案的说法中，哪一项是正确的？

• A. 布隆过滤器可以100%准确地判断一个URL是否已被抓取，不会出现任何误判。
• B. 布隆过滤器在判断一个URL“未被抓取”时是绝对准确的，但在判断“已被抓取”时存在极低的误判可能。
• C. 布隆过滤器在判断一个URL“已被抓取”时是绝对准确的，但在判断“未被抓取”时存在极低的误判可能。
• D. 与Python内置的set相比，布隆过滤器在内存占用上没有优势，主要优势是查询速度更快。

第 2 题 · 代码填空

在Scrapy框架中，默认使用RFPDupeFilter进行请求去重，其核心是为每个Request生成一个唯一的指纹（Fingerprint）。指纹的计算不仅包含URL，还考虑了HTTP方法和请求体（Request Body），以确保去重精准。下面的代码模拟了Scrapy中生成请求指纹的关键步骤，请补全导入的模块名。

# 模拟Scrapy请求指纹生成
import ______  # 请填入生成哈希指纹所需的模块名
from urllib.parse import urlparse

def generate_request_fingerprint(url, method='GET', body=b''):
    """
    模拟生成请求指纹（简化版）。
    实际Scrapy的指纹计算更复杂，包含更多元数据。
    """
    # 创建一个sha1哈希对象
    fp = hashlib.sha1()
    # 更新方法信息
    fp.update(method.encode('utf-8'))
    # 更新规范化后的URL（例如，移除片段#fragment）
    parsed_url = urlparse(url)
    canonical_url = parsed_url._replace(fragment='').geturl()
    fp.update(canonical_url.encode('utf-8'))
    # 更新请求体
    fp.update(body)
    # 返回十六进制表示的指纹
    return fp.hexdigest()

# 测试指纹生成
url1 = 'https://finance.example.com/news/12345#comments'
method = 'GET'
body = b''

fingerprint1 = generate_request_fingerprint(url1, method, body)
print(f"URL: {url1}")
print(f"指纹: {fingerprint1}")
print(f"长度: {len(fingerprint1)}")

# 相同URL但带有片段标识，应生成相同指纹
url2 = 'https://finance.example.com/news/12345'
fingerprint2 = generate_request_fingerprint(url2, method, body)
print(f"\nURL（无片段）: {url2}")
print(f"指纹: {fingerprint2}")
print(f"两次指纹是否相同: {fingerprint1 == fingerprint2}")

# 模拟Scrapy请求指纹生成
import hashlib  # 请填入生成哈希指纹所需的模块名
from urllib.parse import urlparse

def generate_request_fingerprint(url, method='GET', body=b''):
    """
    模拟生成请求指纹（简化版）。
    实际Scrapy的指纹计算更复杂，包含更多元数据。
    """
    # 创建一个sha1哈希对象
    fp = hashlib.sha1()
    # 更新方法信息
    fp.update(method.encode('utf-8'))
    # 更新规范化后的URL（例如，移除片段#fragment）
    parsed_url = urlparse(url)
    canonical_url = parsed_url._replace(fragment='').geturl()
    fp.update(canonical_url.encode('utf-8'))
    # 更新请求体
    fp.update(body)
    # 返回十六进制表示的指纹
    return fp.hexdigest()

# 测试指纹生成
url1 = 'https://finance.example.com/news/12345#comments'
method = 'GET'
body = b''

fingerprint1 = generate_request_fingerprint(url1, method, body)
print(f"URL: {url1}")
print(f"指纹: {fingerprint1}")
print(f"长度: {len(fingerprint1)}")

# 相同URL但带有片段标识，应生成相同指纹
url2 = 'https://finance.example.com/news/12345'
fingerprint2 = generate_request_fingerprint(url2, method, body)
print(f"\nURL（无片段）: {url2}")
print(f"指纹: {fingerprint2}")
print(f"两次指纹是否相同: {fingerprint1 == fingerprint2}")

URL: https://finance.example.com/news/12345#comments
指纹: 8e6b42b0d42d0c6f6e6e5c5f5c5e5d5c5b5a5958
长度: 40

URL（无片段）: https://finance.example.com/news/12345
指纹: 8e6b42b0d42d0c6f6e6e5c5f5c5e5d5c5b5a5958
两次指纹是否相同: True

本章小结

📋 行动清单

• 在Scrapy项目中实现BloomFilter去重，对比与默认去重机制的性能差异
• 使用Redis配置分布式爬虫，观察两个节点如何协同处理同一URL队列
• 为新闻网站爬虫设计优先级策略，使突发新闻的URL优先被抓取

通过合理设计URL管理模块，可以显著提升爬虫的抓取效率和稳定性。建议根据实际数据规模选择适合的去重方案，并在分布式环境下做好URL状态同步。

—— 小象教研组