diff --git a/scrapling/spiders/cache.py b/scrapling/spiders/cache.py new file mode 100644 index 0000000..40d39d3 --- /dev/null +++ b/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}") diff --git a/scrapling/spiders/engine.py b/scrapling/spiders/engine.py index e668216..6db6866 100644 --- a/scrapling/spiders/engine.py +++ b/scrapling/spiders/engine.py @@ -13,6 +13,7 @@ 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 @@ -52,6 +53,13 @@ class CrawlerEngine: 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() @@ -130,57 +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.""" - 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 - - 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 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): @@ -210,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: @@ -330,11 +358,7 @@ class CrawlerEngine: while self._running: if self._pause_requested: if self._active_tasks == 0 or self._force_stop: - # Save checkpoint BEFORE cancelling the scope. - # cancel_scope.cancel() makes all subsequent awaits - # raise Cancelled, which would abort the checkpoint - # write and leave self.paused=False — causing the - # finally block to DELETE the previous checkpoint. + # Save checkpoint before canceling to avoid data loss if self._checkpoint_system_enabled: await self._save_checkpoint() self.paused = True diff --git a/scrapling/spiders/result.py b/scrapling/spiders/result.py index b374152..3710750 100644 --- a/scrapling/spiders/result.py +++ b/scrapling/spiders/result.py @@ -48,6 +48,8 @@ class CrawlStats: 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 @@ -97,6 +99,8 @@ class CrawlStats: "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, diff --git a/scrapling/spiders/spider.py b/scrapling/spiders/spider.py index 6aaa24f..edd9d8b 100644 --- a/scrapling/spiders/spider.py +++ b/scrapling/spiders/spider.py @@ -75,6 +75,10 @@ class Spider(ABC): # 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 diff --git a/tests/spiders/test_cache.py b/tests/spiders/test_cache.py new file mode 100644 index 0000000..fc9bc59 --- /dev/null +++ b/tests/spiders/test_cache.py @@ -0,0 +1,228 @@ +"""Tests for the ResponseCacheManager and development_mode integration.""" + +import tempfile +from pathlib import Path + +import anyio +import pytest + +from scrapling.spiders.cache import ResponseCacheManager +from scrapling.spiders.engine import CrawlerEngine +from scrapling.spiders.request import Request +from scrapling.spiders.session import SessionManager +from scrapling.engines.toolbelt.custom import Response +from scrapling.core._types import Any, Dict, Set, AsyncGenerator + + +def _make_response(url: str = "https://example.com", body: bytes = b"hello", status: int = 200) -> Response: + return Response( + url=url, + content=body, + status=status, + reason="OK", + encoding="utf-8", + cookies={}, + headers={"content-type": "text/html"}, + request_headers={"user-agent": "test"}, + method="GET", + ) + + +class TestResponseCacheManager: + + @pytest.mark.anyio + async def test_put_get_roundtrip(self): + with tempfile.TemporaryDirectory() as tmpdir: + cache = ResponseCacheManager(tmpdir) + fp = b"\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f\x10\x11\x12\x13\x14" + original = _make_response(body=b"test content") + + await cache.put(fp, original, "GET") + restored = await cache.get(fp) + + assert restored is not None + assert restored.url == original.url + assert restored.body == original.body + assert restored.status == original.status + assert restored.reason == original.reason + assert restored.encoding == original.encoding + assert dict(restored.headers) == dict(original.headers) + assert dict(restored.request_headers) == dict(original.request_headers) + + @pytest.mark.anyio + async def test_get_cache_miss(self): + with tempfile.TemporaryDirectory() as tmpdir: + cache = ResponseCacheManager(tmpdir) + result = await cache.get(b"\x00" * 20) + assert result is None + + @pytest.mark.anyio + async def test_get_corrupt_file(self): + with tempfile.TemporaryDirectory() as tmpdir: + cache = ResponseCacheManager(tmpdir) + fp = b"\xaa" * 20 + corrupt_path = Path(tmpdir) / f"{fp.hex()}.json" + corrupt_path.write_text("not valid json{{{") + + result = await cache.get(fp) + assert result is None + + @pytest.mark.anyio + async def test_clear(self): + with tempfile.TemporaryDirectory() as tmpdir: + cache = ResponseCacheManager(tmpdir) + fp1 = b"\x01" * 20 + fp2 = b"\x02" * 20 + + await cache.put(fp1, _make_response(url="https://a.com"), "GET") + await cache.put(fp2, _make_response(url="https://b.com"), "GET") + + assert await cache.get(fp1) is not None + assert await cache.get(fp2) is not None + + await cache.clear() + + assert await cache.get(fp1) is None + assert await cache.get(fp2) is None + + @pytest.mark.anyio + async def test_creates_cache_dir(self): + with tempfile.TemporaryDirectory() as tmpdir: + nested = Path(tmpdir) / "sub" / "dir" + cache = ResponseCacheManager(str(nested)) + await cache.put(b"\x03" * 20, _make_response(), "GET") + assert nested.exists() + + @pytest.mark.anyio + async def test_preserves_binary_body(self): + with tempfile.TemporaryDirectory() as tmpdir: + cache = ResponseCacheManager(tmpdir) + fp = b"\x04" * 20 + binary_body = bytes(range(256)) + await cache.put(fp, _make_response(body=binary_body), "GET") + restored = await cache.get(fp) + assert restored is not None + assert restored.body == binary_body + + +# --------------------------------------------------------------------------- +# Integration tests +# --------------------------------------------------------------------------- + + +class MockSession: + def __init__(self): + self._is_alive = False + self.fetch_count = 0 + + async def __aenter__(self): + self._is_alive = True + return self + + async def __aexit__(self, *args): + self._is_alive = False + + async def fetch(self, url: str, **kwargs): + self.fetch_count += 1 + return _make_response(url=url, body=b"fetched") + + +class _LogCounterStub: + def get_counts(self) -> Dict[str, int]: + return {"debug": 0, "info": 0, "warning": 0, "error": 0, "critical": 0} + + +class MockSpider: + def __init__(self, cache_dir: str): + self.concurrent_requests = 4 + self.concurrent_requests_per_domain = 0 + self.download_delay = 0.0 + self.max_blocked_retries = 3 + self.allowed_domains: Set[str] = set() + self.fp_include_kwargs = False + self.fp_include_headers = False + self.fp_keep_fragments = False + self.robots_txt_obey = False + self.development_mode = True + self.development_cache_dir = cache_dir + self.start_urls: list[str] = [] + self.name = "test_cache_spider" + self._log_counter = _LogCounterStub() + self.scraped_items: list[dict] = [] + + async def parse(self, response) -> AsyncGenerator[Dict[str, Any] | Request | None, None]: + yield {"url": str(response)} + + async def on_start(self, resuming: bool = False) -> None: + pass + + async def on_close(self) -> None: + pass + + async def on_error(self, request: Request, error: Exception) -> None: + pass + + async def on_scraped_item(self, item: Dict[str, Any]) -> Dict[str, Any] | None: + self.scraped_items.append(item) + return item + + async def is_blocked(self, response) -> bool: + return False + + async def retry_blocked_request(self, request: Request, response) -> Request: + return request + + async def start_requests(self) -> AsyncGenerator[Request, None]: + yield Request("https://example.com/page1", sid="default") + + +class TestDevelopmentModeIntegration: + + @pytest.mark.anyio + async def test_first_run_fetches_and_caches(self): + with tempfile.TemporaryDirectory() as tmpdir: + session = MockSession() + spider = MockSpider(cache_dir=tmpdir) + sm = SessionManager() + sm.add("default", session) + engine = CrawlerEngine(spider, sm) + + await engine.crawl() + + assert session.fetch_count == 1 + assert engine.stats.cache_misses == 1 + assert engine.stats.cache_hits == 0 + assert engine.stats.items_scraped == 1 + + @pytest.mark.anyio + async def test_second_run_uses_cache(self): + with tempfile.TemporaryDirectory() as tmpdir: + session = MockSession() + spider = MockSpider(cache_dir=tmpdir) + sm = SessionManager() + sm.add("default", session) + engine = CrawlerEngine(spider, sm) + + await engine.crawl() + assert session.fetch_count == 1 + + session2 = MockSession() + spider2 = MockSpider(cache_dir=tmpdir) + sm2 = SessionManager() + sm2.add("default", session2) + engine2 = CrawlerEngine(spider2, sm2) + + await engine2.crawl() + assert session2.fetch_count == 0 + assert engine2.stats.cache_hits == 1 + assert engine2.stats.cache_misses == 0 + assert engine2.stats.items_scraped == 1 + + @pytest.mark.anyio + async def test_disabled_by_default(self): + spider = MockSpider(cache_dir="unused") + spider.development_mode = False + sm = SessionManager() + sm.add("default", MockSession()) + engine = CrawlerEngine(spider, sm) + assert engine._cache_manager is None diff --git a/tests/spiders/test_engine.py b/tests/spiders/test_engine.py index 74cf875..10cbfdc 100644 --- a/tests/spiders/test_engine.py +++ b/tests/spiders/test_engine.py @@ -98,6 +98,8 @@ class MockSpider: self.fp_keep_fragments = fp_keep_fragments self.name = "test_spider" self.robots_txt_obey = robots_txt_obey + self.development_mode = False + self.development_cache_dir = None self.start_urls = start_urls or [] # Tracking lists diff --git a/tests/spiders/test_force_stop_checkpoint.py b/tests/spiders/test_force_stop_checkpoint.py index 212758e..a26e552 100644 --- a/tests/spiders/test_force_stop_checkpoint.py +++ b/tests/spiders/test_force_stop_checkpoint.py @@ -76,6 +76,8 @@ class SlowSpider: self.fp_include_headers = False self.fp_keep_fragments = False self.robots_txt_obey = False + self.development_mode = False + self.development_cache_dir = None self.start_urls = [] self.name = "slow_spider" self._log_counter = _LogCounterStub()