Async web crawler with politeness and caching

spidey crawl https://news.ycombinator.com \
    --depth 2 \
    --concurrency 50 \
    --rate 5 \
    --out hn.db

spidey export hn.db --format jsonl > hn.jsonl

spidey stats hn.db
# pages_total=1240, pages_per_minute=180, errors=12, avg_latency=380ms

src/spidey/
├── __init__.py
├── cli.py
├── crawler.py               # main orchestration (TaskGroup)
├── fetcher.py               # httpx wrapper with retries
├── parser.py                # HTML -> links + extracted data
├── frontier.py              # priority queue of URLs
├── robots.py                # robots.txt cache + check
├── politeness.py            # per-host rate limiter
├── dedupe.py                # URL canonicalisation + seen-set (bloom filter)
├── cache.py                 # HTTP cache (ETag, If-Modified-Since)
├── storage/
│   ├── sqlite.py
│   └── postgres.py
├── extractors/
│   ├── base.py
│   ├── meta.py              # OG tags, title, description
│   └── article.py           # readability-like text extraction
└── config.py
tests/
├── conftest.py
├── test_robots.py
├── test_politeness.py
├── test_dedupe.py
├── test_fetcher.py
└── test_integration.py

async def crawl(seeds: list[str], *, depth: int, concurrency: int):
    frontier = Frontier(seeds)
    async with httpx.AsyncClient(http2=True, follow_redirects=True) as client, \
               asyncio.TaskGroup() as tg:
        for _ in range(concurrency):
            tg.create_task(worker(client, frontier, depth))

async def worker(client, frontier, depth):
    while True:
        url, current_depth = await frontier.next()
        if url is None:
            break
        if current_depth > depth:
            continue
        try:
            await politeness.wait(host_of(url))
            if not await robots.allowed(url):
                continue
            r = await fetcher.fetch(client, url)
            page = await parser.parse(url, r.text)
            await storage.save(page)
            if current_depth < depth:
                for link in page.links:
                    if dedupe.is_new(link):
                        await frontier.put(link, current_depth + 1)
        except Exception as e:
            log.exception("worker", url=url, err=str(e))

class HostLimiter:
    def __init__(self, rate: float):
        self._rate = rate
        self._last: dict[str, float] = {}
        self._locks: dict[str, asyncio.Lock] = defaultdict(asyncio.Lock)

    async def wait(self, host: str):
        async with self._locks[host]:
            delta = time.monotonic() - self._last.get(host, 0)
            wait = max(0, 1 / self._rate - delta)
            if wait:
                await asyncio.sleep(wait)
            self._last[host] = time.monotonic()

from pybloom_live import ScalableBloomFilter

class Dedupe:
    def __init__(self):
        self.bloom = ScalableBloomFilter(initial_capacity=1_000_000, error_rate=0.0001)

    def is_new(self, url: str) -> bool:
        canon = canonicalise(url)
        if canon in self.bloom:
            return False
        self.bloom.add(canon)
        return True

from tenacity import retry, retry_if_exception_type, wait_exponential_jitter, stop_after_attempt

@retry(
    retry=retry_if_exception_type((httpx.HTTPError, httpx.TimeoutException)),
    wait=wait_exponential_jitter(initial=1, max=30),
    stop=stop_after_attempt(3),
    reraise=True,
)
async def fetch(client: httpx.AsyncClient, url: str) -> httpx.Response:
    r = await client.get(url, timeout=15)
    if r.status_code == 429:
        retry_after = float(r.headers.get("Retry-After", "5"))
        await asyncio.sleep(retry_after)
        raise httpx.HTTPError("429")
    r.raise_for_status()
    return r

from selectolax.parser import HTMLParser

def parse(url: str, html: str) -> Page:
    tree = HTMLParser(html)
    title = tree.css_first("title")
    links = [a.attributes.get("href") for a in tree.css("a[href]")]
    abs_links = [urljoin(url, l) for l in links if l]
    text = tree.body.text(separator="\n", strip=True) if tree.body else ""
    return Page(url=url, title=title.text() if title else "", links=abs_links, text=text)

try:
    async with asyncio.TaskGroup() as tg:
        for _ in range(concurrency):
            tg.create_task(worker(...))
except* KeyboardInterrupt:
    print("Saving state and shutting down...")
    await frontier.persist()