Merge branch 'dev' into fix/fetcher-session-state-corruption

scrapling/spiders/cache.py

@@ -0,0 +1,79 @@
+from base64 import b64encode, b64decode
+from pathlib import Path
+
+import orjson
+import anyio
+from anyio import Path as AsyncPath
+
+from scrapling.core.utils import log
+from scrapling.core._types import Dict, Optional, Any
+from scrapling.engines.toolbelt.custom import Response
+
+
+class ResponseCacheManager:
+    """Caches HTTP responses to disk for replay during spider development."""
+
+    def __init__(self, cache_dir: str | Path):
+        self._cache_dir = AsyncPath(cache_dir)
+
+    def _cache_path(self, fingerprint: bytes) -> AsyncPath:
+        return self._cache_dir / f"{fingerprint.hex()}.json"
+
+    async def get(self, fingerprint: bytes) -> Optional[Response]:
+        path = self._cache_path(fingerprint)
+        if not await path.exists():
+            return None
+
+        try:
+            async with await anyio.open_file(path, "rb") as f:
+                data: Dict[str, Any] = orjson.loads(await f.read())
+
+            return Response(
+                url=data["url"],
+                content=b64decode(data["content"]),
+                status=data["status"],
+                reason=data["reason"],
+                encoding=data["encoding"],
+                cookies=data["cookies"],
+                headers=data["headers"],
+                request_headers=data["request_headers"],
+                method=data["method"],
+            )
+        except Exception as e:
+            log.warning(f"Failed to read cached response for {fingerprint.hex()}: {e}")
+            return None
+
+    async def put(self, fingerprint: bytes, response: Response, method: str = "GET") -> None:
+        await self._cache_dir.mkdir(parents=True, exist_ok=True)
+        temp_path = self._cache_path(fingerprint).with_suffix(".tmp")
+
+        try:
+            serialized = orjson.dumps(
+                {
+                    "url": response.url,
+                    "content": b64encode(response.body).decode("ascii"),
+                    "status": response.status,
+                    "reason": response.reason,
+                    "encoding": response.encoding,
+                    "cookies": dict(response.cookies) if isinstance(response.cookies, dict) else {},
+                    "headers": dict(response.headers),
+                    "request_headers": dict(response.request_headers),
+                    "method": method,
+                }
+            )
+            async with await anyio.open_file(temp_path, "wb") as f:
+                await f.write(serialized)
+
+            await temp_path.rename(self._cache_path(fingerprint))
+        except Exception as e:
+            if await temp_path.exists():
+                await temp_path.unlink()
+            log.warning(f"Failed to cache response for {fingerprint.hex()}: {e}")
+
+    async def clear(self) -> None:
+        if not await self._cache_dir.exists():
+            return
+        async for entry in self._cache_dir.iterdir():
+            if entry.suffix == ".json":
+                await entry.unlink()
+        log.info(f"Cleared response cache at {self._cache_dir}")

scrapling/spiders/engine.py

@@ -1,16 +1,19 @@
 import json
 import pprint
 from pathlib import Path
+from urllib.parse import urlparse
 
 import anyio
 from anyio import Path as AsyncPath
 from anyio import create_task_group, CapacityLimiter, create_memory_object_stream, EndOfStream
 
 from scrapling.core.utils import log
-from scrapling.spiders.request import Request
 from scrapling.spiders.scheduler import Scheduler
 from scrapling.spiders.session import SessionManager
+from scrapling.spiders.request import Request, Response
+from scrapling.spiders.robotstxt import RobotsTxtManager
 from scrapling.spiders.result import CrawlStats, ItemList
+from scrapling.spiders.cache import ResponseCacheManager
 from scrapling.spiders.checkpoint import CheckpointManager, CheckpointData
 from scrapling.core._types import Dict, Union, Optional, TYPE_CHECKING, Any, AsyncGenerator
 
@@ -41,10 +44,29 @@ class CrawlerEngine:
         )
         self.stats = CrawlStats()
 
+        if self.spider.robots_txt_obey:
+
+            async def _fetch_robots(url: str, sid: str) -> Response:
+                return await self.session_manager.fetch(Request(url, sid=sid))
+
+            self._robots_manager: Optional[RobotsTxtManager] = RobotsTxtManager(_fetch_robots)
+        else:
+            self._robots_manager = None
+
+        if self.spider.development_mode:
+            cache_dir = self.spider.development_cache_dir or f".scrapling_cache/{self.spider.name}"
+            self._cache_manager: Optional[ResponseCacheManager] = ResponseCacheManager(cache_dir)
+            log.warning("Development mode enabled -- responses will be cached to disk and replayed on subsequent runs")
+        else:
+            self._cache_manager = None
+
         self._global_limiter = CapacityLimiter(spider.concurrent_requests)
         self._domain_limiters: dict[str, CapacityLimiter] = {}
         self._allowed_domains: set[str] = spider.allowed_domains or set()
 
