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

Implement scrapy.utils.response.get_response_class

This commit is contained in:
Adrián Chaves 2022-06-30 15:15:52 +02:00
parent 19a00bd4ef
commit fd2317bd6d
20 changed files with 540 additions and 316 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -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)

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

@ -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)

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

@ -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):

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

@ -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]}"

4
scrapy/core/downloader/webclient.py
View File

@ -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):

6
scrapy/core/http2/stream.py
View File

@ -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,
)

10
scrapy/downloadermiddlewares/decompression.py
View File

@ -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):

8
scrapy/downloadermiddlewares/httpcompression.py
View File

@ -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):

6
scrapy/extensions/httpcache.py
View File

@ -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

3
scrapy/http/request/form.py
View File

@ -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 <form> element found in {response}")

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

@ -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):

17
scrapy/linkextractors/lxmlhtml.py
View File

@ -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)

174
scrapy/responsetypes.py
View File

@ -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()

186
scrapy/utils/response.py
View File

@ -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
"""

7
tests/test_downloader_handlers.py
View File

@ -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;'

8
tests/test_downloadermiddleware_httpcompression.py
View File

@ -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'<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)
request = Request("http://www.example.com/index")

14
tests/test_http_response.py
View File

@ -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'<html><body><base href="https://example.net"></body></html>'
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"""
<html>
<head><base href="http://www.example.com/img/" target="_blank"></head>
<body>blahablsdfsal&amp;</body>
</html>""")
self.assertEqual(resp.base_url, "http://www.example.com/img/")
resp2 = HtmlResponse("http://www.example.com", body=b"""
<html><body>blahablsdfsal&amp;</body></html>""")
self.assertEqual(resp2.base_url, "http://www.example.com")
def test_html_encoding(self):
body = b"""<html><head><title>Some page</title><meta http-equiv="Content-Type" content="text/html; charset=iso-8859-1">

132
tests/test_responsetypes.py
View File

@ -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'<html><head><title>Hello</title></head>', HtmlResponse),
# https://codersblock.com/blog/the-smallest-valid-html5-page/
(b'<!DOCTYPE html>\n<title>.</title>', HtmlResponse),
(b'<?xml version="1.0" encoding="utf-8"', XmlResponse),
]
for source, cls in mappings:
@ -95,88 +112,9 @@ class ResponseTypesTest(unittest.TestCase):
retcls = responsetypes.from_headers(source)
assert retcls is cls, f"{source} ==> {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'<html><head><title>Hello</title></head>'}, HtmlResponse),
({'body': b'<?xml version="1.0" encoding="utf-8"'}, XmlResponse),
({'body': b'Some plain text data\1\2 with tabs and\n null bytes\0'}, Response),
# https://codersblock.com/blog/the-smallest-valid-html5-page/
({'body': b'<!DOCTYPE html>\n<title>.</title>'}, 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'<!DOCTYPE html>\n<title>.</title>', '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 <html>'}, TextResponse),
({'body': b'this is not <?xml'}, TextResponse),
]
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_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__":

238
tests/test_utils_response.py
View File

@ -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'<html><head><title>Hello</title></head>'}, HtmlResponse),
({'body': b'<?xml version="1.0" encoding="utf-8"'}, XmlResponse),
(
{'body': b'Some plain text data\1\2 with tabs and\n null bytes\0'},
Response,
),
# https://codersblock.com/blog/the-smallest-valid-html5-page/
({'body': b'<!DOCTYPE html>\n<title>.</title>'}, 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'<!DOCTYPE html>\n<title>.</title>',
'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 <html>'}, TextResponse),
({'body': b'this is not <?xml'}, TextResponse),
)
@pytest.mark.parametrize(
'kwargs,response_class',
(
*PRE_XTRACTMIME_SCENARIOS,
*POST_XTRACTMIME_SCENARIOS,
),
)
def test_get_response_class_http(kwargs, response_class):
if 'headers' in kwargs:
kwargs['http_headers'] = kwargs.pop('headers')
if 'filename' in kwargs:
assert 'url' not in kwargs
kwargs['url'] = kwargs.pop('filename')
assert get_response_class(**kwargs) == response_class
class ResponseUtilsTest(unittest.TestCase):
dummy_response = TextResponse(url='http://example.org/', body=b'dummy_response')
@ -67,18 +287,6 @@ class ResponseUtilsTest(unittest.TestCase):
self.assertEqual(get_meta_refresh(r2), (None, None))
self.assertEqual(get_meta_refresh(r3), (None, None))
def test_get_base_url(self):
resp = HtmlResponse("http://www.example.com", body=b"""
<html>
<head><base href="http://www.example.com/img/" target="_blank"></head>
<body>blahablsdfsal&amp;</body>
</html>""")
self.assertEqual(get_base_url(resp), "http://www.example.com/img/")
resp2 = HtmlResponse("http://www.example.com", body=b"""
<html><body>blahablsdfsal&amp;</body></html>""")
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')

5
tox.ini
View File

@ -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