طراحی خزنده توزیعشده با Redis، فیلتر بلوم (Bloom Filter) و صفهای مقیاسپذیر Celery

import hashlib
from urllib.parse import urlparse, urlunparse, parse_qsl, urlencode
import redis
from celery import Celery

redis_client = redis.Redis(host='localhost', port=6379, db=0)
celery_app = Celery('distributed_crawler', broker='redis://localhost:6379/1')

# Atomic Token-Bucket / Window Rate Limiter in Lua
LUA_RATE_LIMIT_SCRIPT = """
local key = KEYS[1]
local limit = tonumber(ARGV[1])
local window_ms = tonumber(ARGV[2])
local current = redis.call('INCR', key)
if current == 1 then
    redis.call('PEXPIRE', key, window_ms)
end
if current > limit then
    return 0
end
return 1
"""
rate_limiter = redis_client.register_script(LUA_RATE_LIMIT_SCRIPT)

class RedisBloomDeduplicator:
    def __init__(self, key: str = 'crawler:bloom', bit_size: int = 1 << 30, hash_count: int = 7):
        self.key = key
        self.bit_size = bit_size
        self.hash_count = hash_count

    @staticmethod
    def canonicalize_url(url: str) -> str:
        parsed = urlparse(url.strip())
        filtered_query = sorted(
            [(k, v) for k, v in parse_qsl(parsed.query) if not k.startswith(('utm_', 'fbclid', 'gclid'))]
        )
        clean_path = parsed.path.rstrip('/') or '/'
        return urlunparse((parsed.scheme.lower(), parsed.netloc.lower(), clean_path, '', urlencode(filtered_query), ''))

    def _get_bit_offsets(self, canonical_url: str):
        digest = hashlib.sha256(canonical_url.encode('utf-8')).digest()
        h1 = int.from_bytes(digest[:8], 'big')
        h2 = int.from_bytes(digest, 'big')
        return [(h1 + i * h2) % self.bit_size for i in range(self.hash_count)]

    def check_and_add(self, url: str) -> bool:
        """Returns True if URL is new, False if already seen."""
        clean_url = self.canonicalize_url(url)
        offsets = self._get_bit_offsets(clean_url)
        pipe = redis_client.pipeline()
        for offset in offsets:
            pipe.setbit(self.key, offset, 1)
        previous_bits = pipe.execute()
        return not all(previous_bits)

deduplicator = RedisBloomDeduplicator()

@celery_app.task(bind=True, queue='crawl_frontier', max_retries=5)
def crawl_url_task(self, url: str):
    domain = urlparse(url).netloc
    allowed = rate_limiter(keys=[f'ratelimit:{domain}'], args=[5, 1000])
    if not allowed:
        # Re-queue with jitter delay if domain rate limit is exceeded
        raise self.retry(countdown=1.5)

    # Perform HTTP fetch and extract child links here
    discovered_links = [f'{url}/page-1', f'{url}/page-2']
    for link in discovered_links:
        if deduplicator.check_and_add(link):
            crawl_url_task.apply_async(args=[link], queue='crawl_frontier')