Python脚本爬虫去重机制如何设计

wen 实用脚本 2026-07-19 184

本文目录导读：

Python脚本爬虫去重机制如何设计

基础去重方案
使用Redis实现分布式去重
Bloom Filter去重（高效内存使用）
综合去重方案
内容去重（检测重复内容）
完整爬虫去重示例
使用建议

我来详细说明Python爬虫去重机制的设计方案：

基础去重方案

内存集合去重（适合小规模）

class SimpleDeduplicator:
    def __init__(self):
        self.seen = set()
    def is_duplicate(self, item):
        if item in self.seen:
            return True
        self.seen.add(item)
        return False
    def clear(self):
        self.seen.clear()

哈希去重（节省内存）

import hashlib
class HashDeduplicator:
    def __init__(self):
        self.seen = set()
    def _hash(self, url):
        return hashlib.md5(url.encode()).hexdigest()
    def is_duplicate(self, url):
        url_hash = self._hash(url)
        if url_hash in self.seen:
            return True
        self.seen.add(url_hash)
        return False

使用Redis实现分布式去重

import redis
import hashlib
class RedisDeduplicator:
    def __init__(self, redis_host='localhost', redis_port=6379, key='crawler:visited'):
        self.client = redis.Redis(host=redis_host, port=redis_port)
        self.key = key
    def is_duplicate(self, url):
        url_hash = hashlib.md5(url.encode()).hexdigest()
        return self.client.sismember(self.key, url_hash)
    def add(self, url):
        url_hash = hashlib.md5(url.encode()).hexdigest()
        self.client.sadd(self.key, url_hash)
    def batch_add(self, urls):
        hashes = [hashlib.md5(url.encode()).hexdigest() for url in urls]
        self.client.sadd(self.key, *hashes)

Bloom Filter去重（高效内存使用）

from pybloom_live import BloomFilter
class BloomDeduplicator:
    def __init__(self, capacity=100000, error_rate=0.001):
        self.bloom = BloomFilter(capacity, error_rate)
    def is_duplicate(self, url):
        return url in self.bloom
    def add(self, url):
        self.bloom.add(url)

自定义Bloom Filter实现

import math
import hashlib
from bitarray import bitarray
class CustomBloomFilter:
    def __init__(self, n, fpr):
        """
        n: 预计元素数量
        fpr: 期望误判率
        """
        self.n = n
        self.fpr = fpr
        self.m = self._calc_m(n, fpr)  # 位数组大小
        self.k = self._calc_k(n, self.m)  # 哈希函数数量
        self.bit_array = bitarray(self.m)
        self.bit_array.setall(0)
    def _calc_m(self, n, p):
        return int(-n * math.log(p) / (math.log(2) ** 2))
    def _calc_k(self, n, m):
        return int((m / n) * math.log(2))
    def _get_hashes(self, item):
        hashes = []
        for i in range(self.k):
            hash_val = hashlib.md5(f"{item}{i}".encode()).hexdigest()
            hashes.append(int(hash_val, 16) % self.m)
        return hashes
    def add(self, item):
        for hash_val in self._get_hashes(item):
            self.bit_array[hash_val] = 1
    def is_duplicate(self, item):
        for hash_val in self._get_hashes(item):
            if not self.bit_array[hash_val]:
                return False
        return True

综合去重方案

import re
from urllib.parse import urlparse, urlunparse
from collections import deque
class AdvancedDeduplicator:
    def __init__(self, 
                 method='bloom', 
                 max_size=1000000,
                 redis_config=None):
        self.method = method
        self.method = method
        self.fingerprints = set() if method == 'set' else None
        if method == 'bloom':
            self.deduplicator = BloomDeduplicator(max_size)
        elif method == 'redis':
            self.deduplicator = RedisDeduplicator(**redis_config)
        elif method == 'hash':
            self.deduplicator = HashDeduplicator()
        else:  # set
            self.deduplicator = SimpleDeduplicator()
    def normalize_url(self, url):
        """URL标准化处理"""
        parsed = urlparse(url)
        # 去除片段标识
        clean_url = parsed._replace(fragment='')
        # 标准化路径
        path = re.sub(r'/$', '', clean_url.path)
        clean_url = clean_url._replace(path=path)
        # 标准化查询参数（排序）
        if clean_url.query:
            params = sorted(clean_url.query.split('&'))
            clean_url = clean_url._replace(query='&'.join(params))
        return urlunparse(clean_url)
    def extract_fingerprint(self, url):
        """提取URL指纹"""
        normalized_url = self.normalize_url(url)
        return normalized_url
    def is_duplicate(self, url):
        fingerprint = self.extract_fingerprint(url)
        return self.deduplicator.is_duplicate(fingerprint)
    def add(self, url):
        fingerprint = self.extract_fingerprint(url)
        self.deduplicator.add(fingerprint)

