feat(spiders): add a development mode
This commit is contained in:
Karim shoair
+80
-56
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user