diff --git a/scrapy/core/downloader/handlers/datauri.py b/scrapy/core/downloader/handlers/datauri.py index a45b4ff3c..3c09dd246 100644 --- a/scrapy/core/downloader/handlers/datauri.py +++ b/scrapy/core/downloader/handlers/datauri.py @@ -1,8 +1,8 @@ from w3lib.url import parse_data_uri from scrapy.http import TextResponse -from scrapy.responsetypes import responsetypes from scrapy.utils.decorators import defers +from scrapy.utils.response import get_response_class class DataURIDownloadHandler: @@ -11,7 +11,10 @@ class DataURIDownloadHandler: @defers def download_request(self, request, spider): 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(),), + ) resp_kwargs = {} if (issubclass(respcls, TextResponse) diff --git a/scrapy/core/downloader/handlers/file.py b/scrapy/core/downloader/handlers/file.py index 0d94e3df0..ffff915fa 100644 --- a/scrapy/core/downloader/handlers/file.py +++ b/scrapy/core/downloader/handlers/file.py @@ -1,7 +1,7 @@ from w3lib.url import file_uri_to_path -from scrapy.responsetypes import responsetypes from scrapy.utils.decorators import defers +from scrapy.utils.response import get_response_class class FileDownloadHandler: @@ -12,5 +12,5 @@ class FileDownloadHandler: filepath = file_uri_to_path(request.url) with open(filepath, 'rb') as fo: body = fo.read() - respcls = responsetypes.from_args(filename=filepath, body=body) + respcls = get_response_class(url=request.url, body=body) return respcls(url=request.url, body=body) diff --git a/scrapy/core/downloader/handlers/ftp.py b/scrapy/core/downloader/handlers/ftp.py index a495874bd..397ff7b98 100644 --- a/scrapy/core/downloader/handlers/ftp.py +++ b/scrapy/core/downloader/handlers/ftp.py @@ -36,9 +36,9 @@ from twisted.internet.protocol import ClientCreator, Protocol from twisted.protocols.ftp import CommandFailed, FTPClient from scrapy.http import Response -from scrapy.responsetypes import responsetypes from scrapy.utils.httpobj import urlparse_cached from scrapy.utils.python import to_bytes +from scrapy.utils.response import get_response_class class ReceivedDataProtocol(Protocol): @@ -105,7 +105,7 @@ class FTPDownloadHandler: protocol.close() headers = {"local filename": protocol.filename or '', "size": protocol.size} body = to_bytes(protocol.filename or protocol.body.read()) - respcls = responsetypes.from_args(url=request.url, body=body) + respcls = get_response_class(url=request.url, body=body) return respcls(url=request.url, status=200, body=body, headers=headers) def _failed(self, result, request): diff --git a/scrapy/core/downloader/handlers/http11.py b/scrapy/core/downloader/handlers/http11.py index 38935667d..c40389aae 100644 --- a/scrapy/core/downloader/handlers/http11.py +++ b/scrapy/core/downloader/handlers/http11.py @@ -24,8 +24,8 @@ from scrapy.core.downloader.contextfactory import load_context_factory_from_sett from scrapy.core.downloader.webclient import _parse from scrapy.exceptions import ScrapyDeprecationWarning, StopDownload from scrapy.http import Headers -from scrapy.responsetypes import responsetypes from scrapy.utils.python import to_bytes, to_unicode +from scrapy.utils.response import get_response_class logger = logging.getLogger(__name__) @@ -449,7 +449,7 @@ class ScrapyAgent: def _cb_bodydone(self, result, request, url): headers = self._headers_from_twisted_response(result["txresponse"]) - respcls = responsetypes.from_args(headers=headers, url=url, body=result["body"]) + respcls = get_response_class(http_headers=headers, url=url, body=result["body"]) try: version = result["txresponse"].version protocol = f"{to_unicode(version[0])}/{version[1]}.{version[2]}" diff --git a/scrapy/core/downloader/webclient.py b/scrapy/core/downloader/webclient.py index 7d048c1e4..a97ef7027 100644 --- a/scrapy/core/downloader/webclient.py +++ b/scrapy/core/downloader/webclient.py @@ -9,7 +9,7 @@ from twisted.internet.protocol import ClientFactory from scrapy.http import Headers from scrapy.utils.httpobj import urlparse_cached from scrapy.utils.python import to_bytes, to_unicode -from scrapy.responsetypes import responsetypes +from scrapy.utils.response import get_response_class def _parsed_url_args(parsed): @@ -112,7 +112,7 @@ class ScrapyHTTPClientFactory(ClientFactory): request.meta['download_latency'] = self.headers_time - self.start_time status = int(self.status) headers = Headers(self.response_headers) - respcls = responsetypes.from_args(headers=headers, url=self._url, body=body) + respcls = get_response_class(http_headers=headers, url=self._url, body=body) return respcls(url=self._url, status=status, headers=headers, body=body, protocol=to_unicode(self.version)) def _set_connection_attributes(self, request): diff --git a/scrapy/core/http2/stream.py b/scrapy/core/http2/stream.py index 5c393c027..a36c8c36c 100644 --- a/scrapy/core/http2/stream.py +++ b/scrapy/core/http2/stream.py @@ -14,7 +14,7 @@ from twisted.web.client import ResponseFailed from scrapy.http import Request from scrapy.http.headers import Headers -from scrapy.responsetypes import responsetypes +from scrapy.utils.response import get_response_class if TYPE_CHECKING: from scrapy.core.http2.protocol import H2ClientProtocol @@ -450,8 +450,8 @@ class Stream: generated response instance""" body = self._response['body'].getvalue() - response_cls = responsetypes.from_args( - headers=self._response['headers'], + response_cls = get_response_class( + http_headers=self._response['headers'], url=self._request.url, body=body, ) diff --git a/scrapy/downloadermiddlewares/decompression.py b/scrapy/downloadermiddlewares/decompression.py index 0fcf8fb8c..389755d12 100644 --- a/scrapy/downloadermiddlewares/decompression.py +++ b/scrapy/downloadermiddlewares/decompression.py @@ -10,7 +10,7 @@ import zipfile from io import BytesIO from tempfile import mktemp -from scrapy.responsetypes import responsetypes +from scrapy.utils.response import get_response_class logger = logging.getLogger(__name__) @@ -36,7 +36,7 @@ class DecompressionMiddleware: return body = tar_file.extractfile(tar_file.members[0]).read() - respcls = responsetypes.from_args(filename=tar_file.members[0].name, body=body) + respcls = get_response_class(url=tar_file.members[0].name, body=body) return response.replace(body=body, cls=respcls) def _is_zip(self, response): @@ -48,7 +48,7 @@ class DecompressionMiddleware: namelist = zip_file.namelist() body = zip_file.read(namelist[0]) - respcls = responsetypes.from_args(filename=namelist[0], body=body) + respcls = get_response_class(url=namelist[0], body=body) return response.replace(body=body, cls=respcls) def _is_gzip(self, response): @@ -58,7 +58,7 @@ class DecompressionMiddleware: except IOError: return - respcls = responsetypes.from_args(body=body) + respcls = get_response_class(body=body) return response.replace(body=body, cls=respcls) def _is_bzip2(self, response): @@ -67,7 +67,7 @@ class DecompressionMiddleware: except IOError: return - respcls = responsetypes.from_args(body=body) + respcls = get_response_class(body=body) return response.replace(body=body, cls=respcls) def process_response(self, request, response, spider): diff --git a/scrapy/downloadermiddlewares/httpcompression.py b/scrapy/downloadermiddlewares/httpcompression.py index 4e7feeeaf..11407ca46 100644 --- a/scrapy/downloadermiddlewares/httpcompression.py +++ b/scrapy/downloadermiddlewares/httpcompression.py @@ -4,9 +4,9 @@ import zlib from scrapy.exceptions import NotConfigured from scrapy.http import Response, TextResponse -from scrapy.responsetypes import responsetypes from scrapy.utils.deprecate import ScrapyDeprecationWarning from scrapy.utils.gz import gunzip +from scrapy.utils.response import get_response_class ACCEPTED_ENCODINGS = [b'gzip', b'deflate'] @@ -63,8 +63,10 @@ class HttpCompressionMiddleware: if self.stats: self.stats.inc_value('httpcompression/response_bytes', len(decoded_body), spider=spider) self.stats.inc_value('httpcompression/response_count', spider=spider) - 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(cls=respcls, body=decoded_body) if issubclass(respcls, TextResponse): diff --git a/scrapy/extensions/httpcache.py b/scrapy/extensions/httpcache.py index 843e14812..a5fe87232 100644 --- a/scrapy/extensions/httpcache.py +++ b/scrapy/extensions/httpcache.py @@ -10,10 +10,10 @@ from weakref import WeakKeyDictionary from w3lib.http import headers_raw_to_dict, headers_dict_to_raw 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 logger = logging.getLogger(__name__) @@ -240,7 +240,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) response = respcls(url=url, headers=headers, status=status, body=body) return response @@ -299,7 +299,7 @@ class FilesystemCacheStorage: url = metadata.get('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) response = respcls(url=url, headers=headers, status=status, body=body) return response diff --git a/scrapy/http/request/form.py b/scrapy/http/request/form.py index 0c947565a..77341962f 100644 --- a/scrapy/http/request/form.py +++ b/scrapy/http/request/form.py @@ -15,7 +15,6 @@ from w3lib.html import strip_html5_whitespace from scrapy.http.request import Request from scrapy.http.response.text import TextResponse from scrapy.utils.python import to_bytes, is_listlike -from scrapy.utils.response import get_base_url FormRequestTypeVar = TypeVar("FormRequestTypeVar", bound="FormRequest") @@ -98,7 +97,7 @@ def _get_form( formxpath: Optional[str], ) -> FormElement: """Find the wanted form element within the given response.""" - root = create_root_node(response.text, HTMLParser, base_url=get_base_url(response)) + root = create_root_node(response.text, HTMLParser, base_url=response.base_url) forms = root.xpath('//form') if not forms: raise ValueError(f"No
element found in {response}") diff --git a/scrapy/http/response/text.py b/scrapy/http/response/text.py index 89516b9b6..e1cae274f 100644 --- a/scrapy/http/response/text.py +++ b/scrapy/http/response/text.py @@ -13,12 +13,11 @@ from urllib.parse import urljoin import parsel from w3lib.encoding import (html_body_declared_encoding, html_to_unicode, http_content_type_encoding, resolve_encoding) -from w3lib.html import strip_html5_whitespace +from w3lib.html import get_base_url, strip_html5_whitespace from scrapy.http import Request from scrapy.http.response import Response from scrapy.utils.python import memoizemethod_noargs, to_unicode -from scrapy.utils.response import get_base_url _NONE = object() @@ -32,6 +31,7 @@ class TextResponse(Response): def __init__(self, *args, **kwargs): self._encoding = kwargs.pop('encoding', None) + self._cached_base_url = None self._cached_benc = None self._cached_ubody = None self._cached_selector = None @@ -42,6 +42,7 @@ class TextResponse(Response): self._url = to_unicode(url, self.encoding) else: super()._set_url(url) + self._cached_base_url = None def _set_body(self, body): self._body = b'' # used by encoding detection @@ -52,6 +53,7 @@ class TextResponse(Response): self._body = body.encode(self._encoding) else: super()._set_body(body) + self._cached_base_url = None @property def encoding(self): @@ -85,10 +87,21 @@ 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""" + 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): """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): diff --git a/scrapy/linkextractors/lxmlhtml.py b/scrapy/linkextractors/lxmlhtml.py index b5d2585a8..6c41758a1 100644 --- a/scrapy/linkextractors/lxmlhtml.py +++ b/scrapy/linkextractors/lxmlhtml.py @@ -13,7 +13,6 @@ from scrapy.link import Link from scrapy.linkextractors import FilteringLinkExtractor 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 lxml/src/lxml/html/__init__.py @@ -82,8 +81,12 @@ class LxmlParserLinkExtractor: return self._deduplicate_if_needed(links) def extract_links(self, response): - base_url = get_base_url(response) - return self._extract_links(response.selector, response.url, response.encoding, base_url) + return self._extract_links( + response.selector, + response.url, + response.encoding, + response.base_url, + ) def _process_links(self, links): """ Normalize and filter extracted links @@ -148,7 +151,6 @@ class LxmlLinkExtractor(FilteringLinkExtractor): Duplicate links are omitted. """ - base_url = get_base_url(response) if self.restrict_xpaths: docs = [ subdoc @@ -159,6 +161,11 @@ class LxmlLinkExtractor(FilteringLinkExtractor): 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)) return unique_list(all_links) diff --git a/scrapy/responsetypes.py b/scrapy/responsetypes.py index 1bf21ca81..7308e0350 100644 --- a/scrapy/responsetypes.py +++ b/scrapy/responsetypes.py @@ -2,140 +2,13 @@ This module implements a class which returns the appropriate Response class based on different criteria. """ -from mimetypes import MimeTypes -from pkgutil import get_data -from io import StringIO -from urllib.parse import urlparse from warnings import warn -from xtractmime import ( - RESOURCE_HEADER_BUFFER_LENGTH, - extract_mime, - is_binary_data, -) -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 HtmlResponse, Response, TextResponse, XmlResponse +from scrapy.http import Response from scrapy.utils.misc import load_object from scrapy.utils.python import binary_is_text, to_bytes, to_unicode - - -_CONTENT_ENCODING_MIME_TYPES = { - 'br': b'application/brotli', - 'deflate': b'application/zip', - 'gzip': b'application/gzip', -} -_MIME_TYPES = MimeTypes() -_scrapy_mime_data = get_data('scrapy', 'mime.types') or b'' -_MIME_TYPES.readfp(StringIO(_scrapy_mime_data.decode())) - - -def _is_other_text_mime_type(mime_type): - value = ( - mime_type.startswith(b'text/') - or is_json_mime_type(mime_type) - or is_javascript_mime_type(mime_type) - or mime_type in ( - b'application/x-json', - b'application/json-amazonui-streaming', - b'application/x-javascript', - ) - ) - return value - - -_PRIORITIZED_MIME_TYPE_CHECKERS = ( - is_html_mime_type, - is_xml_mime_type, - _is_other_text_mime_type, -) - - -def _content_type_from_metadata(*, headers=None, url_path=None, filename=None): - if headers and b'Content-Type' in headers: - content_type_mime_type = headers.getlist(b'Content-Type')[-1] - else: - content_type_mime_type = None - - if headers and b'Content-Disposition' in headers: - _filename = ( - headers.get(b"Content-Disposition") - .split(b";")[-1] - .split(b"=")[-1] - .strip(b"\"'") - .decode() - ) - content_disposition_mime_type = _mime_type_from_path(_filename) - else: - content_disposition_mime_type = None - - filename_mime_type = _mime_type_from_path(filename) if filename else None - url_mime_type = _mime_type_from_path(url_path) if url_path else None - - candidate_mime_types = tuple( - mime_type - for mime_type in ( - content_type_mime_type, - content_disposition_mime_type, - filename_mime_type, - url_mime_type, - ) - if mime_type is not None - ) - for mime_type_checker in _PRIORITIZED_MIME_TYPE_CHECKERS: - for candidate_mime_type in candidate_mime_types: - if mime_type_checker(candidate_mime_type): - return candidate_mime_type - return ( - content_type_mime_type - or content_disposition_mime_type - or filename_mime_type - or url_mime_type - ) - - -def _mime_type_from_path(path): - mimetype, encoding = _MIME_TYPES.guess_type(path, strict=False) - encoding_mime_type = _CONTENT_ENCODING_MIME_TYPES.get(encoding, None) - if encoding_mime_type: - return encoding_mime_type - if mimetype: - return mimetype.encode() - return None - - -def _remove_nul_byte_from_text(text): - """Return the text with removed null byte (b'\x00') if there are no other - binary bytes in the text, otherwise return the text as-is. - - Based on https://github.com/scrapy/scrapy/issues/2481 - """ - for index in range(len(text)): - if ( - text[index:index + 1] != b'\x00' - and is_binary_data(text[index:index + 1]) - ): - return text - - return text.replace(b'\x00', b'') - - -def _response_type_from_mime_type(mime_type): - 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_other_text_mime_type(mime_type): - return TextResponse - return Response +from scrapy.utils.response import _MIME_TYPES class ResponseTypes: @@ -159,11 +32,14 @@ class ResponseTypes: def __init__(self): self.classes = {} + self.mimetypes = _MIME_TYPES for mimetype, cls in self.CLASSES.items(): self.classes[mimetype] = load_object(cls) def from_mimetype(self, mimetype): """Return the most appropriate Response class for the given mimetype""" + warn('ResponseTypes.from_mimetype is deprecated, ' + 'please use scrapy.utils.response.get_response_class instead', ScrapyDeprecationWarning) if mimetype is None: return Response elif mimetype in self.classes: @@ -176,7 +52,7 @@ class ResponseTypes: """Return the most appropriate Response class from an HTTP Content-Type header """ warn('ResponseTypes.from_content_type is deprecated, ' - 'please use ResponseTypes.from_args instead', ScrapyDeprecationWarning) + 'please use scrapy.utils.response.get_response_class instead', ScrapyDeprecationWarning) if content_encoding: return Response mimetype = to_unicode(content_type).split(';')[0].strip().lower() @@ -184,7 +60,7 @@ class ResponseTypes: def from_content_disposition(self, content_disposition): warn('ResponseTypes.from_content_disposition is deprecated, ' - 'please use ResponseTypes.from_args instead', ScrapyDeprecationWarning) + 'please use scrapy.utils.response.get_response_class instead', ScrapyDeprecationWarning) try: filename = to_unicode( content_disposition, encoding='latin-1', errors='replace' @@ -197,7 +73,7 @@ class ResponseTypes: """Return the most appropriate Response class by looking at the HTTP headers""" warn('ResponseTypes.from_headers is deprecated, ' - 'please use ResponseTypes.from_args instead', ScrapyDeprecationWarning) + 'please use scrapy.utils.response.get_response_class instead', ScrapyDeprecationWarning) cls = Response if b'Content-Type' in headers: cls = self.from_content_type( @@ -211,8 +87,8 @@ class ResponseTypes: def from_filename(self, filename): """Return the most appropriate Response class from a file name""" warn('ResponseTypes.from_filename is deprecated, ' - 'please use ResponseTypes.from_args instead', ScrapyDeprecationWarning) - mimetype, encoding = _MIME_TYPES.guess_type(filename) + 'please use scrapy.utils.response.get_response_class instead', ScrapyDeprecationWarning) + mimetype, encoding = self.mimetypes.guess_type(filename) if mimetype and not encoding: return self.from_mimetype(mimetype) else: @@ -224,7 +100,7 @@ class ResponseTypes: it's not meant to be used except for special cases where response types cannot be guess using more straightforward methods.""" warn('ResponseTypes.from_body is deprecated, ' - 'please use ResponseTypes.from_args instead', ScrapyDeprecationWarning) + 'please use scrapy.utils.response.get_response_class instead', ScrapyDeprecationWarning) chunk = body[:5000] chunk = to_bytes(chunk) if not binary_is_text(chunk): @@ -241,22 +117,18 @@ class ResponseTypes: def from_args(self, headers=None, url=None, filename=None, body=None): """Guess the most appropriate Response class based on the given arguments.""" - body = body or b'' - body = _remove_nul_byte_from_text(body[:RESOURCE_HEADER_BUFFER_LENGTH]) - url_parts = urlparse(url) if url else url - content_type = _content_type_from_metadata( - headers=headers, - url_path=url_parts.path if url_parts else url_parts, - filename=filename, - ) - content_types = (content_type,) if content_type else None - http_origin = not url or url_parts.scheme in ("http", "https") - mime_type = extract_mime( - body, - content_types=content_types, - http_origin=http_origin, - ) - return _response_type_from_mime_type(mime_type) + warn('ResponseTypes.from_args is deprecated, ' + 'please use scrapy.utils.response.get_response_class instead', ScrapyDeprecationWarning) + cls = Response + if headers is not None: + cls = self.from_headers(headers) + if cls is Response and url is not None: + cls = self.from_filename(url) + if cls is Response and filename is not None: + cls = self.from_filename(filename) + if cls is Response and body is not None: + cls = self.from_body(body) + return cls responsetypes = ResponseTypes() diff --git a/scrapy/utils/response.py b/scrapy/utils/response.py index 741dce350..6d1b575b8 100644 --- a/scrapy/utils/response.py +++ b/scrapy/utils/response.py @@ -6,27 +6,159 @@ import os import re import tempfile import webbrowser -from typing import Any, Callable, Iterable, Optional, Tuple, Union +from io import StringIO +from mimetypes import MimeTypes +from pkgutil import get_data +from typing import ( + Any, + Callable, + Iterable, + Optional, + Sequence, + Tuple, + Type, + Union, +) +from urllib.parse import urlparse +from warnings import warn from weakref import WeakKeyDictionary -import scrapy -from scrapy.http.response import Response - from twisted.web import http -from scrapy.utils.python import to_bytes, to_unicode -from scrapy.utils.decorators import deprecated from w3lib import html +from xtractmime import ( + RESOURCE_HEADER_BUFFER_LENGTH as BODY_LIMIT, + extract_mime, + is_binary_data, +) +from xtractmime.mimegroups import ( + is_html_mime_type, + is_javascript_mime_type, + is_json_mime_type, + is_xml_mime_type, +) + +import scrapy +from scrapy.exceptions import ScrapyDeprecationWarning +from scrapy.http import ( + Headers, + HtmlResponse, + Response, + TextResponse, + XmlResponse, +) +from scrapy.utils.decorators import deprecated +from scrapy.utils.python import to_bytes, to_unicode _baseurl_cache: "WeakKeyDictionary[Response, str]" = WeakKeyDictionary() +_CONTENT_ENCODING_MIME_TYPES = { + 'br': b'application/brotli', + 'deflate': b'application/zip', + 'gzip': b'application/gzip', +} +_MIME_TYPES = MimeTypes() +_mime_overrides = get_data('scrapy', 'mime.types') or b'' +_MIME_TYPES.readfp(StringIO(_mime_overrides.decode())) -def get_base_url(response: "scrapy.http.response.text.TextResponse") -> str: +def _is_other_text_mime_type(mime_type): + return ( + mime_type.startswith(b'text/') + or is_json_mime_type(mime_type) + or is_javascript_mime_type(mime_type) + or mime_type in ( + b'application/x-json', + b'application/json-amazonui-streaming', + b'application/x-javascript', + ) + ) + + +_PRIORITIZED_MIME_TYPE_CHECKERS = ( + is_html_mime_type, + is_xml_mime_type, + _is_other_text_mime_type, +) + + +def _get_best_mime_type(mime_types): + candidate_mime_types = tuple( + mime_type + for mime_type in mime_types + if mime_type is not None + ) + for mime_type_checker in _PRIORITIZED_MIME_TYPE_CHECKERS: + for candidate_mime_type in candidate_mime_types: + if mime_type_checker(candidate_mime_type): + return candidate_mime_type + return mime_types[0] + + +def _get_http_header_mime_types(headers: Headers) -> Sequence[bytes]: + mime_types = [] + if b'Content-Type' in headers: + mime_types.append(headers[b'Content-Type'].split(b';')[0]) + if b'Content-Disposition' in headers: + path = ( + headers.get(b"Content-Disposition") + .split(b";")[-1] + .split(b"=")[-1] + .strip(b"\"'") + .decode() + ) + mime_types.append(_get_mime_type_from_path(path)) + return mime_types + + +def _get_mime_type_from_path(path): + mimetype, encoding = _MIME_TYPES.guess_type(path, strict=False) + encoding_mime_type = _CONTENT_ENCODING_MIME_TYPES.get(encoding, None) + if encoding_mime_type: + return encoding_mime_type + if mimetype: + return mimetype.encode() + return None + + +def _get_response_class_from_mime_type(mime_type): + 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_other_text_mime_type(mime_type): + return TextResponse + return Response + + +def _remove_nul_byte_from_text(text): + """Return the text with removed null byte (b'\x00') if there are no other + binary bytes in the text, otherwise return the text as-is. + + Based on https://github.com/scrapy/scrapy/issues/2481 + """ + for index in range(len(text)): + if ( + text[index:index + 1] != b'\x00' + and is_binary_data(text[index:index + 1]) + ): + return text + + return text.replace(b'\x00', b'') + + +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] + warn( + ( + "scrapy.utils.response.get_base_url is deprecated, use " + "scrapy.http.TextResponse.base_url instead." + ), + ScrapyDeprecationWarning, + stacklevel=2, + ) + return response.base_url _metaref_cache: "WeakKeyDictionary[Response, Union[Tuple[None, None], Tuple[float, str]]]" = WeakKeyDictionary() @@ -44,6 +176,38 @@ def get_meta_refresh( return _metaref_cache[response] +def get_response_class( + *, + url: str = None, + body: bytes = None, + declared_mime_types: Sequence[bytes] = None, + http_headers: Headers = None, +) -> Type[Response]: + """Guess the most appropriate Response class based on the given + arguments.""" + mime_types = list(declared_mime_types or []) + if http_headers: + mime_types.extend(_get_http_header_mime_types(http_headers)) + if url is not None: + url_parts = urlparse(url) + http_origin = url_parts.scheme in ("http", "https") + mime_types.append(_get_mime_type_from_path(url_parts.path)) + else: + http_origin = True + body = _remove_nul_byte_from_text((body or b'')[:BODY_LIMIT]) + if mime_types: + best_mime_type = _get_best_mime_type(mime_types) + content_types = (best_mime_type,) if best_mime_type else best_mime_type + else: + content_types = None + mime_type = extract_mime( + body, + content_types=content_types, + http_origin=http_origin, + ) + return _get_response_class_from_mime_type(mime_type) + + def response_status_message(status: Union[bytes, float, int, str]) -> str: """Return status code plus status text descriptive message """ diff --git a/tests/test_downloader_handlers.py b/tests/test_downloader_handlers.py index 72f52121e..e9f890391 100644 --- a/tests/test_downloader_handlers.py +++ b/tests/test_downloader_handlers.py @@ -27,7 +27,6 @@ from scrapy.core.downloader.handlers.s3 import S3DownloadHandler from scrapy.exceptions import NotConfigured, ScrapyDeprecationWarning from scrapy.http import Headers, HtmlResponse, Request from scrapy.http.response.text import TextResponse -from scrapy.responsetypes import responsetypes from scrapy.spiders import Spider from scrapy.utils.misc import create_instance from scrapy.utils.python import to_bytes @@ -1190,7 +1189,7 @@ class DataURITestCase(unittest.TestCase): def test_default_mediatype_encoding(self): def _test(response): self.assertEqual(response.text, 'A brief note') - self.assertEqual(type(response), responsetypes.from_mimetype("text/plain")) + self.assertIsInstance(response, TextResponse) self.assertEqual(response.encoding, "US-ASCII") request = Request("data:,A%20brief%20note") @@ -1199,7 +1198,7 @@ class DataURITestCase(unittest.TestCase): def test_default_mediatype(self): def _test(response): self.assertEqual(response.text, '\u038e\u03a3\u038e') - self.assertEqual(type(response), responsetypes.from_mimetype("text/plain")) + self.assertIsInstance(response, TextResponse) self.assertEqual(response.encoding, "iso-8859-7") request = Request("data:;charset=iso-8859-7,%be%d3%be") @@ -1217,7 +1216,7 @@ class DataURITestCase(unittest.TestCase): def test_mediatype_parameters(self): def _test(response): self.assertEqual(response.text, '\u038e\u03a3\u038e') - self.assertEqual(type(response), responsetypes.from_mimetype("text/plain")) + self.assertIsInstance(response, TextResponse) self.assertEqual(response.encoding, "utf-8") request = Request('data:text/plain;foo=%22foo;bar%5C%22%22;' diff --git a/tests/test_downloadermiddleware_httpcompression.py b/tests/test_downloadermiddleware_httpcompression.py index 40e9f3a96..4a670742a 100644 --- a/tests/test_downloadermiddleware_httpcompression.py +++ b/tests/test_downloadermiddleware_httpcompression.py @@ -8,8 +8,8 @@ from scrapy.spiders import Spider from scrapy.http import Response, Request, HtmlResponse from scrapy.downloadermiddlewares.httpcompression import HttpCompressionMiddleware, ACCEPTED_ENCODINGS from scrapy.exceptions import NotConfigured, ScrapyDeprecationWarning -from scrapy.responsetypes import responsetypes 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 from w3lib.encoding import resolve_encoding @@ -247,7 +247,11 @@ class HttpCompressionTest(TestCase): } plainbody = (b'Some page' b'') - 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) request = Request("http://www.example.com/index") diff --git a/tests/test_http_response.py b/tests/test_http_response.py index 2986f884f..05f2380e9 100644 --- a/tests/test_http_response.py +++ b/tests/test_http_response.py @@ -500,7 +500,7 @@ class TextResponseTest(BaseResponseTest): ) def test_urljoin_with_base_url(self): - """Test urljoin shortcut which also evaluates base-url through get_base_url().""" + """Test urljoin shortcut which also evaluates base-url.""" body = b'' joined = self.response_class('http://www.example.com', body=body).urljoin('/test') absolute = 'https://example.net/test' @@ -697,6 +697,18 @@ class HtmlResponseTest(TextResponseTest): response_class = HtmlResponse + def test_base_url(self): + resp = HtmlResponse("http://www.example.com", body=b""" + + + blahablsdfsal& + """) + self.assertEqual(resp.base_url, "http://www.example.com/img/") + + resp2 = HtmlResponse("http://www.example.com", body=b""" + blahablsdfsal&""") + self.assertEqual(resp2.base_url, "http://www.example.com") + def test_html_encoding(self): body = b"""Some page diff --git a/tests/test_responsetypes.py b/tests/test_responsetypes.py index 075b0ad22..3d5217385 100644 --- a/tests/test_responsetypes.py +++ b/tests/test_responsetypes.py @@ -1,5 +1,7 @@ import unittest +import pytest + from scrapy.http import ( Headers, HtmlResponse, @@ -7,24 +9,37 @@ from scrapy.http import ( TextResponse, XmlResponse, ) -from scrapy.responsetypes import _MIME_TYPES, responsetypes, ResponseTypes +from scrapy.responsetypes import responsetypes +from .test_utils_response import ( + POST_XTRACTMIME_SCENARIOS, + PRE_XTRACTMIME_SCENARIOS, +) -class PreXtractmimeResponseTypes(ResponseTypes): - def from_args(self, headers=None, url=None, filename=None, body=None): - cls = Response - if headers is not None: - cls = self.from_headers(headers) - if cls is Response and url is not None: - cls = self.from_filename(url) - if cls is Response and filename is not None: - cls = self.from_filename(filename) - if cls is Response and body is not None: - cls = self.from_body(body) - return cls - - -_PRE_XTRACTMIME_RESPONSE_TYPES = PreXtractmimeResponseTypes() +@pytest.mark.parametrize( + "kwargs,response_class", + ( + *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 class ResponseTypesTest(unittest.TestCase): @@ -77,6 +92,8 @@ class ResponseTypesTest(unittest.TestCase): (b'\x03\x02\xdf\xdd\x23', Response), (b'Some plain text\ndata with tabs\t and null bytes\0', TextResponse), (b'Hello', HtmlResponse), + # https://codersblock.com/blog/the-smallest-valid-html5-page/ + (b'\n.', HtmlResponse), (b' {retcls} != {cls}" - def test_from_args_pre_xtractmime(self): - """Each of the following test cases remains unaffected after - using xtractmime for MIME sniffing""" - mappings = [ - ({'url': 'http://www.example.com/data.csv'}, TextResponse), - ({'headers': Headers({'Content-Type': ['text/html; charset=utf-8']}), - 'url': 'http://www.example.com/item/'}, HtmlResponse), - ({'body': b'\x01\x02', 'headers': Headers({'Content-Disposition': ['attachment; filename="data.xml.gz"']}), - 'url': 'http://www.example.com/page/'}, Response), - ({'body': b'Some plain\0 text data with\0 tabs and null bytes\0'}, TextResponse), - ({'body': b'\x03\x02\xdf\xdd\x23', 'headers': Headers({'Content-Encoding': 'UTF-8'})}, - Response), - ({'body': b'\x00\x01\xff', 'url': '://www.example.com/item/', - 'headers': Headers({'Content-Type': ['text/plain']})}, TextResponse), - ({'url': 'http://www.example.com/item/file.html'}, HtmlResponse), - ({'body': b'Hello'}, HtmlResponse), - ({'body': b'\n.'}, HtmlResponse), - ({'body': b'\x01\x02', 'headers': Headers({'Content-Type': ['application/pdf']})}, Response), - ({'headers': Headers({'Content-Type': ['application/x-json']})}, TextResponse), - ({'headers': Headers({'Content-Type': ['application/x-javascript']})}, TextResponse), - ({'headers': Headers({'Content-Type': ['application/json-amazonui-streaming']})}, TextResponse), - ({'headers': Headers({'Content-Disposition': ['attachment; filename="data.xml.gz"']}), - 'url': 'http://www.example.com/page/'}, Response), - ({'headers': Headers({'Content-Type': ['application/pdf']})}, Response), - ({'url': 'http://www.example.com/page/file.html', - 'headers': Headers({'Content-Type': 'application/octet-stream'})}, HtmlResponse), - ({'url': 'http://www.example.com/item/file.xml', - 'headers': Headers({'Content-Type': 'application/octet-stream'})}, XmlResponse), - ({'url': 'http://www.example.com/item/file.html', - 'headers': Headers({'Content-Type': 'application/octet-stream'})}, HtmlResponse), - ({'url': 'http://www.example.com/item/file.xml', - 'headers': Headers({'Content-Disposition': ['attachment; filename="data.xml.gz"'], - 'Content-Type': 'application/octet-stream'})}, XmlResponse), - ({'url': 'http://www.example.com/item/file.pdf'}, Response), - ({'filename': 'file.pdf'}, Response), - ({'body': b'\n.', 'url': 'http://www.example.com'}, HtmlResponse), - ] - for source, cls in mappings: - old_cls = _PRE_XTRACTMIME_RESPONSE_TYPES.from_args(**source) - new_cls = responsetypes.from_args(**source) - message = f"{source} ==> {old_cls} (old) != {new_cls} (current)" - assert old_cls == new_cls, message - assert new_cls is cls, f"{source} ==> {new_cls} != {cls}" - - def test_from_args_post_xtractmime(self): - """Each of the following test cases got affected after - using xtractmime for MIME sniffing""" - mappings = [ - ({'body': b'\x00\x01\xff', 'url': 'http://www.example.com/item/', - 'headers': Headers({'Content-Type': ['text/plain']})}, Response), - ({'filename': '/tmp/temp^'}, TextResponse), - ({'body': b'%PDF-1.4'}, Response), - ({'headers': Headers({'Content-Type': ['application/ecmascript']})}, TextResponse), - ({'headers': Headers({'Content-Type': ['application/ld+json']})}, TextResponse), - ({'headers': Headers({'Content-Encoding': ['zip'], 'Content-Type': ['text/html']})}, - HtmlResponse), - ({'headers': Headers({'Content-Encoding': ['zip'], 'Content-Type': ['text/plain']})}, - TextResponse), - ({'body': b'Non HTML', 'headers': Headers({'Content-Encoding': ['zip'], - 'Content-Type': ['text/html']})}, HtmlResponse), - ({'body': b'Some plain text', 'headers': Headers({'Content-Type': 'application/octet-stream'})}, - Response), - ({'body': b'\x0c\x1b'}, TextResponse), - ({'body': b'this is not '}, TextResponse), - ({'body': b'this is not {old_cls} (old) == {new_cls} (current)" - assert old_cls != new_cls, message - assert new_cls is cls, f"{source} ==> {new_cls} != {cls}" - def test_custom_mime_types_loaded(self): - """Check that mime.types files shipped with Scrapy are loaded.""" - self.assertEqual( - _MIME_TYPES.guess_type('x.scrapytest')[0], - 'x-scrapy/test', - ) + # check that mime.types files shipped with scrapy are loaded + self.assertEqual(responsetypes.mimetypes.guess_type('x.scrapytest')[0], 'x-scrapy/test') if __name__ == "__main__": diff --git a/tests/test_utils_response.py b/tests/test_utils_response.py index 0a09f6109..f67fdb40e 100644 --- a/tests/test_utils_response.py +++ b/tests/test_utils_response.py @@ -2,15 +2,235 @@ import os import unittest from urllib.parse import urlparse -from scrapy.http import Response, TextResponse, HtmlResponse +import pytest + +from scrapy.http import HtmlResponse, Response, TextResponse, XmlResponse +from scrapy.http.headers import Headers from scrapy.utils.python import to_bytes -from scrapy.utils.response import (response_httprepr, open_in_browser, - get_meta_refresh, get_base_url, response_status_message) +from scrapy.utils.response import ( + get_meta_refresh, + get_response_class, + open_in_browser, + response_httprepr, + response_status_message, +) __doctests__ = ['scrapy.utils.response'] +# Scenarios that work the same with the previously-used, deprecated +# scrapy.responsetypes.responsetypes.from_args +PRE_XTRACTMIME_SCENARIOS = ( + ( + { + 'url': 'http://www.example.com/data.csv', + }, + TextResponse, + ), + ( + { + 'url': 'http://www.example.com/item/', + 'headers': Headers({'Content-Type': ['text/html; charset=utf-8']}), + }, + HtmlResponse, + ), + ( + { + 'url': 'http://www.example.com/page/', + 'headers': Headers( + { + 'Content-Disposition': [ + 'attachment; filename="data.xml.gz"', + ] + } + ), + 'body': b'\x01\x02', + }, + Response, + ), + ( + {'body': b'Some plain\0 text data with\0 tabs and null bytes\0'}, + TextResponse, + ), + ( + { + 'body': b'\x03\x02\xdf\xdd\x23', + 'headers': Headers({'Content-Encoding': 'UTF-8'}), + }, + Response, + ), + ( + { + 'body': b'\x00\x01\xff', + 'url': '://www.example.com/item/', + 'headers': Headers({'Content-Type': ['text/plain']}), + }, + TextResponse, + ), + ({'url': 'http://www.example.com/item/file.html'}, HtmlResponse), + ({'body': b'Hello'}, HtmlResponse), + ({'body': b'\n.'}, HtmlResponse), + ( + { + 'body': b'\x01\x02', + 'headers': Headers({'Content-Type': ['application/pdf']}), + }, + Response, + ), + ( + {'headers': Headers({'Content-Type': ['application/x-json']})}, + TextResponse, + ), + ( + {'headers': Headers({'Content-Type': ['application/x-javascript']})}, + TextResponse, + ), + ( + { + 'headers': Headers( + {'Content-Type': ['application/json-amazonui-streaming']} + ) + }, + TextResponse, + ), + ( + { + 'headers': Headers( + {'Content-Disposition': ['attachment; filename="data.xml.gz"']} + ), + 'url': 'http://www.example.com/page/', + }, + Response, + ), + ({'headers': Headers({'Content-Type': ['application/pdf']})}, Response), + ( + { + 'url': 'http://www.example.com/page/file.html', + 'headers': Headers({'Content-Type': 'application/octet-stream'}), + }, + HtmlResponse, + ), + ( + { + 'url': 'http://www.example.com/item/file.xml', + 'headers': Headers({'Content-Type': 'application/octet-stream'}), + }, + XmlResponse, + ), + ( + { + 'url': 'http://www.example.com/item/file.html', + 'headers': Headers({'Content-Type': 'application/octet-stream'}), + }, + HtmlResponse, + ), + ( + { + 'url': 'http://www.example.com/item/file.xml', + 'headers': Headers( + { + 'Content-Disposition': [ + 'attachment; filename="data.xml.gz"' + ], + 'Content-Type': 'application/octet-stream', + } + ), + }, + XmlResponse, + ), + ({'url': 'http://www.example.com/item/file.pdf'}, Response), + ({'filename': 'file.pdf'}, Response), + ( + { + 'body': b'\n.', + 'url': 'http://www.example.com', + }, + HtmlResponse, + ), +) + +# Scenarios that work differently with the previously-used, deprecated +# scrapy.responsetypes.responsetypes.from_args +POST_XTRACTMIME_SCENARIOS = ( + ( + { + 'body': b'\x00\x01\xff', + 'url': 'http://www.example.com/item/', + 'headers': Headers({'Content-Type': ['text/plain']}), + }, + Response, + ), + ({'filename': '/tmp/temp^'}, TextResponse), + ({'body': b'%PDF-1.4'}, Response), + ( + {'headers': Headers({'Content-Type': ['application/ecmascript']})}, + TextResponse, + ), + ( + {'headers': Headers({'Content-Type': ['application/ld+json']})}, + TextResponse, + ), + ( + { + 'headers': Headers( + {'Content-Encoding': ['zip'], 'Content-Type': ['text/html']} + ) + }, + HtmlResponse, + ), + ( + { + 'headers': Headers( + {'Content-Encoding': ['zip'], 'Content-Type': ['text/plain']} + ) + }, + TextResponse, + ), + ( + { + 'body': b'Non HTML', + 'headers': Headers( + {'Content-Encoding': ['zip'], 'Content-Type': ['text/html']} + ), + }, + HtmlResponse, + ), + ( + { + 'body': b'Some plain text', + 'headers': Headers({'Content-Type': 'application/octet-stream'}), + }, + Response, + ), + ({'body': b'\x0c\x1b'}, TextResponse), + ({'body': b'this is not '}, TextResponse), + ({'body': b'this is not - - blahablsdfsal& - """) - self.assertEqual(get_base_url(resp), "http://www.example.com/img/") - - resp2 = HtmlResponse("http://www.example.com", body=b""" - blahablsdfsal&""") - self.assertEqual(get_base_url(resp2), "http://www.example.com") - def test_response_status_message(self): self.assertEqual(response_status_message(200), '200 OK') self.assertEqual(response_status_message(404), '404 Not Found') diff --git a/tox.ini b/tox.ini index a65395a6e..dba61590a 100644 --- a/tox.ini +++ b/tox.ini @@ -63,7 +63,8 @@ commands = pytest --flake8 {posargs:docs scrapy tests} [testenv:pylint] -basepython = python3 +# extra deps require Python 3.8 or lower +basepython = python3.8 deps = {[testenv:extra-deps]deps} pylint==2.12.2 @@ -118,6 +119,8 @@ setenv = {[pinned]setenv} [testenv:extra-deps] +# reppy requires Python 3.8 or lower +basepython = python3.8 deps = {[testenv]deps} boto