Source code for scrapy.downloadermiddlewares.redirect

from __future__ import annotations

import logging
from typing import TYPE_CHECKING, Any, List, Union, cast
from urllib.parse import urljoin

from w3lib.url import safe_url_string

from scrapy import Request, Spider
from scrapy.crawler import Crawler
from scrapy.exceptions import IgnoreRequest, NotConfigured
from scrapy.http import HtmlResponse, Response
from scrapy.settings import BaseSettings
from scrapy.utils.httpobj import urlparse_cached
from scrapy.utils.response import get_meta_refresh

if TYPE_CHECKING:
    # typing.Self requires Python 3.11
    from typing_extensions import Self

logger = logging.getLogger(__name__)


def _build_redirect_request(
    source_request: Request, *, 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"]

    return redirect_request


class BaseRedirectMiddleware:
    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")

    @classmethod
    def from_crawler(cls, crawler: Crawler) -> Self:
        return cls(crawler.settings)

    def _redirect(
        self, redirected: Request, request: Request, spider: Spider, reason: Any
    ) -> 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": spider},
            )
            return redirected
        logger.debug(
            "Discarding %(request)s: max redirections reached",
            {"request": request},
            extra={"spider": spider},
        )
        raise IgnoreRequest("max redirections reached")

    def _redirect_request_using_get(
        self, request: Request, redirect_url: str
    ) -> Request:
        redirect_request = _build_redirect_request(
            request,
            url=redirect_url,
            method="GET",
            body="",
        )
        redirect_request.headers.pop("Content-Type", None)
        redirect_request.headers.pop("Content-Length", None)
        return redirect_request


[docs]class RedirectMiddleware(BaseRedirectMiddleware): """ Handle redirection of requests based on response status and meta-refresh html tag. """ def process_response( self, request: Request, response: Response, spider: Spider ) -> Union[Request, Response]: if ( request.meta.get("dont_redirect", False) or response.status in getattr(spider, "handle_httpstatus_list", []) or response.status in request.meta.get("handle_httpstatus_list", []) or request.meta.get("handle_httpstatus_all", False) ): return response allowed_status = (301, 302, 303, 307, 308) if "Location" not in response.headers or response.status not in allowed_status: return response assert response.headers["Location"] is not None location = safe_url_string(response.headers["Location"]) if response.headers["Location"].startswith(b"//"): request_scheme = urlparse_cached(request).scheme location = request_scheme + "://" + location.lstrip("/") redirected_url = urljoin(request.url, location) redirected = _build_redirect_request(request, url=redirected_url) if urlparse_cached(redirected).scheme not in {"http", "https"}: return response if response.status in (301, 307, 308) or request.method == "HEAD": return self._redirect(redirected, request, spider, response.status) redirected = self._redirect_request_using_get(request, redirected_url) return self._redirect(redirected, request, spider, response.status)
[docs]class MetaRefreshMiddleware(BaseRedirectMiddleware): enabled_setting = "METAREFRESH_ENABLED" def __init__(self, settings: BaseSettings): super().__init__(settings) self._ignore_tags: List[str] = settings.getlist("METAREFRESH_IGNORE_TAGS") self._maxdelay: int = settings.getint("METAREFRESH_MAXDELAY") def process_response( self, request: Request, response: Response, spider: Spider ) -> Union[Request, Response]: if ( request.meta.get("dont_redirect", False) or request.method == "HEAD" or not isinstance(response, HtmlResponse) or urlparse_cached(request).scheme not in {"http", "https"} ): return response interval, url = get_meta_refresh(response, ignore_tags=self._ignore_tags) if not url: return response redirected = self._redirect_request_using_get(request, url) if urlparse_cached(redirected).scheme not in {"http", "https"}: return response if cast(float, interval) < self._maxdelay: return self._redirect(redirected, request, spider, "meta refresh") return response