diff --git a/scrapy/spiders/sitemap.py b/scrapy/spiders/sitemap.py index cb5779d74..d87ffccaf 100644 --- a/scrapy/spiders/sitemap.py +++ b/scrapy/spiders/sitemap.py @@ -72,31 +72,53 @@ class SitemapSpider(Spider): def _parse_sitemap(self, response: Response) -> Iterable[Request]: if response.url.endswith("/robots.txt"): - for url in sitemap_urls_from_robots(response.text, base_url=response.url): - yield Request(url, callback=self._parse_sitemap) - else: - body = self._get_sitemap_body(response) - if body is None: - logger.warning( - "Ignoring invalid sitemap: %(response)s", - {"response": response}, - extra={"spider": self}, - ) - return + urls = list(sitemap_urls_from_robots(response.body, base_url=response.url)) + return (Request(url, callback=self._parse_sitemap) for url in urls) - s = Sitemap(body) - it = self.sitemap_filter(s) + body = self._get_sitemap_body(response) + if not body: + logger.warning( + "Ignoring invalid sitemap: %(response)s", + {"response": response}, + extra={"spider": self}, + ) + return () - if s.type == "sitemapindex": - for loc in iterloc(it, self.sitemap_alternate_links): - if any(x.search(loc) for x in self._follow): - yield Request(loc, callback=self._parse_sitemap) - elif s.type == "urlset": - for loc in iterloc(it, self.sitemap_alternate_links): - for r, c in self._cbs: - if r.search(loc): - yield Request(loc, callback=c) - break + s = Sitemap(body) + + if s.type == "sitemapindex": + urls = list(self._get_urls_from_sitemapindex(self.sitemap_filter(s))) + return (Request(loc, callback=self._parse_sitemap) for loc in urls) + + if s.type == "urlset": + url_callback_pairs = list( + self._get_urls_and_callbacks_from_urlset(self.sitemap_filter(s)) + ) + return (Request(loc, callback=c) for loc, c in url_callback_pairs) + + logger.warning( + "Ignoring invalid sitemap: %(response)s", + {"response": response}, + extra={"spider": self}, + ) + + return () + + def _get_urls_from_sitemapindex( + self, it: Iterable[dict[str, Any]] + ) -> Iterable[str]: + for loc in iterloc(it, self.sitemap_alternate_links): + if any(x.search(loc) for x in self._follow): + yield loc + + def _get_urls_and_callbacks_from_urlset( + self, it: Iterable[dict[str, Any]] + ) -> Iterable[tuple[str, CallbackT]]: + for loc in iterloc(it, self.sitemap_alternate_links): + for r, c in self._cbs: + if r.search(loc): + yield loc, c + break def _get_sitemap_body(self, response: Response) -> bytes | None: """Return the sitemap body contained in the given response, @@ -140,8 +162,9 @@ def regex(x: re.Pattern[str] | str) -> re.Pattern[str]: def iterloc(it: Iterable[dict[str, Any]], alt: bool = False) -> Iterable[str]: for d in it: - yield d["loc"] + if loc := d["loc"]: + yield loc # Also consider alternate URLs (xhtml:link rel="alternate") - if alt and "alternate" in d: - yield from d["alternate"] + if alt and (alt_list := d.get("alternate")): + yield from alt_list diff --git a/scrapy/utils/sitemap.py b/scrapy/utils/sitemap.py index c0100ea62..042ed6281 100644 --- a/scrapy/utils/sitemap.py +++ b/scrapy/utils/sitemap.py @@ -7,11 +7,15 @@ SitemapSpider, its API is subject to change without notice. from __future__ import annotations +import warnings +from io import BytesIO, StringIO from typing import TYPE_CHECKING, Any from urllib.parse import urljoin import lxml.etree +from scrapy.exceptions import ScrapyDeprecationWarning + if TYPE_CHECKING: from collections.abc import Iterable, Iterator @@ -20,40 +24,112 @@ class Sitemap: """Class to parse Sitemap (type=urlset) and Sitemap Index (type=sitemapindex) files""" + __slots__ = ("type", "xmliter") + def __init__(self, xmltext: str | bytes): - xmlp = lxml.etree.XMLParser( - recover=True, remove_comments=True, resolve_entities=False + if isinstance(xmltext, str): + warnings.warn( + "Passing `str` type as `xmltext` is deprecated, use `bytes`", + ScrapyDeprecationWarning, + stacklevel=2, + ) + xmltext = xmltext.encode() + + self.xmliter = lxml.etree.iterparse( + BytesIO(xmltext), + recover=True, + remove_comments=True, + resolve_entities=False, + remove_blank_text=True, + collect_ids=False, + remove_pis=True, + events=("start", "end"), ) - self._root = lxml.etree.fromstring(xmltext, parser=xmlp) - rt = self._root.tag - assert isinstance(rt, str) - self.type = rt.split("}", 1)[1] if "}" in rt else rt + _, root = next(self.xmliter) + self.type = self._get_tag_name(root) def __iter__(self) -> Iterator[dict[str, Any]]: - for elem in self._root.getchildren(): - d: dict[str, Any] = {} - for el in elem.getchildren(): - tag = el.tag - assert isinstance(tag, str) - name = tag.split("}", 1)[1] if "}" in tag else tag + for event, elem in self.xmliter: + if event == "start": + continue - if name == "link": - if "href" in el.attrib: - d.setdefault("alternate", []).append(el.get("href")) - else: - d[name] = el.text.strip() if el.text else "" + if self._get_tag_name(elem) not in {"url", "sitemap"}: + continue - if "loc" in d: + if d := self._process_sitemap_element(elem): yield d + def _process_sitemap_element( + self, elem: lxml.etree._Element + ) -> dict[str, Any] | None: + d: dict[str, Any] = {} + alternate: list[str] = [] + has_loc = False + + for el in elem: + try: + tag_name = self._get_tag_name(el) + if not tag_name: + continue + + if tag_name == "link": + if href := el.get("href"): + alternate.append(href) + else: + d[tag_name] = el.text.strip() if el.text else "" + if not has_loc and tag_name == "loc": + has_loc = True + finally: + el.clear() + elem.clear() + parent = elem.getparent() + if parent is not None: + while elem.getprevious() is not None: + del parent[0] + + if not has_loc: + return None + + if alternate: + d["alternate"] = alternate + + return d + + @staticmethod + def _get_tag_name(elem: lxml.etree._Element) -> str: + if TYPE_CHECKING: + assert isinstance(elem.tag, str) + _, _, localname = elem.tag.partition("}") + return localname or elem.tag + def sitemap_urls_from_robots( - robots_text: str, base_url: str | None = None + robots_text: str | bytes, + base_url: str | None = None, ) -> Iterable[str]: - """Return an iterator over all sitemap urls contained in the given - robots.txt file - """ - for line in robots_text.splitlines(): - if line.lstrip().lower().startswith("sitemap:"): - url = line.split(":", 1)[1].strip() + if isinstance(robots_text, bytes): + for line in BytesIO(robots_text): + if line.lstrip()[:8].lower() == b"sitemap:": + try: + url = line.partition(b":")[2].strip().decode() + except UnicodeDecodeError: + continue + yield urljoin(base_url or "", url) + + else: + yield from _sitemap_urls_from_robots_str(robots_text, base_url) + + +def _sitemap_urls_from_robots_str( + robots_text: str, + base_url: str | None = None, +) -> Iterable[str]: + warnings.warn( + "Passing `str` type as `robots_text` is deprecated, use `bytes`", + ScrapyDeprecationWarning, + stacklevel=2, + ) + for line in StringIO(robots_text): + if line.lstrip()[:8].lower() == "sitemap:": + url = line.partition(":")[2].strip() yield urljoin(base_url or "", url) diff --git a/tests/test_spider_sitemap.py b/tests/test_spider_sitemap.py index d2664b19e..57c209615 100644 --- a/tests/test_spider_sitemap.py +++ b/tests/test_spider_sitemap.py @@ -1,6 +1,7 @@ from __future__ import annotations import gzip +import re import warnings from datetime import datetime from io import BytesIO @@ -43,6 +44,9 @@ class TestSitemapSpider(TestSpider): r = Response(url="http://www.example.com/favicon.ico", body=self.BODY) self.assertSitemapBody(r, None) + r = XmlResponse(url="http://www.example.com/", body=b"") + self.assertSitemapBody(r, b"") + def test_get_sitemap_body_gzip_headers(self): r = Response( url="http://www.example.com/sitemap", @@ -85,6 +89,20 @@ Sitemap: /sitemap-relative-url.xml "http://www.example.com/sitemap-relative-url.xml", ] + def test_get_sitemap_urls_from_robotstxt_skips_invalid_utf8_urls(self): + robots = ( + b"User-agent: *\n" + b"Sitemap: http://example.com/\xff.xml\n" + b"Sitemap: http://example.com/ok.xml\n" + ) + + r = TextResponse(url="http://www.example.com/robots.txt", body=robots) + spider = self.spider_class("example.com") + + assert [req.url for req in spider._parse_sitemap(r)] == [ + "http://example.com/ok.xml", + ] + def test_alternate_url_locs(self): sitemap = b""" + + http://www.example.com/english/ + http://www.example.com/portuguese/ + """ + r = TextResponse(url="http://www.example.com/sitemap.xml", body=sitemap) + + class _RuleSpider(self.spider_class): # type: ignore[name-defined,misc] + sitemap_rules = [(rule, "parse")] + + spider = _RuleSpider("example.com") + urls = [req.url for req in spider._parse_sitemap(r)] + assert urls == result + + def test_parse_sitemap_empty_body(self): + r = XmlResponse(url="http://www.example.com/sitemap.xml", body=b"") + spider = self.spider_class("example.com") + + with LogCapture() as lc: + results = list(spider._parse_sitemap(r)) + + assert not results + + lc.check( + ( + "scrapy.spiders.sitemap", + "WARNING", + "Ignoring invalid sitemap: <200 http://www.example.com/sitemap.xml>", + ) + ) + + def test_parse_sitemap_not_sitemap(self): + body = b""" + + sometext + sometext2 + """ + r = XmlResponse(url="http://www.example.com/random.xml", body=body) + spider = self.spider_class("example.com") + + results = list(spider._parse_sitemap(r)) + + assert not results + + @pytest.mark.parametrize( + ("follow", "result"), + [ + (r"1.xml", ["http://www.example.com/sitemap1.xml"]), + (re.compile(r"sitemap\d"), ["http://www.example.com/sitemap1.xml"]), + (r"nonexistent", []), + ], + ) + def test_sitemap_follow(self, follow, result): + sitemap = b""" + + + http://www.example.com/sitemap1.xml + + """ + r = TextResponse(url="http://www.example.com/sitemap.xml", body=sitemap) + + class _FollowSpider(self.spider_class): + sitemap_follow = [follow] + + spider = _FollowSpider("example.com") + urls = [req.url for req in spider._parse_sitemap(r)] + assert urls == result + def test_compression_bomb_setting(self): settings = {"DOWNLOAD_MAXSIZE": 10_000_000} crawler = get_crawler(settings_dict=settings) diff --git a/tests/test_utils_sitemap.py b/tests/test_utils_sitemap.py index 464a31777..f2bcd7541 100644 --- a/tests/test_utils_sitemap.py +++ b/tests/test_utils_sitemap.py @@ -1,3 +1,5 @@ +import warnings + from scrapy.utils.sitemap import Sitemap, sitemap_urls_from_robots @@ -156,7 +158,7 @@ def test_sitemap_wrong_ns2(): def test_sitemap_urls_from_robots(): - robots = """User-agent: * + robots = b"""User-agent: * Disallow: /aff/ Disallow: /wl/ @@ -182,6 +184,40 @@ Disallow: /forum/active/ ] +def test_sitemap_urls_from_robots_str_compat(): + robots = """User-agent: * +Disallow: /aff/ +Disallow: /wl/ + +# Search and shopping refining +Disallow: /s*/*facet +Disallow: /s*/*tags + +# Sitemap files +Sitemap: http://example.com/sitemap.xml +Sitemap: http://example.com/sitemap-product-index.xml +Sitemap: HTTP://example.com/sitemap-uppercase.xml +Sitemap: /sitemap-relative-url.xml + +# Forums +Disallow: /forum/search/ +Disallow: /forum/active/ +""" + + with warnings.catch_warnings(record=True) as w: + assert list( + sitemap_urls_from_robots(robots, base_url="http://example.com") + ) == [ + "http://example.com/sitemap.xml", + "http://example.com/sitemap-product-index.xml", + "http://example.com/sitemap-uppercase.xml", + "http://example.com/sitemap-relative-url.xml", + ] + assert "Passing `str` type as `robots_text` is deprecated, use `bytes`" in str( + w[0].message + ) + + def test_sitemap_blanklines(): """Assert we can deal with starting blank lines before tag""" s = Sitemap(