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

Remove deprecated offsite spider middleware. (#6926) 

* Remove deprecated offsite spider middleware.

* Add a deprecation notice for the offsite spider middleware.
This commit is contained in:
Andrey Rakhmatullin 2025-06-29 02:35:20 +05:00 committed by GitHub
parent 3019393686
commit 2c1c10e923
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 11 additions and 225 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
docs/news.rst
View File

@ -1233,6 +1233,17 @@ Security bug fixes
.. _defusedxml: https://github.com/tiran/defusedxml
Deprecations
~~~~~~~~~~~~
- ``scrapy.spidermiddlewares.offsite.OffsiteMiddleware`` (a spider
middleware) is now deprecated and not enabled by default. The new
downloader middleware with the same functionality,
:class:`scrapy.downloadermiddlewares.offsite.OffsiteMiddleware`, is enabled
instead.
(:issue:`2241`, :issue:`6358`)
Bug fixes
~~~~~~~~~

120
scrapy/spidermiddlewares/offsite.py
View File

@ -1,120 +0,0 @@
"""
Offsite Spider Middleware
See documentation in docs/topics/spider-middleware.rst
"""
from __future__ import annotations
import logging
import re
import warnings
from typing import TYPE_CHECKING
from scrapy import Spider, signals
from scrapy.exceptions import ScrapyDeprecationWarning
from scrapy.spidermiddlewares.base import BaseSpiderMiddleware
from scrapy.utils.httpobj import urlparse_cached
warnings.warn(
"The scrapy.spidermiddlewares.offsite module is deprecated, use "
"scrapy.downloadermiddlewares.offsite instead.",
ScrapyDeprecationWarning,
)
if TYPE_CHECKING:
# typing.Self requires Python 3.11
from typing_extensions import Self
from scrapy.crawler import Crawler
from scrapy.http import Request, Response
from scrapy.statscollectors import StatsCollector
logger = logging.getLogger(__name__)
class OffsiteMiddleware(BaseSpiderMiddleware):
crawler: Crawler
def __init__(self, stats: StatsCollector): # pylint: disable=super-init-not-called
self.stats: StatsCollector = stats
@classmethod
def from_crawler(cls, crawler: Crawler) -> Self:
assert crawler.stats
o = cls(crawler.stats)
o.crawler = crawler
crawler.signals.connect(o.spider_opened, signal=signals.spider_opened)
return o
def get_processed_request(
self, request: Request, response: Response | None
) -> Request | None:
if response is None:
# skip start requests for backward compatibility
return request
assert self.crawler.spider
if (
request.dont_filter
or request.meta.get("allow_offsite")
or self.should_follow(request, self.crawler.spider)
):
return request
domain = urlparse_cached(request).hostname
if domain and domain not in self.domains_seen:
self.domains_seen.add(domain)
logger.debug(
"Filtered offsite request to %(domain)r: %(request)s",
{"domain": domain, "request": request},
extra={"spider": self.crawler.spider},
)
self.stats.inc_value("offsite/domains", spider=self.crawler.spider)
self.stats.inc_value("offsite/filtered", spider=self.crawler.spider)
return None
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))
def get_host_regex(self, spider: Spider) -> re.Pattern[str]:
"""Override this method to implement a different offsite policy"""
allowed_domains = getattr(spider, "allowed_domains", None)
if not allowed_domains:
return re.compile("") # allow all by default
url_pattern = re.compile(r"^https?://.*$")
port_pattern = re.compile(r":\d+$")
domains = []
for domain in allowed_domains:
if domain is None:
continue
if url_pattern.match(domain):
message = (
"allowed_domains accepts only domains, not URLs. "
f"Ignoring URL entry {domain} in allowed_domains."
)
warnings.warn(message, URLWarning)
elif port_pattern.search(domain):
message = (
"allowed_domains accepts only domains without ports. "
f"Ignoring entry {domain} in allowed_domains."
)
warnings.warn(message, PortWarning)
else:
domains.append(re.escape(domain))
regex = rf"^(.*\.)?({'|'.join(domains)})$"
return re.compile(regex)
def spider_opened(self, spider: Spider) -> None:
self.host_regex: re.Pattern[str] = self.get_host_regex(spider)
self.domains_seen: set[str] = set()
class URLWarning(Warning):
pass
class PortWarning(Warning):
pass

