robots_cache.py

"""
Thread-safe per-domain robots.txt cache.

The first thread that asks about a domain becomes that domain's fetcher.
Other threads for the same domain wait on a per-domain Event instead of
launching duplicate network requests. The actual HTTP request happens outside
the cache lock, so unrelated domains are not blocked by slow I/O.
"""

import threading
from typing import Dict
from urllib.parse import urlparse
from urllib.robotparser import RobotFileParser

import requests


class RobotsCache:
    USER_AGENT = "CrawlGraph/1.0"

    def __init__(self, timeout: float = 5.0) -> None:
        self._cache: Dict[str, RobotFileParser] = {}
        self._inflight: Dict[str, threading.Event] = {}
        self._lock = threading.Lock()
        self._timeout = timeout

    def _fetch_robots(self, domain: str) -> RobotFileParser:
        """Fetch and parse robots.txt for domain. Called outside the lock."""
        rp = RobotFileParser()
        try:
            for scheme in ("https", "http"):
                try:
                    resp = requests.get(
                        f"{scheme}://{domain}/robots.txt",
                        timeout=self._timeout,
                        headers={"User-Agent": self.USER_AGENT},
                    )
                    if resp.status_code == 200:
                        rp.parse(resp.text.splitlines())
                        return rp
                    break
                except requests.exceptions.SSLError:
                    continue
        except Exception:
            pass

        # On any error, allow crawling to avoid false blocks.
        rp.allow_all = True
        return rp

    def can_fetch(self, url: str) -> bool:
        domain = urlparse(url).netloc.lower()

        with self._lock:
            if domain in self._cache:
                return self._cache[domain].can_fetch(self.USER_AGENT, url)

            if domain in self._inflight:
                ready = self._inflight[domain]
                fetcher = False
            else:
                ready = threading.Event()
                self._inflight[domain] = ready
                fetcher = True

        if not fetcher:
            ready.wait()
            with self._lock:
                return self._cache[domain].can_fetch(self.USER_AGENT, url)

        rp = self._fetch_robots(domain)

        with self._lock:
            self._cache[domain] = rp
            ready.set()
            del self._inflight[domain]
            return rp.can_fetch(self.USER_AGENT, url)