Spider Classes¶

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)

@abstractmethod
async def parse(self, response: "Response") -> AsyncGenerator[Dict[str, Any] | Request | None, None]:
    """Default callback for processing responses"""
    raise NotImplementedError(f"{self.__class__.__name__} must implement parse() method")
    yield  # Make this a generator for type checkers

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 is_blocked(self, response: "Response") -> bool:
    """Check if the response is blocked. Users should override this for custom detection logic."""
    if response.status in BLOCKED_CODES:
        return True
    return False

async def retry_blocked_request(self, request: Request, response: "Response") -> Request:
    """Users should override this to prepare the blocked request before retrying, if needed."""
    return request

def __repr__(self) -> str:
    """String representation of the spider."""
    return f"<{self.__class__.__name__} '{self.name}'>"

def configure_sessions(self, manager: SessionManager) -> None:
    """Configure sessions for this spider.

    Override this method to add custom sessions.
    The default implementation creates a FetcherSession session.

    The first session added becomes the default for `start_requests()` unless specified otherwise.

    :param manager: SessionManager to configure
    """
    from scrapling.fetchers import FetcherSession

    manager.add("default", FetcherSession())

def pause(self):
    """Request graceful shutdown of the crawling process."""
    if self._engine:
        self._engine.request_pause()
    else:
        raise RuntimeError("No active crawl to stop")

def start(self, use_uvloop: bool = False, **backend_options: Any) -> CrawlResult:
    """Run the spider and return results.

    This is the main entry point for running a spider.
    Handles async execution internally via anyio.

    Pressing Ctrl+C will initiate graceful shutdown (waits for active tasks to complete).
    Pressing Ctrl+C a second time will force immediate stop.

    If crawldir is set, a checkpoint will also be saved on graceful shutdown,
    allowing you to resume the crawl later by running the spider again.

    :param use_uvloop: Whether to use the faster uvloop/winloop event loop implementation, if available.
    :param backend_options: Asyncio backend options to be used with `anyio.run`
    """
    backend_options = backend_options or {}
    if use_uvloop:
        backend_options.update({"use_uvloop": True})

    # Set up SIGINT handler for graceful shutdown
    self._setup_signal_handler()
    try:
        return anyio.run(self.__run, backend="asyncio", backend_options=backend_options)
    finally:
        self._restore_signal_handler()

async def stream(self) -> AsyncGenerator[Dict[str, Any], None]:
    """Stream items as they're scraped. Ideal for long-running spiders or building applications on top of the spiders.

    Must be called from an async context. Yields items one by one as they are scraped.
    Access `spider.stats` during iteration for real-time statistics.

    Note: SIGINT handling for pause/resume is not available in stream mode.
    """
    token = set_logger(self.logger)
    try:
        self._engine = CrawlerEngine(self, self._session_manager, self.crawldir, self._interval)
        async for item in self._engine:
            yield item
    finally:
        self._engine = None
        reset_logger(token)
        if self.log_file:
            for handler in self.logger.handlers:
                if isinstance(handler, logging.FileHandler):
                    handler.close()

def copy(self) -> "Request":
    """Create a copy of this request."""
    return Request(
        url=self.url,
        sid=self.sid,
        callback=self.callback,
        priority=self.priority,
        dont_filter=self.dont_filter,
        meta=self.meta.copy(),
        _retry_count=self._retry_count,
        **self._session_kwargs,
    )

