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

Merge 82e280dde1 into d2842a205c

This commit is contained in:
Akshay Sharma 2026-06-19 22:56:18 +05:00 committed by GitHub
parent d2842a205c 82e280dde1
commit 1eb7f7abe2
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
19 changed files with 1060 additions and 83 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
docs/topics/request-response.rst
View File

@ -1201,6 +1201,8 @@ TextResponse objects
.. autoattribute:: TextResponse.attributes
.. autoattribute:: TextResponse.base_url
:class:`TextResponse` objects support the following methods in addition to
the standard :class:`Response` ones:

5
pyproject.toml
View File

@ -22,6 +22,7 @@ dependencies = [
"service_identity>=23.1.0",
"tldextract",
"w3lib>=1.17.0",
"xtractmime>=0.2.0",
"zope.interface>=5.1.0",
# Platform-specific dependencies
'PyDispatcher>=2.0.5; platform_python_implementation == "CPython"',
@ -132,6 +133,8 @@ module = [
"pytest_twisted",
"robotexclusionrulesparser",
"testfixtures",
"xtractmime",
"xtractmime.*",
"zope.interface.*",
]
ignore_missing_imports = true
@ -274,6 +277,8 @@ markers = [
]
filterwarnings = [
"ignore::DeprecationWarning:twisted.web.static",
"ignore:scrapy.responsetypes is deprecated",
"ignore:scrapy.utils.response.get_base_url is deprecated",
]
[tool.ruff.lint]

7
scrapy/core/downloader/handlers/datauri.py
View File

@ -6,7 +6,7 @@ from w3lib.url import parse_data_uri
from scrapy.core.downloader.handlers.base import BaseDownloadHandler
from scrapy.http import Response, TextResponse
from scrapy.responsetypes import responsetypes
from scrapy.utils.response import get_response_class
if TYPE_CHECKING:
from scrapy import Request
@ -15,7 +15,10 @@ if TYPE_CHECKING:
class DataURIDownloadHandler(BaseDownloadHandler):
async def download_request(self, request: Request) -> Response:
uri = parse_data_uri(request.url)
respcls = responsetypes.from_mimetype(uri.media_type)
respcls = get_response_class(
body=uri.data,
declared_mime_types=(uri.media_type.encode(),),
)
if issubclass(respcls, TextResponse) and uri.media_type.split("/")[0] == "text":
charset = uri.media_type_parameters.get("charset")

4
scrapy/core/downloader/handlers/file.py
View File

@ -6,8 +6,8 @@ from typing import TYPE_CHECKING
from w3lib.url import file_uri_to_path
from scrapy.core.downloader.handlers.base import BaseDownloadHandler
from scrapy.responsetypes import responsetypes
from scrapy.utils.asyncio import run_in_thread
from scrapy.utils.response import get_response_class
if TYPE_CHECKING:
from scrapy import Request
@ -18,5 +18,5 @@ class FileDownloadHandler(BaseDownloadHandler):
async def download_request(self, request: Request) -> Response:
filepath = file_uri_to_path(request.url)
body = await run_in_thread(Path(filepath).read_bytes)
respcls = responsetypes.from_args(filename=filepath, body=body)
respcls = get_response_class(url=request.url, body=body)
return respcls(url=request.url, body=body)

4
scrapy/core/downloader/handlers/ftp.py
View File

@ -41,9 +41,9 @@ from twisted.internet.protocol import ClientCreator, Protocol
from scrapy.core.downloader.handlers.base import BaseDownloadHandler
from scrapy.exceptions import NotConfigured
from scrapy.http import Response
from scrapy.responsetypes import responsetypes
from scrapy.utils.defer import maybe_deferred_to_future
from scrapy.utils.httpobj import urlparse_cached
from scrapy.utils.response import get_response_class
if TYPE_CHECKING:
from twisted.protocols.ftp import FTPClient
@ -122,6 +122,6 @@ class FTPDownloadHandler(BaseDownloadHandler):
protocol.close()
headers = {"local filename": protocol.filename or b"", "size": protocol.size}
body = protocol.filename or protocol.body.read()
respcls = responsetypes.from_args(url=request.url, body=body)
respcls = get_response_class(url=request.url, body=body)
# hints for Headers-related types may need to be fixed to not use AnyStr
return respcls(url=request.url, status=200, body=body, headers=headers) # type: ignore[arg-type]

1
scrapy/core/downloader/handlers/http11.py
View File

@ -67,7 +67,6 @@ if TYPE_CHECKING:
from scrapy.crawler import Crawler
logger = logging.getLogger(__name__)
_T = TypeVar("_T")

8
scrapy/downloadermiddlewares/httpcompression.py
View File

@ -8,7 +8,6 @@ from typing import TYPE_CHECKING, Any
from scrapy import Request, Spider, signals
from scrapy.exceptions import IgnoreRequest, NotConfigured
from scrapy.http import Response, TextResponse
from scrapy.responsetypes import responsetypes
from scrapy.utils._compression import (
_DecompressionMaxSizeExceeded,
_inflate,
@ -18,6 +17,7 @@ from scrapy.utils._compression import (
from scrapy.utils.decorators import _warn_spider_arg
from scrapy.utils.deprecate import warn_on_deprecated_spider_attribute
from scrapy.utils.gz import gunzip
from scrapy.utils.response import get_response_class
if TYPE_CHECKING:
# typing.Self requires Python 3.11
@ -139,8 +139,10 @@ class HttpCompressionMiddleware:
len(decoded_body),
)
self.stats.inc_value("httpcompression/response_count")
respcls = responsetypes.from_args(
headers=response.headers, url=response.url, body=decoded_body
respcls = get_response_class(
http_headers=response.headers,
url=response.url,
body=decoded_body,
)
kwargs: dict[str, Any] = {"body": decoded_body}
if issubclass(respcls, TextResponse):

6
scrapy/extensions/httpcache.py
View File

@ -13,10 +13,10 @@ from weakref import WeakKeyDictionary
from w3lib.http import headers_dict_to_raw, headers_raw_to_dict
from scrapy.http import Headers, Response
from scrapy.responsetypes import responsetypes
from scrapy.utils.httpobj import urlparse_cached
from scrapy.utils.project import data_path
from scrapy.utils.python import to_bytes, to_unicode
from scrapy.utils.response import get_response_class
if TYPE_CHECKING:
import os
@ -277,7 +277,7 @@ class DbmCacheStorage:
status = data["status"]
headers = Headers(data["headers"])
body = data["body"]
respcls = responsetypes.from_args(headers=headers, url=url, body=body)
respcls = get_response_class(http_headers=headers, url=url, body=body)
return respcls(url=url, headers=headers, status=status, body=body)
def store_response(
@ -345,7 +345,7 @@ class FilesystemCacheStorage:
url = metadata["response_url"]
status = metadata["status"]
headers = Headers(headers_raw_to_dict(rawheaders))
respcls = responsetypes.from_args(headers=headers, url=url, body=body)
respcls = get_response_class(http_headers=headers, url=url, body=body)
return respcls(url=url, headers=headers, status=status, body=body)
def store_response(

25
scrapy/http/response/text.py
View File

@ -20,11 +20,10 @@ from w3lib.encoding import (
read_bom,
resolve_encoding,
)
from w3lib.html import strip_html5_whitespace
from w3lib.html import get_base_url, strip_html5_whitespace
from scrapy.http.response import Response
from scrapy.utils.python import memoizemethod_noargs, to_unicode
from scrapy.utils.response import get_base_url
if TYPE_CHECKING:
from collections.abc import Callable, Iterable, Mapping
@ -44,6 +43,7 @@ class TextResponse(Response):
attributes: tuple[str, ...] = (*Response.attributes, "encoding")
__slots__ = (
"_cached_base_url",
"_cached_benc",
"_cached_decoded_json",
"_cached_selector",
@ -53,6 +53,7 @@ class TextResponse(Response):
def __init__(self, *args: Any, **kwargs: Any):
self._encoding: str | None = kwargs.pop("encoding", None)
self._cached_base_url: str | None = None
self._cached_benc: str | None = None
self._cached_ubody: str | None = None
self._cached_selector: Selector | None = None
@ -70,6 +71,7 @@ class TextResponse(Response):
self._body = body.encode(self._encoding)
else:
super()._set_body(body)
self._cached_base_url = None
@property
def encoding(self) -> str:
@ -100,10 +102,27 @@ class TextResponse(Response):
self._cached_ubody = html_to_unicode(charset, self.body)[1]
return self._cached_ubody
@property
def base_url(self) -> str:
"""Base URL for any relative URL in the response.
It defaults to the response :attr:`~scrapy.http.Response.url`, but HTML
responses may include a `base element
<https://developer.mozilla.org/en-US/docs/Web/HTML/Element/base>`_ with
a different base URL.
"""
if self._cached_base_url is None:
self._cached_base_url = get_base_url(
self.text[:4096],
self.url,
self.encoding,
)
return self._cached_base_url
def urljoin(self, url: str) -> str:
"""Join this Response's url with a possible relative url to form an
absolute interpretation of the latter."""
return urljoin(get_base_url(self), url)
return urljoin(self.base_url, url)
@memoizemethod_noargs
def _headers_encoding(self) -> str | None:

15
scrapy/linkextractors/lxmlhtml.py
View File

@ -21,7 +21,6 @@ from scrapy.link import Link
from scrapy.linkextractors import IGNORED_EXTENSIONS, _is_valid_url, _matches
from scrapy.utils.misc import arg_to_iter, rel_has_nofollow
from scrapy.utils.python import unique as unique_list
from scrapy.utils.response import get_base_url
from scrapy.utils.url import url_has_any_extension, url_is_from_any_domain
if TYPE_CHECKING:
@ -139,9 +138,11 @@ class LxmlParserLinkExtractor:
return self._deduplicate_if_needed(links)
def extract_links(self, response: TextResponse) -> list[Link]:
base_url = get_base_url(response)
return self._extract_links(
response.selector, response.url, response.encoding, base_url
response.selector,
response.url,
response.encoding,
response.base_url,
)
def _process_links(self, links: list[Link]) -> list[Link]:
@ -268,7 +269,6 @@ class LxmlLinkExtractor:
Duplicate links are omitted if the ``unique`` attribute is set to ``True``,
otherwise they are returned.
"""
base_url = get_base_url(response)
if self.restrict_xpaths:
docs = [
subdoc for x in self.restrict_xpaths for subdoc in response.xpath(x)
@ -277,7 +277,12 @@ class LxmlLinkExtractor:
docs = [response.selector]
all_links = []
for doc in docs:
links = self._extract_links(doc, response.url, response.encoding, base_url)
links = self._extract_links(
doc,
response.url,
response.encoding,
response.base_url,
)
all_links.extend(self._process_links(links))
if self.link_extractor.unique:
return unique_list(all_links, key=self.link_extractor.link_key)

24
scrapy/responsetypes.py
View File

@ -5,17 +5,27 @@ based on different criteria.
from __future__ import annotations
from io import StringIO
from mimetypes import MimeTypes
from pkgutil import get_data
from typing import TYPE_CHECKING, ClassVar
from warnings import warn
from scrapy.exceptions import ScrapyDeprecationWarning
from scrapy.http import Response
from scrapy.utils.misc import load_object
from scrapy.utils.python import binary_is_text, to_bytes, to_unicode
from scrapy.utils.response import _MIME_TYPES
warn(
(
"scrapy.responsetypes is deprecated, use "
"scrapy.utils.response.get_response_class instead"
),
ScrapyDeprecationWarning,
stacklevel=2,
)
if TYPE_CHECKING:
from collections.abc import Mapping
from mimetypes import MimeTypes
class ResponseTypes:
@ -38,13 +48,7 @@ class ResponseTypes:
def __init__(self) -> None:
self.classes: dict[str, type[Response]] = {}
self.mimetypes: MimeTypes = MimeTypes()
mimedata = get_data("scrapy", "mime.types")
if not mimedata:
raise ValueError(
"The mime.types file is not found in the Scrapy installation"
)
self.mimetypes.readfp(StringIO(mimedata.decode("utf8")))
self.mimetypes: MimeTypes = _MIME_TYPES
for mimetype, cls in self.CLASSES.items():
self.classes[mimetype] = load_object(cls)

4
scrapy/utils/_download_handlers.py
View File

@ -14,7 +14,6 @@ from twisted.python.failure import Failure
from twisted.web.client import ResponseFailed
from twisted.web.error import SchemeNotSupported
from scrapy import responsetypes
from scrapy.exceptions import (
CannotResolveHostError,
DownloadCancelledError,
@ -25,6 +24,7 @@ from scrapy.exceptions import (
UnsupportedURLSchemeError,
)
from scrapy.utils.log import logger
from scrapy.utils.response import get_response_class
if TYPE_CHECKING:
from collections.abc import Iterator
@ -102,7 +102,7 @@ def make_response(
protocol: str | None = None,
stop_download: StopDownload | None = None,
) -> Response:
respcls = responsetypes.responsetypes.from_args(headers=headers, url=url, body=body)
respcls = get_response_class(http_headers=headers, url=url, body=body)
response = respcls(
url=url,
status=status,

8
scrapy/utils/python.py
View File

@ -167,6 +167,14 @@ def binary_is_text(data: bytes) -> bool:
"""Returns ``True`` if the given ``data`` argument (a ``bytes`` object)
does not contain unprintable control characters.
"""
warnings.warn(
(
"scrapy.utils.python.binary_is_text is deprecated, use "
"xtractmime.is_binary_data instead."
),
ScrapyDeprecationWarning,
stacklevel=2,
)
if not isinstance(data, bytes):
raise TypeError(f"data must be bytes, got '{type(data).__name__}'")
return all(c not in _BINARYCHARS for c in data)

203
scrapy/utils/response.py
View File

@ -9,37 +9,151 @@ import os
import re
import tempfile
import webbrowser
from io import StringIO
from mimetypes import MimeTypes
from pkgutil import get_data
from typing import TYPE_CHECKING, Any
from urllib.parse import urlparse
from warnings import warn
from weakref import WeakKeyDictionary
from twisted.web import http
from w3lib import html
from xtractmime import RESOURCE_HEADER_BUFFER_LENGTH as BODY_LIMIT
from xtractmime import extract_mime
from xtractmime.mimegroups import (
is_html_mime_type,
is_javascript_mime_type,
is_json_mime_type,
is_xml_mime_type,
)
from scrapy.exceptions import ScrapyDeprecationWarning
from scrapy.http import (
Headers,
HtmlResponse,
JsonResponse,
Response,
TextResponse,
XmlResponse,
)
from scrapy.utils.python import to_bytes, to_unicode
if TYPE_CHECKING:
from collections.abc import Callable, Iterable
from scrapy.http import Response, TextResponse
_baseurl_cache: WeakKeyDictionary[Response, str] = WeakKeyDictionary()
def get_base_url(response: TextResponse) -> str:
"""Return the base url of the given response, joined with the response url"""
if response not in _baseurl_cache:
text = response.text[0:4096]
_baseurl_cache[response] = html.get_base_url(
text, response.url, response.encoding
)
return _baseurl_cache[response]
from collections.abc import Callable, Iterable, Sequence
_ENCODING_MIME_TYPE_MAP = {
b"br": b"application/brotli",
b"compress": b"application/x-compress",
b"deflate": b"application/zip",
b"gzip": b"application/gzip",
b"zstd": b"application/zstd",
}
_ENCODING_MIME_TYPES = {*_ENCODING_MIME_TYPE_MAP.values()}
_MIME_TYPES = MimeTypes()
_mime_overrides = get_data("scrapy", "mime.types") or b""
_MIME_TYPES.readfp(StringIO(_mime_overrides.decode()))
_metaref_cache: WeakKeyDictionary[Response, tuple[None, None] | tuple[float, str]] = (
WeakKeyDictionary()
)
def _is_compressed_mime_type(mime_type: bytes) -> bool:
return mime_type in _ENCODING_MIME_TYPES
def _is_other_text_mime_type(mime_type: bytes) -> bool:
return (
mime_type.startswith(b"text/")
or mime_type == b"application/x-javascript"
or is_javascript_mime_type(mime_type)
)
def _get_encoding_or_mime_type_from_headers(
headers: Headers,
) -> tuple[bytes | None, bytes | None]:
if b"Content-Encoding" in headers:
encodings = [
item.strip()
for item in b",".join(headers.getlist(b"Content-Encoding")).split(b",")
if item.strip().lower() != b"identity"
]
if encodings:
return encodings[-1], None
if (
b"Content-Type" in headers
and headers[b"Content-Type"]
and headers[b"Content-Type"].split(b";")[0].strip().lower()
not in (
b"",
b"unknown/unknown",
b"application/unknown",
b"*/*",
)
):
return None, headers[b"Content-Type"]
content_disposition = headers.get(b"Content-Disposition")
if content_disposition:
path = (
content_disposition.split(b";")[-1].split(b"=")[-1].strip(b"\"'").decode()
)
encoding, mime_type = _get_encoding_or_mime_type_from_path(path)
if encoding:
return encoding, None
return None, mime_type
return None, None
def _get_mime_type_from_encoding(encoding: bytes) -> bytes:
return _ENCODING_MIME_TYPE_MAP.get(encoding) or b"application/" + encoding
def _get_encoding_or_mime_type_from_path(
path: str,
) -> tuple[bytes | None, bytes | None]:
mimetype, encoding = _MIME_TYPES.guess_type(path, strict=False)
if encoding:
return encoding.encode(), None
if mimetype:
return None, mimetype.encode()
return None, None
def _get_response_class_from_mime_type(mime_type: bytes | None) -> type[Response]:
if not mime_type:
return Response
if is_html_mime_type(mime_type):
return HtmlResponse
if is_xml_mime_type(mime_type):
return XmlResponse
if is_json_mime_type(mime_type) or (
mime_type
in (
b"application/x-json",
b"application/json-amazonui-streaming",
)
):
return JsonResponse
if _is_other_text_mime_type(mime_type):
return TextResponse
return Response
def get_base_url(response: TextResponse) -> str:
"""Return the base url of the given response, joined with the response url"""
warn(
(
"scrapy.utils.response.get_base_url is deprecated, use "
"scrapy.http.TextResponse.base_url instead."
),
ScrapyDeprecationWarning,
stacklevel=2,
)
return response.base_url
def get_meta_refresh(
response: TextResponse,
ignore_tags: Iterable[str] = ("script", "noscript"),
@ -48,11 +162,65 @@ def get_meta_refresh(
if response not in _metaref_cache:
text = response.text[0:4096]
_metaref_cache[response] = html.get_meta_refresh(
text, get_base_url(response), response.encoding, ignore_tags=ignore_tags
text, response.base_url, response.encoding, ignore_tags=ignore_tags
)
return _metaref_cache[response]
def get_response_class(
*,
url: str | None = None,
body: bytes | None = None,
declared_mime_types: Sequence[bytes] | None = None,
http_headers: Headers | None = None,
) -> type[Response]:
"""Guess the most appropriate Response class based on the given
arguments."""
mime_type = next(iter(declared_mime_types or []), None)
encoding = None # as in compression (e.g. gzip), not charset
if http_headers:
encoding, header_mime_type = _get_encoding_or_mime_type_from_headers(
http_headers
)
if encoding is None and mime_type is None:
mime_type = header_mime_type
if url is not None:
url_parts = urlparse(url)
http_origin = url_parts.scheme in ("http", "https")
if not http_origin and not encoding:
encoding, path_mime_type = _get_encoding_or_mime_type_from_path(
url_parts.path
)
if encoding is None and mime_type is None:
mime_type = path_mime_type
else:
http_origin = True
body = (body or b"")[:BODY_LIMIT]
if encoding:
content_types = (_get_mime_type_from_encoding(encoding),)
elif mime_type:
content_types = (mime_type,)
else:
content_types = None
mime_type = extract_mime(
body,
content_types=content_types,
http_origin=http_origin,
)
cls = _get_response_class_from_mime_type(mime_type)
if cls is not Response or not content_types or encoding or not http_origin:
return cls
# In scenarios where there was a declared Content-Type, no
# Content-Encoding, HTTP/HTTPS was used, and xtractmime determined the
# output to be binary, repeat MIME extraction ignoring the declared
# Content-Type, so that the body is taken into account.
mime_type = extract_mime(
body,
http_origin=http_origin,
)
return _get_response_class_from_mime_type(mime_type)
def response_status_message(status: bytes | float | str) -> str:
"""Return status code plus status text descriptive message"""
status_int = int(status)
@ -91,9 +259,6 @@ def open_in_browser(
if "item name" not in response.body:
open_in_browser(response)
"""
# circular imports
from scrapy.http import HtmlResponse, TextResponse # noqa: PLC0415
# XXX: this implementation is a bit dirty and could be improved
body = response.body
if isinstance(response, HtmlResponse):

BIN
tests/sample_data/compressed/html-br-gzip.bin Normal file
View File

Binary file not shown.

2
tests/test_downloader_handlers_http_base.py
View File

@ -552,7 +552,7 @@ class TestHttpBase(ABC):
"""Tests choosing of correct response type
in case of Content-Type is empty but body contains text.
"""
body = b"Some plain text\ndata with tabs\t and null bytes\0"
body = b"Some plain text\ndata with tabs\t"
request = Request(
mockserver.url("/nocontenttype", is_secure=self.is_secure), body=body
)

34
tests/test_downloadermiddleware_httpcompression.py
View File

@ -13,9 +13,9 @@ from scrapy.downloadermiddlewares.httpcompression import (
)
from scrapy.exceptions import IgnoreRequest, NotConfigured
from scrapy.http import HtmlResponse, Request, Response
from scrapy.responsetypes import responsetypes
from scrapy.spiders import Spider
from scrapy.utils.gz import gunzip
from scrapy.utils.response import get_response_class
from scrapy.utils.test import get_crawler
from tests import tests_datadir
@ -29,6 +29,7 @@ FORMAT = {
"gzip-deflate": ("html-gzip-deflate.bin", "gzip, deflate"),
"gzip-deflate-gzip": ("html-gzip-deflate-gzip.bin", "gzip, deflate, gzip"),
"br": ("html-br.bin", "br"),
"br,gzip": ("html-br-gzip.bin", "br,gzip"),
# $ zstd raw.html --content-size -o html-zstd-static-content-size.bin
"zstd-static-content-size": ("html-zstd-static-content-size.bin", "zstd"),
# $ zstd raw.html --no-content-size -o html-zstd-static-no-content-size.bin
@ -155,6 +156,31 @@ class TestHttpCompression:
self.assertStatsEqual("httpcompression/response_count", 1)
self.assertStatsEqual("httpcompression/response_bytes", 74837)
def test_process_response_br_gzip(self):
_skip_if_no_br()
response = self._getresponse("br,gzip")
request = response.request
assert response.headers["Content-Encoding"] == b"br,gzip"
newresponse = self.mw.process_response(request, response)
assert newresponse is not response
assert newresponse.body.startswith(b"<!DOCTYPE")
assert "Content-Encoding" not in newresponse.headers
self.assertStatsEqual("httpcompression/response_count", 1)
self.assertStatsEqual("httpcompression/response_bytes", 74837)
def test_process_response_br_gzip_header_list(self):
_skip_if_no_br()
response = self._getresponse("br,gzip")
request = response.request
assert response.headers["Content-Encoding"] == b"br,gzip"
response.headers.setlist("Content-Encoding", [b"br", b"gzip"])
newresponse = self.mw.process_response(request, response)
assert newresponse is not response
assert newresponse.body.startswith(b"<!DOCTYPE")
assert "Content-Encoding" not in newresponse.headers
self.assertStatsEqual("httpcompression/response_count", 1)
self.assertStatsEqual("httpcompression/response_bytes", 74837)
def test_process_response_br_unsupported(self):
try:
try:
@ -405,8 +431,10 @@ class TestHttpCompression:
b"<html><head><title>Some page</title>"
b'<meta http-equiv="Content-Type" content="text/html; charset=gb2312">'
)
respcls = responsetypes.from_args(
url="http://www.example.com/index", headers=headers, body=plainbody
respcls = get_response_class(
url="http://www.example.com/index",
http_headers=headers,
body=plainbody,
)
response = respcls(
"http://www.example.com/index", headers=headers, body=plainbody

69
tests/test_responsetypes.py
View File

@ -1,3 +1,9 @@
from __future__ import annotations
from typing import Any
import pytest
from scrapy.http import (
Headers,
HtmlResponse,
@ -8,6 +14,43 @@ from scrapy.http import (
)
from scrapy.responsetypes import responsetypes
from .test_utils_response import POST_XTRACTMIME_SCENARIOS, PRE_XTRACTMIME_SCENARIOS
def _unmark(item: Any) -> Any:
return pytest.param(*item.values)
@pytest.mark.parametrize(
("kwargs", "response_class"),
[
*(
item if not hasattr(item, "marks") else _unmark(item)
for item in PRE_XTRACTMIME_SCENARIOS
),
*(
pytest.param(
kwargs,
response_class,
marks=pytest.mark.xfail(
strict=True,
reason=(
"Expected failure of deprecated "
"scrapy.responsetypes.responsetypes.from_args, works "
"with its replacement "
"scrapy.utils.response.get_response_class"
),
),
)
for kwargs, response_class in POST_XTRACTMIME_SCENARIOS
),
],
)
def test_from_args(kwargs, response_class):
assert responsetypes.from_args(**kwargs) == response_class, (
f"{responsetypes.from_args(**kwargs)=} != {response_class=}"
)
class TestResponseTypes:
def test_from_filename(self):
@ -93,32 +136,6 @@ class TestResponseTypes:
retcls = responsetypes.from_headers(source)
assert retcls is cls, f"{source} ==> {retcls} != {cls}"
def test_from_args(self):
# TODO: add more tests that check precedence between the different arguments
mappings = [
({"url": "http://www.example.com/data.csv"}, TextResponse),
# headers takes precedence over url
(
{
"headers": Headers({"Content-Type": ["text/html; charset=utf-8"]}),
"url": "http://www.example.com/item/",
},
HtmlResponse,
),
(
{
"headers": Headers(
{"Content-Disposition": ['attachment; filename="data.xml.gz"']}
),
"url": "http://www.example.com/page/",
},
Response,
),
]
for source, cls in mappings:
retcls = responsetypes.from_args(**source)
assert retcls is cls, f"{source} ==> {retcls} != {cls}"
def test_custom_mime_types_loaded(self):
# check that mime.types files shipped with scrapy are loaded
assert responsetypes.mimetypes.guess_type("x.scrapytest")[0] == "x-scrapy/test"

722
tests/test_utils_response.py
View File

@ -1,15 +1,22 @@
from itertools import chain
from pathlib import Path
from time import process_time
from urllib.parse import urlparse
import pytest
from xtractmime import BINARY_BYTES, RESOURCE_HEADER_BUFFER_LENGTH
from scrapy.http import HtmlResponse, Response, TextResponse
from scrapy.http import HtmlResponse, JsonResponse, Response, TextResponse, XmlResponse
from scrapy.http.headers import Headers
from scrapy.responsetypes import ResponseTypes
from scrapy.utils.misc import load_object
from scrapy.utils.python import to_bytes
from scrapy.utils.response import (
_get_encoding_or_mime_type_from_headers,
_remove_html_comments,
get_base_url,
get_meta_refresh,
get_response_class,
open_in_browser,
response_status_message,
)
@ -22,6 +29,719 @@ def _read_browser_output(burl: str):
return Path(path).read_bytes()
# https://mimesniff.spec.whatwg.org/#interpreting-the-resource-metadata
PRE_XTRACTMIME_HTML_STARTS = (
b"<!DOCTYPE HTML",
b"<HTML",
)
POST_XTRACTMIME_HTML_STARTS = (
b"<HEAD",
b"<SCRIPT",
b"<IFRAME",
b"<H1",
b"<DIV",
b"<FONT",
b"<TABLE",
b"<A",
b"<STYLE",
b"<TITLE",
b"<B",
b"<BODY",
b"<BR",
b"<P",
b"<!--",
)
NON_BINARY_ASCII_BYTES = (
byte for byte in (bytes([byte]) for byte in range(128)) if byte not in BINARY_BYTES
)
# https://mimesniff.spec.whatwg.org/#whitespace-byte
WHITESPACE_BYTES = (
b"\t",
b"\n",
b"\x0c",
b"\r",
b" ",
)
def odd_capitalize(value: bytes) -> bytes:
"""Make odd bytes lowecase and even bytes uppercase.
>>> odd_capitalize(b'foobar')
b'fOoBaR'
"""
return b"".join(
bytes([byte]).lower() if index % 2 == 0 else bytes([byte]).upper()
for index, byte in enumerate(value)
)
# Scenarios that work the same with the previously-used, deprecated
# scrapy.responsetypes.responsetypes.from_args
PRE_XTRACTMIME_SCENARIOS = (
# Content-Type determines the type for the HTTP protocol.
*(
(
{
"url": f"{protocol}://example.com/foo",
"headers": Headers(
{"Content-Type": content_type + content_type_parameters}
),
},
response_class,
)
for protocol in ("http", "https")
# Make sure that MIME parameters do not break response class choice.
for content_type_parameters in ("", "; foo=bar")
for content_type, response_class in (
("text/plain", TextResponse),
("text/html", HtmlResponse),
("text/xml", XmlResponse),
*(
(mime_type, load_object(class_path))
for mime_type, class_path in ResponseTypes.CLASSES.items()
if mime_type
not in (
# “Note that XHTML is best parsed as XML”
# https://lxml.de/parsing.html
"application/xhtml+xml",
"application/vnd.wap.xhtml+xml",
)
),
# JavaScript MIME types should trigger a TextResponse.
#
# https://mimesniff.spec.whatwg.org/#javascript-mime-type
*(
(mime_type, TextResponse)
for mime_type in (
"application/javascript",
"application/x-javascript",
"text/ecmascript",
"text/javascript",
"text/javascript1.0",
"text/javascript1.1",
"text/javascript1.2",
"text/javascript1.3",
"text/javascript1.4",
"text/javascript1.5",
"text/jscript",
"text/livescript",
"text/x-ecmascript",
"text/x-javascript",
# Unofficial
"application/x-javascript",
)
),
# JSON MIME types should trigger a JsonResponse.
#
# https://mimesniff.spec.whatwg.org/#json-mime-type
*(
(mime_type, JsonResponse)
for mime_type in (
"application/json",
# Unofficial
"application/json-amazonui-streaming",
"application/x-json",
)
),
)
),
# Content-Type triumphs body, except for:
#
# - Binary content mislabeled as plain text due to an Apache bug
# https://mimesniff.spec.whatwg.org/#check-for-apache-bug-flag
# https://mimesniff.spec.whatwg.org/#rules-for-text-or-binary
#
# - Feeds mislabeled as HTML
# https://mimesniff.spec.whatwg.org/#rules-for-distinguishing-if-a-resource-is-a-feed-or-html
*(
(
{
"body": body,
"headers": Headers({"Content-Type": [content_type]}),
},
response_class,
)
for body, content_type, response_class in (
*(
(b"\x00\x01\xff", content_type, TextResponse)
for content_type in (
# text/plain variants *not* affected by the Apache bug
"text/plain; charset=Iso-8859-1",
"text/plain; charset=utf-8",
"text/plain; charset=windows-1252",
)
),
)
),
# Content-Type triumphs Content-Disposition.
*(
(
{
"url": f"{protocol}://example.com/a",
"headers": Headers(
{
"Content-Disposition": [
f'attachment; filename="a.{file_extension}"',
],
"Content-Type": [content_type],
}
),
},
response_class,
)
for protocol in ("http", "https")
for file_extension, content_type, response_class in (
("html", "application/json", JsonResponse),
("xml", "application/json", JsonResponse),
)
),
# Compressed content should be of type Response until uncompressed.
(
{
"headers": Headers(
{
"Content-Encoding": ["zip"],
"Content-Type": ["text/html"],
}
)
},
Response,
),
# We take the file extension of URL paths into account, except for HTTP
# responses, because “they are unreliable and easily spoofed”.
#
# https://mimesniff.spec.whatwg.org/#interpreting-the-resource-metadata
*(
(
{"url": f"{protocol}://example.com/a.{extension}"},
response_class,
)
for protocol in ("file", "ftp")
for extension, response_class in (
("gz", Response),
("html", HtmlResponse),
("json", JsonResponse),
("pdf", Response),
("txt", TextResponse),
("xml", XmlResponse),
)
),
# Unlike in a web browser, where an attachment Content-Disposition header
# causes the response to be downloaded, and hence MIME sniffing becomes
# irrelevant, in Scrapy those responses are handled the same as any, and
# hence we take the file extension from Content-Disposition into account
# to choose a response class, as a fallback when there is no Content-Type.
*(
(
{
"url": f"{protocol}://example.com/a",
"headers": Headers(
{
"Content-Disposition": [
'attachment; filename="a.xml"',
]
}
),
},
XmlResponse,
)
for protocol in ("http", "https")
),
*(
(
{
"url": f"{protocol}://example.com/a",
"headers": Headers(
{
"Content-Disposition": [
'attachment; filename="a.html"',
],
"Content-Type": "text/xml",
}
),
},
XmlResponse,
)
for protocol in ("http", "https")
),
*(
(
{
"url": f"{protocol}://example.com/a",
"body": b"<?xml",
"headers": Headers(
{
"Content-Disposition": [
'attachment; filename="a.html"',
],
}
),
},
HtmlResponse,
)
for protocol in ("http", "https")
),
# Without anything else, the body determines the response class.
*(
({"body": body}, response_class)
for body, response_class in (
(b"<html><head><title>Hello</title></head>", HtmlResponse),
(b'<?xml version="1.0" encoding="utf-8"', XmlResponse),
# https://codersblock.com/blog/the-smallest-valid-html5-page/
(b"<!DOCTYPE html>\n<title>.</title>", HtmlResponse),
# https://mimesniff.spec.whatwg.org/#identifying-a-resource-with-an-unknown-mime-type
*(
(prefix + start + b">", HtmlResponse)
for prefix in (
b"",
*(byte for byte in WHITESPACE_BYTES if byte != b"\x0c"),
)
for start in (
set_case(start)
for set_case in (bytes.lower, bytes.upper, odd_capitalize)
for start in PRE_XTRACTMIME_HTML_STARTS
)
),
*(
(prefix + b"<?xml", XmlResponse)
for prefix in (
b"",
*(byte for byte in WHITESPACE_BYTES if byte != b"\x0c"),
)
),
(b"\xfe\xffab", TextResponse),
(b"\xff\xfeab", TextResponse),
(b"\xef\xbb\xbfa", TextResponse),
(b"\x00\x00\x01\x00", Response),
(b"\x00\x00\x02\x00", Response),
(b"\x89PNG\r\n\x1a\n", Response),
(b"MThd\x00\x00\x00\x06", Response),
(b"\x00\x00\x00\x0cftypmp4a", Response),
(b"\x00\x00\x00\x14ftypabcdefghmp4a", Response),
(
# https://github.com/mathiasbynens/small/blob/267b39f682598eebb0dafe7590b1504be79b5cad/webm.webm
(
b"\x1aE\xdf\xa3@ B\x86\x81\x01B\xf7\x81\x01B\xf2\x81\x04B"
b"\xf3\x81\x08B\x82@\x04webm"
),
Response,
),
(
# https://github.com/mathiasbynens/small/blob/267b39f682598eebb0dafe7590b1504be79b5cad/mp3.mp3
(
b"\xff\xe3\x18\xc4\x00\x00\x00\x03H\x00\x00\x00\x00LAME3.9"
b"8.2\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
b"\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
b"\x00\x00\x00\x00\x00\x00\x00\x00"
),
Response,
),
(b"\x1f\x8b\x08", Response),
(b"PK\x03\x04", Response),
(b"Rar \x1a\x07\x00", Response),
# A body is considered binary if its header (first 1445 bytes)
# contains any binary data byte.
*((byte, Response) for byte in BINARY_BYTES[1:]),
*(
(byte, TextResponse)
for byte in NON_BINARY_ASCII_BYTES
if byte not in (b"\x0c", b"\x1b")
),
*(
(b"a" * (RESOURCE_HEADER_BUFFER_LENGTH - 1) + byte, Response)
for byte in BINARY_BYTES[1:]
),
(b"a" * RESOURCE_HEADER_BUFFER_LENGTH + BINARY_BYTES[0], TextResponse),
)
),
# A Content-Type whose essence is "unknown/unknown", "application/unknown",
# or "*/*" has the same effect as no Content-Type being defined.
#
# https://mimesniff.spec.whatwg.org/#mime-type-sniffing-algorithm
*(
(
{
"body": b"<?xml",
"headers": Headers(
{"Content-Type": [content_type + content_type_suffix]}
),
},
XmlResponse,
)
for content_type_suffix in ("", "; foo=bar")
for content_type in (
"unknown/unknown",
"application/unknown",
"*/*",
)
),
*(
(
{
"url": f"{protocol}://example.com/a",
"headers": Headers(
{
"Content-Disposition": [
'attachment; filename="a.xml"',
],
"Content-Type": [content_type + content_type_suffix],
}
),
},
XmlResponse,
)
for protocol in ("http", "https")
for content_type_suffix in ("", "; foo=bar")
for content_type in (
"unknown/unknown",
"application/unknown",
"*/*",
)
),
# Content triumphs Content-Type when using HTTP or HTTPS and the
# Content-Type is unknown or binary while the content is plain text. This
# is a conscious divergence from the MIME Sniffing Standard for a better
# web scraping experience.
*(
(
{
"url": f"{protocol}://example.com/foo",
"headers": Headers({"Content-Type": content_type}),
"body": body,
},
TextResponse,
)
for protocol in ("http", "https")
for body in (
b"",
b"a",
b"var a = 'b';",
b'{"a": "b"}',
b'.a {b: "c"}',
)
for content_type in (
"application/octet-stream",
"application/pdf",
"application/custom",
"application/bad-custom-json", # Should end in +json
"application/bad-custom-text", # Should start with text/
"application/bad-custom-xml", # Should end in +xml
)
),
)
# Scenarios that work differently with the previously-used, deprecated
# scrapy.responsetypes.responsetypes.from_args
POST_XTRACTMIME_SCENARIOS = (
# Content-Type determines the type for the HTTP protocol.
*(
(
{
"url": f"{protocol}://example.com/foo",
"headers": Headers({"Content-Type": content_type}),
},
response_class,
)
for protocol in ("http", "https")
# Make sure that MIME parameters do not break response class choice.
for content_type_parameters in ("", "; foo=bar")
for content_type, response_class in (
# “Note that XHTML is best parsed as XML”
# https://lxml.de/parsing.html
("application/xhtml+xml", XmlResponse),
("application/vnd.wap.xhtml+xml", XmlResponse),
# JavaScript MIME types should trigger a TextResponse.
#
# https://mimesniff.spec.whatwg.org/#javascript-mime-type
*(
(mime_type, TextResponse)
for mime_type in (
"application/ecmascript",
"application/x-ecmascript",
)
),
# JSON MIME types should trigger a TextResponse.
#
# https://mimesniff.spec.whatwg.org/#json-mime-type
*(
(mime_type, JsonResponse)
for mime_type in (
"application/foo+json",
"application/ld+json",
"text/json",
)
),
# XML MIME types should trigger an XmlResponse.
#
# https://mimesniff.spec.whatwg.org/#xml-mime-type
*((mime_type, XmlResponse) for mime_type in ("application/foo+xml",)),
)
),
# Content-Type triumphs body, except for:
#
# - Binary content mislabeled as plain text due to an Apache bug
# https://mimesniff.spec.whatwg.org/#check-for-apache-bug-flag
# https://mimesniff.spec.whatwg.org/#rules-for-text-or-binary
#
# - Feeds mislabeled as HTML
# https://mimesniff.spec.whatwg.org/#rules-for-distinguishing-if-a-resource-is-a-feed-or-html
*(
(
{
"body": body,
"headers": Headers({"Content-Type": [content_type]}),
},
response_class,
)
for body, content_type, response_class in (
*(
(b"\x00\x01\xff", content_type, Response)
for content_type in (
"text/plain",
"text/plain; charset=ISO-8859-1",
"text/plain; charset=iso-8859-1",
"text/plain; charset=UTF-8",
)
),
(b"\x00\x01\xff", "text/json", JsonResponse),
*(
(body, "text/html", XmlResponse)
for body in (
b"<rss",
b"<feed",
(
b"<rdf:RDF "
b"... http://purl.org/rss/1.0/ "
b"... http://www.w3.org/1999/02/22-rdf-syntax-ns#"
),
(
b"<rdf:RDF "
b"... http://www.w3.org/1999/02/22-rdf-syntax-ns# "
b"... http://purl.org/rss/1.0/"
),
)
),
)
),
# Compressed content should be of type Response until uncompressed.
(
{
"headers": Headers(
{
"Content-Disposition": [
'attachment; filename="a.html"',
],
"Content-Encoding": ["zip"],
}
)
},
Response,
),
(
{
"body": b"<html>",
"headers": Headers(
{
"Content-Encoding": ["zip"],
}
),
},
Response,
),
# If the body is empty, it contains no binary data bytes, hence body-based
# MIME type detection must interpret the result as text.
#
# https://mimesniff.spec.whatwg.org/#identifying-a-resource-with-an-unknown-mime-type
({}, TextResponse),
# We take the file extension of URL paths into account, except for HTTP
# responses, because “they are unreliable and easily spoofed”.
#
# https://mimesniff.spec.whatwg.org/#interpreting-the-resource-metadata
*(
(
{"url": f"{protocol}://example.com/a.{extension}"},
response_class,
)
for protocol in ("file", "ftp")
for extension, response_class in (
# “Note that XHTML is best parsed as XML”
# https://lxml.de/parsing.html
("xhtml", XmlResponse),
)
),
*(
(
{"url": f"{protocol}://example.com/a.html"},
response_class,
)
for protocol, response_class in (
*((protocol, TextResponse) for protocol in ("http", "https")),
)
),
# File extension triumphs body.
(
{
"body": b"<html>",
"headers": Headers(
{
"Content-Disposition": [
'attachment; filename="a.gz"',
],
}
),
},
Response,
),
(
{
"body": b"<html>",
"url": "file:///a.gz",
},
Response,
),
# Without anything else, the body determines the response class.
*(
({"body": body}, response_class)
for body, response_class in (
# https://mimesniff.spec.whatwg.org/#identifying-a-resource-with-an-unknown-mime-type
*(
(start + b">", HtmlResponse)
for start in (
set_case(start)
for set_case in (bytes.lower, bytes.upper, odd_capitalize)
for start in POST_XTRACTMIME_HTML_STARTS
)
),
*(
(start + b" ", HtmlResponse)
for start in (
set_case(start)
for set_case in (bytes.lower, bytes.upper, odd_capitalize)
for start in chain(
PRE_XTRACTMIME_HTML_STARTS,
POST_XTRACTMIME_HTML_STARTS,
)
)
),
*(
(b"\x0c" + start + b">", HtmlResponse)
for start in (
set_case(start)
for set_case in (bytes.lower, bytes.upper, odd_capitalize)
for start in PRE_XTRACTMIME_HTML_STARTS
)
),
(b"\x0c<?xml", XmlResponse),
(b"%PDF-", Response),
(b"%!PS-Adobe-", Response),
(b"BM", Response),
(b"GIF87a", Response),
(b"GIF89a", Response),
(b"RIFFabcdWEBPVP", Response),
(b"\xff\xd8\xff", Response),
(b"FORMabcdAIFF", Response),
(b"ID3", Response),
(b"OggS\x00", Response),
(b"RIFFabcdAVI ", Response),
(b"RIFFabcdWAVE", Response),
# A body is considered binary if its header (first 1445 bytes)
# contains any binary data byte.
(BINARY_BYTES[0], Response),
*((byte, TextResponse) for byte in (b"\x0c", b"\x1b")),
(b"a" * (RESOURCE_HEADER_BUFFER_LENGTH - 1) + BINARY_BYTES[0], Response),
*(
(b"a" * RESOURCE_HEADER_BUFFER_LENGTH + byte, TextResponse)
for byte in BINARY_BYTES[1:]
),
# HTML and XML detection does not allow for unexpected content
# before document start.
(b"a<html>", TextResponse),
(b"a<?xml", TextResponse),
)
),
# Content triumphs Content-Type when using HTTP or HTTPS and the
# Content-Type is known and binary while the content is plain text. This is
# a conscious divergence from the MIME Sniffing Standard for a better web
# scraping experience.
*(
(
{
"url": (
f"{protocol}://example.com/foo"
if use_header
else f"{protocol}://example.com/foo.{file_extension}"
),
"headers": (
Headers({"Content-Type": content_type}) if use_header else Headers()
),
"body": b"\x00",
},
Response,
)
for protocol, use_header in (
("http", True),
("https", True),
("file", False),
("ftp", False),
)
for content_type, file_extension in (
("application/octet-stream", "bin"),
("application/pdf", "pdf"),
)
),
*(
(
{
"url": f"{protocol}://example.com/foo.{file_extension}",
"body": body,
},
Response,
)
for protocol in ("file", "ftp")
for body in (b"", b"a")
for file_extension in ("bin", "pdf")
),
)
@pytest.mark.parametrize(
("kwargs", "response_class"),
[
*PRE_XTRACTMIME_SCENARIOS,
*POST_XTRACTMIME_SCENARIOS,
],
)
def test_get_response_class_http(kwargs, response_class):
kwargs = dict(kwargs)
if "headers" in kwargs:
kwargs["http_headers"] = kwargs.pop("headers")
assert get_response_class(**kwargs) == response_class
@pytest.mark.parametrize(
("headers", "expected"),
[
*(
(
Headers({"Content-Encoding": content_encoding_header}),
(encoding, None),
)
for content_encoding_header, encoding in (
(["gzip"], b"gzip"),
(["gzip", "compress"], b"compress"),
(["deflate, br"], b"br"),
)
),
],
)
def test_get_encoding_or_mime_type_from_headers(headers, expected):
assert _get_encoding_or_mime_type_from_headers(headers) == expected
def test_open_in_browser():
url = "http://www.example.com/some/page.html"
body = (