BaseHandler.send_message() - Code Metrics - Inspection of "fix typo last_modif{,i}ed" - binux/pyspider - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 39eece...c8d455 )

by Roy

created 2016-04-20 20:07 UTC

BaseHandler.send_message() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	1
dl	0
loc	3
rs	10

#!/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 2014-02-16 23:12:48

import sys
import inspect
import functools
import fractions

import six
from six import add_metaclass, iteritems

from pyspider.libs.url import (
    quote_chinese, _build_url, _encode_params,
    _encode_multipart_formdata, curl_to_arguments)
from pyspider.libs.utils import md5string
from pyspider.libs.ListIO import ListO
from pyspider.libs.response import rebuild_response
from pyspider.libs.pprint import pprint
from pyspider.processor import ProcessorResult


def catch_status_code_error(func):
    """
    Non-200 response will been regarded as fetch failed and will not pass to callback.
    Use this decorator to override this feature.
    """
    func._catch_status_code_error = True
    return func


def not_send_status(func):
    """
    Do not send process status package back to scheduler.

    It's used by callbacks like on_message, on_result etc...
    """
    @functools.wraps(func)
    def wrapper(self, response, task):
        self._extinfo['not_send_status'] = True
        function = func.__get__(self, self.__class__)
        return self._run_func(function, response, task)
    return wrapper


def config(_config=None, **kwargs):
    """
    A decorator for setting the default kwargs of `BaseHandler.crawl`.
    Any self.crawl with this callback will use this config.
    """
    if _config is None:
        _config = {}
    _config.update(kwargs)

    def wrapper(func):
        func._config = _config
        return func
    return wrapper


class NOTSET(object):
    pass


def every(minutes=NOTSET, seconds=NOTSET):
    """
    method will been called every minutes or seconds
    """
    def wrapper(func):
        # mark the function with variable 'is_cronjob=True', the function would be
        # collected into the list Handler._cron_jobs by meta class
        func.is_cronjob = True

        # collect interval and unify to seconds, it's used in meta class. See the
        # comments in meta class.
        func.tick = minutes * 60 + seconds
        return func

    if inspect.isfunction(minutes):
        func = minutes
        minutes = 1
        seconds = 0
        return wrapper(func)

    if minutes is NOTSET:
        if seconds is NOTSET:
            minutes = 1
            seconds = 0
        else:
            minutes = 0
    if seconds is NOTSET:
        seconds = 0

    return wrapper


class BaseHandlerMeta(type):

    def __new__(cls, name, bases, attrs):
        # A list of all functions which is marked as 'is_cronjob=True'
        cron_jobs = []

        # The min_tick is the greatest common divisor(GCD) of the interval of cronjobs
        # this value would be queried by scheduler when the project initial loaded.
        # Scheudler may only send _on_cronjob task every min_tick seconds. It can reduce
        # the number of tasks sent from scheduler.
        min_tick = 0

        for each in attrs.values():
            if inspect.isfunction(each) and getattr(each, 'is_cronjob', False):
                cron_jobs.append(each)
                min_tick = fractions.gcd(min_tick, each.tick)
        newcls = type.__new__(cls, name, bases, attrs)
        newcls._cron_jobs = cron_jobs
        newcls._min_tick = min_tick
        return newcls


