Advanced Features

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)
        return self._domain_limiters[domain]
    return self._global_limiter

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)

def request_pause(self) -> None:
    """Request a graceful pause of the crawl.

    First call: requests graceful pause (waits for active tasks).
    Second call: forces immediate stop.
    """
    if self._force_stop:
        return  # Already forcing stop

    if self._pause_requested:
        # Second Ctrl+C - force stop
        self._force_stop = True
        log.warning("Force stop requested, cancelling immediately...")
    else:
        self._pause_requested = True
        log.info(
            "Pause requested, waiting for in-flight requests to complete (press Ctrl+C again to force stop)..."
        )

async def _save_checkpoint(self) -> None:
    """Save current state to checkpoint files."""
    requests, seen = self.scheduler.snapshot()
    data = CheckpointData(requests=requests, seen=seen)
    await self._checkpoint_manager.save(data)
    self._last_checkpoint_time = anyio.current_time()

def __aiter__(self) -> AsyncGenerator[dict, None]:
    return self._stream()

async def _stream(self) -> AsyncGenerator[dict, None]:
    """Async generator that runs crawl and yields items."""
    send, recv = create_memory_object_stream[dict](100)
    self._item_stream = send

    async def run():
        try:
            await self.crawl()
        finally:
            await send.aclose()

    async with create_task_group() as tg:
        tg.start_soon(run)
        try:
            async for item in recv:
                yield item
        except EndOfStream:
            pass

async def on_start(self, resuming: bool = False) -> None:
    """Called before crawling starts. Override for setup logic.

    :param resuming: It's enabled if the spider is resuming from a checkpoint, left for the user to use.
    """
    if resuming:
        self.logger.debug("Resuming spider from checkpoint")
    else:
        self.logger.debug("Starting spider")

async def on_close(self) -> None:
    """Called after crawling finishes. Override for cleanup logic."""
    self.logger.debug("Spider closed")

async def on_error(self, request: Request, error: Exception) -> None:
    """
    Handle request errors for all spider requests.

    Override for custom error handling.
    """
    pass

async def on_scraped_item(self, item: Dict[str, Any]) -> Dict[str, Any] | None:
    """A hook to be overridden by users to do some processing on scraped items, return `None` to drop the item silently."""
    return item

async def start_requests(self) -> AsyncGenerator[Request, None]:
    """Generate initial requests to start the crawl.

    By default, this generates Request objects for each URL in `start_urls`
    using the session manager's default session and `parse()` as callback.

    Override this method for more control over initial requests
    (e.g., to add custom headers, use different callbacks, etc.)
    """
    if not self.start_urls:
        raise RuntimeError(
            "Spider has no starting point, either set `start_urls` or override `start_requests` function."
        )

    for url in self.start_urls:
        yield Request(url, sid=self._session_manager.default_session_id)

@dataclass
class CrawlStats:
    """Statistics for a crawl run."""

    requests_count: int = 0
    concurrent_requests: int = 0
    concurrent_requests_per_domain: int = 0
    failed_requests_count: int = 0
    offsite_requests_count: int = 0
    response_bytes: int = 0
    items_scraped: int = 0
    items_dropped: int = 0
    start_time: float = 0.0
    end_time: float = 0.0
    download_delay: float = 0.0
    blocked_requests_count: int = 0
    custom_stats: Dict = field(default_factory=dict)
    response_status_count: Dict = field(default_factory=dict)
    domains_response_bytes: Dict = field(default_factory=dict)
    sessions_requests_count: Dict = field(default_factory=dict)
    proxies: List[str | Dict | Tuple] = field(default_factory=list)
    log_levels_counter: Dict = field(default_factory=dict)

self.logger = logging.getLogger(f"scrapling.spiders.{self.name}")
self.logger.setLevel(self.logging_level)
self.logger.handlers.clear()
self.logger.propagate = False  # Don't propagate to parent 'scrapling' logger

formatter = logging.Formatter(
    fmt=self.logging_format.format(spider_name=self.name), datefmt=self.logging_date_format
)

# Add a log counter handler to track log counts by level
self._log_counter = LogCounterHandler()
self.logger.addHandler(self._log_counter)

console_handler = logging.StreamHandler()
console_handler.setFormatter(formatter)
self.logger.addHandler(console_handler)

if self.log_file:
    Path(self.log_file).parent.mkdir(parents=True, exist_ok=True)
    file_handler = logging.FileHandler(self.log_file)
    file_handler.setFormatter(formatter)
    self.logger.addHandler(file_handler)