105
tests/test_spidermiddleware_offsite.py
View File

@ -1,105 +0,0 @@
import warnings
from urllib.parse import urlparse
from scrapy.http import Request, Response
from scrapy.spidermiddlewares.offsite import OffsiteMiddleware, PortWarning, URLWarning
from scrapy.spiders import Spider
from scrapy.utils.test import get_crawler
class TestOffsiteMiddleware:
def setup_method(self):
crawler = get_crawler(Spider)
self.spider = crawler.spider = crawler._create_spider(**self._get_spiderargs())
self.mw = OffsiteMiddleware.from_crawler(crawler)
self.mw.spider_opened(self.spider)
def _get_spiderargs(self):
return {
"name": "foo",
"allowed_domains": ["scrapytest.org", "scrapy.org", "scrapy.test.org"],
}
def test_process_spider_output(self):
res = Response("http://scrapytest.org")
onsite_reqs = [
Request("http://scrapytest.org/1"),
Request("http://scrapy.org/1"),
Request("http://sub.scrapy.org/1"),
Request("http://offsite.tld/letmepass", dont_filter=True),
Request("http://offsite-2.tld/allow", meta={"allow_offsite": True}),
Request("http://scrapy.test.org/"),
Request("http://scrapy.test.org:8000/"),
]
offsite_reqs = [
Request("http://scrapy2.org"),
Request("http://offsite.tld/"),
Request("http://offsite.tld/scrapytest.org"),
Request("http://offsite.tld/rogue.scrapytest.org"),
Request("http://rogue.scrapytest.org.haha.com"),
Request("http://roguescrapytest.org"),
Request("http://test.org/"),
Request("http://notscrapy.test.org/"),
]
reqs = onsite_reqs + offsite_reqs
out = list(self.mw.process_spider_output(res, reqs, self.spider))
assert out == onsite_reqs
class TestOffsiteMiddleware2(TestOffsiteMiddleware):
def _get_spiderargs(self):
return {"name": "foo", "allowed_domains": None}
def test_process_spider_output(self):
res = Response("http://scrapytest.org")
reqs = [Request("http://a.com/b.html"), Request("http://b.com/1")]
out = list(self.mw.process_spider_output(res, reqs, self.spider))
assert out == reqs
class TestOffsiteMiddleware3(TestOffsiteMiddleware2):
def _get_spiderargs(self):
return {"name": "foo"}
class TestOffsiteMiddleware4(TestOffsiteMiddleware3):
def _get_spiderargs(self):
bad_hostname = urlparse("http:////scrapytest.org").hostname
return {
"name": "foo",
"allowed_domains": ["scrapytest.org", None, bad_hostname],
}
def test_process_spider_output(self):
res = Response("http://scrapytest.org")
reqs = [Request("http://scrapytest.org/1")]
out = list(self.mw.process_spider_output(res, reqs, self.spider))
assert out == reqs
class TestOffsiteMiddleware5(TestOffsiteMiddleware4):
def test_get_host_regex(self):
self.spider.allowed_domains = [
"http://scrapytest.org",
"scrapy.org",
"scrapy.test.org",
]
with warnings.catch_warnings(record=True) as w:
warnings.simplefilter("always")
self.mw.get_host_regex(self.spider)
assert issubclass(w[-1].category, URLWarning)
class TestOffsiteMiddleware6(TestOffsiteMiddleware4):
def test_get_host_regex(self):
self.spider.allowed_domains = [
"scrapytest.org:8000",
"scrapy.org",
"scrapy.test.org",
]
with warnings.catch_warnings(record=True) as w:
warnings.simplefilter("always")
self.mw.get_host_regex(self.spider)
assert issubclass(w[-1].category, PortWarning)