<s> __author__ = '<STR_LIT>' <EOL> import datetime <EOL> import logging <EOL> from twisted . python . failure import Failure <EOL> import scrapy <EOL> from ants . utils import log <EOL> import downloader <EOL> from ants . utils . extension import ExtensionManager <EOL> from ants . signalmanager import SignalManager <EOL> from ants . http import Response , Request <EOL> from ants import signals <EOL> from ants . utils . misc import load_object <EOL> import json <EOL> from ants . mail import MailSender <EOL> class EngineServer ( ) : <EOL> def __init__ ( self , spider , cluster_manager , scheduler ) : <EOL> self . spider = spider <EOL> self . scheduler = scheduler <EOL> self . cluster_manager = cluster_manager <EOL> self . status = EngineStatusServer ( ) <EOL> self . distribute_index = <NUM_LIT:0> <EOL> def add_request ( self , request ) : <EOL> request . spider_name = self . spider . name <EOL> request . hash_code = hash ( <EOL> self . spider . name + str ( request ) + datetime . datetime . now ( ) . strftime ( '<STR_LIT>' ) ) <EOL> self . scheduler . push_queue_request ( request ) <EOL> self . status . waiting_list . append ( request ) <EOL> def start ( self ) : <EOL> self . cluster_manager . init_all_node ( self . spider . name , self . run ) <EOL> def run ( self ) : <EOL> for request in self . spider . start_requests ( ) : <EOL> self . add_request ( request ) <EOL> def log_distributed_request ( self , request , node_info ) : <EOL> self . status . waiting_list . remove ( request ) <EOL> self . status . distribute_dict . setdefault ( str ( node_info . name ) , list ( ) ) . append ( request ) <EOL> def accept_request_result ( self , node_name , request_hash_code , msg ) : <EOL> node_name = str ( node_name ) <EOL> del_index = None <EOL> for index , request in enumerate ( self . status . distribute_dict [ node_name ] ) : <EOL> if request . hash_code == request_hash_code : <EOL> if msg == '<STR_LIT>' : <EOL> self . status . ok_crawled_dict . setdefault ( node_name , list ( ) ) . append ( request ) <EOL> else : <EOL> self . status . not_ok_crawled_dict . setdefault ( node_name , list ( ) ) . append ( request ) <EOL> del_index = index <EOL> break <EOL> if del_index is not None : <EOL> del self . status . distribute_dict [ node_name ] [ del_index ] <EOL> if not self . status . waiting_list and self . status . all_node_empty ( ) : <EOL> self . cluster_manager . stop_all_node ( self . spider . name ) <EOL> def stop ( self ) : <EOL> logging . info ( self . spider . name + "<STR_LIT>" ) <EOL> result = self . status . make_readable_status ( ) <EOL> result [ '<STR_LIT>' ] = datetime . datetime . now ( ) . strftime ( '<STR_LIT>' ) <EOL> dump_file_name = '<STR_LIT>' + self . spider . name + '<STR_LIT>' + result [ '<STR_LIT>' ] + '<STR_LIT>' + result [ '<STR_LIT>' ] + '<STR_LIT>' <EOL> body = json . dumps ( result , indent = <NUM_LIT:4> ) <EOL> with open ( dump_file_name , '<STR_LIT:w>' ) as f : <EOL> f . write ( body ) <EOL> MailSender ( ) . from_settings ( self . cluster_manager . settings ) . send ( self . cluster_manager . settings . get ( '<STR_LIT>' ) , <EOL> subject = self . spider . name , <EOL> body = body ) <EOL> class EngineClient ( ) : <EOL> def __init__ ( self , spider , node_manager , schedule ) : <EOL> self . settings = node_manager . settings <EOL> self . node_manager = node_manager <EOL> self . status = EngineStatusClient ( ) <EOL> self . stats = load_object ( self . settings [ '<STR_LIT>' ] ) ( self ) <EOL> self . spider = spider <EOL> self . scheduler = schedule <EOL> self . signals = SignalManager ( self ) <EOL> self . downloader = downloader . Downloader ( self ) <EOL> self . extension_manager = ExtensionManager ( self ) <EOL> self . scraper = scrapy . Scraper ( self , spider ) <EOL> def stop ( self ) : <EOL> self . signals . send_catch_log ( signal = signals . spider_closed ) <EOL> self . signals . disconnect_all ( ) <EOL> def accept_request ( self , request ) : <EOL> if hasattr ( request , "<STR_LIT>" ) : <EOL> callback = getattr ( request , "<STR_LIT>" ) <EOL> if isinstance ( callback , str ) : <EOL> request . callback = getattr ( self . spider , callback ) <EOL> request . spider_name = self . spider . name <EOL> self . scheduler . push_queue_request ( request ) <EOL> self . status . waiting_list . append ( request ) <EOL> def add_request ( self , source_request , request ) : <EOL> request . spider_name = self . spider . name <EOL> request . source_hash_code = source_request . hash_code <EOL> request . node_name = self . node_manager . node_info . name <EOL> if hasattr ( request , "<STR_LIT>" ) : <EOL> callback = getattr ( request , "<STR_LIT>" ) <EOL> if type ( callback ) . __name__ == '<STR_LIT>' : <EOL> request . callback = callback . __name__ <EOL> self . node_manager . send_request_to_master ( request ) <EOL> def send_request_result ( self , request , msg = '<STR_LIT>' ) : <EOL> self . status . running_list . remove ( request ) <EOL> self . status . crawled_num += <NUM_LIT:1> <EOL> self . node_manager . send_result_to_master ( self . spider . name , <EOL> self . node_manager . node_info . name , <EOL> request . hash_code , <EOL> msg ) <EOL> def crawl_request ( self , request ) : <EOL> self . status . waiting_list . remove ( request ) <EOL> self . status . running_list . append ( request ) <EOL> d = self . _download ( request , self . spider ) <EOL> d . addBoth ( self . _handle_downloader_output , request , self . spider ) <EOL> d . addErrback ( log . spider_log , spider = self . spider ) <EOL> return d <EOL> def _handle_downloader_output ( self , response , request , spider ) : <EOL> assert isinstance ( response , ( Request , Response , Failure ) ) , response <EOL> if isinstance ( response , Request ) : <EOL> self . add_request ( request , response ) <EOL> return <EOL> d = self . scraper . enqueue_scrape ( response , request , spider ) <EOL> d . addErrback ( log . spider_log , spider = spider ) <EOL> return d <EOL> def download ( self , request , spider ) : <EOL> d = self . _download ( request , spider ) <EOL> d . addBoth ( self . _downloaded , request , spider ) <EOL> return d <EOL> def _downloaded ( self , response , request , spider ) : <EOL> return self . download ( response , spider ) if isinstance ( response , Request ) else response <EOL> def _download ( self , request , spider ) : <EOL> def _on_success ( response ) : <EOL> assert isinstance ( response , ( Response , Request ) ) <EOL> if isinstance ( response , Response ) : <EOL> response . request = request <EOL> logging . info ( spider . name + '<STR_LIT::>' + '<STR_LIT>' + response . url ) <EOL> self . signals . send_catch_log ( signal = signals . response_received , <EOL> response = response , request = request , spider = spider ) <EOL> return response <EOL> dwld = self . downloader . fetch ( request , spider ) <EOL> dwld . addCallbacks ( _on_success ) <EOL> return dwld <EOL> class EngineStatus ( object ) : <EOL> STATUS_INIT = <NUM_LIT:0> <EOL> STATUS_RUNNING = <NUM_LIT:1> <EOL> STATUS_STOP = <NUM_LIT:2> <EOL> def __init__ ( self ) : <EOL> self . status = self . STATUS_INIT <EOL> self . start_time = datetime . datetime . now ( ) <EOL> class EngineStatusServer ( EngineStatus ) : <EOL> def __init__ ( self ) : <EOL> super ( EngineStatusServer , self ) . __init__ ( ) <EOL> self . ok_crawled_dict = dict ( ) <EOL> self . not_ok_crawled_dict = dict ( ) <EOL> self . distribute_dict = dict ( ) <EOL> self . waiting_list = list ( ) <EOL> def all_node_empty ( self ) : <EOL> for v in self . distribute_dict . itervalues ( ) : <EOL> if v : <EOL> return False <EOL> return True <EOL> def make_readable_status ( self ) : <EOL> data = dict ( ) <EOL> data [ '<STR_LIT>' ] = self . start_time . strftime ( '<STR_LIT>' ) <EOL> data [ '<STR_LIT>' ] = { k : len ( v ) for k , v in self . ok_crawled_dict . iteritems ( ) } <EOL> data [ '<STR_LIT>' ] = { k : len ( v ) for k , v in self . not_ok_crawled_dict . iteritems ( ) } <EOL> data [ '<STR_LIT>' ] = { k : len ( v ) for k , v in self . distribute_dict . iteritems ( ) } <EOL> data [ '<STR_LIT>' ] = len ( self . waiting_list ) <EOL> return data <EOL> class EngineStatusClient ( EngineStatus ) : <EOL> def __init__ ( self ) : <EOL> super ( EngineStatusClient , self ) . __init__ ( ) <EOL> self . crawled_num = <NUM_LIT:0> <EOL> self . running_list = list ( ) <EOL> self . waiting_list = list ( ) </s>