def update_fingerprint(
    self,
    include_kwargs: bool = False,
    include_headers: bool = False,
    keep_fragments: bool = False,
) -> bytes:
    """Generate a unique fingerprint for deduplication.

    Caches the result in self._fp after first computation.
    """
    if self._fp is not None:
        return self._fp

    post_data = self._session_kwargs.get("data", {})
    body = b""
    if post_data:
        if isinstance(post_data, dict | list | tuple):
            body = urlencode(post_data).encode()
        elif isinstance(post_data, str):
            body = post_data.encode()
        elif isinstance(post_data, BytesIO):
            body = post_data.getvalue()
        elif isinstance(post_data, bytes):
            body = post_data
    else:
        post_data = self._session_kwargs.get("json", {})
        body = orjson.dumps(post_data) if post_data else b""

    data: Dict[str, str | Tuple] = {
        "sid": self.sid,
        "body": body.hex(),
        "method": self._session_kwargs.get("method", "GET"),
        "url": canonicalize_url(self.url, keep_fragments=keep_fragments),
    }

    if include_kwargs:
        filtered_kwargs = {
            key.lower(): _stable_value_repr(value)
            for key, value in self._session_kwargs.items()
            if key.lower() not in ("data", "json")
        }
        data["kwargs"] = tuple(sorted(filtered_kwargs.items()))

    if include_headers:
        headers = self._session_kwargs.get("headers") or self._session_kwargs.get("extra_headers") or {}
        processed_headers = {}
        # Some header normalization
        for key, value in headers.items():
            processed_headers[_convert_to_bytes(key.lower()).hex()] = _convert_to_bytes(value).hex()
        data["headers"] = tuple(processed_headers.items())

    fp = hashlib.sha1(orjson.dumps(data, option=orjson.OPT_SORT_KEYS), usedforsecurity=False).digest()
    self._fp = fp
    return fp

def __repr__(self) -> str:
    callback_name = getattr(self.callback, "__name__", None) or "None"
    return f"<Request({self.url}) priority={self.priority} callback={callback_name}>"

def __str__(self) -> str:
    return self.url

def __lt__(self, other: object) -> bool:
    """Compare requests by priority"""
    if not isinstance(other, Request):
        return NotImplemented
    return self.priority < other.priority

def __gt__(self, other: object) -> bool:
    """Compare requests by priority"""
    if not isinstance(other, Request):
        return NotImplemented
    return self.priority > other.priority

def __eq__(self, other: object) -> bool:
    """Requests are equal if they have the same fingerprint."""
    if not isinstance(other, Request):
        return NotImplemented
    if self._fp is None or other._fp is None:
        raise RuntimeError("Cannot compare requests before generating their fingerprints!")
    return self._fp == other._fp

def __getstate__(self) -> dict[str, Any]:
    """Prepare state for pickling - store callback as name string for pickle compatibility."""
    state = self.__dict__.copy()
    state["_callback_name"] = getattr(self.callback, "__name__", None) if self.callback is not None else None
    state["callback"] = None  # Don't pickle the actual callable
    return state

def __setstate__(self, state: dict[str, Any]) -> None:
    """Restore state from pickle - callback restored later via _restore_callback()."""
    self._callback_name: str | None = state.pop("_callback_name", None)
    self.__dict__.update(state)

def __len__(self) -> int:
    return len(self.items)

def __iter__(self) -> Iterator[dict[str, Any]]:
    return iter(self.items)

def increment_status(self, status: int) -> None:
    self.response_status_count[f"status_{status}"] = self.response_status_count.get(f"status_{status}", 0) + 1

def increment_response_bytes(self, domain: str, count: int) -> None:
    self.response_bytes += count
    self.domains_response_bytes[domain] = self.domains_response_bytes.get(domain, 0) + count

def increment_requests_count(self, sid: str) -> None:
    self.requests_count += 1
    self.sessions_requests_count[sid] = self.sessions_requests_count.get(sid, 0) + 1

def to_dict(self) -> dict[str, Any]:
    return {
        "items_scraped": self.items_scraped,
        "items_dropped": self.items_dropped,
        "elapsed_seconds": round(self.elapsed_seconds, 2),
        "download_delay": round(self.download_delay, 2),
        "concurrent_requests": self.concurrent_requests,
        "concurrent_requests_per_domain": self.concurrent_requests_per_domain,
        "requests_count": self.requests_count,
        "requests_per_second": round(self.requests_per_second, 2),
        "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,
        "domains_response_bytes": self.domains_response_bytes,
        "proxies": self.proxies,
        "custom_stats": self.custom_stats,
        "log_count": self.log_levels_counter,
    }

