RomanNum3ral
/
scrapy
mirror of https://github.com/scrapy/scrapy.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge b8cedd8d27 into b78ab3d6c8

This commit is contained in:
Diogo Castro 2026-06-23 13:32:10 +00:00 committed by GitHub
parent b78ab3d6c8 b8cedd8d27
commit 7a0cae68b1
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 394 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
docs/topics/downloader-middleware.rst
View File

@ -856,6 +856,21 @@ OffsiteMiddleware
:attr:`~scrapy.Spider.allowed_domains` attribute, or the
attribute is empty, the offsite middleware will allow all requests.
If the spider defines a :attr:`~scrapy.Spider.disallowed_domains`
attribute, any request whose host name matches one of the domains in
that list (or their subdomains) will be filtered out, regardless of
:attr:`~scrapy.Spider.allowed_domains`. This is useful when you want
to allow most domains but block a few specific ones.
If a request matches both :attr:`~scrapy.Spider.allowed_domains` and
:attr:`~scrapy.Spider.disallowed_domains`, it will be filtered out
(i.e. :attr:`~scrapy.Spider.disallowed_domains` takes precedence).
Both :attr:`~scrapy.Spider.allowed_domains` and
:attr:`~scrapy.Spider.disallowed_domains` must contain valid domain
names only (not URLs or domains with ports). Invalid entries will cause
the spider to close with reason ``invalid_domain_configuration``.
.. reqmeta:: allow_offsite
If the request has the :attr:`~scrapy.Request.dont_filter` attribute set to

12
docs/topics/spiders.rst
View File

@ -74,6 +74,18 @@ scrapy.Spider
Let's say your target url is ``https://www.example.com/1.html``,
then add ``'example.com'`` to the list.
.. attribute:: disallowed_domains
An optional list of strings containing domains that this spider is
not allowed to crawl. Requests for URLs belonging to the domain names
specified in this list (or their subdomains) will be filtered out when
:class:`~scrapy.downloadermiddlewares.offsite.OffsiteMiddleware` is
enabled.
This is useful when you want to allow all domains except a few
specific ones. For example, to block requests to ``ads.example.com``,
add ``'ads.example.com'`` to the list.
.. autoattribute:: start_urls
.. attribute:: custom_settings

85
scrapy/downloadermiddlewares/offsite.py
View File

