pyspider.fetcher.Fetcher.check_output() - Code Metrics - Inspection of "Revert "new fetcher ghost.py"" - binux/pyspider - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 9592b7...7ffc4b )

by Roy

created 2015-12-13 18:56 UTC

pyspider.fetcher.Fetcher.check_output() B

↳ Parent: pyspider.fetcher.Fetcher

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	6
dl	0
loc	11
rs	8

#!/usr/bin/env python
# -*- encoding: utf-8 -*-
# vim: set et sw=4 ts=4 sts=4 ff=unix fenc=utf8:
# Author: Binux<[email protected]>
#         http://binux.me
# Created on 2012-12-17 11:07:19

from __future__ import unicode_literals

import six
import copy
import time
import json
import logging
import threading
import tornado.ioloop
import tornado.httputil
import tornado.httpclient
import pyspider

from six.moves import queue, http_cookies
from requests import cookies
from six.moves.urllib.parse import urljoin, urlsplit
from tornado.curl_httpclient import CurlAsyncHTTPClient
from tornado.simple_httpclient import SimpleAsyncHTTPClient
from pyspider.libs import utils, dataurl, counter
from .cookie_utils import extract_cookies_to_jar
logger = logging.getLogger('fetcher')


class MyCurlAsyncHTTPClient(CurlAsyncHTTPClient):

    def free_size(self):
        return len(self._free_list)

    def size(self):
        return len(self._curls) - self.free_size()


class MySimpleAsyncHTTPClient(SimpleAsyncHTTPClient):

    def free_size(self):
        return self.max_clients - self.size()

    def size(self):
        return len(self.active)

fetcher_output = {
    "status_code": int,
    "orig_url": str,
    "url": str,
    "headers": dict,
    "content": str,
    "cookies": dict,
}


