<s> import logging <EOL> import os <EOL> import time <EOL> import dateutil . parser <EOL> import requests <EOL> import streamcorpus <EOL> import trollius as asyncio <EOL> logger = logging . getLogger ( __name__ ) <EOL> class Fetcher ( object ) : <EOL> headers = { <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> def __init__ ( self , timeout = <NUM_LIT> ) : <EOL> self . session = requests . Session ( ) <EOL> self . timeout = timeout <EOL> def get_async ( self , urls , callback ) : <EOL> def get ( url , headers ) : <EOL> resp = requests . get ( url , headers = headers ) <EOL> if resp : <EOL> si = self . process_response ( resp ) <EOL> callback ( si , resp . url ) <EOL> @ asyncio . coroutine <EOL> def run ( urls , concurrency , loop ) : <EOL> urls = list ( urls ) <EOL> pending = [ loop . run_in_executor ( None , get , url , self . headers ) for url in urls [ : concurrency ] ] <EOL> rest = urls [ concurrency : ] <EOL> while pending : <EOL> done , pending = yield asyncio . From ( asyncio . wait ( pending , return_when = asyncio . FIRST_COMPLETED ) ) <EOL> while rest and len ( pending ) < concurrency : <EOL> pending . add ( loop . run_in_executor ( None , get , rest . pop ( ) , self . headers ) ) <EOL> for future in done : <EOL> try : <EOL> si = future . result ( ) <EOL> except Exception : <EOL> logger . info ( '<STR_LIT>' , exc_info = True ) <EOL> loop = asyncio . get_event_loop ( ) <EOL> loop . run_until_complete ( run ( urls , concurrency = <NUM_LIT:32> , loop = loop ) ) <EOL> def get ( self , url ) : <EOL> req = requests . Request ( '<STR_LIT:GET>' , url , headers = self . headers ) <EOL> try : <EOL> prepped = self . session . prepare_request ( req ) <EOL> resp = self . session . send ( prepped , timeout = self . timeout ) <EOL> except Exception : <EOL> logger . info ( '<STR_LIT>' , url , exc_info = True ) <EOL> return None <EOL> return self . process_response ( resp , url ) <EOL> def process_response ( self , resp ) : <EOL> logger . info ( '<STR_LIT>' , len ( resp . content ) , resp . url ) <EOL> last_modified = resp . headers . get ( '<STR_LIT>' ) <EOL> if last_modified : <EOL> try : <EOL> last_modified = dateutil . parser . parse ( last_modified ) <EOL> last_modified = int ( last_modified . strftime ( '<STR_LIT:%s>' ) ) <EOL> except Exception : <EOL> last_modified = None <EOL> if not last_modified : <EOL> last_modified = int ( time . time ( ) ) <EOL> si = streamcorpus . make_stream_item ( <EOL> last_modified , <EOL> resp . url <EOL> ) <EOL> si . original_url = resp . url <EOL> si . body . raw = resp . content <EOL> media_type = resp . headers . get ( '<STR_LIT>' ) <EOL> try : <EOL> media_type = ( media_type <EOL> . decode ( '<STR_LIT:utf8>' , '<STR_LIT:ignore>' ) <EOL> . encode ( '<STR_LIT:utf8>' , '<STR_LIT:ignore>' ) ) <EOL> except Exception : <EOL> media_type = repr ( media_type ) <EOL> si . body . media_type = media_type <EOL> si . body . encoding = resp . apparent_encoding <EOL> return si <EOL> class ChunkRoller ( object ) : <EOL> def __init__ ( self , chunk_dir , chunk_max = <NUM_LIT> ) : <EOL> self . chunk_dir = chunk_dir <EOL> self . chunk_max = chunk_max <EOL> self . t_path = os . path . join ( chunk_dir , '<STR_LIT>' ) <EOL> self . o_chunk = None <EOL> def add ( self , si ) : <EOL> '''<STR_LIT>''' <EOL> if self . o_chunk is None : <EOL> if os . path . exists ( self . t_path ) : <EOL> os . remove ( self . t_path ) <EOL> self . o_chunk = streamcorpus . Chunk ( self . t_path , mode = '<STR_LIT:wb>' ) <EOL> self . o_chunk . add ( si ) <EOL> logger . debug ( '<STR_LIT>' , len ( self . o_chunk ) ) <EOL> if len ( self . o_chunk ) == self . chunk_max : <EOL> self . close ( ) <EOL> def close ( self ) : <EOL> if self . o_chunk : <EOL> self . o_chunk . close ( ) <EOL> o_path = os . path . join ( <EOL> self . chunk_dir , <EOL> '<STR_LIT>' % ( len ( self . o_chunk ) , self . o_chunk . md5_hexdigest ) <EOL> ) <EOL> os . rename ( self . t_path , o_path ) <EOL> self . o_chunk = None <EOL> logger . info ( '<STR_LIT>' , o_path ) </s>
