Spider 类¶

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 is not None else b""

    params = self._session_kwargs.get("params")
    url = update_url_params(self.url, params) if params else self.url

    data: Dict[str, str | Tuple] = {
        "sid": self.sid,
        "body": body.hex(),
        "method": self._session_kwargs.get("method", "GET"),
        "url": canonicalize_url(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(sorted(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),
        "autothrottle_enabled": self.autothrottle_enabled,
        "autothrottle_delays": {domain: round(delay, 2) for domain, delay in self.autothrottle_delays.items()},
        "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 to_csv(self, path: Union[str, Path], *, fields: Optional[Iterable[str]] = None, delimiter: str = ","):
    """Export items to a CSV file.

    Items that don't share the same keys are still written, with the missing cells left empty, and any value
    that isn't a scalar (a nested dictionary or a list) is written as JSON.

    :param path: Path to the output file
    :param fields: The columns to write, defaulting to every key found in the items, in the order they appeared
    :param delimiter: The character separating the columns
    """
    columns = list(fields) if fields is not None else list({key: None for item in self for key in item})

    file = Path(path)
    file.parent.mkdir(parents=True, exist_ok=True)
    with open(file, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=columns, delimiter=delimiter, extrasaction="ignore")
        writer.writeheader()
        for item in self:
            writer.writerow({column: _stringify(item.get(column)) for column in columns})

    log.info("Saved %d items to %s", len(self), path)

def to_xml(self, path: Union[str, Path], *, root_tag: str = "items", item_tag: str = "item", indent: bool = True):
    """Export items to an XML file.

    Each item becomes an element whose children are named after the item's keys. Keys that aren't valid XML
    names are rewritten and keep the original in a `name` attribute, and any value that isn't a scalar
    (a nested dictionary or a list) is written as JSON.

    :param path: Path to the output file
    :param root_tag: The name of the element wrapping all the items
    :param item_tag: The name of the element wrapping every item
    :param indent: Pretty-print the file instead of writing it on a single line
    """
    root = Element(root_tag)
    for item in self:
        element = SubElement(root, item_tag)
        for key, value in item.items():
            tag = _xml_tag(key)
            child = SubElement(element, tag)
            if tag != str(key):
                child.set("name", _XML_FORBIDDEN_CHARS.sub("", str(key)))
            child.text = _XML_FORBIDDEN_CHARS.sub("", _stringify(value))

    tree = ElementTree(root)
    if indent:
        indent_tree(tree, space="  ")

    file = Path(path)
    file.parent.mkdir(parents=True, exist_ok=True)
    tree.write(file, encoding="utf-8", xml_declaration=True)
    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