<s> from __future__ import absolute_import <EOL> import types <EOL> import logging <EOL> import time <EOL> try : <EOL> from urlparse import urljoin <EOL> except ImportError : <EOL> from urllib . parse import urljoin <EOL> from random import randint <EOL> from six . moves import queue <EOL> from copy import deepcopy <EOL> import six <EOL> import os <EOL> from weblib import metric <EOL> from contextlib import contextmanager <EOL> from traceback import format_exc <EOL> import multiprocessing <EOL> import threading <EOL> from datetime import datetime <EOL> import threading <EOL> from threading import Thread <EOL> from collections import deque <EOL> from grab . base import Grab <EOL> from grab . error import GrabInvalidUrl <EOL> from grab . spider . error import ( SpiderError , SpiderMisuseError , FatalError , <EOL> NoTaskHandler , NoDataHandler , <EOL> SpiderConfigurationError ) <EOL> from grab . spider . task import Task <EOL> from grab . spider . data import Data <EOL> from grab . spider . transport . multicurl import MulticurlTransport <EOL> from grab . proxylist import ProxyList , BaseProxySource <EOL> from grab . util . misc import camel_case_to_underscore <EOL> from weblib . encoding import make_str , make_unicode <EOL> from grab . base import GLOBAL_STATE <EOL> from grab . stat import Stat , Timer <EOL> from grab . spider . parser_pipeline import ParserPipeline <EOL> from grab . spider . cache_pipeline import CachePipeline <EOL> from grab . spider . deprecated import DeprecatedThingsSpiderMixin <EOL> from grab . util . warning import warn <EOL> DEFAULT_TASK_PRIORITY = <NUM_LIT:100> <EOL> DEFAULT_NETWORK_STREAM_NUMBER = <NUM_LIT:3> <EOL> DEFAULT_TASK_TRY_LIMIT = <NUM_LIT:5> <EOL> DEFAULT_NETWORK_TRY_LIMIT = <NUM_LIT:5> <EOL> RANDOM_TASK_PRIORITY_RANGE = ( <NUM_LIT:50> , <NUM_LIT:100> ) <EOL> NULL = object ( ) <EOL> logger = logging . getLogger ( '<STR_LIT>' ) <EOL> logger_verbose = logging . getLogger ( '<STR_LIT>' ) <EOL> logger_verbose . setLevel ( logging . FATAL ) <EOL> class SpiderMetaClass ( type ) : <EOL> """<STR_LIT>""" <EOL> def __new__ ( cls , name , bases , namespace ) : <EOL> if '<STR_LIT:Meta>' not in namespace : <EOL> for base in bases : <EOL> if hasattr ( base , '<STR_LIT:Meta>' ) : <EOL> meta = type ( '<STR_LIT:Meta>' , ( object , ) , dict ( base . Meta . __dict__ ) ) <EOL> meta . abstract = False <EOL> namespace [ '<STR_LIT:Meta>' ] = meta <EOL> break <EOL> if '<STR_LIT:Meta>' not in namespace : <EOL> namespace [ '<STR_LIT:Meta>' ] = type ( '<STR_LIT:Meta>' , ( object , ) , { } ) <EOL> if not hasattr ( namespace [ '<STR_LIT:Meta>' ] , '<STR_LIT>' ) : <EOL> namespace [ '<STR_LIT:Meta>' ] . abstract = False <EOL> return super ( SpiderMetaClass , cls ) . __new__ ( cls , name , bases , namespace ) <EOL> @ six . add_metaclass ( SpiderMetaClass ) <EOL> class Spider ( DeprecatedThingsSpiderMixin ) : <EOL> """<STR_LIT>""" <EOL> initial_urls = None <EOL> class Meta : <EOL> abstract = True <EOL> @ classmethod <EOL> def update_spider_config ( cls , config ) : <EOL> pass <EOL> @ classmethod <EOL> def get_spider_name ( cls ) : <EOL> if hasattr ( cls , '<STR_LIT>' ) : <EOL> return cls . spider_name <EOL> else : <EOL> return camel_case_to_underscore ( cls . __name__ ) <EOL> def __init__ ( self , thread_number = None , <EOL> network_try_limit = None , task_try_limit = None , <EOL> request_pause = NULL , <EOL> priority_mode = '<STR_LIT>' , <EOL> meta = None , <EOL> only_cache = False , <EOL> config = None , <EOL> slave = None , <EOL> args = None , <EOL> taskq = None , <EOL> network_result_queue = None , <EOL> parser_result_queue = None , <EOL> is_parser_idle = None , <EOL> shutdown_event = None , <EOL> mp_mode = False , <EOL> parser_pool_size = None , <EOL> parser_mode = False , <EOL> parser_requests_per_process = <NUM_LIT> , <EOL> http_api_port = None , <EOL> ) : <EOL> """<STR_LIT>""" <EOL> if slave is not None : <EOL> raise SpiderConfigurtionError ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . http_api_port = http_api_port <EOL> self . mp_mode = mp_mode <EOL> if self . mp_mode : <EOL> from multiprocessing import Process , Event , Queue <EOL> else : <EOL> from multiprocessing . dummy import Process , Event , Queue <EOL> if network_result_queue is not None : <EOL> self . network_result_queue = network_result_queue <EOL> else : <EOL> self . network_result_queue = Queue ( ) <EOL> self . parser_result_queue = parser_result_queue <EOL> self . is_parser_idle = is_parser_idle <EOL> if shutdown_event is not None : <EOL> self . shutdown_event = shutdown_event <EOL> else : <EOL> self . shutdown_event = Event ( ) <EOL> if not self . mp_mode and parser_pool_size and parser_pool_size > <NUM_LIT:1> : <EOL> raise SpiderConfigurationError ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . parser_pool_size = parser_pool_size <EOL> self . parser_mode = parser_mode <EOL> self . parser_requests_per_process = parser_requests_per_process <EOL> self . stat = Stat ( ) <EOL> self . timer = Timer ( ) <EOL> self . task_queue = taskq <EOL> if args is None : <EOL> self . args = { } <EOL> else : <EOL> self . args = args <EOL> if config is not None : <EOL> self . config = config <EOL> else : <EOL> self . config = { } <EOL> if meta : <EOL> self . meta = meta <EOL> else : <EOL> self . meta = { } <EOL> self . thread_number = ( <EOL> thread_number or <EOL> int ( self . config . get ( '<STR_LIT>' , <EOL> DEFAULT_NETWORK_STREAM_NUMBER ) ) ) <EOL> self . task_try_limit = ( <EOL> task_try_limit or <EOL> int ( self . config . get ( '<STR_LIT>' , DEFAULT_TASK_TRY_LIMIT ) ) ) <EOL> self . network_try_limit = ( <EOL> network_try_limit or <EOL> int ( self . config . get ( '<STR_LIT>' , <EOL> DEFAULT_NETWORK_TRY_LIMIT ) ) ) <EOL> self . _grab_config = { } <EOL> if priority_mode not in [ '<STR_LIT>' , '<STR_LIT>' ] : <EOL> raise SpiderMisuseError ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> else : <EOL> self . priority_mode = priority_mode <EOL> self . only_cache = only_cache <EOL> self . cache_pipeline = None <EOL> self . work_allowed = True <EOL> if request_pause is not NULL : <EOL> warn ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . proxylist_enabled = None <EOL> self . proxylist = None <EOL> self . proxy = None <EOL> self . proxy_auto_change = False <EOL> self . interrupted = False <EOL> def setup_cache ( self , backend = '<STR_LIT>' , database = None , use_compression = True , <EOL> ** kwargs ) : <EOL> if database is None : <EOL> raise SpiderMisuseError ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . cache_enabled = True <EOL> mod = __import__ ( '<STR_LIT>' % backend , <EOL> globals ( ) , locals ( ) , [ '<STR_LIT:foo>' ] ) <EOL> cache = mod . CacheBackend ( database = database , <EOL> use_compression = use_compression , <EOL> spider = self , ** kwargs ) <EOL> self . cache_pipeline = CachePipeline ( self , cache ) <EOL> def setup_queue ( self , backend = '<STR_LIT>' , ** kwargs ) : <EOL> logger . debug ( '<STR_LIT>' % backend ) <EOL> mod = __import__ ( '<STR_LIT>' % backend , <EOL> globals ( ) , locals ( ) , [ '<STR_LIT:foo>' ] ) <EOL> self . task_queue = mod . QueueBackend ( spider_name = self . get_spider_name ( ) , <EOL> ** kwargs ) <EOL> def add_task ( self , task , raise_error = False ) : <EOL> """<STR_LIT>""" <EOL> if self . parser_mode : <EOL> self . parser_result_queue . put ( ( task , None ) ) <EOL> return <EOL> if self . task_queue is None : <EOL> raise SpiderMisuseError ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> if task . priority is None or not task . priority_is_custom : <EOL> task . priority = self . generate_task_priority ( ) <EOL> task . priority_is_custom = False <EOL> else : <EOL> task . priority_is_custom = True <EOL> try : <EOL> if not task . url . startswith ( ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ) ) : <EOL> if self . base_url is None : <EOL> msg = '<STR_LIT>' '<STR_LIT>' % ( task . name , task . url ) <EOL> raise SpiderError ( msg ) <EOL> else : <EOL> warn ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> task . url = urljoin ( self . base_url , task . url ) <EOL> if task . grab_config : <EOL> task . grab_config [ '<STR_LIT:url>' ] = task . url <EOL> except Exception as ex : <EOL> self . stat . collect ( '<STR_LIT>' , task . url ) <EOL> if raise_error : <EOL> raise <EOL> else : <EOL> logger . error ( '<STR_LIT>' , exc_info = ex ) <EOL> return False <EOL> self . task_queue . put ( task , task . priority , schedule_time = task . schedule_time ) <EOL> return True <EOL> def stop ( self ) : <EOL> """<STR_LIT>""" <EOL> logger_verbose . debug ( '<STR_LIT>' ) <EOL> self . work_allowed = False <EOL> def load_proxylist ( self , source , source_type = None , proxy_type = '<STR_LIT:http>' , <EOL> auto_init = True , auto_change = True , <EOL> ** kwargs ) : <EOL> self . proxylist = ProxyList ( ) <EOL> if isinstance ( source , BaseProxySource ) : <EOL> self . proxylist . set_source ( source ) <EOL> elif isinstance ( source , six . string_types ) : <EOL> if source_type == '<STR_LIT>' : <EOL> self . proxylist . load_file ( source , proxy_type = proxy_type ) <EOL> elif source_type == '<STR_LIT:url>' : <EOL> self . proxylist . load_url ( source , proxy_type = proxy_type ) <EOL> else : <EOL> raise SpiderMisuseError ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> % source_type ) <EOL> else : <EOL> raise SpiderMisuseError ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> % source ) <EOL> self . proxylist_enabled = True <EOL> self . proxy = None <EOL> if not auto_change and auto_init : <EOL> self . proxy = self . proxylist . get_random_proxy ( ) <EOL> self . proxy_auto_change = auto_change <EOL> def process_next_page ( self , grab , task , xpath , <EOL> resolve_base = False , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> next_url = grab . doc . select ( xpath ) . text ( ) <EOL> except IndexError : <EOL> return False <EOL> else : <EOL> url = grab . make_url_absolute ( next_url , resolve_base = resolve_base ) <EOL> page = task . get ( '<STR_LIT>' , <NUM_LIT:1> ) + <NUM_LIT:1> <EOL> grab2 = grab . clone ( ) <EOL> grab2 . setup ( url = url ) <EOL> task2 = task . clone ( task_try_count = <NUM_LIT:1> , grab = grab2 , <EOL> page = page , ** kwargs ) <EOL> self . add_task ( task2 ) <EOL> return True <EOL> def render_stats ( self , timing = True ) : <EOL> out = [ '<STR_LIT>' ] <EOL> out . append ( '<STR_LIT>' ) <EOL> items = sorted ( self . stat . counters . items ( ) , <EOL> key = lambda x : x [ <NUM_LIT:0> ] , reverse = True ) <EOL> for item in items : <EOL> out . append ( '<STR_LIT>' % item ) <EOL> out . append ( '<STR_LIT>' ) <EOL> out . append ( '<STR_LIT>' ) <EOL> col_sizes = [ ( x , len ( y ) ) for x , y in self . stat . collections . items ( ) ] <EOL> col_sizes = sorted ( col_sizes , key = lambda x : x [ <NUM_LIT:1> ] , reverse = True ) <EOL> for col_size in col_sizes : <EOL> out . append ( '<STR_LIT>' % col_size ) <EOL> out . append ( '<STR_LIT>' ) <EOL> if '<STR_LIT>' in self . stat . counters : <EOL> out . append ( '<STR_LIT>' % <EOL> metric . format_traffic_value ( <EOL> self . stat . counters [ '<STR_LIT>' ] ) ) <EOL> out . append ( '<STR_LIT>' % self . task_queue . size ( ) <EOL> if self . task_queue else '<STR_LIT>' ) <EOL> out . append ( '<STR_LIT>' % self . thread_number ) <EOL> elapsed = self . timer . timers [ '<STR_LIT>' ] <EOL> hours , seconds = divmod ( elapsed , <NUM_LIT> ) <EOL> minutes , seconds = divmod ( seconds , <NUM_LIT> ) <EOL> out . append ( '<STR_LIT>' % ( <EOL> hours , minutes , seconds ) ) <EOL> out . append ( '<STR_LIT>' % <EOL> datetime . utcnow ( ) . strftime ( '<STR_LIT>' ) ) <EOL> if timing : <EOL> out . append ( '<STR_LIT>' ) <EOL> out . append ( self . render_timing ( ) ) <EOL> return '<STR_LIT:\n>' . join ( out ) + '<STR_LIT:\n>' <EOL> def render_timing ( self ) : <EOL> out = [ '<STR_LIT>' ] <EOL> out . append ( '<STR_LIT>' % GLOBAL_STATE [ '<STR_LIT>' ] ) <EOL> time_items = [ ( x , y ) for x , y in self . timer . timers . items ( ) ] <EOL> time_items = sorted ( time_items , key = lambda x : x [ <NUM_LIT:1> ] ) <EOL> for time_item in time_items : <EOL> out . append ( '<STR_LIT>' % time_item ) <EOL> return '<STR_LIT:\n>' . join ( out ) + '<STR_LIT:\n>' <EOL> def prepare ( self ) : <EOL> """<STR_LIT>""" <EOL> def prepare_parser ( self ) : <EOL> """<STR_LIT>""" <EOL> def shutdown ( self ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def update_grab_instance ( self , grab ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def create_grab_instance ( self , ** kwargs ) : <EOL> if self . _grab_config and kwargs : <EOL> merged_config = deepcopy ( self . _grab_config ) <EOL> merged_config . update ( kwargs ) <EOL> grab = Grab ( ** merged_config ) <EOL> elif self . _grab_config and not kwargs : <EOL> grab = Grab ( ** self . _grab_config ) <EOL> else : <EOL> grab = Grab ( ** kwargs ) <EOL> return grab <EOL> def task_generator ( self ) : <EOL> """<STR_LIT>""" <EOL> if False : <EOL> yield '<STR_LIT>' <EOL> return <EOL> def check_task_limits ( self , task ) : <EOL> """<STR_LIT>""" <EOL> if task . task_try_count > self . task_try_limit : <EOL> return False , '<STR_LIT>' <EOL> if task . network_try_count > self . network_try_limit : <EOL> return False , '<STR_LIT>' <EOL> return True , None <EOL> def generate_task_priority ( self ) : <EOL> if self . priority_mode == '<STR_LIT>' : <EOL> return DEFAULT_TASK_PRIORITY <EOL> else : <EOL> return randint ( * RANDOM_TASK_PRIORITY_RANGE ) <EOL> def task_generator_thread_wrapper ( self , task_generator ) : <EOL> """<STR_LIT>""" <EOL> while True : <EOL> with self . timer . log_time ( '<STR_LIT>' ) : <EOL> queue_size = self . task_queue . size ( ) <EOL> min_limit = self . thread_number * <NUM_LIT:10> <EOL> if queue_size < min_limit : <EOL> with self . timer . log_time ( '<STR_LIT>' ) : <EOL> logger_verbose . debug ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % ( queue_size , min_limit ) ) <EOL> try : <EOL> for x in six . moves . range ( min_limit - queue_size ) : <EOL> item = next ( task_generator ) <EOL> logger_verbose . debug ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . process_handler_result ( item ) <EOL> except StopIteration : <EOL> logger_verbose . debug ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> break <EOL> else : <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> def start_task_generators ( self ) : <EOL> """<STR_LIT>""" <EOL> logger_verbose . debug ( '<STR_LIT>' ) <EOL> if self . initial_urls : <EOL> for url in self . initial_urls : <EOL> self . add_task ( Task ( '<STR_LIT>' , url = url ) ) <EOL> self . _task_generator_list = [ ] <EOL> th = Thread ( target = self . task_generator_thread_wrapper , <EOL> args = [ self . task_generator ( ) ] ) <EOL> th . daemon = True <EOL> th . start ( ) <EOL> self . _task_generator_list . append ( th ) <EOL> def get_task_from_queue ( self ) : <EOL> start = time . time ( ) <EOL> try : <EOL> with self . timer . log_time ( '<STR_LIT>' ) : <EOL> return self . task_queue . get ( ) <EOL> except queue . Empty : <EOL> size = self . task_queue . size ( ) <EOL> if size : <EOL> logger_verbose . debug ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % size ) <EOL> return True <EOL> else : <EOL> logger_verbose . debug ( '<STR_LIT>' ) <EOL> return None <EOL> def setup_grab_for_task ( self , task ) : <EOL> grab = self . create_grab_instance ( ) <EOL> if task . grab_config : <EOL> grab . load_config ( task . grab_config ) <EOL> else : <EOL> grab . setup ( url = task . url ) <EOL> grab . config [ '<STR_LIT>' ] = grab . common_headers ( ) <EOL> self . update_grab_instance ( grab ) <EOL> return grab <EOL> def is_valid_network_response_code ( self , code , task ) : <EOL> """<STR_LIT>""" <EOL> return ( code < <NUM_LIT> or code == <NUM_LIT> or <EOL> code in task . valid_status ) <EOL> def process_handler_error ( self , func_name , ex , task ) : <EOL> self . stat . inc ( '<STR_LIT>' % ex . __class__ . __name__ . lower ( ) ) <EOL> if hasattr ( ex , '<STR_LIT>' ) : <EOL> logger . error ( '<STR_LIT>' % func_name ) <EOL> logger . error ( ex . tb ) <EOL> else : <EOL> logger . error ( '<STR_LIT>' % func_name , exc_info = ex ) <EOL> try : <EOL> ex_str = six . text_type ( ex ) <EOL> except TypeError : <EOL> try : <EOL> ex_str = ex . decode ( '<STR_LIT:utf-8>' , '<STR_LIT:ignore>' ) <EOL> except TypeError : <EOL> ex_str = str ( ex ) <EOL> task_url = task . url if task is not None else None <EOL> self . stat . collect ( '<STR_LIT>' , '<STR_LIT>' % ( <EOL> func_name , ex . __class__ . __name__ , ex_str , task_url ) ) <EOL> if isinstance ( ex , FatalError ) : <EOL> raise ex <EOL> def find_data_handler ( self , data ) : <EOL> try : <EOL> return getattr ( data , '<STR_LIT>' ) <EOL> except AttributeError : <EOL> try : <EOL> handler = getattr ( self , '<STR_LIT>' % data . handler_key ) <EOL> except AttributeError : <EOL> raise NoDataHandler ( '<STR_LIT>' <EOL> % data . handler_key ) <EOL> else : <EOL> return handler <EOL> def is_valid_network_result ( self , res ) : <EOL> if res [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) : <EOL> return True <EOL> if res [ '<STR_LIT>' ] : <EOL> res_code = res [ '<STR_LIT>' ] . response . code <EOL> if self . is_valid_network_response_code ( res_code , res [ '<STR_LIT>' ] ) : <EOL> return True <EOL> return False <EOL> def run_parser ( self ) : <EOL> """<STR_LIT>""" <EOL> self . is_parser_idle . clear ( ) <EOL> if self . parser_mode : <EOL> self . stat = Stat ( logging_period = None ) <EOL> self . prepare_parser ( ) <EOL> process_request_count = <NUM_LIT:0> <EOL> try : <EOL> recent_task_time = time . time ( ) <EOL> while True : <EOL> try : <EOL> result = self . network_result_queue . get ( block = False ) <EOL> except queue . Empty : <EOL> self . is_parser_idle . set ( ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> self . is_parser_idle . clear ( ) <EOL> logger_verbose . debug ( '<STR_LIT>' ) <EOL> if self . shutdown_event . is_set ( ) : <EOL> logger_verbose . debug ( '<STR_LIT>' ) <EOL> return <EOL> else : <EOL> process_request_count += <NUM_LIT:1> <EOL> recent_task_time = time . time ( ) <EOL> if self . parser_mode : <EOL> self . stat . reset ( ) <EOL> try : <EOL> handler = self . find_task_handler ( result [ '<STR_LIT>' ] ) <EOL> except NoTaskHandler as ex : <EOL> ex . tb = format_exc ( ) <EOL> self . parser_result_queue . put ( ( ex , result [ '<STR_LIT>' ] ) ) <EOL> self . stat . inc ( '<STR_LIT>' ) <EOL> else : <EOL> self . process_network_result_with_handler ( <EOL> result , handler ) <EOL> self . stat . inc ( '<STR_LIT>' ) <EOL> finally : <EOL> if self . parser_mode : <EOL> data = { <EOL> '<STR_LIT:type>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : self . stat . counters , <EOL> '<STR_LIT>' : self . stat . collections , <EOL> } <EOL> self . parser_result_queue . put ( ( data , <EOL> result [ '<STR_LIT>' ] ) ) <EOL> if self . parser_mode : <EOL> if self . parser_requests_per_process : <EOL> if ( process_request_count >= <EOL> self . parser_requests_per_process ) : <EOL> break <EOL> except Exception as ex : <EOL> logging . error ( '<STR_LIT>' , exc_info = ex ) <EOL> raise <EOL> def process_network_result_with_handler ( self , result , handler ) : <EOL> handler_name = getattr ( handler , '<STR_LIT>' , '<STR_LIT>' ) <EOL> try : <EOL> with self . timer . log_time ( '<STR_LIT>' ) : <EOL> with self . timer . log_time ( '<STR_LIT>' % handler_name ) : <EOL> handler_result = handler ( result [ '<STR_LIT>' ] , result [ '<STR_LIT>' ] ) <EOL> if handler_result is None : <EOL> pass <EOL> else : <EOL> for something in handler_result : <EOL> self . parser_result_queue . put ( ( something , <EOL> result [ '<STR_LIT>' ] ) ) <EOL> except NoDataHandler as ex : <EOL> ex . tb = format_exc ( ) <EOL> self . parser_result_queue . put ( ( ex , result [ '<STR_LIT>' ] ) ) <EOL> except Exception as ex : <EOL> ex . tb = format_exc ( ) <EOL> self . parser_result_queue . put ( ( ex , result [ '<STR_LIT>' ] ) ) <EOL> def find_task_handler ( self , task ) : <EOL> if task . origin_task_generator is not None : <EOL> return self . handler_for_inline_task <EOL> callback = task . get ( '<STR_LIT>' ) <EOL> if callback : <EOL> return callback <EOL> else : <EOL> try : <EOL> handler = getattr ( self , '<STR_LIT>' % task . name ) <EOL> except AttributeError : <EOL> raise NoTaskHandler ( '<STR_LIT>' <EOL> '<STR_LIT>' % task . name ) <EOL> else : <EOL> return handler <EOL> def handler_for_inline_task ( self , grab , task ) : <EOL> if isinstance ( task , types . GeneratorType ) : <EOL> coroutines_stack = [ ] <EOL> sendval = None <EOL> origin_task_generator = task <EOL> target = origin_task_generator <EOL> else : <EOL> coroutines_stack = task . coroutines_stack <EOL> sendval = grab <EOL> origin_task_generator = task . origin_task_generator <EOL> target = origin_task_generator <EOL> while True : <EOL> try : <EOL> result = target . send ( sendval ) <EOL> if isinstance ( result , types . GeneratorType ) : <EOL> coroutines_stack . append ( target ) <EOL> sendval = None <EOL> target = result <EOL> origin_task_generator = target <EOL> else : <EOL> new_task = result <EOL> new_task . origin_task_generator = origin_task_generator <EOL> new_task . coroutines_stack = coroutines_stack <EOL> self . add_task ( new_task ) <EOL> return <EOL> except StopIteration : <EOL> if coroutines_stack : <EOL> target = coroutines_stack . pop ( ) <EOL> origin_task_generator = target <EOL> else : <EOL> return <EOL> def log_network_result_stats ( self , res , from_cache = False ) : <EOL> self . stat . inc ( '<STR_LIT>' ) <EOL> self . stat . inc ( '<STR_LIT>' ) <EOL> self . stat . inc ( '<STR_LIT>' % res [ '<STR_LIT>' ] . name ) <EOL> if ( res [ '<STR_LIT>' ] . network_try_count == <NUM_LIT:1> and <EOL> res [ '<STR_LIT>' ] . task_try_count == <NUM_LIT:1> ) : <EOL> self . stat . inc ( '<STR_LIT>' % res [ '<STR_LIT>' ] . name ) <EOL> if res [ '<STR_LIT>' ] and res [ '<STR_LIT>' ] . response : <EOL> resp = res [ '<STR_LIT>' ] . response <EOL> self . timer . inc_timer ( '<STR_LIT>' , resp . name_lookup_time ) <EOL> self . timer . inc_timer ( '<STR_LIT>' , resp . connect_time ) <EOL> self . timer . inc_timer ( '<STR_LIT>' , resp . total_time ) <EOL> if from_cache : <EOL> self . stat . inc ( '<STR_LIT>' , <EOL> resp . download_size ) <EOL> self . stat . inc ( '<STR_LIT>' , <EOL> resp . upload_size ) <EOL> else : <EOL> self . stat . inc ( '<STR_LIT>' , resp . download_size ) <EOL> self . stat . inc ( '<STR_LIT>' , resp . upload_size ) <EOL> def process_grab_proxy ( self , task , grab ) : <EOL> "<STR_LIT>" <EOL> if task . use_proxylist : <EOL> if self . proxylist_enabled : <EOL> grab . setup ( connection_reuse = False ) <EOL> if self . proxy_auto_change : <EOL> self . proxy = self . change_proxy ( task , grab ) <EOL> def change_proxy ( self , task , grab ) : <EOL> proxy = self . proxylist . get_random_proxy ( ) <EOL> grab . setup ( proxy = proxy . get_address ( ) , <EOL> proxy_userpwd = proxy . get_userpwd ( ) , <EOL> proxy_type = proxy . proxy_type ) <EOL> return proxy <EOL> def submit_task_to_transport ( self , task , grab ) : <EOL> if self . only_cache : <EOL> self . stat . inc ( '<STR_LIT>' ) <EOL> else : <EOL> grab_config_backup = grab . dump_config ( ) <EOL> self . process_grab_proxy ( task , grab ) <EOL> self . stat . inc ( '<STR_LIT>' ) <EOL> self . stat . inc ( '<STR_LIT>' % task . name ) <EOL> with self . timer . log_time ( '<STR_LIT>' ) : <EOL> logger_verbose . debug ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> try : <EOL> self . transport . start_task_processing ( <EOL> task , grab , grab_config_backup ) <EOL> except GrabInvalidUrl : <EOL> logger . debug ( '<STR_LIT>' % ( <EOL> task . name , task . url ) ) <EOL> self . stat . collect ( '<STR_LIT>' , task . url ) <EOL> def start_api_thread ( self ) : <EOL> from grab . spider . http_api import HttpApiThread <EOL> proc = HttpApiThread ( self ) <EOL> proc . start ( ) <EOL> return proc <EOL> def is_ready_to_shutdown ( self ) : <EOL> return ( <EOL> not self . parser_result_queue . qsize ( ) <EOL> and all ( x [ '<STR_LIT>' ] . is_set ( ) <EOL> for x in self . parser_pipeline . parser_pool ) <EOL> and not any ( x . isAlive ( ) for x in self . _task_generator_list ) <EOL> and not self . transport . get_active_threads_number ( ) <EOL> and not self . task_queue . size ( ) <EOL> and not self . network_result_queue . qsize ( ) <EOL> and ( self . cache_pipeline is None <EOL> or ( self . cache_pipeline . is_idle ( ) <EOL> and self . cache_pipeline . input_queue . qsize ( ) == <NUM_LIT:0> <EOL> and self . cache_pipeline . result_queue . qsize ( ) == <NUM_LIT:0> ) ) <EOL> ) <EOL> def run ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . mp_mode : <EOL> from multiprocessing import Process , Event , Queue <EOL> else : <EOL> from multiprocessing . dummy import Process , Event , Queue <EOL> self . timer . start ( '<STR_LIT>' ) <EOL> self . transport = MulticurlTransport ( self . thread_number ) <EOL> if self . http_api_port : <EOL> http_api_proc = self . start_api_thread ( ) <EOL> else : <EOL> http_api_proc = None <EOL> self . parser_result_queue = Queue ( ) <EOL> self . parser_pipeline = ParserPipeline ( <EOL> bot = self , <EOL> mp_mode = self . mp_mode , <EOL> pool_size = self . parser_pool_size , <EOL> shutdown_event = self . shutdown_event , <EOL> network_result_queue = self . network_result_queue , <EOL> parser_result_queue = self . parser_result_queue , <EOL> requests_per_process = self . parser_requests_per_process , <EOL> ) <EOL> network_result_queue_limit = max ( <NUM_LIT:10> , self . thread_number * <NUM_LIT:2> ) <EOL> try : <EOL> self . prepare ( ) <EOL> if self . task_queue is None : <EOL> self . setup_queue ( ) <EOL> with self . timer . log_time ( '<STR_LIT>' ) : <EOL> self . start_task_generators ( ) <EOL> pending_tasks = deque ( ) <EOL> while self . work_allowed : <EOL> free_threads = self . transport . get_free_threads_number ( ) <EOL> if ( self . transport . get_free_threads_number ( ) <EOL> and ( self . network_result_queue . qsize ( ) <EOL> < network_result_queue_limit ) <EOL> and ( self . cache_pipeline is None <EOL> or self . cache_pipeline . has_free_resources ( ) ) ) : <EOL> if pending_tasks : <EOL> task = pending_tasks . popleft ( ) <EOL> else : <EOL> task = self . get_task_from_queue ( ) <EOL> if task is None : <EOL> if not pending_tasks and self . is_ready_to_shutdown ( ) : <EOL> self . shutdown_event . set ( ) <EOL> self . stop ( ) <EOL> break <EOL> elif isinstance ( task , bool ) and ( task is True ) : <EOL> if not self . transport . get_active_threads_number ( ) : <EOL> time . sleep ( <NUM_LIT> ) <EOL> else : <EOL> logger_verbose . debug ( '<STR_LIT>' <EOL> % task ) <EOL> task . network_try_count += <NUM_LIT:1> <EOL> is_valid , reason = self . check_task_limits ( task ) <EOL> if is_valid : <EOL> task_grab = self . setup_grab_for_task ( task ) <EOL> if self . cache_pipeline : <EOL> self . cache_pipeline . input_queue . put ( <EOL> ( '<STR_LIT>' , ( task , task_grab ) ) , <EOL> ) <EOL> else : <EOL> self . submit_task_to_transport ( task , task_grab ) <EOL> else : <EOL> self . log_rejected_task ( task , reason ) <EOL> handler = task . get_fallback_handler ( self ) <EOL> if handler : <EOL> handler ( task ) <EOL> with self . timer . log_time ( '<STR_LIT>' ) : <EOL> logger_verbose . debug ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . transport . process_handlers ( ) <EOL> logger_verbose . debug ( '<STR_LIT>' ) <EOL> results = [ ( x , False ) for x in <EOL> self . transport . iterate_results ( ) ] <EOL> if self . cache_pipeline : <EOL> while True : <EOL> try : <EOL> action , result = self . cache_pipeline . result_queue . get ( False ) <EOL> except queue . Empty : <EOL> break <EOL> else : <EOL> assert action in ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> if action == '<STR_LIT>' : <EOL> results . append ( ( result , True ) ) <EOL> elif action == '<STR_LIT>' : <EOL> task = result <EOL> task_grab = self . setup_grab_for_task ( task ) <EOL> if ( self . transport . get_free_threads_number ( ) <EOL> and ( self . network_result_queue . qsize ( ) <EOL> < network_result_queue_limit ) ) : <EOL> self . submit_task_to_transport ( task , task_grab ) <EOL> else : <EOL> pending_tasks . append ( task ) <EOL> if ( not results <EOL> and ( task is None or bool ( task ) == True ) <EOL> and not self . transport . get_active_threads_number ( ) <EOL> and not self . parser_result_queue . qsize ( ) <EOL> and ( self . cache_pipeline is None <EOL> or ( self . cache_pipeline . input_queue . qsize ( ) == <NUM_LIT:0> <EOL> and self . cache_pipeline . is_idle ( ) <EOL> and self . cache_pipeline . result_queue . qsize ( ) == <NUM_LIT:0> ) ) <EOL> ) : <EOL> time . sleep ( <NUM_LIT> ) <EOL> for result , from_cache in results : <EOL> if self . cache_pipeline and not from_cache : <EOL> if result [ '<STR_LIT>' ] : <EOL> self . cache_pipeline . input_queue . put ( <EOL> ( '<STR_LIT>' , ( result [ '<STR_LIT>' ] , result [ '<STR_LIT>' ] ) ) <EOL> ) <EOL> self . log_network_result_stats ( <EOL> result , from_cache = from_cache ) <EOL> if self . is_valid_network_result ( result ) : <EOL> self . network_result_queue . put ( result ) <EOL> else : <EOL> self . log_failed_network_result ( result ) <EOL> if self . network_try_limit > <NUM_LIT:0> : <EOL> result [ '<STR_LIT>' ] . refresh_cache = True <EOL> result [ '<STR_LIT>' ] . setup_grab_config ( <EOL> result [ '<STR_LIT>' ] ) <EOL> self . add_task ( result [ '<STR_LIT>' ] ) <EOL> if from_cache : <EOL> self . stat . inc ( '<STR_LIT>' % result [ '<STR_LIT>' ] . name ) <EOL> self . stat . inc ( '<STR_LIT>' ) <EOL> while True : <EOL> try : <EOL> p_res , p_task = self . parser_result_queue . get ( block = False ) <EOL> except queue . Empty : <EOL> break <EOL> else : <EOL> self . stat . inc ( '<STR_LIT>' ) <EOL> self . process_handler_result ( p_res , p_task ) <EOL> if not self . shutdown_event . is_set ( ) : <EOL> self . parser_pipeline . check_pool_health ( ) <EOL> logger_verbose . debug ( '<STR_LIT>' ) <EOL> except KeyboardInterrupt : <EOL> logger . info ( '<STR_LIT>' <EOL> % os . getpid ( ) ) <EOL> self . interrupted = True <EOL> raise <EOL> finally : <EOL> self . timer . stop ( '<STR_LIT>' ) <EOL> self . stat . print_progress_line ( ) <EOL> self . shutdown ( ) <EOL> if http_api_proc : <EOL> http_api_proc . server . shutdown ( ) <EOL> http_api_proc . join ( ) <EOL> if self . task_queue : <EOL> self . task_queue . clear ( ) <EOL> self . shutdown_event . set ( ) <EOL> self . parser_pipeline . shutdown ( ) <EOL> logger . debug ( '<STR_LIT>' % os . getpid ( ) ) <EOL> def log_failed_network_result ( self , res ) : <EOL> if res [ '<STR_LIT>' ] : <EOL> msg = '<STR_LIT>' % res [ '<STR_LIT>' ] . response . code <EOL> else : <EOL> msg = res [ '<STR_LIT>' ] <EOL> self . stat . inc ( '<STR_LIT>' % msg ) <EOL> def log_rejected_task ( self , task , reason ) : <EOL> logger_verbose . debug ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> % ( task . name , reason ) ) <EOL> if reason == '<STR_LIT>' : <EOL> self . stat . collect ( '<STR_LIT>' , <EOL> task . url ) <EOL> elif reason == '<STR_LIT>' : <EOL> self . stat . collect ( '<STR_LIT>' , <EOL> task . url ) <EOL> else : <EOL> raise SpiderError ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> % reason ) <EOL> def process_handler_result ( self , result , task = None ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( result , Task ) : <EOL> self . add_task ( result ) <EOL> elif isinstance ( result , Data ) : <EOL> handler = self . find_data_handler ( result ) <EOL> try : <EOL> data_result = handler ( ** result . storage ) <EOL> if data_result is None : <EOL> pass <EOL> else : <EOL> for something in data_result : <EOL> self . process_handler_result ( something , task ) <EOL> except Exception as ex : <EOL> self . process_handler_error ( '<STR_LIT>' % result . handler_key , ex , <EOL> task ) <EOL> elif result is None : <EOL> pass <EOL> elif isinstance ( result , Exception ) : <EOL> handler = self . find_task_handler ( task ) <EOL> handler_name = getattr ( handler , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . process_handler_error ( handler_name , result , task ) <EOL> elif isinstance ( result , dict ) : <EOL> if result . get ( '<STR_LIT:type>' ) == '<STR_LIT>' : <EOL> for name , count in result [ '<STR_LIT>' ] . items ( ) : <EOL> self . stat . inc ( name , count ) <EOL> for name , items in result [ '<STR_LIT>' ] . items ( ) : <EOL> for item in items : <EOL> self . stat . collect ( name , item ) <EOL> else : <EOL> raise SpiderError ( '<STR_LIT>' % result ) <EOL> else : <EOL> raise SpiderError ( '<STR_LIT>' % result ) </s>