class Fetcher(object):
    user_agent = "pyspider/%s (+http://pyspider.org/)" % pyspider.__version__
    default_options = {
        'method': 'GET',
        'headers': {
        },
        'use_gzip': True,
        'timeout': 120,
    }
    phantomjs_proxy = None

    def __init__(self, inqueue, outqueue, poolsize=100, proxy=None, async=True):
        self.inqueue = inqueue
        self.outqueue = outqueue

        self.poolsize = poolsize
        self._running = False
        self._quit = False
        self.proxy = proxy
        self.async = async
        self.ioloop = tornado.ioloop.IOLoop()

        # binding io_loop to http_client here
        if self.async:
            self.http_client = MyCurlAsyncHTTPClient(max_clients=self.poolsize,
                                                     io_loop=self.ioloop)
        else:
            self.http_client = tornado.httpclient.HTTPClient(
                MyCurlAsyncHTTPClient, max_clients=self.poolsize
            )

        self._cnt = {
            '5m': counter.CounterManager(
                lambda: counter.TimebaseAverageWindowCounter(30, 10)),
            '1h': counter.CounterManager(
                lambda: counter.TimebaseAverageWindowCounter(60, 60)),
        }

    def send_result(self, type, task, result):
        '''Send fetch result to processor'''
        if self.outqueue:
            try:
                self.outqueue.put((task, result))
            except Exception as e:
                logger.exception(e)

    def fetch(self, task, callback=None):
        '''Do one fetch'''
        url = task.get('url', 'data:,')
        if callback is None:
            callback = self.send_result
        if url.startswith('data:'):
            return self.data_fetch(url, task, callback)
        elif task.get('fetch', {}).get('fetch_type') in ('js', 'phantomjs'):
            return self.phantomjs_fetch(url, task, callback)
        else:
            return self.http_fetch(url, task, callback)

    def sync_fetch(self, task):
        '''Synchronization fetch'''
        wait_result = threading.Condition()
        _result = {}

        def callback(type, task, result):
            wait_result.acquire()
            _result['type'] = type
            _result['task'] = task
            _result['result'] = result
            wait_result.notify()
            wait_result.release()

        wait_result.acquire()
        self.fetch(task, callback=callback)
        while 'result' not in _result:
            wait_result.wait()
        wait_result.release()
        return _result['result']

    def data_fetch(self, url, task, callback):
        '''A fake fetcher for dataurl'''
        self.on_fetch('data', task)
        result = {}
        result['orig_url'] = url
        result['content'] = dataurl.decode(url)
        result['headers'] = {}
        result['status_code'] = 200
        result['url'] = url
        result['cookies'] = {}
        result['time'] = 0
        result['save'] = task.get('fetch', {}).get('save')
        if len(result['content']) < 70:
            logger.info("[200] %s:%s %s 0s", task.get('project'), task.get('taskid'), url)
        else:
            logger.info(
                "[200] %s:%s data:,%s...[content:%d] 0s",
                task.get('project'), task.get('taskid'),
                result['content'][:70],
                len(result['content'])
            )

        callback('data', task, result)
        self.on_result('data', task, result)
        return task, result

    def handle_error(self, type, url, task, start_time, callback, error):
        result = {
            'status_code': getattr(error, 'code', 599),
            'error': utils.text(error),
            'content': "",
            'time': time.time() - start_time,
            'orig_url': url,
            'url': url,
        }
        logger.error("[%d] %s:%s %s, %r %.2fs",
                     result['status_code'], task.get('project'), task.get('taskid'),
                     url, error, result['time'])
        callback(type, task, result)
        self.on_result(type, task, result)
        return task, result

    allowed_options = ['method', 'data', 'timeout', 'cookies', 'use_gzip', 'validate_cert']

    def http_fetch(self, url, task, callback):
        '''HTTP fetcher'''
        start_time = time.time()

        self.on_fetch('http', task)
        fetch = copy.deepcopy(self.default_options)
        fetch['url'] = url
        fetch['headers'] = tornado.httputil.HTTPHeaders(fetch['headers'])
        fetch['headers']['User-Agent'] = self.user_agent
        task_fetch = task.get('fetch', {})
        for each in self.allowed_options:
            if each in task_fetch:
                fetch[each] = task_fetch[each]
        fetch['headers'].update(task_fetch.get('headers', {}))

        if task.get('track'):
            track_headers = tornado.httputil.HTTPHeaders(
                task.get('track', {}).get('fetch', {}).get('headers') or {})
            track_ok = task.get('track', {}).get('process', {}).get('ok', False)
        else:
            track_headers = {}
            track_ok = False
        # proxy
        proxy_string = None
        if isinstance(task_fetch.get('proxy'), six.string_types):
            proxy_string = task_fetch['proxy']
        elif self.proxy and task_fetch.get('proxy', True):
            proxy_string = self.proxy
        if proxy_string:
            if '://' not in proxy_string:
                proxy_string = 'http://' + proxy_string
            proxy_splited = urlsplit(proxy_string)
            if proxy_splited.username:
                fetch['proxy_username'] = proxy_splited.username
                if six.PY2:
                    fetch['proxy_username'] = fetch['proxy_username'].encode('utf8')
            if proxy_splited.password:
                fetch['proxy_password'] = proxy_splited.password
                if six.PY2:
                    fetch['proxy_password'] = fetch['proxy_password'].encode('utf8')
            fetch['proxy_host'] = proxy_splited.hostname.encode('utf8')
            if six.PY2:
                fetch['proxy_host'] = fetch['proxy_host'].encode('utf8')
            fetch['proxy_port'] = proxy_splited.port or 8080

        # etag
        if task_fetch.get('etag', True):
            _t = None
            if isinstance(task_fetch.get('etag'), six.string_types):
                _t = task_fetch.get('etag')
            elif track_ok:
                _t = track_headers.get('etag')
            if _t and 'If-None-Match' not in fetch['headers']:
                fetch['headers']['If-None-Match'] = _t
        # last modifed
        if task_fetch.get('last_modified', True):
            _t = None
            if isinstance(task_fetch.get('last_modifed'), six.string_types):
                _t = task_fetch.get('last_modifed')
            elif track_ok:
                _t = track_headers.get('last-modified')
            if _t and 'If-Modified-Since' not in fetch['headers']:
                fetch['headers']['If-Modified-Since'] = _t

        session = cookies.RequestsCookieJar()

        # fix for tornado request obj
        if 'Cookie' in fetch['headers']:
            c = http_cookies.SimpleCookie()
            try:
                c.load(fetch['headers']['Cookie'])
            except AttributeError:
                c.load(utils.utf8(fetch['headers']['Cookie']))
            for key in c:
                session.set(key, c[key])
            del fetch['headers']['Cookie']
        fetch['follow_redirects'] = False
        if 'timeout' in fetch:
            fetch['connect_timeout'] = fetch['request_timeout'] = fetch['timeout']
            del fetch['timeout']
        if 'data' in fetch:
            fetch['body'] = fetch['data']
            del fetch['data']
        if 'cookies' in fetch:
            session.update(fetch['cookies'])
            del fetch['cookies']

        store = {}
        store['max_redirects'] = task_fetch.get('max_redirects', 5)

        def handle_response(response):
            extract_cookies_to_jar(session, response.request, response.headers)
            if (response.code in (301, 302, 303, 307)
                    and response.headers.get('Location')
                    and task_fetch.get('allow_redirects', True)):
                if store['max_redirects'] <= 0:
                    error = tornado.httpclient.HTTPError(
                        599, 'Maximum (%d) redirects followed' % task_fetch.get('max_redirects', 5),
                        response)
                    return handle_error(error)
                if response.code in (302, 303):
                    fetch['method'] = 'GET'
                    if 'body' in fetch:
                        del fetch['body']
                fetch['url'] = urljoin(fetch['url'], response.headers['Location'])
                fetch['request_timeout'] -= time.time() - start_time
                if fetch['request_timeout'] < 0:
                    fetch['request_timeout'] = 0.1
                fetch['connect_timeout'] = fetch['request_timeout']
                store['max_redirects'] -= 1
                return make_request(fetch)

            result = {}
            result['orig_url'] = url
            result['content'] = response.body or ''
            result['headers'] = dict(response.headers)
            result['status_code'] = response.code
            result['url'] = response.effective_url or url
            result['cookies'] = session.get_dict()
            result['time'] = time.time() - start_time
            result['save'] = task_fetch.get('save')
            if response.error:
                result['error'] = utils.text(response.error)
            if 200 <= response.code < 300:
                logger.info("[%d] %s:%s %s %.2fs", response.code,
                            task.get('project'), task.get('taskid'),
                            url, result['time'])
            else:
                logger.warning("[%d] %s:%s %s %.2fs", response.code,
                               task.get('project'), task.get('taskid'),
                               url, result['time'])
            callback('http', task, result)
            self.on_result('http', task, result)
            return task, result

        handle_error = lambda x: self.handle_error('http',
                                                   url, task, start_time, callback, x)

        def make_request(fetch):
            try:
                request = tornado.httpclient.HTTPRequest(**fetch)
                cookie_header = cookies.get_cookie_header(session, request)
                if cookie_header:
                    request.headers['Cookie'] = cookie_header
                if self.async:
                    self.http_client.fetch(request, handle_response)
                else:
                    return handle_response(self.http_client.fetch(request))
            except tornado.httpclient.HTTPError as e:
                if e.response:
                    return handle_response(e.response)
                else:
                    return handle_error(e)
            except Exception as e:
                logger.exception(fetch)
                return handle_error(e)

        return make_request(fetch)

    def phantomjs_fetch(self, url, task, callback):
        '''Fetch with phantomjs proxy'''
        start_time = time.time()

        self.on_fetch('phantomjs', task)
        if not self.phantomjs_proxy:
            result = {
                "orig_url": url,
                "content": "phantomjs is not enabled.",
                "headers": {},
                "status_code": 501,
                "url": url,
                "cookies": {},
                "time": 0,
                "save": task.get('fetch', {}).get('save')
            }
            logger.warning("[501] %s:%s %s 0s", task.get('project'), task.get('taskid'), url)
            callback('http', task, result)
            self.on_result('http', task, result)
            return task, result

        request_conf = {
            'follow_redirects': False
        }

        fetch = copy.deepcopy(self.default_options)
        fetch['url'] = url
        fetch['headers'] = tornado.httputil.HTTPHeaders(fetch['headers'])
        fetch['headers']['User-Agent'] = self.user_agent
        task_fetch = task.get('fetch', {})
        for each in task_fetch:
            if each != 'headers':
                fetch[each] = task_fetch[each]
        fetch['headers'].update(task_fetch.get('headers', {}))

        if 'timeout' in fetch:
            request_conf['connect_timeout'] = fetch['timeout']
            request_conf['request_timeout'] = fetch['timeout'] + 1

        session = cookies.RequestsCookieJar()
        request = tornado.httpclient.HTTPRequest(url=fetch['url'])
        if fetch.get('cookies'):
            session.update(fetch['cookies'])
            if 'Cookie' in request.headers:
                del request.headers['Cookie']
            fetch['headers']['Cookie'] = cookies.get_cookie_header(session, request)

        def handle_response(response):
            if not response.body:
                return handle_error(Exception('no response from phantomjs'))

            try:
                result = json.loads(utils.text(response.body))
                if response.error:
                    result['error'] = utils.text(response.error)
            except Exception as e:
                return handle_error(e)

            if result.get('status_code', 200):
                logger.info("[%d] %s:%s %s %.2fs", result['status_code'],
                            task.get('project'), task.get('taskid'), url, result['time'])
            else:
                logger.error("[%d] %s:%s %s, %r %.2fs", result['status_code'],
                             task.get('project'), task.get('taskid'),
                             url, result['content'], result['time'])
            callback('phantomjs', task, result)
            self.on_result('phantomjs', task, result)
            return task, result

        handle_error = lambda x: self.handle_error('phantomjs',
                                                   url, task, start_time, callback, x)

        fetch['headers'] = dict(fetch['headers'])
        try:
            request = tornado.httpclient.HTTPRequest(
                url="%s" % self.phantomjs_proxy, method="POST",
                body=json.dumps(fetch), **request_conf)
            if self.async:
                self.http_client.fetch(request, handle_response)
            else:
                return handle_response(self.http_client.fetch(request))
        except tornado.httpclient.HTTPError as e:
            if e.response:
                return handle_response(e.response)
            else:
                return handle_error(e)
        except Exception as e:
            return handle_error(e)

    def run(self):
        '''Run loop'''
        logger.info("fetcher starting...")

        def queue_loop():
            if not self.outqueue or not self.inqueue:
                return
            while not self._quit:
                try:
                    if self.outqueue.full():
                        break
                    if self.http_client.free_size() <= 0:
                        break
                    task = self.inqueue.get_nowait()
                    # FIXME: decode unicode_obj should used after data selete from
                    # database, it's used here for performance
                    task = utils.decode_unicode_obj(task)
                    self.fetch(task)
                except queue.Empty:
                    break
                except KeyboardInterrupt:
                    break
                except Exception as e:
                    logger.exception(e)
                    break

        tornado.ioloop.PeriodicCallback(queue_loop, 100, io_loop=self.ioloop).start()
        self._running = True

        try:
            self.ioloop.start()
        except KeyboardInterrupt:
            pass

        logger.info("fetcher exiting...")

    def quit(self):
        '''Quit fetcher'''
        self._running = False
        self._quit = True
        self.ioloop.stop()

    def size(self):
        return self.http_client.size()

    def xmlrpc_run(self, port=24444, bind='127.0.0.1', logRequests=False):
        '''Run xmlrpc server'''
        import umsgpack
        try:
            from xmlrpc.server import SimpleXMLRPCServer
            from xmlrpc.client import Binary
        except ImportError:
            from SimpleXMLRPCServer import SimpleXMLRPCServer
            from xmlrpclib import Binary

        server = SimpleXMLRPCServer((bind, port), allow_none=True, logRequests=logRequests)
        server.register_introspection_functions()
        server.register_multicall_functions()

        server.register_function(self.quit, '_quit')
        server.register_function(self.size)

        def sync_fetch(task):
            result = self.sync_fetch(task)
            result = Binary(umsgpack.packb(result))
            return result
        server.register_function(sync_fetch, 'fetch')

        def dump_counter(_time, _type):
            return self._cnt[_time].to_dict(_type)
        server.register_function(dump_counter, 'counter')

        server.timeout = 0.5
        while not self._quit:
            server.handle_request()
        server.server_close()

    def on_fetch(self, type, task):
        '''Called before task fetch'''
        pass

    def on_result(self, type, task, result):
        '''Called after task fetched'''
        status_code = result.get('status_code', 599)
        if status_code != 599:
            status_code = (int(status_code) / 100 * 100)
        self._cnt['5m'].event((task.get('project'), status_code), +1)
        self._cnt['1h'].event((task.get('project'), status_code), +1)

        if type == 'http' and result.get('time'):
            content_len = len(result.get('content', ''))
            self._cnt['5m'].event((task.get('project'), 'speed'),
                                  float(content_len) / result.get('time'))
            self._cnt['1h'].event((task.get('project'), 'speed'),
                                  float(content_len) / result.get('time'))
            self._cnt['5m'].event((task.get('project'), 'time'), result.get('time'))
            self._cnt['1h'].event((task.get('project'), 'time'), result.get('time'))


