from __future__ import annotations
import logging
import re
import warnings
# Iterable is needed at the run time for the SitemapSpider._parse_sitemap() annotation
from collections.abc import AsyncIterator, Iterable, Sequence # noqa: TC003
from typing import TYPE_CHECKING, Any, Literal, Self, cast
from urllib.parse import urljoin
from scrapy import signals
from scrapy.exceptions import ScrapyDeprecationWarning
from scrapy.http import Request, Response, XmlResponse
from scrapy.responsetypes import responsetypes
from scrapy.spiders import Spider
from scrapy.utils._compression import (
_DecompressionMaxSizeExceeded,
gunzip,
gzip_magic_number,
)
from scrapy.utils._sitemap import Sitemap, sitemap_urls_from_robots
from scrapy.utils.httpobj import urlparse_cached
if TYPE_CHECKING:
from scrapy.crawler import Crawler
from scrapy.http.request import CallbackT
from scrapy.robotstxt import RobotParser
logger = logging.getLogger(__name__)
[docs]
class SitemapSpider(Spider):
"""Spider that crawls a site by discovering its URLs using `sitemaps
<https://www.sitemaps.org/index.html>`_.
It supports nested sitemaps and discovering sitemap URLs from `robots.txt
<https://www.robotstxt.org/>`_.
"""
sitemap_urls: Sequence[str] = ()
"""URLs pointing to the sitemaps whose URLs you want to crawl.
You can also point to a `robots.txt <https://www.robotstxt.org/>`_ and it
will be parsed to extract sitemap URLs from it.
"""
sitemap_rules: Sequence[tuple[re.Pattern[str] | str, str | CallbackT]] = [
("", "parse")
]
"""``(regex, callback)`` tuples where:
- ``regex`` is a regular expression to match URLs extracted from
sitemaps. ``regex`` can be either a str or a compiled regex object.
- ``callback`` is the callback to use for processing the URLs that match
the regular expression. ``callback`` can be a string (indicating the
name of a spider method) or a callable.
For example:
.. code-block:: python
sitemap_rules = [("/product/", "parse_product")]
Rules are applied in order, and only the first one that matches will be
used.
The default value makes all URLs found in sitemaps be processed with the
:meth:`~scrapy.Spider.parse` callback.
"""
sitemap_follow: Sequence[re.Pattern[str] | str] = [""]
"""Regexes of sitemap URLs that should be followed: those linked from
`sitemap index files <https://www.sitemaps.org/protocol.html#index>`_, and
those declared through the ``Sitemap`` directive of a :file:`robots.txt`
file.
By default, all sitemaps are followed.
"""
sitemap_alternate_links: bool = False
"""Specifies if alternate links for one ``url`` should be followed. These are
links for the same website in another language passed within the same
``url`` block.
For example:
.. code-block:: xml
<url>
<loc>http://example.com/</loc>
<xhtml:link rel="alternate" hreflang="de" href="http://example.com/de"/>
</url>
When enabled, this would retrieve both URLs. When disabled, only
``http://example.com/`` would be retrieved.
"""
_max_size: int
_warn_size: int
_obey_robotstxt: bool
@classmethod
def from_crawler(cls, crawler: Crawler, *args: Any, **kwargs: Any) -> Self:
spider = super().from_crawler(crawler, *args, **kwargs)
spider._max_size = getattr(
spider, "download_maxsize", spider.settings.getint("DOWNLOAD_MAXSIZE")
)
spider._warn_size = getattr(
spider, "download_warnsize", spider.settings.getint("DOWNLOAD_WARNSIZE")
)
spider._obey_robotstxt = crawler.settings.getbool("ROBOTSTXT_OBEY")
if spider._obey_robotstxt:
crawler.signals.connect(spider._robots_parsed, signals.robots_parsed)
return spider
def __init__(self, *a: Any, **kw: Any):
super().__init__(*a, **kw)
self._cbs: list[tuple[re.Pattern[str], CallbackT]] = []
for r, c in self.sitemap_rules:
if isinstance(c, str):
c = cast("CallbackT", getattr(self, c)) # noqa: PLW2901
self._cbs.append((regex(r), c))
self._follow: list[re.Pattern[str]] = [regex(x) for x in self.sitemap_follow]
self._robots_sitemaps: dict[str, list[str]] = {}
self._signal_only_netlocs: set[str] = set()
def _robots_parsed(
self, robotparser: RobotParser, request: Request, response: Response
) -> None:
netloc = urlparse_cached(request).netloc
urls = [urljoin(request.url, url) for url in robotparser.sitemaps()]
if netloc in self._signal_only_netlocs:
# No request for this host's robots.txt was sent from start(), so
# _parse_sitemap() will never run for it; schedule its sitemaps
# directly instead of stashing them for a response that never arrives.
self._signal_only_netlocs.discard(netloc)
for url in urls:
if any(x.search(url) for x in self._follow):
self.crawler.engine.crawl(
self.sitemap_request(
url,
self._parse_sitemap,
{"loc": url},
source="robotstxt",
response=response,
)
)
else:
self._robots_sitemaps[netloc] = urls
async def start(self) -> AsyncIterator[Any]:
sitemap_netlocs = {
urlparse_cached(url).netloc
for url in self.sitemap_urls
if not url.endswith("/robots.txt")
}
for url in self.sitemap_urls:
netloc = urlparse_cached(url).netloc
if (
self._obey_robotstxt
and url.endswith("/robots.txt")
and netloc in sitemap_netlocs
):
# RobotsTxtMiddleware will fetch and parse this robots.txt
# anyway, to check the other sitemap_urls requests to the same
# host, so rely on the robots_parsed signal for it instead of
# requesting it a second time here.
self._signal_only_netlocs.add(netloc)
continue
yield Request(url, self._parse_sitemap)
[docs]
def sitemap_filter(
self, entries: Iterable[dict[str, Any]]
) -> Iterable[dict[str, Any]]:
"""Yield the sitemap entries from *entries* that should be processed.
Override it to select sitemap entries based on their attributes. For
example, given the following sitemap entry:
.. code-block:: xml
<url>
<loc>http://example.com/</loc>
<lastmod>2005-01-01</lastmod>
</url>
You can filter entries by date as follows:
.. code-block:: python
from datetime import datetime
from scrapy.spiders import SitemapSpider
class FilteredSitemapSpider(SitemapSpider):
name = "filtered_sitemap_spider"
allowed_domains = ["example.com"]
sitemap_urls = ["http://example.com/sitemap.xml"]
def sitemap_filter(self, entries):
for entry in entries:
date_time = datetime.strptime(entry["lastmod"], "%Y-%m-%d")
if date_time.year >= 2005:
yield entry
This would retrieve only entries modified on 2005 and the following
years.
Entries are dict objects extracted from the sitemap document. Usually,
the key is the tag name and the value is the text inside it.
It's important to notice that:
- as the ``loc`` attribute is required, entries without this tag are
discarded
- alternate links are stored in a list with the key ``alternate``
(see :attr:`sitemap_alternate_links`)
- namespaces are removed, so lxml tags named as ``{namespace}tagname``
become only ``tagname``
The default implementation yields all entries, observing other
attributes and their settings.
"""
yield from entries
[docs]
def sitemap_request(
self,
loc: str,
callback: CallbackT,
entry: dict[str, Any],
*,
source: Literal["robotstxt", "sitemapindex", "urlset"],
response: Response,
) -> Request:
"""Return the request for the *loc* URL of the sitemap *entry*, to be
parsed with *callback*.
.. versionadded:: VERSION
Override this method to build those requests differently, e.g. to
carry sitemap data and request metadata over to *callback*:
.. code-block:: python
def sitemap_request(self, loc, callback, entry, *, source, response):
meta = {
"lastmod": entry.get("lastmod"),
"site": response.meta.get("site"),
}
return Request(loc, callback, meta=meta)
*entry* is the sitemap entry that *loc* comes from, as described in
:meth:`sitemap_filter`. When :attr:`sitemap_alternate_links` is
enabled, the same entry is used for every alternate URL.
*source* tells where *loc* comes from: ``"urlset"`` for a URL to
crawl, ``"sitemapindex"`` for a sitemap linked from a sitemap index,
and ``"robotstxt"`` for a sitemap linked from a :file:`robots.txt`
file, where *entry* only has ``loc``.
*response* is the response that *entry* was read from.
"""
return Request(loc, callback=callback)
def _parse_sitemap(self, response: Response) -> Iterable[Request]:
if response.url.endswith("/robots.txt"):
urls = list(sitemap_urls_from_robots(response.body, base_url=response.url))
netloc = urlparse_cached(response).netloc
for url in self._robots_sitemaps.pop(netloc, ()):
if url not in urls:
urls.append(url)
return (
self.sitemap_request(
url,
self._parse_sitemap,
{"loc": url},
source="robotstxt",
response=response,
)
for url in urls
if any(x.search(url) for x in self._follow)
)
body = self._get_sitemap_body(response)
if not body:
logger.warning(
"Ignoring invalid sitemap: %(response)s",
{"response": response},
extra={"spider": self},
)
return ()
s = Sitemap(body)
if s.type == "sitemapindex":
index_entries = list(
self._get_urls_from_sitemapindex(self.sitemap_filter(s), response.url)
)
return (
self.sitemap_request(
loc,
self._parse_sitemap,
entry,
source="sitemapindex",
response=response,
)
for loc, entry in index_entries
)
if s.type == "urlset":
urlset_entries = list(
self._get_urls_and_callbacks_from_urlset(
self.sitemap_filter(s), response.url
)
)
return (
self.sitemap_request(loc, c, entry, source="urlset", response=response)
for loc, c, entry in urlset_entries
)
logger.warning(
"Ignoring invalid sitemap: %(response)s",
{"response": response},
extra={"spider": self},
)
return ()
def _get_urls_from_sitemapindex(
self, it: Iterable[dict[str, Any]], base_url: str
) -> Iterable[tuple[str, dict[str, Any]]]:
for loc, entry in _iterlocs(it, self.sitemap_alternate_links):
loc = urljoin(base_url, loc) # noqa: PLW2901
if any(x.search(loc) for x in self._follow):
yield loc, entry
def _get_urls_and_callbacks_from_urlset(
self, it: Iterable[dict[str, Any]], base_url: str
) -> Iterable[tuple[str, CallbackT, dict[str, Any]]]:
for loc, entry in _iterlocs(it, self.sitemap_alternate_links):
loc = urljoin(base_url, loc) # noqa: PLW2901
for r, c in self._cbs:
if r.search(loc):
yield loc, c, entry
break
def _get_sitemap_body(self, response: Response) -> bytes | None:
"""Return the sitemap body contained in the given response,
or None if the response is not a sitemap.
"""
# Download handlers other than the built-in ones may use their own
# response classes, so the content type, the URL and the body are taken
# into account as well.
if (
isinstance(response, XmlResponse)
or responsetypes.from_args(
headers=response.headers, url=response.url, body=response.body
)
is XmlResponse
):
return response.body
if gzip_magic_number(response):
uncompressed_size = len(response.body)
max_size = response.meta.get("download_maxsize", self._max_size)
# circular import
from scrapy.utils._download_handlers import get_warnsize # noqa: PLC0415
warn_size = get_warnsize(response.meta, self._warn_size)
try:
body = gunzip(response.body, max_size=max_size)
except _DecompressionMaxSizeExceeded:
return None
if uncompressed_size < warn_size <= len(body):
logger.warning(
f"{response} body size after decompression ({len(body)} B) "
f"is larger than the download warning size ({warn_size} B)."
)
return body
# actual gzipped sitemap files are decompressed above ;
# if we are here (response body is not gzipped)
# and have a response for .xml.gz,
# it usually means that it was already gunzipped
# by HttpCompression middleware,
# the HTTP response being sent with "Content-Encoding: gzip"
# without actually being a .xml.gz file in the first place,
# merely XML gzip-compressed on the fly,
# in other word, here, we have plain XML
if response.url.endswith(".xml") or response.url.endswith(".xml.gz"):
return response.body
return None
def regex(x: re.Pattern[str] | str) -> re.Pattern[str]:
if isinstance(x, str):
return re.compile(x)
return x
def iterloc(it: Iterable[dict[str, Any]], alt: bool = False) -> Iterable[str]:
warnings.warn(
"scrapy.spiders.sitemap.iterloc() is deprecated.",
ScrapyDeprecationWarning,
stacklevel=2,
)
return (loc for loc, _ in _iterlocs(it, alt))
def _iterlocs(
it: Iterable[dict[str, Any]], alt: bool = False
) -> Iterable[tuple[str, dict[str, Any]]]:
for d in it:
if loc := d["loc"]:
yield loc, d
# Also consider alternate URLs (xhtml:link rel="alternate")
if alt and (alt_list := d.get("alternate")):
for loc in alt_list:
yield loc, d