+        if self.spider.robots_txt_obey:
+            self._domain_delays: dict[str, float] = {}
+
         self._active_tasks: int = 0
         self._running: bool = False
         self._items: ItemList = ItemList()
@@ -68,11 +90,42 @@ class CrawlerEngine:
                 return True
         return False
 
+    async def _get_domain_delay(self, request: Request) -> float:
+        """Resolve the effective download delay for a domain.
+
+        Takes the max of the spider's configured delay and any robots.txt
+        directives (Crawl-delay / Request-rate). Result is cached per domain.
+        """
+        robots_manager = self._robots_manager
+        if robots_manager is None:
+            return self.spider.download_delay
+
+        domain = request.domain
+
+        if domain in self._domain_delays:
+            return self._domain_delays[domain]
+
+        # For domains covered by _prefetch_robots_txt this is a local parser read.
+        # Domains discovered mid-crawl (not in start_urls) will fetch here.
+        c_delay, r_rate = await robots_manager.get_delay_directives(request.url, request.sid)
+
+        delay = self.spider.download_delay
+
+        if r_rate:
+            req_count, period = r_rate
+            if req_count > 0:
+                delay = max(delay, period / req_count)
+
+        if c_delay is not None:
+            delay = max(delay, c_delay)
+
+        self._domain_delays[domain] = delay
+        return delay
+
     def _rate_limiter(self, domain: str) -> CapacityLimiter:
         """Get or create a per-domain concurrency limiter if enabled, otherwise use the global limiter."""
         if self.spider.concurrent_requests_per_domain:
-            if domain not in self._domain_limiters:
-                self._domain_limiters[domain] = CapacityLimiter(self.spider.concurrent_requests_per_domain)
+            self._domain_limiters.setdefault(domain, CapacityLimiter(self.spider.concurrent_requests_per_domain))
             return self._domain_limiters[domain]
         return self._global_limiter
 
@@ -85,47 +138,8 @@ class CrawlerEngine:
         if not request.sid:
             request.sid = self.session_manager.default_session_id
 
-    async def _process_request(self, request: Request) -> None:
-        """Download and process a single request."""
-        async with self._rate_limiter(request.domain):
-            if self.spider.download_delay:
-                await anyio.sleep(self.spider.download_delay)
-
-            if request._session_kwargs.get("proxy"):
-                self.stats.proxies.append(request._session_kwargs["proxy"])
-            if request._session_kwargs.get("proxies"):
-                self.stats.proxies.append(dict(request._session_kwargs["proxies"]))
-            try:
-                response = await self.session_manager.fetch(request)
-                self.stats.increment_requests_count(request.sid or self.session_manager.default_session_id)
-                self.stats.increment_response_bytes(request.domain, len(response.body))
-                self.stats.increment_status(response.status)
-
-            except Exception as e:
-                self.stats.failed_requests_count += 1
-                await self.spider.on_error(request, e)
-                return
-
-        if await self.spider.is_blocked(response):
-            self.stats.blocked_requests_count += 1
-            if request._retry_count < self.spider.max_blocked_retries:
-                retry_request = request.copy()
-                retry_request._retry_count += 1
-                retry_request.priority -= 1  # Don't retry immediately
-                retry_request.dont_filter = True
-                retry_request._session_kwargs.pop("proxy", None)
-                retry_request._session_kwargs.pop("proxies", None)
-
-                new_request = await self.spider.retry_blocked_request(retry_request, response)
-                self._normalize_request(new_request)
-                await self.scheduler.enqueue(new_request)
-                log.info(
-                    f"Scheduled blocked request for retry ({retry_request._retry_count}/{self.spider.max_blocked_retries}): {request.url}"
-                )
-            else:
-                log.warning(f"Max retries exceeded for blocked request: {request.url}")
-            return
-
+    async def _run_callbacks(self, request: Request, response: Response) -> None:
+        """Dispatch response to the request's callback and process yielded items/requests."""
         callback = request.callback if request.callback else self.spider.parse
         try:
             async for result in callback(response):
@@ -155,6 +169,75 @@ class CrawlerEngine:
             log.error(msg, exc_info=e)
             await self.spider.on_error(request, e)
 