@add_metaclass(BaseHandlerMeta)
class BaseHandler(object):
    """
    BaseHandler for all scripts.

    `BaseHandler.run` is the main method to handler the task.
    """
    crawl_config = {}
    project_name = None
    _cron_jobs = []
    _min_tick = 0
    __env__ = {'not_inited': True}
    retry_delay = {}

    def _reset(self):
        """
        reset before each task
        """
        self._extinfo = {}
        self._messages = []
        self._follows = []
        self._follows_keys = set()

    def _run_func(self, function, *arguments):
        """
        Running callback function with requested number of arguments
        """
        args, varargs, keywords, defaults = inspect.getargspec(function)
        return function(*arguments[:len(args) - 1])

    def _run_task(self, task, response):
        """
        Finding callback specified by `task['callback']`
        raising status error for it if needed.
        """
        process = task.get('process', {})
        callback = process.get('callback', '__call__')
        if not hasattr(self, callback):
            raise NotImplementedError("self.%s() not implemented!" % callback)

        function = getattr(self, callback)
        # do not run_func when 304
        if response.status_code == 304 and not getattr(function, '_catch_status_code_error', False):
            return None
        if not getattr(function, '_catch_status_code_error', False):
            response.raise_for_status()
        return self._run_func(function, response, task)

    def run_task(self, module, task, response):
        """
        Processing the task, catching exceptions and logs, return a `ProcessorResult` object
        """
        logger = module.logger
        result = None
        exception = None
        stdout = sys.stdout
        self.task = task
        if isinstance(response, dict):
            response = rebuild_response(response)
        self.response = response
        self.save = (task.get('track') or {}).get('save', {})

        try:
            if self.__env__.get('enable_stdout_capture', True):
                sys.stdout = ListO(module.log_buffer)
            self._reset()
            result = self._run_task(task, response)
            if inspect.isgenerator(result):
                for r in result:
                    self._run_func(self.on_result, r, response, task)
            else:
                self._run_func(self.on_result, result, response, task)
        except Exception as e:
            logger.exception(e)
            exception = e
        finally:
            follows = self._follows
            messages = self._messages
            logs = list(module.log_buffer)
            extinfo = self._extinfo
            save = self.save

            sys.stdout = stdout
            self.task = None
            self.response = None
            self.save = None

        module.log_buffer[:] = []
        return ProcessorResult(result, follows, messages, logs, exception, extinfo, save)

    def _crawl(self, url, **kwargs):
        """
        real crawl API

        checking kwargs, and repack them to each sub-dict
        """
        task = {}

        assert len(url) < 1024, "Maximum (1024) URL length error."

        if kwargs.get('callback'):
            callback = kwargs['callback']
            if isinstance(callback, six.string_types) and hasattr(self, callback):
                func = getattr(self, callback)
            elif six.callable(callback) and six.get_method_self(callback) is self:
                func = callback
                kwargs['callback'] = func.__name__
            else:
                raise NotImplementedError("self.%s() not implemented!" % callback)
            if hasattr(func, '_config'):
                for k, v in iteritems(func._config):
                    if isinstance(v, dict) and isinstance(kwargs.get(k), dict):
                        kwargs[k].update(v)
                    else:
                        kwargs.setdefault(k, v)

        for k, v in iteritems(self.crawl_config):
            if isinstance(v, dict) and isinstance(kwargs.get(k), dict):
                kwargs[k].update(v)
            else:
                kwargs.setdefault(k, v)

        url = quote_chinese(_build_url(url.strip(), kwargs.pop('params', None)))
        if kwargs.get('files'):
            assert isinstance(
                kwargs.get('data', {}), dict), "data must be a dict when using with files!"
            content_type, data = _encode_multipart_formdata(kwargs.pop('data', {}),
                                                            kwargs.pop('files', {}))
            kwargs.setdefault('headers', {})
            kwargs['headers']['Content-Type'] = content_type
            kwargs['data'] = data
        if kwargs.get('data'):
            kwargs['data'] = _encode_params(kwargs['data'])
        if kwargs.get('data'):
            kwargs.setdefault('method', 'POST')

        schedule = {}
        for key in ('priority', 'retries', 'exetime', 'age', 'itag', 'force_update',
                    'auto_recrawl'):
            if key in kwargs:
                schedule[key] = kwargs.pop(key)
        task['schedule'] = schedule

        fetch = {}
        for key in (
                'method',
                'headers',
                'data',
                'timeout',
                'allow_redirects',
                'cookies',
                'proxy',
                'etag',
                'last_modifed',
                'last_modified',
                'save',
                'js_run_at',
                'js_script',
                'js_viewport_width',
                'js_viewport_height',
                'load_images',
                'fetch_type',
                'use_gzip',
                'validate_cert',
                'max_redirects',
                'robots_txt'
        ):
            if key in kwargs:
                fetch[key] = kwargs.pop(key)
        task['fetch'] = fetch

        process = {}
        for key in ('callback', ):
            if key in kwargs:
                process[key] = kwargs.pop(key)
        task['process'] = process

        task['project'] = self.project_name
        task['url'] = url
        if 'taskid' in kwargs:
            task['taskid'] = kwargs.pop('taskid')
        else:
            task['taskid'] = self.get_taskid(task)

        if kwargs:
            raise TypeError('crawl() got unexpected keyword argument: %s' % kwargs.keys())

        cache_key = "%(project)s:%(taskid)s" % task
        if cache_key not in self._follows_keys:
            self._follows_keys.add(cache_key)
            self._follows.append(task)
        return task

    def get_taskid(self, task):
        '''Generate taskid by information of task md5(url) by default, override me'''
        return md5string(task['url'])

    # apis
    def crawl(self, url, **kwargs):
        '''
        available params:
          url
          callback

          method
          params
          data
          files
          headers
          timeout
          allow_redirects
          cookies
          proxy
          etag
          last_modified
          auto_recrawl

          fetch_type
          js_run_at
          js_script
          js_viewport_width
          js_viewport_height
          load_images

          priority
          retries
          exetime
          age
          itag

          save
          taskid

          full documents: http://pyspider.readthedocs.org/en/latest/apis/self.crawl/
        '''

        if isinstance(url, six.string_types) and url.startswith('curl '):
            curl_kwargs = curl_to_arguments(url)
            url = curl_kwargs.pop('urls')
            for k, v in iteritems(curl_kwargs):
                kwargs.setdefault(k, v)

        if isinstance(url, six.string_types):
            return self._crawl(url, **kwargs)
        elif hasattr(url, "__iter__"):
            result = []
            for each in url:
                result.append(self._crawl(each, **kwargs))
            return result

    def is_debugger(self):
        """Return true if running in debugger"""
        return self.__env__.get('debugger')

    def send_message(self, project, msg, url='data:,on_message'):
        """Send messages to other project."""
        self._messages.append((project, msg, url))

    def on_message(self, project, msg):
        """Receive message from other project, override me."""
        pass

    def on_result(self, result):
        """Receiving returns from other callback, override me."""
        if not result:
            return
        assert self.task, "on_result can't outside a callback."
        if self.is_debugger():
            pprint(result)
        if self.__env__.get('result_queue'):
            self.__env__['result_queue'].put((self.task, result))

    @not_send_status
    def _on_message(self, response):
        project, msg = response.save
        return self.on_message(project, msg)

    @not_send_status
    def _on_cronjob(self, response, task):
        if (not response.save
                or not isinstance(response.save, dict)
                or 'tick' not in response.save):
            return

        # When triggered, a '_on_cronjob' task is sent from scheudler with 'tick' in
        # Response.save. Scheduler may at least send the trigger task every GCD of the
        # inverval of the cronjobs. The method should check the tick for each cronjob
        # function to confirm the execute interval.
        for cronjob in self._cron_jobs:
            if response.save['tick'] % cronjob.tick != 0:
                continue
            function = cronjob.__get__(self, self.__class__)
            self._run_func(function, response, task)

    def _on_get_info(self, response, task):
        """Sending runtime infomation about this script."""
        for each in response.save or []:
            if each == 'min_tick':
                self.save[each] = self._min_tick
            elif each == 'retry_delay':
                if not isinstance(self.retry_delay, dict):
                    self.retry_delay = {'': self.retry_delay}
                self.save[each] = self.retry_delay

    @not_send_status
    def on_finished(self, response, task):
        pass


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 2014-02-16 23:12:48
7
8			import sys
9			import inspect
10			import functools
11			import fractions
12
13			import six
14			from six import add_metaclass, iteritems
15
16			from pyspider.libs.url import (
17			quote_chinese, _build_url, _encode_params,
18			_encode_multipart_formdata, curl_to_arguments)
19			from pyspider.libs.utils import md5string
20			from pyspider.libs.ListIO import ListO
21			from pyspider.libs.response import rebuild_response
22			from pyspider.libs.pprint import pprint
23			from pyspider.processor import ProcessorResult
24
25
26			def catch_status_code_error(func):
27			"""
28			Non-200 response will been regarded as fetch failed and will not pass to callback.
29			Use this decorator to override this feature.
30			"""
31			func._catch_status_code_error = True
32			return func
33
34
35			def not_send_status(func):
36			"""
37			Do not send process status package back to scheduler.
38
39			It's used by callbacks like on_message, on_result etc...
40			"""
41			@functools.wraps(func)
42			def wrapper(self, response, task):
43			self._extinfo['not_send_status'] = True
44			function = func.__get__(self, self.__class__)
45			return self._run_func(function, response, task)
46			return wrapper
47
48
49			def config(_config=None, **kwargs):
50			"""
51			A decorator for setting the default kwargs of `BaseHandler.crawl`.
52			Any self.crawl with this callback will use this config.
53			"""
54			if _config is None:
55			_config = {}
56			_config.update(kwargs)
57
58			def wrapper(func):
59			func._config = _config
60			return func
61			return wrapper
62
63
64			class NOTSET(object):
65			pass
66
67
68			def every(minutes=NOTSET, seconds=NOTSET):
69			"""
70			method will been called every minutes or seconds
71			"""
72			def wrapper(func):
73			# mark the function with variable 'is_cronjob=True', the function would be
74			# collected into the list Handler._cron_jobs by meta class
75			func.is_cronjob = True
76
77			# collect interval and unify to seconds, it's used in meta class. See the
78			# comments in meta class.
79			func.tick = minutes * 60 + seconds
80			return func
81
82			if inspect.isfunction(minutes):
83			func = minutes
84			minutes = 1
85			seconds = 0
86			return wrapper(func)
87
88			if minutes is NOTSET:
89			if seconds is NOTSET:
90			minutes = 1
91			seconds = 0
92			else:
93			minutes = 0
94			if seconds is NOTSET:
95			seconds = 0
96
97			return wrapper
98
99
100			class BaseHandlerMeta(type):
101
102			def __new__(cls, name, bases, attrs):
103			# A list of all functions which is marked as 'is_cronjob=True'
104			cron_jobs = []
105
106			# The min_tick is the greatest common divisor(GCD) of the interval of cronjobs
107			# this value would be queried by scheduler when the project initial loaded.
108			# Scheudler may only send _on_cronjob task every min_tick seconds. It can reduce
109			# the number of tasks sent from scheduler.
110			min_tick = 0
111
112			for each in attrs.values():
113			if inspect.isfunction(each) and getattr(each, 'is_cronjob', False):
114			cron_jobs.append(each)
115			min_tick = fractions.gcd(min_tick, each.tick)
116			newcls = type.__new__(cls, name, bases, attrs)
117			newcls._cron_jobs = cron_jobs
118			newcls._min_tick = min_tick
119			return newcls
120
121
122			@add_metaclass(BaseHandlerMeta)
123			class BaseHandler(object):
124			"""
125			BaseHandler for all scripts.
126
127			`BaseHandler.run` is the main method to handler the task.
128			"""
129			crawl_config = {}
130			project_name = None
131			_cron_jobs = []
132			_min_tick = 0
133			__env__ = {'not_inited': True}
134			retry_delay = {}
135
136			def _reset(self):
137			"""
138			reset before each task
139			"""
140			self._extinfo = {}
141			self._messages = []
142			self._follows = []
143			self._follows_keys = set()
144
145			def _run_func(self, function, *arguments):
146			"""
147			Running callback function with requested number of arguments
148			"""
149			args, varargs, keywords, defaults = inspect.getargspec(function)
150			return function(*arguments[:len(args) - 1])
151
152			def _run_task(self, task, response):
153			"""
154			Finding callback specified by `task['callback']`
155			raising status error for it if needed.
156			"""
157			process = task.get('process', {})
158			callback = process.get('callback', '__call__')
159			if not hasattr(self, callback):
160			raise NotImplementedError("self.%s() not implemented!" % callback)
161
162			function = getattr(self, callback)
163			# do not run_func when 304
164			if response.status_code == 304 and not getattr(function, '_catch_status_code_error', False):
165			return None
166			if not getattr(function, '_catch_status_code_error', False):
167			response.raise_for_status()
168			return self._run_func(function, response, task)
169
170			def run_task(self, module, task, response):
171			"""
172			Processing the task, catching exceptions and logs, return a `ProcessorResult` object
173			"""
174			logger = module.logger
175			result = None
176			exception = None
177			stdout = sys.stdout
178			self.task = task
179			if isinstance(response, dict):
180			response = rebuild_response(response)
181			self.response = response
182			self.save = (task.get('track') or {}).get('save', {})
183
184			try:
185			if self.__env__.get('enable_stdout_capture', True):
186			sys.stdout = ListO(module.log_buffer)
187			self._reset()
188			result = self._run_task(task, response)
189			if inspect.isgenerator(result):
190			for r in result:
191			self._run_func(self.on_result, r, response, task)
192			else:
193			self._run_func(self.on_result, result, response, task)
194			except Exception as e:
195			logger.exception(e)
196			exception = e
197			finally:
198			follows = self._follows
199			messages = self._messages
200			logs = list(module.log_buffer)
201			extinfo = self._extinfo
202			save = self.save
203
204			sys.stdout = stdout
205			self.task = None
206			self.response = None
207			self.save = None
208
209			module.log_buffer[:] = []
210			return ProcessorResult(result, follows, messages, logs, exception, extinfo, save)
211
212			def _crawl(self, url, **kwargs):
213			"""
214			real crawl API
215
216			checking kwargs, and repack them to each sub-dict
217			"""
218			task = {}
219
220			assert len(url) < 1024, "Maximum (1024) URL length error."
221
222			if kwargs.get('callback'):
223			callback = kwargs['callback']
224			if isinstance(callback, six.string_types) and hasattr(self, callback):
225			func = getattr(self, callback)
226			elif six.callable(callback) and six.get_method_self(callback) is self:
227			func = callback
228			kwargs['callback'] = func.__name__
229			else:
230			raise NotImplementedError("self.%s() not implemented!" % callback)
231			if hasattr(func, '_config'):
232			for k, v in iteritems(func._config):
233			if isinstance(v, dict) and isinstance(kwargs.get(k), dict):
234			kwargs[k].update(v)
235			else:
236			kwargs.setdefault(k, v)
237
238			for k, v in iteritems(self.crawl_config):
239			if isinstance(v, dict) and isinstance(kwargs.get(k), dict):
240			kwargs[k].update(v)
241			else:
242			kwargs.setdefault(k, v)
243
244			url = quote_chinese(_build_url(url.strip(), kwargs.pop('params', None)))
245			if kwargs.get('files'):
246			assert isinstance(
247			kwargs.get('data', {}), dict), "data must be a dict when using with files!"
248			content_type, data = _encode_multipart_formdata(kwargs.pop('data', {}),
249			kwargs.pop('files', {}))
250			kwargs.setdefault('headers', {})
251			kwargs['headers']['Content-Type'] = content_type
252			kwargs['data'] = data
253			if kwargs.get('data'):
254			kwargs['data'] = _encode_params(kwargs['data'])
255			if kwargs.get('data'):
256			kwargs.setdefault('method', 'POST')
257
258			schedule = {}
259			for key in ('priority', 'retries', 'exetime', 'age', 'itag', 'force_update',
260			'auto_recrawl'):
261			if key in kwargs:
262			schedule[key] = kwargs.pop(key)
263			task['schedule'] = schedule
264
265			fetch = {}
266			for key in (
267			'method',
268			'headers',
269			'data',
270			'timeout',
271			'allow_redirects',
272			'cookies',
273			'proxy',
274			'etag',
275			'last_modifed',
276			'last_modified',
277			'save',
278			'js_run_at',
279			'js_script',
280			'js_viewport_width',
281			'js_viewport_height',
282			'load_images',
283			'fetch_type',
284			'use_gzip',
285			'validate_cert',
286			'max_redirects',
287			'robots_txt'
288			):
289			if key in kwargs:
290			fetch[key] = kwargs.pop(key)
291			task['fetch'] = fetch
292
293			process = {}
294			for key in ('callback', ):
295			if key in kwargs:
296			process[key] = kwargs.pop(key)
297			task['process'] = process
298
299			task['project'] = self.project_name
300			task['url'] = url
301			if 'taskid' in kwargs:
302			task['taskid'] = kwargs.pop('taskid')
303			else:
304			task['taskid'] = self.get_taskid(task)
305
306			if kwargs:
307			raise TypeError('crawl() got unexpected keyword argument: %s' % kwargs.keys())
308
309			cache_key = "%(project)s:%(taskid)s" % task
310			if cache_key not in self._follows_keys:
311			self._follows_keys.add(cache_key)
312			self._follows.append(task)
313			return task
314
315			def get_taskid(self, task):
316			'''Generate taskid by information of task md5(url) by default, override me'''
317			return md5string(task['url'])
318
319			# apis
320			def crawl(self, url, **kwargs):
321			'''
322			available params:
323			url
324			callback
325
326			method
327			params
328			data
329			files
330			headers
331			timeout
332			allow_redirects
333			cookies
334			proxy
335			etag
336			last_modified
337			auto_recrawl
338
339			fetch_type
340			js_run_at
341			js_script
342			js_viewport_width
343			js_viewport_height
344			load_images
345
346			priority
347			retries
348			exetime
349			age
350			itag
351
352			save
353			taskid
354
355			full documents: http://pyspider.readthedocs.org/en/latest/apis/self.crawl/
356			'''
357
358			if isinstance(url, six.string_types) and url.startswith('curl '):
359			curl_kwargs = curl_to_arguments(url)
360			url = curl_kwargs.pop('urls')
361			for k, v in iteritems(curl_kwargs):
362			kwargs.setdefault(k, v)
363
364			if isinstance(url, six.string_types):
365			return self._crawl(url, **kwargs)
366			elif hasattr(url, "__iter__"):
367			result = []
368			for each in url:
369			result.append(self._crawl(each, **kwargs))
370			return result
371
372			def is_debugger(self):
373			"""Return true if running in debugger"""
374			return self.__env__.get('debugger')
375
376			def send_message(self, project, msg, url='data:,on_message'):
377			"""Send messages to other project."""
378			self._messages.append((project, msg, url))
379
380			def on_message(self, project, msg):
381			"""Receive message from other project, override me."""
382			pass
383
384			def on_result(self, result):
385			"""Receiving returns from other callback, override me."""
386			if not result:
387			return
388			assert self.task, "on_result can't outside a callback."
389			if self.is_debugger():
390			pprint(result)
391			if self.__env__.get('result_queue'):
392			self.__env__['result_queue'].put((self.task, result))
393
394			@not_send_status
395			def _on_message(self, response):
396			project, msg = response.save
397			return self.on_message(project, msg)
398
399			@not_send_status
400			def _on_cronjob(self, response, task):
401			if (not response.save
402			or not isinstance(response.save, dict)
403			or 'tick' not in response.save):
404			return
405
406			# When triggered, a '_on_cronjob' task is sent from scheudler with 'tick' in
407			# Response.save. Scheduler may at least send the trigger task every GCD of the
408			# inverval of the cronjobs. The method should check the tick for each cronjob
409			# function to confirm the execute interval.
410			for cronjob in self._cron_jobs:
411			if response.save['tick'] % cronjob.tick != 0:
412			continue
413			function = cronjob.__get__(self, self.__class__)
414			self._run_func(function, response, task)
415
416			def _on_get_info(self, response, task):
417			"""Sending runtime infomation about this script."""
418			for each in response.save or []:
419			if each == 'min_tick':
420			self.save[each] = self._min_tick
421			elif each == 'retry_delay':
422			if not isinstance(self.retry_delay, dict):
423			self.retry_delay = {'': self.retry_delay}
424			self.save[each] = self.retry_delay
425
426			@not_send_status
427			def on_finished(self, response, task):
428			pass
429

binux / pyspider

Push — master ( 39eece...c8d455 )

BaseHandler.send_message() A

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like