1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
|
"""
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
|
