PY3 port scrapy.utils.response

This commit is contained in:
Mikhail Korobov 2015-08-01 00:35:43 +05:00
parent d05cf6e0af
commit 7f927f68e1
3 changed files with 49 additions and 30 deletions

View File

@ -2,7 +2,6 @@
This module provides some useful functions for working with
scrapy.http.Response objects
"""
import os
import re
import weakref
@ -11,6 +10,7 @@ import tempfile
from twisted.web import http
from twisted.web.http import RESPONSES
from scrapy.utils.python import to_bytes
from w3lib import html
from scrapy.utils.decorators import deprecated
@ -27,10 +27,11 @@ def get_base_url(response):
"""Return the base url of the given response, joined with the response url"""
if response not in _baseurl_cache:
text = response.body_as_unicode()[0:4096]
_baseurl_cache[response] = html.get_base_url(text, response.url, \
_baseurl_cache[response] = html.get_base_url(text, response.url,
response.encoding)
return _baseurl_cache[response]
_noscript_re = re.compile(u'<noscript>.*?</noscript>', re.IGNORECASE | re.DOTALL)
_script_re = re.compile(u'<script.*?>.*?</script>', re.IGNORECASE | re.DOTALL)
_metaref_cache = weakref.WeakKeyDictionary()
@ -40,10 +41,11 @@ def get_meta_refresh(response):
text = response.body_as_unicode()[0:4096]
text = _noscript_re.sub(u'', text)
text = _script_re.sub(u'', text)
_metaref_cache[response] = html.get_meta_refresh(text, response.url, \
_metaref_cache[response] = html.get_meta_refresh(text, response.url,
response.encoding)
return _metaref_cache[response]
def response_status_message(status):
"""Return status code plus status text descriptive message
@ -55,19 +57,21 @@ def response_status_message(status):
"""
return '%s %s' % (status, http.responses.get(int(status)))
def response_httprepr(response):
"""Return raw HTTP representation (as string) of the given response. This
"""Return raw HTTP representation (as bytes) of the given response. This
is provided only for reference, since it's not the exact stream of bytes
that was received (that's not exposed by Twisted).
"""
s = "HTTP/1.1 %d %s\r\n" % (response.status, RESPONSES.get(response.status, ''))
s = b"HTTP/1.1 " + to_bytes(str(response.status)) + b" " + \
to_bytes(RESPONSES.get(response.status, b'')) + b"\r\n"
if response.headers:
s += response.headers.to_string() + "\r\n"
s += "\r\n"
s += response.headers.to_string() + b"\r\n"
s += b"\r\n"
s += response.body
return s
def open_in_browser(response, _openfunc=webbrowser.open):
"""Open the given response in a local web browser, populating the <base>
tag for external links to work
@ -76,14 +80,15 @@ def open_in_browser(response, _openfunc=webbrowser.open):
# XXX: this implementation is a bit dirty and could be improved
body = response.body
if isinstance(response, HtmlResponse):
if '<base' not in body:
body = body.replace('<head>', '<head><base href="%s">' % response.url)
if b'<base' not in body:
repl = '<head><base href="%s">' % response.url
body = body.replace(b'<head>', to_bytes(repl))
ext = '.html'
elif isinstance(response, TextResponse):
ext = '.txt'
else:
raise TypeError("Unsupported response type: %s" % \
response.__class__.__name__)
raise TypeError("Unsupported response type: %s" %
response.__class__.__name__)
fd, fname = tempfile.mkstemp(ext)
os.write(fd, body)
os.close(fd)

View File

@ -42,7 +42,6 @@ tests/test_stats.py
tests/test_utils_iterators.py
tests/test_utils_log.py
tests/test_utils_reqser.py
tests/test_utils_response.py
tests/test_utils_template.py
tests/test_webclient.py

View File

@ -3,53 +3,59 @@ import unittest
from six.moves.urllib.parse import urlparse
from scrapy.http import Response, TextResponse, HtmlResponse
from scrapy.utils.response import response_httprepr, open_in_browser, get_meta_refresh
from scrapy.utils.python import to_bytes
from scrapy.utils.response import (response_httprepr, open_in_browser,
get_meta_refresh, get_base_url)
__doctests__ = ['scrapy.utils.response']
class ResponseUtilsTest(unittest.TestCase):
dummy_response = TextResponse(url='http://example.org/', body='dummy_response')
dummy_response = TextResponse(url='http://example.org/', body=b'dummy_response')
def test_response_httprepr(self):
r1 = Response("http://www.example.com")
self.assertEqual(response_httprepr(r1), 'HTTP/1.1 200 OK\r\n\r\n')
self.assertEqual(response_httprepr(r1), b'HTTP/1.1 200 OK\r\n\r\n')
r1 = Response("http://www.example.com", status=404, headers={"Content-type": "text/html"}, body="Some body")
self.assertEqual(response_httprepr(r1), 'HTTP/1.1 404 Not Found\r\nContent-Type: text/html\r\n\r\nSome body')
r1 = Response("http://www.example.com", status=404, headers={"Content-type": "text/html"}, body=b"Some body")
self.assertEqual(response_httprepr(r1), b'HTTP/1.1 404 Not Found\r\nContent-Type: text/html\r\n\r\nSome body')
r1 = Response("http://www.example.com", status=6666, headers={"Content-type": "text/html"}, body="Some body")
self.assertEqual(response_httprepr(r1), 'HTTP/1.1 6666 \r\nContent-Type: text/html\r\n\r\nSome body')
r1 = Response("http://www.example.com", status=6666, headers={"Content-type": "text/html"}, body=b"Some body")
self.assertEqual(response_httprepr(r1), b'HTTP/1.1 6666 \r\nContent-Type: text/html\r\n\r\nSome body')
def test_open_in_browser(self):
url = "http:///www.example.com/some/page.html"
body = "<html> <head> <title>test page</title> </head> <body>test body</body> </html>"
body = b"<html> <head> <title>test page</title> </head> <body>test body</body> </html>"
def browser_open(burl):
path = urlparse(burl).path
if not os.path.exists(path):
path = burl.replace('file://', '')
bbody = open(path).read()
assert '<base href="%s">' % url in bbody, "<base> tag not added"
with open(path, "rb") as f:
bbody = f.read()
self.assertIn(b'<base href="' + to_bytes(url) + b'">', bbody)
return True
response = HtmlResponse(url, body=body)
assert open_in_browser(response, _openfunc=browser_open), \
"Browser not called"
self.assertRaises(TypeError, open_in_browser, Response(url, body=body), \
debug=True)
resp = Response(url, body=body)
self.assertRaises(TypeError, open_in_browser, resp, debug=True)
def test_get_meta_refresh(self):
r1 = HtmlResponse("http://www.example.com", body="""
r1 = HtmlResponse("http://www.example.com", body=b"""
<html>
<head><title>Dummy</title><meta http-equiv="refresh" content="5;url=http://example.org/newpage" /></head>
<body>blahablsdfsal&amp;</body>
</html>""")
r2 = HtmlResponse("http://www.example.com", body="""
r2 = HtmlResponse("http://www.example.com", body=b"""
<html>
<head><title>Dummy</title><noScript>
<meta http-equiv="refresh" content="5;url=http://example.org/newpage" /></head>
</noSCRIPT>
<body>blahablsdfsal&amp;</body>
</html>""")
r3 = HtmlResponse("http://www.example.com", body="""
r3 = HtmlResponse("http://www.example.com", body=b"""
<noscript><meta http-equiv="REFRESH" content="0;url=http://www.example.com/newpage</noscript>
<script type="text/javascript">
if(!checkCookies()){
@ -61,5 +67,14 @@ class ResponseUtilsTest(unittest.TestCase):
self.assertEqual(get_meta_refresh(r2), (None, None))
self.assertEqual(get_meta_refresh(r3), (None, None))
if __name__ == "__main__":
unittest.main()
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")