内容去重（检测重复内容）

import hashlib
from simhash import Simhash
class ContentDeduplicator:
    def __init__(self, threshold=3):
        self.threshold = threshold  # SimHash汉明距离阈值
        self.simhashes = []
    def simhash_distance(self, hash1, hash2):
        """计算SimHash汉明距离"""
        x = hash1 ^ hash2
        return bin(x).count('1')
    def is_duplicate_content(self, content):
        """判断内容是否重复"""
        current_hash = Simhash(content).value
        for existing_hash in self.simhashes:
            distance = self.simhash_distance(current_hash, existing_hash)
            if distance <= self.threshold:
                return True
        self.simhashes.append(current_hash)
        return False

完整爬虫去重示例

import requests
from urllib.parse import urljoin
from bs4 import BeautifulSoup
class Crawler:
    def __init__(self):
        self.deduplicator = AdvancedDeduplicator(method='bloom', max_size=1000000)
        self.queue = deque()
        self.content_dedup = ContentDeduplicator()
    def extract_links(self, url, html):
        """提取页面链接"""
        soup = BeautifulSoup(html, 'html.parser')
        links = []
        for a_tag in soup.find_all('a', href=True):
            href = a_tag['href']
            full_url = urljoin(url, href)
            # 过滤非HTTP链接
            if not full_url.startswith(('http://', 'https://')):
                continue
            # 去重检查
            if not self.deduplicator.is_duplicate(full_url):
                links.append(full_url)
        return links
    def crawl(self, start_url, max_pages=100):
        """开始爬取"""
        self.queue.append(start_url)
        crawled_count = 0
        while self.queue and crawled_count < max_pages:
            url = self.queue.popleft()
            if self.deduplicator.is_duplicate(url):
                continue
            try:
                response = requests.get(url, timeout=10)
                response.raise_for_status()
                # 内容去重
                if self.content_dedup.is_duplicate_content(response.text):
                    print(f"重复内容跳过: {url}")
                    continue
                # 标记为已访问
                self.deduplicator.add(url)
                crawled_count += 1
                print(f"爬取: {url} ({crawled_count}/{max_pages})")
                # 提取新链接
                new_links = self.extract_links(url, response.text)
                for link in new_links:
                    self.queue.append(link)
            except Exception as e:
                print(f"爬取失败 {url}: {str(e)}")

使用建议

场景选择

小规模爬虫（<1万URL）：使用内存Set
中等规模（1-100万）：使用Bloom Filter
大规模分布式（>100万）：使用Redis去重需求**：添加SimHash

优化策略

# 1. 多级去重
class MultiLayerDeduplicator:
    def __init__(self):
        self.bloom = BloomDeduplicator()
        self.redis_set = RedisDeduplicator()
    def is_duplicate(self, url):
        # 先在Bloom Filter检查（快速）
        if not self.bloom.is_duplicate(url):
            self.bloom.add(url)
            return False
        # Bloom Filter认为重复时，再到Redis确认
        return self.redis_set.is_duplicate(url)
# 2. 定时清理
class TimedCleanDeduplicator:
    def __init__(self, clean_interval=3600):
        self.deduplicator = AdvancedDeduplicator()
        self.clean_interval = clean_interval
        self.last_clean = time.time()
    def check_and_clean(self):
        if time.time() - self.last_clean > self.clean_interval:
            self.deduplicator.deduplicator.clear()
            self.last_clean = time.time()

选择合适的去重方案取决于你的爬虫规模、运行环境和性能要求，建议从小规模开始,根据实际需求逐步升级。

本文地址： https://www.bmjup.com/post/24778.html