@ -7,6 +7,7 @@ from typing import TYPE_CHECKING
from scrapy import Request, Spider, signals
from scrapy.exceptions import IgnoreRequest
from scrapy.utils.decorators import _warn_spider_arg
from scrapy.utils.defer import _schedule_coro
from scrapy.utils.httpobj import urlparse_cached
if TYPE_CHECKING:
@ -37,7 +38,23 @@ class OffsiteMiddleware:
return o
def spider_opened(self, spider: Spider) -> None:
self.host_regex: re.Pattern[str] = self.get_host_regex(spider)
try:
self.host_regex: re.Pattern[str] = self.get_host_regex(spider)
self.disallowed_host_regex: re.Pattern[str] | None = (
self._get_disallowed_host_regex(spider)
)
except ValueError as exc:
logger.error(
"Invalid domain configuration: %(error)s",
{"error": exc},
extra={"spider": spider},
)
assert self.crawler.engine
_schedule_coro(
self.crawler.engine.close_spider_async(
reason="invalid_domain_configuration"
)
)
def request_scheduled(self, request: Request, spider: Spider) -> None:
self.process_request(request)
@ -64,20 +81,66 @@ class OffsiteMiddleware:
raise IgnoreRequest
def should_follow(self, request: Request, spider: Spider) -> bool:
regex = self.host_regex
# hostname can be None for wrong urls (like javascript links)
host = urlparse_cached(request).hostname or ""
return bool(regex.search(host))
if self.disallowed_host_regex and self.disallowed_host_regex.search(host):
return False
return bool(self.host_regex.search(host))
@staticmethod
def _process_domains(
domains_list: list[str | None],
domains_type: str,
) -> list[str]:
"""Process a domains list and return a list of valid, regex-escaped domains.
Raises ``ValueError`` on invalid entries (``None``, URLs, or
domains with ports) so that the spider fails fast on
misconfigured domains.
"""
url_pattern = re.compile(r"^https?://.*$")
protocol_relative_pattern = re.compile(r"^//")
port_pattern = re.compile(r":\d+$")
valid_domains: list[str] = []
for domain in domains_list:
if domain is None:
raise ValueError(f"{domains_type} contains empty value.")
if url_pattern.match(domain) or protocol_relative_pattern.match(domain):
raise ValueError(
f"{domains_type} accepts only domains, not URLs. "
f"Got URL entry {domain} in {domains_type}."
)
if port_pattern.search(domain):
raise ValueError(
f"{domains_type} accepts only domains without ports. "
f"Got entry {domain} in {domains_type}."
)
valid_domains.append(re.escape(domain))
return valid_domains
def get_host_regex(self, spider: Spider) -> re.Pattern[str]:
"""Override this method to implement a different offsite policy"""
"""Override this method to implement a different offsite policy.
Returns a compiled regular expression object that matches the hosts
that are allowed to be crawled.
"""
allowed_domains = getattr(spider, "allowed_domains", None)
if not allowed_domains:
return re.compile("") # allow all by default
domains = []
for domain in allowed_domains:
if domain is None:
continue
domains.append(re.escape(domain))
regex = rf"^(.*\.)?({'|'.join(domains)})$"
return re.compile(regex)
domains = self._process_domains(allowed_domains, "allowed_domains")
return re.compile(rf"^(.*\.)?({'|'.join(domains)})$")
def _get_disallowed_host_regex(self, spider: Spider) -> re.Pattern[str] | None:
"""Build a regex that positively matches disallowed hosts.
Returns ``None`` when there are no disallowed domains, meaning
nothing should be blocked via this mechanism.
"""
disallowed_domains = getattr(spider, "disallowed_domains", None)
if not disallowed_domains:
return None
domains = self._process_domains(disallowed_domains, "disallowed_domains")
return re.compile(rf"^(.*\.)?({'|'.join(domains)})$")

307
tests/test_downloadermiddleware_offsite.py
View File

