"""
This modules implements the CrawlSpider which is the recommended spider to use
for scraping typical websites that requires crawling pages.
See documentation in docs/topics/spiders.rst
"""
from __future__ import annotations
import copy
import warnings
from collections.abc import AsyncIterator, Callable
from typing import TYPE_CHECKING, Any, Self, TypeAlias, TypeVar, cast
from scrapy.exceptions import ScrapyDeprecationWarning
from scrapy.http import HtmlResponse, Request, Response
from scrapy.link import Link
from scrapy.linkextractors import LinkExtractor
from scrapy.spiders import Spider
from scrapy.utils.deprecate import method_is_overridden
from scrapy.utils.python import global_object_name
from scrapy.utils.spider import _await_spider_output, iterate_spider_output
if TYPE_CHECKING:
from collections.abc import Iterable, Sequence
from twisted.python.failure import Failure
from scrapy.crawler import Crawler
from scrapy.http.request import CallbackT
from scrapy.utils.defer import FailureWithRequest
_T = TypeVar("_T")
ProcessLinksT: TypeAlias = Callable[[list[Link]], list[Link]]
ProcessRequestT: TypeAlias = Callable[[Request, Response], Request | None]
def _identity(x: _T) -> _T:
return x
def _identity_process_request(request: Request, response: Response) -> Request | None:
return request
_CallableT = TypeVar("_CallableT", bound="Callable[..., Any]")
def _get_method(method: _CallableT | str | None, spider: Spider) -> _CallableT | None:
if callable(method):
return method
if isinstance(method, str):
return getattr(spider, method, None)
return None
_default_link_extractor = LinkExtractor()
[docs]
class Rule:
"""A link-following rule for :class:`CrawlSpider`, to be used in its
:attr:`~scrapy.spiders.CrawlSpider.rules` attribute.
It defines which links to extract from a response, what to do with the
responses those links produce, and whether to keep extracting links from
them.
.. versionadded:: VERSION
The *name* and *from_rules* parameters.
.. versionchanged:: VERSION
*errback* may now be a coroutine function (``async def``) or an
asynchronous generator function.
*callback*, *process_links*, *process_request* and *errback* may be
specified as a string instead of a callable, in which case the spider
method with that name is used.
*link_extractor* is a :class:`LinkExtractor
<scrapy.linkextractors.lxmlhtml.LxmlLinkExtractor>` object which defines
how links will be extracted from each crawled page. Each produced link will
be used to generate a :class:`~scrapy.Request` object, which will contain
the link's text in its ``meta`` dictionary (under the ``link_text`` key).
If omitted, a default link extractor created with no arguments will be
used, resulting in all links being extracted.
*callback* is called for each link extracted with *link_extractor*. It is a
:meth:`spider callback <scrapy.Spider.parse>`. If omitted, no callback runs
for the matching responses, which are then only used to extract further
links, as per *follow*.
*cb_kwargs* are the keyword arguments to pass to the callback.
*follow* determines whether links should be followed from each response
extracted with this rule. It defaults to ``True`` if *callback* is
``None``, and to ``False`` otherwise.
*process_links* is called for each list of links extracted from each
response using *link_extractor*. This is mainly used for filtering
purposes.
*process_request* is called for every :class:`~scrapy.Request` extracted by
this rule. It should take said request as first argument and the
:class:`~scrapy.http.Response` from which the request originated as second
argument. It must return a :class:`~scrapy.Request` object or ``None`` (to
filter out the request).
Use *process_request* to set the :attr:`~scrapy.Request.priority` of
requests generated by a rule, e.g. ``process_request=lambda request,
response: request.replace(priority=10)``.
*errback* is called if any exception is raised while processing a request
generated by the rule. It receives a
:class:`Twisted Failure <twisted.python.failure.Failure>` instance as first
parameter.
*name* identifies the rule, to be used as a target of the *from_rules* of
other rules.
*from_rules* is the *name* of another rule, or an iterable of names. If
set, this rule is only applied to responses reached through one of those
rules, instead of to every response.
"""
def __init__(
self,
link_extractor: LinkExtractor | None = None,
callback: CallbackT | str | None = None,
cb_kwargs: dict[str, Any] | None = None,
follow: bool | None = None,
process_links: ProcessLinksT | str | None = None,
process_request: ProcessRequestT | str | None = None,
errback: Callable[[Failure], Any] | str | None = None,
name: str | None = None,
from_rules: Iterable[str] | str | None = None,
):
self.link_extractor: LinkExtractor = link_extractor or _default_link_extractor
self.callback: CallbackT | str | None = callback
self.errback: Callable[[Failure], Any] | str | None = errback
self.cb_kwargs: dict[str, Any] = cb_kwargs or {}
self.process_links: ProcessLinksT | str = process_links or _identity
self.process_request: ProcessRequestT | str = (
process_request or _identity_process_request
)
self.follow: bool = follow if follow is not None else not callback
self.name: str | None = name
if isinstance(from_rules, str):
from_rules = (from_rules,)
self.from_rules: frozenset[str] | None = (
frozenset(from_rules) if from_rules is not None else None
)
def _compile(self, spider: Spider) -> None:
# this replaces method names with methods and we can't express this in type hints
self.callback = _get_method(self.callback, spider)
self.errback = _get_method(self.errback, spider)
self.process_links = cast(
"ProcessLinksT", _get_method(self.process_links, spider)
)
self.process_request = cast(
"ProcessRequestT", _get_method(self.process_request, spider)
)
[docs]
class CrawlSpider(Spider):
"""Spider that follows links based on a set of rules.
You declare a set of :class:`Rule` objects in :attr:`rules`, and this
spider extracts links from every response it gets and follows them
accordingly.
Use it only for crawls that fit that declarative model. As soon as you need
to control which requests are sent, in which order, or which data they
carry along, subclass :class:`~scrapy.Spider` instead and follow links from
your callbacks, e.g. with :meth:`response.follow
<scrapy.http.TextResponse.follow>`. That is usually simpler than bending
:attr:`rules` to a use case they do not cover.
.. warning:: Unlike in other spiders, a request without a callback is not
handled by :meth:`Spider.parse <scrapy.Spider.parse>`. It is handled by
:meth:`parse_start_url`, and its links are followed according to
:attr:`rules`.
That is the way to feed a response of your own to :attr:`rules`. If you
want your own parsing code to run for it instead, set its callback
explicitly; defining :meth:`~scrapy.Spider.parse` is not enough.
Requests generated from :attr:`rules` are not affected, they always get
a callback of their own.
.. reqmeta:: rule
Requests generated from :attr:`rules` carry the index of the matching rule
within :attr:`rules` in their ``rule`` :attr:`Request.meta
<scrapy.Request.meta>` key. :class:`CrawlSpider` needs that key to dispatch
the response to the right rule, so copying it into a request generated by a
different rule sends the response to the wrong callback.
"""
rules: Sequence[Rule] = ()
"""The rules that define how to crawl the site. If multiple rules match the
same link, the first one will be used, according to the order they're
defined in this attribute.
"""
_rules: list[Rule]
_follow_links: bool
def __init__(self, *a: Any, **kw: Any):
super().__init__(*a, **kw)
self._compile_rules()
if method_is_overridden(self.__class__, CrawlSpider, "_parse_response"):
warnings.warn(
"The CrawlSpider._parse_response method, which the "
f"{global_object_name(self.__class__)} class overrides, is "
"deprecated: it will be removed in future Scrapy releases. "
"Please override the CrawlSpider.parse_with_rules method "
"instead.",
ScrapyDeprecationWarning,
stacklevel=2,
)
def _parse(self, response: Response, **kwargs: Any) -> Any:
return self.parse_with_rules(
response=response,
callback=self.parse_start_url,
cb_kwargs=kwargs,
follow=True,
)
[docs]
def parse_start_url(self, response: Response, **kwargs: Any) -> Any:
"""Handle *response*, one of the responses produced for the URLs in the
:attr:`~scrapy.Spider.start_urls` attribute of the spider.
This is a :meth:`spider callback <scrapy.Spider.parse>`; override it to
parse the initial responses.
"""
return ()
def process_results(
self, response: Response, results: Iterable[Any]
) -> Iterable[Any]:
return results
def _build_request(self, rule_index: int, link: Link) -> Request:
return Request(
url=link.url,
callback=self._callback,
errback=self._errback,
meta={
"rule": rule_index,
"link_text": link.text,
"rule_name": self._rules[rule_index].name,
},
)
def _requests_to_follow(self, response: Response) -> Iterable[Request | None]:
if not isinstance(response, HtmlResponse):
return
seen: set[Link] = set()
response_rule_name = (
response.request.meta.get("rule_name") if response.request else None
)
for rule_index, rule in enumerate(self._rules):
if (
rule.from_rules is not None
and response_rule_name not in rule.from_rules
):
continue
links: list[Link] = [
lnk
for lnk in rule.link_extractor.extract_links(response)
if lnk not in seen
]
for link in cast("ProcessLinksT", rule.process_links)(links):
seen.add(link)
request = self._build_request(rule_index, link)
yield cast("ProcessRequestT", rule.process_request)(request, response)
def _callback(self, response: Response, **cb_kwargs: Any) -> Any:
rule = self._rules[cast("int", response.meta["rule"])]
return self.parse_with_rules(
response,
cast("CallbackT", rule.callback),
{**rule.cb_kwargs, **cb_kwargs},
rule.follow,
)
def _errback(self, failure: Failure) -> AsyncIterator[Any]:
request = cast("FailureWithRequest", failure).request
rule = self._rules[cast("int", request.meta["rule"])]
return self._handle_failure(
failure, cast("Callable[[Failure], Any]", rule.errback)
)
async def parse_with_rules(
self,
response: Response,
callback: CallbackT | None,
cb_kwargs: dict[str, Any],
follow: bool = True,
) -> AsyncIterator[Any]:
try:
if callback:
cb_res = await _await_spider_output(
callback(response, **cb_kwargs) or ()
)
cb_res = self.process_results(response, cb_res)
for request_or_item in iterate_spider_output(cb_res):
yield request_or_item
finally:
if follow and self._follow_links:
for request_or_item in self._requests_to_follow(response):
yield request_or_item
def _parse_response(
self,
response: Response,
callback: CallbackT | None,
cb_kwargs: dict[str, Any],
follow: bool = True,
) -> AsyncIterator[Any]:
warnings.warn(
"The CrawlSpider._parse_response method is deprecated: "
"it will be removed in future Scrapy releases. "
"Please use the CrawlSpider.parse_with_rules method instead.",
ScrapyDeprecationWarning,
stacklevel=2,
)
return self.parse_with_rules(response, callback, cb_kwargs, follow)
@staticmethod
async def _handle_failure(
failure: Failure, errback: Callable[[Failure], Any] | None
) -> AsyncIterator[Any]:
if errback:
results = await _await_spider_output(errback(failure) or ())
for request_or_item in iterate_spider_output(results):
yield request_or_item
def _compile_rules(self) -> None:
self._rules = []
for rule in self.rules:
copied_rule = copy.copy(rule)
copied_rule._compile(self)
self._rules.append(copied_rule)
@classmethod
def from_crawler(cls, crawler: Crawler, *args: Any, **kwargs: Any) -> Self:
spider = super().from_crawler(crawler, *args, **kwargs)
spider._follow_links = crawler.settings.getbool("CRAWLSPIDER_FOLLOW_LINKS")
if not spider._follow_links:
warnings.warn(
"The CRAWLSPIDER_FOLLOW_LINKS setting is deprecated."
" You can set follow=False in your rules to achieve the same effect.",
category=ScrapyDeprecationWarning,
stacklevel=2,
)
return spider