def to_json(self, path: Union[str, Path], *, indent: bool = False):
    """Export items to a JSON file.

    :param path: Path to the output file
    :param indent: Pretty-print with 2-space indentation (slightly slower)
    """
    options = orjson.OPT_SERIALIZE_NUMPY
    if indent:
        options |= orjson.OPT_INDENT_2

    file = Path(path)
    file.parent.mkdir(parents=True, exist_ok=True)
    file.write_bytes(orjson.dumps(list(self), option=options))
    log.info("Saved %d items to %s", len(self), path)

def to_jsonl(self, path: Union[str, Path]):
    """Export items as JSON Lines (one JSON object per line).

    :param path: Path to the output file
    """
    Path(path).parent.mkdir(parents=True, exist_ok=True)
    with open(path, "wb") as f:
        for item in self:
            f.write(orjson.dumps(item, option=orjson.OPT_SERIALIZE_NUMPY))
            f.write(b"\n")
    log.info("Saved %d items to %s", len(self), path)

def add(self, session_id: str, session: Session, *, default: bool = False, lazy: bool = False) -> "SessionManager":
    """Register a session instance.

    :param session_id: Name to reference this session in requests
    :param session: Your pre-configured session instance
    :param default: If True, this becomes the default session
    :param lazy: If True, the session will be started only when a request uses its ID.
    """
    if session_id in self._sessions:
        raise ValueError(f"Session '{session_id}' already registered")

    self._sessions[session_id] = session

    if default or self._default_session_id is None:
        self._default_session_id = session_id

    if lazy:
        self._lazy_sessions.add(session_id)

    return self

def remove(self, session_id: str) -> None:
    """Removes a session.

    :param session_id: ID of session to remove
    """
    _ = self.pop(session_id)

def pop(self, session_id: str) -> Session:
    """Remove and returns a session.

    :param session_id: ID of session to remove
    """
    if session_id not in self._sessions:
        raise KeyError(f"Session '{session_id}' not found")

    session = self._sessions.pop(session_id)
    if session_id in self._lazy_sessions:
        self._lazy_sessions.remove(session_id)

    if session and self._default_session_id == session_id:
        self._default_session_id = next(iter(self._sessions), None)

    return session

def get(self, session_id: str) -> Session:
    if session_id not in self._sessions:
        available = ", ".join(self._sessions.keys())
        raise KeyError(f"Session '{session_id}' not found. Available: {available}")
    return self._sessions[session_id]

async def start(self) -> None:
    """Start all sessions that aren't already alive."""
    if self._started:
        return

    for sid, session in self._sessions.items():
        if sid not in self._lazy_sessions and not session._is_alive:
            await session.__aenter__()

    self._started = True

async def close(self) -> None:
    """Close all registered sessions."""
    for sid, session in self._sessions.items():
        if sid in self._lazy_sessions and not session._is_alive:
            continue
        _ = await session.__aexit__(None, None, None)

    self._started = False

async def fetch(self, request: Request) -> Response:
    sid = request.sid if request.sid else self.default_session_id
    session = self.get(sid)

    if session:
        if sid in self._lazy_sessions and not session._is_alive:
            async with self._lazy_lock:
                if not session._is_alive:
                    await session.__aenter__()

        if isinstance(session, FetcherSession):
            client = session._client

            if isinstance(client, _ASyncSessionLogic):
                kwargs = request._session_kwargs.copy()
                method = cast(SUPPORTED_HTTP_METHODS, kwargs.pop("method", "GET"))
                response = await client._make_request(
                    method=method,
                    url=request.url,
                    **kwargs,
                )
            else:
                # Sync session or other types - shouldn't happen in async context
                raise TypeError(f"Session type {type(client)} not supported for async fetch")
        else:
            response = await session.fetch(url=request.url, **request._session_kwargs)

        response.request = request
        # Merge request meta into response meta (response meta takes priority)
        response.meta = {**request.meta, **response.meta}
        return response
    raise RuntimeError("No session found with the request session id")

async def __aenter__(self) -> "SessionManager":
    await self.start()
    return self

async def __aexit__(self, *exc) -> None:
    await self.close()

def __contains__(self, session_id: str) -> bool:
    """Check if a session ID is registered."""
    return session_id in self._sessions

def __len__(self) -> int:
    """Number of registered sessions."""
    return len(self._sessions)