+    async def _process_request(self, request: Request) -> None:
+        """Download and process a single request."""
+        if self._robots_manager:
+            can_fetch = await self._robots_manager.can_fetch(request.url, request.sid)
+            if not can_fetch:
+                self.stats.robots_disallowed_count += 1
+                log.info(f"Request disallowed by robots.txt: {request.url}")
+                return
+            delay = await self._get_domain_delay(request)
+        else:
+            delay = self.spider.download_delay
+
+        if self._cache_manager and request._fp is not None:
+            cached = await self._cache_manager.get(request._fp)
+            if cached is not None:
+                cached.request = request
+                self.stats.cache_hits += 1
+                self.stats.increment_requests_count(request.sid or self.session_manager.default_session_id)
+                self.stats.increment_response_bytes(request.domain, len(cached.body))
+                self.stats.increment_status(cached.status)
+                log.debug(f"Cache hit: {request.url}")
+                await self._run_callbacks(request, cached)
+                return
+
+        async with self._rate_limiter(request.domain):
+            if delay:
+                await anyio.sleep(delay)
+
+            if request._session_kwargs.get("proxy"):
+                self.stats.proxies.append(request._session_kwargs["proxy"])
+            if request._session_kwargs.get("proxies"):
+                self.stats.proxies.append(dict(request._session_kwargs["proxies"]))
+            try:
+                response = await self.session_manager.fetch(request)
+                self.stats.increment_requests_count(request.sid or self.session_manager.default_session_id)
+                self.stats.increment_response_bytes(request.domain, len(response.body))
+                self.stats.increment_status(response.status)
+
+            except Exception as e:
+                self.stats.failed_requests_count += 1
+                await self.spider.on_error(request, e)
+                return
+
+        if self._cache_manager and request._fp is not None:
+            self.stats.cache_misses += 1
+            await self._cache_manager.put(request._fp, response, request._session_kwargs.get("method", "GET"))
+
+        if await self.spider.is_blocked(response):
+            self.stats.blocked_requests_count += 1
+            if request._retry_count < self.spider.max_blocked_retries:
+                retry_request = request.copy()
+                retry_request._retry_count += 1
+                retry_request.priority -= 1  # Don't retry immediately
+                retry_request.dont_filter = True
+                retry_request._session_kwargs.pop("proxy", None)
+                retry_request._session_kwargs.pop("proxies", None)
+
+                new_request = await self.spider.retry_blocked_request(retry_request, response)
+                self._normalize_request(new_request)
+                await self.scheduler.enqueue(new_request)
+                log.info(
+                    f"Scheduled blocked request for retry ({retry_request._retry_count}/{self.spider.max_blocked_retries}): {request.url}"
+                )
+            else:
+                log.warning(f"Max retries exceeded for blocked request: {request.url}")
+            return
+
+        await self._run_callbacks(request, response)
+
     async def _task_wrapper(self, request: Request) -> None:
         """Wrapper to track active task count."""
         try:
@@ -219,6 +302,25 @@ class CrawlerEngine:
 
         return True
 
+    async def _prefetch_robots_txt(self) -> None:
+        """Pre-warm the robots.txt cache before the crawl loop starts.
+
+        Extracts unique domains from start_urls, preserving the original scheme.
+        """
+        if not self._robots_manager or not self.spider.start_urls:
+            return
+
+        # Deduplicate by netloc, preserving the scheme from the first URL per domain
+        seen: set[str] = set()
+        seed_urls: list[str] = []
+        for url in self.spider.start_urls:
+            parsed = urlparse(url)
+            if parsed.netloc not in seen:
+                seen.add(parsed.netloc)
+                seed_urls.append(f"{parsed.scheme}://{parsed.netloc}/")
+
+        await self._robots_manager.prefetch(seed_urls, self.session_manager.default_session_id)
+
     async def crawl(self) -> CrawlStats:
         """Run the spider and return CrawlStats."""
         self._running = True
@@ -227,6 +329,9 @@ class CrawlerEngine:
         self._pause_requested = False
         self._force_stop = False
         self.stats = CrawlStats(start_time=anyio.current_time())
+        self._domain_limiters.clear()
+        if self._robots_manager:
+            self._domain_delays.clear()
 
         # Check for existing checkpoint
         resuming = (await self._restore_from_checkpoint()) if self._checkpoint_system_enabled else False
@@ -238,6 +343,8 @@ class CrawlerEngine:
             self.stats.download_delay = self.spider.download_delay
             await self.spider.on_start(resuming=resuming)
 
+            await self._prefetch_robots_txt()
+
             try:
                 if not resuming:
                     async for request in self.spider.start_requests():
@@ -251,11 +358,7 @@ class CrawlerEngine:
                     while self._running:
                         if self._pause_requested:
                             if self._active_tasks == 0 or self._force_stop:
-                                if self._force_stop:
-                                    log.warning(f"Force stopping with {self._active_tasks} active tasks")
-                                    tg.cancel_scope.cancel()
-
-                                # Only save checkpoint if checkpoint system is enabled
+                                # Save checkpoint before canceling to avoid data loss
                                 if self._checkpoint_system_enabled:
                                     await self._save_checkpoint()
                                     self.paused = True