@ -1,3 +1,6 @@
import logging
from unittest.mock import AsyncMock, patch
import pytest
from scrapy import Request, Spider
@ -113,18 +116,43 @@ def test_process_request_no_allowed_domains(value):
assert mw.process_request(request) is None
def test_process_request_invalid_domains():
@pytest.mark.parametrize(
"allowed_domains",
[
["a.example", None],
["a.example", "http://b.example"],
["a.example", "//c.example"],
["a.example", "c.example:8080"],
],
)
def test_process_request_invalid_domains(allowed_domains, caplog):
crawler = get_crawler(Spider)
allowed_domains = ["a.example", None, "http:////b.example", "//c.example"]
crawler.spider = crawler._create_spider(name="a", allowed_domains=allowed_domains)
mw = OffsiteMiddleware.from_crawler(crawler)
mw.spider_opened(crawler.spider)
request = Request("https://a.example")
assert mw.process_request(request) is None
for letter in ("b", "c"):
request = Request(f"https://{letter}.example")
with pytest.raises(IgnoreRequest):
mw.process_request(request)
crawler.engine = AsyncMock()
with (
patch("scrapy.downloadermiddlewares.offsite._schedule_coro"),
caplog.at_level(logging.ERROR),
):
mw.spider_opened(crawler.spider)
assert "Invalid domain configuration" in caplog.text
def test_invalid_domains_closes_spider(caplog):
crawler = get_crawler(Spider)
crawler.spider = crawler._create_spider(
name="a", allowed_domains=["a.example", None]
)
mw = OffsiteMiddleware.from_crawler(crawler)
mock_engine = AsyncMock()
crawler.engine = mock_engine
with (
patch("scrapy.downloadermiddlewares.offsite._schedule_coro") as mock_schedule,
caplog.at_level(logging.ERROR),
):
mw.spider_opened(crawler.spider)
assert "Invalid domain configuration" in caplog.text
mock_schedule.assert_called_once()
@pytest.mark.parametrize(
@ -201,15 +229,266 @@ def test_request_scheduled_no_allowed_domains(value):
assert mw.request_scheduled(request, crawler.spider) is None
def test_request_scheduled_invalid_domains():
@pytest.mark.parametrize(
"allowed_domains",
[
["a.example", None],
["a.example", "http://b.example"],
["a.example", "//c.example"],
["a.example", "c.example:8080"],
],
)
def test_request_scheduled_invalid_domains(allowed_domains, caplog):
crawler = get_crawler(Spider)
allowed_domains = ["a.example", None, "http:////b.example", "//c.example"]
crawler.spider = crawler._create_spider(name="a", allowed_domains=allowed_domains)
mw = OffsiteMiddleware.from_crawler(crawler)
crawler.engine = AsyncMock()
with (
patch("scrapy.downloadermiddlewares.offsite._schedule_coro"),
caplog.at_level(logging.ERROR),
):
mw.spider_opened(crawler.spider)
assert "Invalid domain configuration" in caplog.text
@pytest.mark.parametrize(
("value", "filtered"),
[
(UNSET, True),
(None, True),
(False, True),
(True, False),
],
)
def test_process_request_disallowed_dont_filter(value, filtered):
crawler = get_crawler(Spider)
crawler.spider = crawler._create_spider(name="a", disallowed_domains=["a.example"])
mw = OffsiteMiddleware.from_crawler(crawler)
mw.spider_opened(crawler.spider)
request = Request("https://a.example")
kwargs = {}
if value is not UNSET:
kwargs["dont_filter"] = value
request = Request("https://a.example", **kwargs)
if filtered:
with pytest.raises(IgnoreRequest):
mw.process_request(request)
else:
assert mw.process_request(request) is None
request2 = Request("https://b.example")
assert mw.process_request(request2) is None
@pytest.mark.parametrize(
("allow_offsite", "dont_filter", "filtered"),
[
(True, UNSET, False),
(True, None, False),
(True, False, False),
(True, True, False),
(False, UNSET, True),
(False, None, True),
(False, False, True),
(False, True, False),
],
)
def test_process_request_disallowed_allow_offsite(allow_offsite, dont_filter, filtered):
crawler = get_crawler(Spider)
crawler.spider = crawler._create_spider(name="a", disallowed_domains=["a.example"])
mw = OffsiteMiddleware.from_crawler(crawler)
mw.spider_opened(crawler.spider)
kwargs = {"meta": {}}
if allow_offsite is not UNSET:
kwargs["meta"]["allow_offsite"] = allow_offsite
if dont_filter is not UNSET:
kwargs["dont_filter"] = dont_filter
request = Request("https://a.example", **kwargs)
if filtered:
with pytest.raises(IgnoreRequest):
mw.process_request(request)
else:
assert mw.process_request(request) is None
@pytest.mark.parametrize(
"value",
[
UNSET,
None,
[],
],
)
def test_process_request_no_disallowed_domains(value):
crawler = get_crawler(Spider)
kwargs = {}
if value is not UNSET:
kwargs["disallowed_domains"] = value
crawler.spider = crawler._create_spider(name="a", **kwargs)
mw = OffsiteMiddleware.from_crawler(crawler)
mw.spider_opened(crawler.spider)
request = Request("https://example.com")
assert mw.process_request(request) is None
@pytest.mark.parametrize(
"disallowed_domains",
[
["a.example", None],
["a.example", "http:////b.example"],
["a.example", "//c.example:8080"],
],
)
def test_process_request_invalid_disallowed_domains(disallowed_domains, caplog):
crawler = get_crawler(Spider)
crawler.spider = crawler._create_spider(
name="a", disallowed_domains=disallowed_domains
)
mw = OffsiteMiddleware.from_crawler(crawler)
crawler.engine = AsyncMock()
with (
patch("scrapy.downloadermiddlewares.offsite._schedule_coro"),
caplog.at_level(logging.ERROR),
):
mw.spider_opened(crawler.spider)
assert "Invalid domain configuration" in caplog.text
@pytest.mark.parametrize(
"value",
[
UNSET,
None,
[],
],
)
def test_request_scheduled_no_disallowed_domains(value):
crawler = get_crawler(Spider)
kwargs = {}
if value is not UNSET:
kwargs["disallowed_domains"] = value
crawler.spider = crawler._create_spider(name="a", **kwargs)
mw = OffsiteMiddleware.from_crawler(crawler)
mw.spider_opened(crawler.spider)
request = Request("https://example.com")
assert mw.request_scheduled(request, crawler.spider) is None
for letter in ("b", "c"):
request = Request(f"https://{letter}.example")
@pytest.mark.parametrize(
("value", "filtered"),
[
(UNSET, True),
(None, True),
(False, True),
(True, False),
],
)
def test_request_scheduled_disallowed_dont_filter(value, filtered):
crawler = get_crawler(Spider)
crawler.spider = crawler._create_spider(name="a", disallowed_domains=["a.example"])
mw = OffsiteMiddleware.from_crawler(crawler)
mw.spider_opened(crawler.spider)
kwargs = {}
if value is not UNSET:
kwargs["dont_filter"] = value
request = Request("https://a.example", **kwargs)
if filtered:
with pytest.raises(IgnoreRequest):
mw.request_scheduled(request, crawler.spider)
else:
assert mw.request_scheduled(request, crawler.spider) is None
request2 = Request("https://b.example")
assert mw.request_scheduled(request2, crawler.spider) is None
@pytest.mark.parametrize(
("allow_offsite", "dont_filter", "filtered"),
[
(True, UNSET, False),
(True, None, False),
(True, False, False),
(True, True, False),
(False, UNSET, True),
(False, None, True),
(False, False, True),
(False, True, False),
],
)
def test_request_scheduled_disallowed_allow_offsite(
allow_offsite, dont_filter, filtered
):
crawler = get_crawler(Spider)
crawler.spider = crawler._create_spider(name="a", disallowed_domains=["a.example"])
mw = OffsiteMiddleware.from_crawler(crawler)
mw.spider_opened(crawler.spider)
kwargs = {"meta": {}}
if allow_offsite is not UNSET:
kwargs["meta"]["allow_offsite"] = allow_offsite
if dont_filter is not UNSET:
kwargs["dont_filter"] = dont_filter
request = Request("https://a.example", **kwargs)
if filtered:
with pytest.raises(IgnoreRequest):
mw.request_scheduled(request, crawler.spider)
else:
assert mw.request_scheduled(request, crawler.spider) is None
@pytest.mark.parametrize(
"disallowed_domains",
[
["a.example", None],
["a.example", "http:////b.example"],
["a.example", "//c.example:8080"],
],
)
def test_request_scheduled_invalid_disallowed_domains(disallowed_domains, caplog):
crawler = get_crawler(Spider)
crawler.spider = crawler._create_spider(
name="a", disallowed_domains=disallowed_domains
)
mw = OffsiteMiddleware.from_crawler(crawler)
crawler.engine = AsyncMock()
with (
patch("scrapy.downloadermiddlewares.offsite._schedule_coro"),
caplog.at_level(logging.ERROR),
):
mw.spider_opened(crawler.spider)
assert "Invalid domain configuration" in caplog.text
@pytest.mark.parametrize(
("url", "filtered"),
[
("http://example.com/page", False),
("http://sub.example.com/page", False),
("http://ads.example.com/page", True),
("http://sub.ads.example.com/page", True),
("http://other.com/page", True),
],
)
def test_process_request_allowed_and_disallowed_domains(url, filtered):
crawler = get_crawler(Spider)
crawler.spider = crawler._create_spider(
name="a",
allowed_domains=["example.com"],
disallowed_domains=["ads.example.com"],
)
mw = OffsiteMiddleware.from_crawler(crawler)
mw.spider_opened(crawler.spider)
request = Request(url)
if filtered:
with pytest.raises(IgnoreRequest):
mw.process_request(request)
else:
assert mw.process_request(request) is None