from __future__ import annotations
import logging
from typing import TYPE_CHECKING, Any, Self, cast
from scrapy import signals
from scrapy.exceptions import IgnoreRequest, NotConfigured
from scrapy.http import HtmlResponse, Response
from scrapy.spidermiddlewares.referer import RefererMiddleware
from scrapy.utils.decorators import _warn_spider_arg
from scrapy.utils.httpobj import urlparse_cached
from scrapy.utils.python import global_object_name
from scrapy.utils.response import _body_size, get_meta_refresh
from scrapy.utils.url import _redirect_url
if TYPE_CHECKING:
from scrapy import Request, Spider
from scrapy.crawler import Crawler
from scrapy.settings import BaseSettings
logger = logging.getLogger(__name__)
# Status codes for which the HTTP standard defines redirection handling. Other
# status codes from REDIRECT_HTTP_CODES get conservative handling, see
# RedirectMiddleware.
_STANDARD_HTTP_CODES = frozenset({301, 302, 303, 307, 308})
class BaseRedirectMiddleware:
crawler: Crawler
enabled_setting: str = "REDIRECT_ENABLED"
def __init__(self, settings: BaseSettings):
if not settings.getbool(self.enabled_setting):
raise NotConfigured
self.max_redirect_times: int = settings.getint("REDIRECT_MAX_TIMES")
self.priority_adjust: int = settings.getint("REDIRECT_PRIORITY_ADJUST")
self._referer_spider_middleware: RefererMiddleware | None = None
@classmethod
def from_crawler(cls, crawler: Crawler) -> Self:
o = cls(crawler.settings)
o.crawler = crawler
crawler.signals.connect(o._spider_opened, signal=signals.spider_opened)
return o
def handle_referer(self, request: Request, response: Response) -> None:
"""Remove, modify or keep the Referer header of *request* based on the
*response* that triggered *request*.
By default, this method finds a run-time instance of
scrapy.spidermiddlewares.referer.RefererMiddleware (or of a subclass)
and uses it to set the right Referer header.
Override this method if you use a different Scrapy component to handle
Referer headers, of if you want to use a custom logic to set the
Referer header on redirects.
"""
request.headers.pop("Referer", None)
if not self._referer_spider_middleware:
return
self._referer_spider_middleware.get_processed_request(request, response)
def _spider_opened(self) -> None:
self._referer_spider_middleware = self.crawler.get_spider_middleware(
RefererMiddleware
)
if self._referer_spider_middleware:
return
redirect_cls = global_object_name(self.__class__)
referer_cls = global_object_name(RefererMiddleware)
if self.__class__ in {RedirectMiddleware, MetaRefreshMiddleware}:
replacement = (
f"replace {redirect_cls} with a subclass that overrides the "
f"handle_referer() method"
)
else:
replacement = (
f"or edit {redirect_cls} (if defined in your code base) to "
f"override the handle_referer() method, or replace "
f"{redirect_cls} with a subclass that overrides the "
f"handle_referer() method."
)
logger.warning(
f"{redirect_cls} found no {referer_cls} instance to handle "
f"Referer header handling, so the Referer header will be removed "
f"on redirects. To set a Referer header on redirects, enable "
f"{referer_cls} (or a subclass), or {replacement}.",
)
def _redirect(
self, redirected: Request, request: Request, reason: str | int
) -> Request:
ttl = request.meta.setdefault("redirect_ttl", self.max_redirect_times)
redirects = request.meta.get("redirect_times", 0) + 1
if ttl and redirects <= self.max_redirect_times:
redirected.meta["redirect_times"] = redirects
redirected.meta["redirect_ttl"] = ttl - 1
redirected.meta["redirect_urls"] = [
*request.meta.get("redirect_urls", []),
request.url,
]
redirected.meta["redirect_reasons"] = [
*request.meta.get("redirect_reasons", []),
reason,
]
redirected.dont_filter = request.dont_filter
redirected.priority = request.priority + self.priority_adjust
logger.debug(
"Redirecting (%(reason)s) to %(redirected)s from %(request)s",
{"reason": reason, "redirected": redirected, "request": request},
extra={"spider": self.crawler.spider},
)
return redirected
logger.debug(
"Discarding %(request)s: max redirections reached",
{"request": request},
extra={"spider": self.crawler.spider},
)
raise IgnoreRequest("max redirections reached")
def _build_redirect_request(
self, source_request: Request, response: Response, *, url: str, **kwargs: Any
) -> Request:
redirect_request = source_request.replace(
url=url,
**kwargs,
cls=None,
cookies=None,
)
if "_scheme_proxy" in redirect_request.meta:
source_request_scheme = urlparse_cached(source_request).scheme
redirect_request_scheme = urlparse_cached(redirect_request).scheme
if source_request_scheme != redirect_request_scheme:
redirect_request.meta.pop("_scheme_proxy")
redirect_request.meta.pop("proxy", None)
redirect_request.meta.pop("_auth_proxy", None)
redirect_request.headers.pop(b"Proxy-Authorization", None)
has_cookie_header = "Cookie" in redirect_request.headers
has_authorization_header = "Authorization" in redirect_request.headers
if has_cookie_header or has_authorization_header:
default_ports = {"http": 80, "https": 443}
parsed_source_request = urlparse_cached(source_request)
source_scheme, source_host, source_port = (
parsed_source_request.scheme,
parsed_source_request.hostname,
parsed_source_request.port
or default_ports.get(parsed_source_request.scheme),
)
parsed_redirect_request = urlparse_cached(redirect_request)
redirect_scheme, redirect_host, redirect_port = (
parsed_redirect_request.scheme,
parsed_redirect_request.hostname,
parsed_redirect_request.port
or default_ports.get(parsed_redirect_request.scheme),
)
if has_cookie_header and (
redirect_scheme not in {source_scheme, "https"}
or source_host != redirect_host
):
del redirect_request.headers["Cookie"]
# https://fetch.spec.whatwg.org/#ref-for-cors-non-wildcard-request-header-name
if has_authorization_header and (
source_scheme != redirect_scheme
or source_host != redirect_host
or source_port != redirect_port
):
del redirect_request.headers["Authorization"]
self.handle_referer(redirect_request, response)
return redirect_request
def _redirect_request_using_get(
self, request: Request, response: Response, redirect_url: str
) -> Request:
redirect_request = self._build_redirect_request(
request,
response,
url=redirect_url,
method="GET",
body="",
)
redirect_request.headers.pop("Content-Type", None)
redirect_request.headers.pop("Content-Length", None)
redirect_request.headers.pop("Content-Encoding", None)
redirect_request.headers.pop("Content-Language", None)
redirect_request.headers.pop("Content-Location", None)
return redirect_request
[docs]
class RedirectMiddleware(BaseRedirectMiddleware):
"""Handle redirection of requests based on response status."""
def __init__(self, settings: BaseSettings):
super().__init__(settings)
self.redirect_http_codes: set[int] = {
int(x) for x in settings.getlist("REDIRECT_HTTP_CODES")
}
def _is_redirect(self, request: Request, response: Response) -> bool:
meta_http_codes = request.meta.get("redirect_http_codes")
http_codes = (
{int(x) for x in meta_http_codes}
if meta_http_codes is not None
else self.redirect_http_codes
)
if response.status not in http_codes:
return False
if response.status in _STANDARD_HTTP_CODES:
return True
# For other status codes, e.g. 201, Location identifies a resource
# rather than a redirection target, and the response body may be that
# resource, which following Location would discard.
return not _body_size(response)
@_warn_spider_arg
def process_response(
self, request: Request, response: Response, spider: Spider | None = None
) -> Request | Response:
if (
request.meta.get("dont_redirect", False)
or response.status
in getattr(self.crawler.spider, "handle_httpstatus_list", ())
or response.status in request.meta.get("handle_httpstatus_list", ())
or request.meta.get("handle_httpstatus_all", False)
):
return response
if "Location" not in response.headers or not self._is_redirect(
request, response
):
return response
assert response.headers["Location"] is not None
redirected_url = _redirect_url(request.url, response.headers["Location"])
redirected = self._build_redirect_request(request, response, url=redirected_url)
if urlparse_cached(redirected).scheme not in {"http", "https"}:
return response
# 307 and 308 responses keep the method and the body of the original
# request as is, and so do 301 and 302 responses, except for POST
# requests, for historical reasons. Any other status code gets the
# handling of 303 responses, a bodyless GET request, which is the safest
# choice for status codes with no defined redirection semantics.
if (response.status in {301, 302} and request.method == "POST") or (
response.status not in {301, 302, 307, 308}
and request.method not in {"GET", "HEAD"}
):
redirected = self._redirect_request_using_get(
request, response, redirected_url
)
return self._redirect(redirected, request, response.status)