@@ -263,6 +366,10 @@ class CrawlerEngine:
                                 else:
                                     log.info("Spider stopped gracefully")
 
+                                if self._force_stop:
+                                    log.warning(f"Force stopping with {self._active_tasks} active tasks")
+                                    tg.cancel_scope.cancel()
+
                                 self._running = False
                                 break
 

scrapling/spiders/result.py

@@ -47,6 +47,9 @@ class CrawlStats:
     concurrent_requests_per_domain: int = 0
     failed_requests_count: int = 0
     offsite_requests_count: int = 0
+    robots_disallowed_count: int = 0
+    cache_hits: int = 0
+    cache_misses: int = 0
     response_bytes: int = 0
     items_scraped: int = 0
     items_dropped: int = 0
@@ -95,6 +98,9 @@ class CrawlStats:
             "sessions_requests_count": self.sessions_requests_count,
             "failed_requests_count": self.failed_requests_count,
             "offsite_requests_count": self.offsite_requests_count,
+            "robots_disallowed_count": self.robots_disallowed_count,
+            "cache_hits": self.cache_hits,
+            "cache_misses": self.cache_misses,
             "blocked_requests_count": self.blocked_requests_count,
             "response_status_count": self.response_status_count,
             "response_bytes": self.response_bytes,

scrapling/spiders/robotstxt.py

@@ -0,0 +1,77 @@
+from urllib.parse import urlparse
+
+from anyio import create_task_group
+from protego import Protego
+
+from scrapling.core._types import Dict, Optional, Callable, Awaitable
+from scrapling.core.utils import log
+
+
+class RobotsTxtManager:
+    """Manages fetching, parsing, and caching of robots.txt files."""
+
+    def __init__(self, fetch_fn: Callable[[str, str], Awaitable]):
+        self._fetch_fn = fetch_fn
+        self._cache: Dict[str, Protego] = {}
+
+    async def _get_parser(self, url: str, sid: str) -> Protego:
+        parsed = urlparse(url)
+        domain = parsed.netloc
+
+        if domain in self._cache:
+            return self._cache[domain]
+
+        scheme = parsed.scheme or "https"
+        robots_url = f"{scheme}://{domain}/robots.txt"
+        content = ""
+        try:
+            response = await self._fetch_fn(robots_url, sid)
+            if response.status == 200:
+                content = response.body.decode(response.encoding, errors="replace")
+        except Exception as e:
+            log.warning(f"Failed to fetch robots.txt for {domain}: {e}")
+
+        try:
+            parser = Protego.parse(content)
+        except Exception as e:
+            log.warning(f"Failed to parse robots.txt for {domain}: {e}")
+            parser = Protego.parse("")
+
+        self._cache[domain] = parser
+        return parser
+
+    async def can_fetch(self, url: str, sid: str) -> bool:
+        """Check if a URL can be fetched according to the domain's robots.txt.
+
+        :param url: The full URL to check
+        :param sid: Session ID for fetching robots.txt if not yet cached
+        """
+        parser = await self._get_parser(url, sid)
+        return parser.can_fetch(url, "*")
+
+    async def get_delay_directives(self, url: str, sid: str) -> tuple[Optional[float], Optional[tuple[int, int]]]:
+        """Return both crawl-delay and request-rate in a single parser lookup.
+
+        :param url: Any URL on the domain to check
+        :param sid: Session ID for fetching robots.txt if not yet cached
+        """
+        parser = await self._get_parser(url, sid)
+        c_delay = parser.crawl_delay("*")
+        rate = parser.request_rate("*")
+        return (
+            float(c_delay) if c_delay is not None else None,
+            (rate.requests, rate.seconds) if rate is not None else None,
+        )
+
+    async def prefetch(self, urls: list[str], sid: str) -> None:
+        """Pre-warm the robots.txt cache for a list of seed URLs concurrently.
+
+        :param urls: Seed URLs whose domains should be pre-fetched (one per domain).
+        :param sid: Session ID to use for the robots.txt fetch requests.
+        """
+        if not urls:
+            return
+        log.debug(f"Pre-fetching robots.txt for {len(urls)} domain(s)")
+        async with create_task_group() as tg:
+            for url in urls:
+                tg.start_soon(self._get_parser, url, sid)

scrapling/spiders/spider.py

@@ -72,6 +72,13 @@ class Spider(ABC):
     start_urls: list[str] = []
     allowed_domains: Set[str] = set()
 
+    # Robots.txt compliance
+    robots_txt_obey: bool = False
+
+    # Development mode
+    development_mode: bool = False
+    development_cache_dir: Optional[str] = None
+
     # Concurrency settings
     concurrent_requests: int = 4
     concurrent_requests_per_domain: int = 0