1			#!/usr/bin/env python
2			# -- encoding: utf-8 --
3			# vim: set et sw=4 ts=4 sts=4 ff=unix fenc=utf8:
4			# Author: Binux<[email protected]>
5			# http://binux.me
6			# Created on 2012-12-17 11:07:19
7
8			from __future__ import unicode_literals
9
10			import six
11			import copy
12			import time
13			import json
14			import logging
15			import threading
16			import tornado.ioloop
17			import tornado.httputil
18			import tornado.httpclient
19			import pyspider
20
21			from six.moves import queue, http_cookies
22			from requests import cookies
23			from six.moves.urllib.parse import urljoin, urlsplit
24			from tornado.curl_httpclient import CurlAsyncHTTPClient
25			from tornado.simple_httpclient import SimpleAsyncHTTPClient
26			from pyspider.libs import utils, dataurl, counter
27			from .cookie_utils import extract_cookies_to_jar
28			logger = logging.getLogger('fetcher')
29
30
31			class MyCurlAsyncHTTPClient(CurlAsyncHTTPClient):
32
33			def free_size(self):
34			return len(self._free_list)
35
36			def size(self):
37			return len(self._curls) - self.free_size()
38
39
40			class MySimpleAsyncHTTPClient(SimpleAsyncHTTPClient):
41
42			def free_size(self):
43			return self.max_clients - self.size()
44
45			def size(self):
46			return len(self.active)
47
48			fetcher_output = {
49			"status_code": int,
50			"orig_url": str,
51			"url": str,
52			"headers": dict,
53			"content": str,
54			"cookies": dict,
55			}
56
57
58			class Fetcher(object):
59			user_agent = "pyspider/%s (+http://pyspider.org/)" % pyspider.__version__
60			default_options = {
61			'method': 'GET',
62			'headers': {
63			},
64			'use_gzip': True,
65			'timeout': 120,
66			}
67			phantomjs_proxy = None
68
69			def __init__(self, inqueue, outqueue, poolsize=100, proxy=None, async=True):
70			self.inqueue = inqueue
71			self.outqueue = outqueue
72
73			self.poolsize = poolsize
74			self._running = False
75			self._quit = False
76			self.proxy = proxy
77			self.async = async
78			self.ioloop = tornado.ioloop.IOLoop()
79
80			# binding io_loop to http_client here
81			if self.async:
82			self.http_client = MyCurlAsyncHTTPClient(max_clients=self.poolsize,
83			io_loop=self.ioloop)
84			else:
85			self.http_client = tornado.httpclient.HTTPClient(
86			MyCurlAsyncHTTPClient, max_clients=self.poolsize
87			)
88
89			self._cnt = {
90			'5m': counter.CounterManager(
91			lambda: counter.TimebaseAverageWindowCounter(30, 10)),
92			'1h': counter.CounterManager(
93			lambda: counter.TimebaseAverageWindowCounter(60, 60)),
94			}
95
96			def send_result(self, type, task, result):
97			'''Send fetch result to processor'''
98			if self.outqueue:
99			try:
100			self.outqueue.put((task, result))
101			except Exception as e:
102			logger.exception(e)
103
104			def fetch(self, task, callback=None):
105			'''Do one fetch'''
106			url = task.get('url', 'data:,')
107			if callback is None:
108			callback = self.send_result
109			if url.startswith('data:'):
110			return self.data_fetch(url, task, callback)
111			elif task.get('fetch', {}).get('fetch_type') in ('js', 'phantomjs'):
112			return self.phantomjs_fetch(url, task, callback)
113			else:
114			return self.http_fetch(url, task, callback)
115
116			def sync_fetch(self, task):
117			'''Synchronization fetch'''
118			wait_result = threading.Condition()
119			_result = {}
120
121			def callback(type, task, result):
122			wait_result.acquire()
123			_result['type'] = type
124			_result['task'] = task
125			_result['result'] = result
126			wait_result.notify()
127			wait_result.release()
128
129			wait_result.acquire()
130			self.fetch(task, callback=callback)
131			while 'result' not in _result:
132			wait_result.wait()
133			wait_result.release()
134			return _result['result']
135
136			def data_fetch(self, url, task, callback):
137			'''A fake fetcher for dataurl'''
138			self.on_fetch('data', task)
139			result = {}
140			result['orig_url'] = url
141			result['content'] = dataurl.decode(url)
142			result['headers'] = {}
143			result['status_code'] = 200
144			result['url'] = url
145			result['cookies'] = {}
146			result['time'] = 0
147			result['save'] = task.get('fetch', {}).get('save')
148			if len(result['content']) < 70:
149			logger.info("[200] %s:%s %s 0s", task.get('project'), task.get('taskid'), url)
150			else:
151			logger.info(
152			"[200] %s:%s data:,%s...[content:%d] 0s",
153			task.get('project'), task.get('taskid'),
154			result['content'][:70],
155			len(result['content'])
156			)
157
158			callback('data', task, result)
159			self.on_result('data', task, result)
160			return task, result
161
162			def handle_error(self, type, url, task, start_time, callback, error):
163			result = {
164			'status_code': getattr(error, 'code', 599),
165			'error': utils.text(error),
166			'content': "",
167			'time': time.time() - start_time,
168			'orig_url': url,
169			'url': url,
170			}
171			logger.error("[%d] %s:%s %s, %r %.2fs",
172			result['status_code'], task.get('project'), task.get('taskid'),
173			url, error, result['time'])
174			callback(type, task, result)
175			self.on_result(type, task, result)
176			return task, result
177
178			allowed_options = ['method', 'data', 'timeout', 'cookies', 'use_gzip', 'validate_cert']
179
180			def http_fetch(self, url, task, callback):
181			'''HTTP fetcher'''
182			start_time = time.time()
183
184			self.on_fetch('http', task)
185			fetch = copy.deepcopy(self.default_options)
186			fetch['url'] = url
187			fetch['headers'] = tornado.httputil.HTTPHeaders(fetch['headers'])
188			fetch['headers']['User-Agent'] = self.user_agent
189			task_fetch = task.get('fetch', {})
190			for each in self.allowed_options:
191			if each in task_fetch:
192			fetch[each] = task_fetch[each]
193			fetch['headers'].update(task_fetch.get('headers', {}))
194
195			if task.get('track'):
196			track_headers = tornado.httputil.HTTPHeaders(
197			task.get('track', {}).get('fetch', {}).get('headers') or {})
198			track_ok = task.get('track', {}).get('process', {}).get('ok', False)
199			else:
200			track_headers = {}
201			track_ok = False
202			# proxy
203			proxy_string = None
204			if isinstance(task_fetch.get('proxy'), six.string_types):
205			proxy_string = task_fetch['proxy']
206			elif self.proxy and task_fetch.get('proxy', True):
207			proxy_string = self.proxy
208			if proxy_string:
209			if '://' not in proxy_string:
210			proxy_string = 'http://' + proxy_string
211			proxy_splited = urlsplit(proxy_string)
212			if proxy_splited.username:
213			fetch['proxy_username'] = proxy_splited.username
214			if six.PY2:
215			fetch['proxy_username'] = fetch['proxy_username'].encode('utf8')
216			if proxy_splited.password:
217			fetch['proxy_password'] = proxy_splited.password
218			if six.PY2:
219			fetch['proxy_password'] = fetch['proxy_password'].encode('utf8')
220			fetch['proxy_host'] = proxy_splited.hostname.encode('utf8')
221			if six.PY2:
222			fetch['proxy_host'] = fetch['proxy_host'].encode('utf8')
223			fetch['proxy_port'] = proxy_splited.port or 8080
224
225			# etag
226			if task_fetch.get('etag', True):
227			_t = None
228			if isinstance(task_fetch.get('etag'), six.string_types):
229			_t = task_fetch.get('etag')
230			elif track_ok:
231			_t = track_headers.get('etag')
232			if _t and 'If-None-Match' not in fetch['headers']:
233			fetch['headers']['If-None-Match'] = _t
234			# last modifed
235			if task_fetch.get('last_modified', True):
236			_t = None
237			if isinstance(task_fetch.get('last_modifed'), six.string_types):
238			_t = task_fetch.get('last_modifed')
239			elif track_ok:
240			_t = track_headers.get('last-modified')
241			if _t and 'If-Modified-Since' not in fetch['headers']:
242			fetch['headers']['If-Modified-Since'] = _t
243
244			session = cookies.RequestsCookieJar()
245
246			# fix for tornado request obj
247			if 'Cookie' in fetch['headers']:
248			c = http_cookies.SimpleCookie()
249			try:
250			c.load(fetch['headers']['Cookie'])
251			except AttributeError:
252			c.load(utils.utf8(fetch['headers']['Cookie']))
253			for key in c:
254			session.set(key, c[key])
255			del fetch['headers']['Cookie']
256			fetch['follow_redirects'] = False
257			if 'timeout' in fetch:
258			fetch['connect_timeout'] = fetch['request_timeout'] = fetch['timeout']
259			del fetch['timeout']
260			if 'data' in fetch:
261			fetch['body'] = fetch['data']
262			del fetch['data']
263			if 'cookies' in fetch:
264			session.update(fetch['cookies'])
265			del fetch['cookies']
266
267			store = {}
268			store['max_redirects'] = task_fetch.get('max_redirects', 5)
269
270			def handle_response(response):
271			extract_cookies_to_jar(session, response.request, response.headers)
272			if (response.code in (301, 302, 303, 307)
273			and response.headers.get('Location')
274			and task_fetch.get('allow_redirects', True)):
275			if store['max_redirects'] <= 0:
276			error = tornado.httpclient.HTTPError(
277			599, 'Maximum (%d) redirects followed' % task_fetch.get('max_redirects', 5),
278			response)
279			return handle_error(error)
280			if response.code in (302, 303):
281			fetch['method'] = 'GET'
282			if 'body' in fetch:
283			del fetch['body']
284			fetch['url'] = urljoin(fetch['url'], response.headers['Location'])
285			fetch['request_timeout'] -= time.time() - start_time
286			if fetch['request_timeout'] < 0:
287			fetch['request_timeout'] = 0.1
288			fetch['connect_timeout'] = fetch['request_timeout']
289			store['max_redirects'] -= 1
290			return make_request(fetch)
291
292			result = {}
293			result['orig_url'] = url
294			result['content'] = response.body or ''
295			result['headers'] = dict(response.headers)
296			result['status_code'] = response.code
297			result['url'] = response.effective_url or url
298			result['cookies'] = session.get_dict()
299			result['time'] = time.time() - start_time
300			result['save'] = task_fetch.get('save')
301			if response.error:
302			result['error'] = utils.text(response.error)
303			if 200 <= response.code < 300:
304			logger.info("[%d] %s:%s %s %.2fs", response.code,
305			task.get('project'), task.get('taskid'),
306			url, result['time'])
307			else:
308			logger.warning("[%d] %s:%s %s %.2fs", response.code,
309			task.get('project'), task.get('taskid'),
310			url, result['time'])
311			callback('http', task, result)
312			self.on_result('http', task, result)
313			return task, result
314
315			handle_error = lambda x: self.handle_error('http',
316			url, task, start_time, callback, x)
317
318			def make_request(fetch):
319			try:
320			request = tornado.httpclient.HTTPRequest(**fetch)
321			cookie_header = cookies.get_cookie_header(session, request)
322			if cookie_header:
323			request.headers['Cookie'] = cookie_header
324			if self.async:
325			self.http_client.fetch(request, handle_response)
326			else:
327			return handle_response(self.http_client.fetch(request))
328			except tornado.httpclient.HTTPError as e:
329			if e.response:
330			return handle_response(e.response)
331			else:
332			return handle_error(e)
333			except Exception as e:
334			logger.exception(fetch)
335			return handle_error(e)
336
337			return make_request(fetch)
338
339			def phantomjs_fetch(self, url, task, callback):
340			'''Fetch with phantomjs proxy'''
341			start_time = time.time()
342
343			self.on_fetch('phantomjs', task)
344			if not self.phantomjs_proxy:
345			result = {
346			"orig_url": url,
347			"content": "phantomjs is not enabled.",
348			"headers": {},
349			"status_code": 501,
350			"url": url,
351			"cookies": {},
352			"time": 0,
353			"save": task.get('fetch', {}).get('save')
354			}
355			logger.warning("[501] %s:%s %s 0s", task.get('project'), task.get('taskid'), url)
356			callback('http', task, result)
357			self.on_result('http', task, result)
358			return task, result
359
360			request_conf = {
361			'follow_redirects': False
362			}
363
364			fetch = copy.deepcopy(self.default_options)
365			fetch['url'] = url
366			fetch['headers'] = tornado.httputil.HTTPHeaders(fetch['headers'])
367			fetch['headers']['User-Agent'] = self.user_agent
368			task_fetch = task.get('fetch', {})
369			for each in task_fetch:
370			if each != 'headers':
371			fetch[each] = task_fetch[each]
372			fetch['headers'].update(task_fetch.get('headers', {}))
373
374			if 'timeout' in fetch:
375			request_conf['connect_timeout'] = fetch['timeout']
376			request_conf['request_timeout'] = fetch['timeout'] + 1
377
378			session = cookies.RequestsCookieJar()
379			request = tornado.httpclient.HTTPRequest(url=fetch['url'])
380			if fetch.get('cookies'):
381			session.update(fetch['cookies'])
382			if 'Cookie' in request.headers:
383			del request.headers['Cookie']
384			fetch['headers']['Cookie'] = cookies.get_cookie_header(session, request)
385
386			def handle_response(response):
387			if not response.body:
388			return handle_error(Exception('no response from phantomjs'))
389
390			try:
391			result = json.loads(utils.text(response.body))
392			if response.error:
393			result['error'] = utils.text(response.error)
394			except Exception as e:
395			return handle_error(e)
396
397			if result.get('status_code', 200):
398			logger.info("[%d] %s:%s %s %.2fs", result['status_code'],
399			task.get('project'), task.get('taskid'), url, result['time'])
400			else:
401			logger.error("[%d] %s:%s %s, %r %.2fs", result['status_code'],
402			task.get('project'), task.get('taskid'),
403			url, result['content'], result['time'])
404			callback('phantomjs', task, result)
405			self.on_result('phantomjs', task, result)
406			return task, result
407
408			handle_error = lambda x: self.handle_error('phantomjs',
409			url, task, start_time, callback, x)
410
411			fetch['headers'] = dict(fetch['headers'])
412			try:
413			request = tornado.httpclient.HTTPRequest(
414			url="%s" % self.phantomjs_proxy, method="POST",
415			body=json.dumps(fetch), **request_conf)
416			if self.async:
417			self.http_client.fetch(request, handle_response)
418			else:
419			return handle_response(self.http_client.fetch(request))
420			except tornado.httpclient.HTTPError as e:
421			if e.response:
422			return handle_response(e.response)
423			else:
424			return handle_error(e)
425			except Exception as e:
426			return handle_error(e)
427
428			def run(self):
429			'''Run loop'''
430			logger.info("fetcher starting...")
431
432			def queue_loop():
433			if not self.outqueue or not self.inqueue:
434			return
435			while not self._quit:
436			try:
437			if self.outqueue.full():
438			break
439			if self.http_client.free_size() <= 0:
440			break
441			task = self.inqueue.get_nowait()
442			# FIXME: decode unicode_obj should used after data selete from
443			# database, it's used here for performance
444			task = utils.decode_unicode_obj(task)
445			self.fetch(task)
446			except queue.Empty:
447			break
448			except KeyboardInterrupt:
449			break
450			except Exception as e:
451			logger.exception(e)
452			break
453
454			tornado.ioloop.PeriodicCallback(queue_loop, 100, io_loop=self.ioloop).start()
455			self._running = True
456
457			try:
458			self.ioloop.start()
459			except KeyboardInterrupt:
460			pass
461
462			logger.info("fetcher exiting...")
463
464			def quit(self):
465			'''Quit fetcher'''
466			self._running = False
467			self._quit = True
468			self.ioloop.stop()
469
470			def size(self):
471			return self.http_client.size()
472
473			def xmlrpc_run(self, port=24444, bind='127.0.0.1', logRequests=False):
474			'''Run xmlrpc server'''
475			import umsgpack
476			try:
477			from xmlrpc.server import SimpleXMLRPCServer
478			from xmlrpc.client import Binary
479			except ImportError:
480			from SimpleXMLRPCServer import SimpleXMLRPCServer
481			from xmlrpclib import Binary
482
483			server = SimpleXMLRPCServer((bind, port), allow_none=True, logRequests=logRequests)
484			server.register_introspection_functions()
485			server.register_multicall_functions()
486
487			server.register_function(self.quit, '_quit')
488			server.register_function(self.size)
489
490			def sync_fetch(task):
491			result = self.sync_fetch(task)
492			result = Binary(umsgpack.packb(result))
493			return result
494			server.register_function(sync_fetch, 'fetch')
495
496			def dump_counter(_time, _type):
497			return self._cnt[_time].to_dict(_type)
498			server.register_function(dump_counter, 'counter')
499
500			server.timeout = 0.5
501			while not self._quit:
502			server.handle_request()
503			server.server_close()
504
505			def on_fetch(self, type, task):
506			'''Called before task fetch'''
507			pass
508
509			def on_result(self, type, task, result):
510			'''Called after task fetched'''
511			status_code = result.get('status_code', 599)
512			if status_code != 599:
513			status_code = (int(status_code) / 100 * 100)
514			self._cnt['5m'].event((task.get('project'), status_code), +1)
515			self._cnt['1h'].event((task.get('project'), status_code), +1)
516
517			if type == 'http' and result.get('time'):
518			content_len = len(result.get('content', ''))
519			self._cnt['5m'].event((task.get('project'), 'speed'),
520			float(content_len) / result.get('time'))
521			self._cnt['1h'].event((task.get('project'), 'speed'),
522			float(content_len) / result.get('time'))
523			self._cnt['5m'].event((task.get('project'), 'time'), result.get('time'))
524			self._cnt['1h'].event((task.get('project'), 'time'), result.get('time'))
525

binux / pyspider

Push — master ( 9592b7...7ffc4b )

pyspider.fetcher.Fetcher.check_output() B

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like