<s> import sys <EOL> try : <EOL> from gevent import monkey <EOL> monkey . patch_all ( ) <EOL> except ImportError : <EOL> sys . exit ( '<STR_LIT>' ) <EOL> import os <EOL> import time <EOL> import datetime <EOL> import logging <EOL> from logging . config import fileConfig <EOL> from optparse import make_option <EOL> from psycopg2 . pool import ThreadedConnectionPool <EOL> import gevent <EOL> from django . core . management . base import BaseCommand <EOL> from django . conf import settings <EOL> from utils . pid import Pid <EOL> from crawler import tasks <EOL> from crawler import auth <EOL> from mturk . main . models import Crawl , RequesterProfile <EOL> log = logging . getLogger ( __name__ ) <EOL> class Command ( BaseCommand ) : <EOL> option_list = BaseCommand . option_list + ( <EOL> make_option ( '<STR_LIT>' , dest = '<STR_LIT>' , type = '<STR_LIT:int>' , default = <NUM_LIT:3> , <EOL> help = '<STR_LIT>' ) , <EOL> make_option ( '<STR_LIT>' , dest = '<STR_LIT>' , metavar = '<STR_LIT>' , <EOL> help = '<STR_LIT>' ) , <EOL> make_option ( '<STR_LIT>' , dest = '<STR_LIT>' , action = '<STR_LIT:store_true>' , <EOL> help = '<STR_LIT>' ) , <EOL> make_option ( '<STR_LIT>' , dest = '<STR_LIT>' , <EOL> help = '<STR_LIT>' ) , <EOL> make_option ( '<STR_LIT>' , dest = '<STR_LIT>' , <EOL> help = '<STR_LIT>' ) , <EOL> ) <EOL> def setup_logging ( self , conf_fname ) : <EOL> "<STR_LIT>" <EOL> fileConfig ( conf_fname ) <EOL> if not os . path . isfile ( conf_fname ) : <EOL> raise IOError ( '<STR_LIT>' % conf_fname ) <EOL> log . info ( '<STR_LIT>' , conf_fname ) <EOL> def setup_debug ( self ) : <EOL> from crawler . debug import debug_listen <EOL> debug_listen ( ) <EOL> def _authenticate_if_possible ( self , email = None , password = None ) : <EOL> """<STR_LIT>""" <EOL> if self . mturk_email and self . mturk_password : <EOL> auth . install_opener ( ) <EOL> resp = auth . authenticate ( <EOL> email = email or self . mturk_email , <EOL> password = password or self . mturk_password <EOL> ) <EOL> if resp . getcode ( ) != <NUM_LIT:200> : <EOL> raise Exception ( '<STR_LIT>' % resp . read ( ) ) <EOL> log . info ( '<STR_LIT>' ) <EOL> return True <EOL> log . warning ( '<STR_LIT>' ) <EOL> return False <EOL> def handle ( self , * args , ** options ) : <EOL> self . mturk_email = getattr ( settings , '<STR_LIT>' , None ) <EOL> self . mturk_password = getattr ( settings , '<STR_LIT>' , None ) <EOL> _start_time = time . time ( ) <EOL> pid = Pid ( '<STR_LIT>' , True ) <EOL> log . info ( '<STR_LIT>' , args , options ) <EOL> if options . get ( '<STR_LIT>' ) : <EOL> self . mturk_email = options [ '<STR_LIT>' ] <EOL> if options . get ( '<STR_LIT>' ) : <EOL> self . mturk_password = options [ '<STR_LIT>' ] <EOL> if options . get ( '<STR_LIT>' , None ) : <EOL> self . setup_logging ( options [ '<STR_LIT>' ] ) <EOL> if options . get ( '<STR_LIT>' , False ) : <EOL> self . setup_debug ( ) <EOL> print '<STR_LIT>' % pid . actual_pid <EOL> print ( '<STR_LIT>' <EOL> '<STR_LIT>' ) % pid . actual_pid <EOL> self . maxworkers = options [ '<STR_LIT>' ] <EOL> if self . maxworkers > <NUM_LIT:9> : <EOL> sys . exit ( '<STR_LIT>' ) <EOL> start_time = datetime . datetime . now ( ) <EOL> hits_available = tasks . hits_mainpage_total ( ) <EOL> groups_available = tasks . hits_groups_total ( ) <EOL> crawl = Crawl . objects . create ( <EOL> start_time = start_time , <EOL> end_time = start_time , <EOL> success = True , <EOL> hits_available = hits_available , <EOL> hits_downloaded = <NUM_LIT:0> , <EOL> groups_available = groups_available , <EOL> groups_downloaded = groups_available ) <EOL> log . debug ( '<STR_LIT>' , crawl . id ) <EOL> reqesters = RequesterProfile . objects . all_as_dict ( ) <EOL> dbpool = ThreadedConnectionPool ( <NUM_LIT:10> , <NUM_LIT> , <EOL> '<STR_LIT>' % ( <EOL> settings . DATABASES [ '<STR_LIT:default>' ] [ '<STR_LIT>' ] , <EOL> settings . DATABASES [ '<STR_LIT:default>' ] [ '<STR_LIT>' ] , <EOL> settings . DATABASES [ '<STR_LIT:default>' ] [ '<STR_LIT>' ] ) ) <EOL> processed_groups = set ( ) <EOL> total_reward = <NUM_LIT:0> <EOL> hitgroups_iter = self . hits_iter ( ) <EOL> for hg_pack in hitgroups_iter : <EOL> jobs = [ ] <EOL> for hg in hg_pack : <EOL> if hg [ '<STR_LIT>' ] in processed_groups : <EOL> log . debug ( '<STR_LIT>' ) <EOL> continue <EOL> processed_groups . add ( hg [ '<STR_LIT>' ] ) <EOL> j = gevent . spawn ( tasks . process_group , <EOL> hg , crawl . id , reqesters , processed_groups , dbpool ) <EOL> jobs . append ( j ) <EOL> total_reward += hg [ '<STR_LIT>' ] * hg [ '<STR_LIT>' ] <EOL> log . debug ( '<STR_LIT>' ) <EOL> gevent . joinall ( <EOL> jobs , timeout = settings . CRAWLER_GROUP_PROCESSING_TIMEOUT ) <EOL> for job in jobs : <EOL> if not job . ready ( ) : <EOL> log . info ( '<STR_LIT>' , job ) <EOL> job . kill ( ) <EOL> if len ( processed_groups ) >= groups_available : <EOL> log . info ( '<STR_LIT>' ) <EOL> break <EOL> gevent . sleep ( <NUM_LIT:1> ) <EOL> dbpool . closeall ( ) <EOL> crawl . groups_downloaded = len ( processed_groups ) <EOL> crawl . end_time = datetime . datetime . now ( ) <EOL> crawl . save ( ) <EOL> work_time = time . time ( ) - _start_time <EOL> log . info ( """<STR_LIT>""" . format ( crawl_id = crawl . id , total_reward = total_reward , <EOL> processed_groups = len ( processed_groups ) , <EOL> groups_available = groups_available , <EOL> work_time = work_time ) ) <EOL> crawl_downloaded_pc = settings . INCOMPLETE_CRAWL_THRESHOLD <EOL> crawl_warning_pc = settings . INCOMPLETE_CRAWL_WARNING_THRESHOLD <EOL> crawl_time_warning = settings . CRAWLER_TIME_WARNING <EOL> downloaded_pc = float ( crawl . groups_downloaded ) / groups_available <EOL> if work_time > crawl_time_warning : <EOL> log . warning ( ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) . format ( work_time , crawl_time_warning ) ) <EOL> if downloaded_pc < crawl_warning_pc : <EOL> log . warning ( ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) . format ( downloaded_pc , crawl_warning_pc ) ) <EOL> if downloaded_pc < crawl_downloaded_pc : <EOL> log . warning ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" . format ( <EOL> downloaded_pc , crawl_downloaded_pc , <EOL> crawl . groups_downloaded , groups_available ) ) <EOL> pid . remove_pid ( ) <EOL> def hits_iter ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _authenticate_if_possible ( ) <EOL> counter = count ( <NUM_LIT:1> , self . maxworkers ) <EOL> for i in counter : <EOL> pages = range ( i , i + self . maxworkers ) <EOL> log . debug ( '<STR_LIT>' . format ( pages ) ) <EOL> jobs = [ gevent . spawn ( tasks . hits_groups_info , page_nr ) <EOL> for page_nr in pages ] <EOL> gevent . joinall ( jobs ) <EOL> hgs = [ ] <EOL> for job in jobs : <EOL> if job . value : <EOL> hgs . extend ( job . value ) <EOL> if not hgs : <EOL> log . info ( '<STR_LIT>' <EOL> '<STR_LIT>' . format ( pages ) ) <EOL> break <EOL> log . debug ( '<STR_LIT>' , len ( hgs ) ) <EOL> yield hgs <EOL> def count ( firstval = <NUM_LIT:0> , step = <NUM_LIT:1> ) : <EOL> "<STR_LIT>" <EOL> while True : <EOL> yield firstval <EOL> firstval += step </s>
