Design a Web Crawler

import time

DEFAULT_DELAY_MS = 1000

async def get_next_url(redis):
    # host_ready: sorted set of host -> next_allowed_fetch_ms
    now_ms = int(time.time() * 1000)
    ready = await redis.zrangebyscore("host_ready", 0, now_ms, start=0, num=1)
    if not ready:
        return None  # no host is polite to fetch yet; caller sleeps briefly
    host = ready[0]
    url = await redis.lpop("frontier:" + host)
    if url is None:
        await redis.zrem("host_ready", host)  # queue drained
        return None
    # block this host until its delay elapses; adjusted again on response
    delay = int(await redis.hget("host_delay", host) or DEFAULT_DELAY_MS)
    await redis.zadd("host_ready", {host: now_ms + delay})
    return url

async def report_result(redis, host, status, elapsed_ms):
    delay = int(await redis.hget("host_delay", host) or DEFAULT_DELAY_MS)
    if status in (429, 503):
        delay = min(delay * 2, 60_000)      # back off hard
    else:
        delay = max(1000, int(delay * 0.9), elapsed_ms * 3)  # adaptive politeness
    await redis.hset("host_delay", host, delay)

from urllib.parse import urlsplit, urlunsplit, parse_qsl, urlencode
from pybloom_live import ScalableBloomFilter

seen = ScalableBloomFilter(initial_capacity=50_000_000, error_rate=0.01)
TRACKING = {"fbclid", "gclid", "ref"}

def canonicalize(url: str) -> str | None:
    s = urlsplit(url)
    if s.scheme not in ("http", "https"):
        return None
    host = s.hostname.lower() if s.hostname else None
    if not host or s.path.count("/") > 8:
        return None  # depth cap against calendar-style traps
    query = urlencode(sorted(
        (k, v) for k, v in parse_qsl(s.query)
        if not k.startswith("utm_") and k not in TRACKING
    ))
    return urlunsplit((s.scheme, host, s.path or "/", query, ""))

def enqueue_if_new(db, redis_pipe, url: str):
    canon = canonicalize(url)
    if canon is None or canon in seen:
        return  # Bloom filter: ~10 bits/URL vs storing full strings
    seen.add(canon)
    # authoritative store confirms; INSERT OR IGNORE handles FP double-checks
    db.execute("INSERT OR IGNORE INTO urls (url, status) VALUES (?, 'queued')", (canon,))
    host = urlsplit(canon).hostname
    redis_pipe.rpush("frontier:" + host, canon)
    redis_pipe.zadd("host_ready", {host: 0}, nx=True)

import asyncio, gzip, hashlib, time
import aiohttp

UA = "MiniCrawler/0.1 (+mailto:you@example.com)"

async def worker(redis, db, session: aiohttp.ClientSession):
    while True:
        url = await get_next_url(redis)
        if url is None:
            await asyncio.sleep(0.05)
            continue
        host = aiohttp.helpers.URL(url).host
        if not await robots_allows(redis, session, host, url):
            continue
        start = time.monotonic()
        try:
            async with session.get(url, timeout=aiohttp.ClientTimeout(total=10),
                                   headers={"User-Agent": UA}) as resp:
                body = await resp.content.read(2_000_000)  # 2 MB cap
                elapsed = int((time.monotonic() - start) * 1000)
                await report_result(redis, host, resp.status, elapsed)
                if resp.status == 200 and "text/html" in resp.headers.get("Content-Type", ""):
                    key = hashlib.sha256(url.encode()).hexdigest()
                    open("pages/" + key + ".html.gz", "wb").write(gzip.compress(body))
                    await parse_and_enqueue(redis, db, url, body)
        except (aiohttp.ClientError, asyncio.TimeoutError):
            await report_result(redis, host, 503, 10_000)