<s> import cPickle as pickle <EOL> import os <EOL> import json <EOL> import logging <EOL> import shutil <EOL> import time <EOL> from collections import defaultdict <EOL> from datetime import datetime , timedelta <EOL> from operator import itemgetter <EOL> import hb <EOL> import settings_local as settings <EOL> log = logging . getLogger ( '<STR_LIT>' ) <EOL> FX = settings . FIREFOX_VERSION <EOL> JSON_DIR = os . path . join ( settings . BASE_DIR , '<STR_LIT>' ) <EOL> PICKLE = settings . path ( '<STR_LIT>' ) <EOL> BACKUP = PICKLE + '<STR_LIT>' <EOL> hbase = hb . Client ( settings . HBASE_HOST , settings . HBASE_PORT , <EOL> settings . HBASE_TABLES [ '<STR_LIT>' ] ) <EOL> continents = json . load ( open ( settings . path ( '<STR_LIT>' ) ) ) <EOL> countries = json . load ( open ( settings . path ( '<STR_LIT>' ) ) ) <EOL> regions = json . load ( open ( settings . path ( '<STR_LIT>' ) ) ) <EOL> geo = ( continents , countries , regions ) <EOL> REDACTED = ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> G = { <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : [ ] , <EOL> '<STR_LIT>' : { } , <EOL> '<STR_LIT:version>' : <NUM_LIT:8> , <EOL> } <EOL> def defaultdict_int ( ) : <EOL> return defaultdict ( int ) <EOL> G [ '<STR_LIT>' ] = dict ( ( k , { } ) for k in continents . values ( ) ) <EOL> for country , continent in continents . items ( ) : <EOL> G [ '<STR_LIT>' ] [ continent ] [ country ] = defaultdict ( defaultdict_int ) <EOL> for country , regions in regions . items ( ) : <EOL> continent = continents [ country ] <EOL> for region in regions : <EOL> G [ '<STR_LIT>' ] [ continent ] [ country ] [ region ] = defaultdict ( int ) <EOL> def row_name ( dt ) : <EOL> """<STR_LIT>""" <EOL> return '<STR_LIT>' % ( FX , dt . strftime ( '<STR_LIT>' ) ) <EOL> def time_sequence ( dt , num = <NUM_LIT:100> ) : <EOL> for i in xrange ( num ) : <EOL> yield dt + timedelta ( minutes = i ) <EOL> def row_sum ( row ) : <EOL> return sum ( row . columns . itervalues ( ) ) if row else <NUM_LIT:0> <EOL> def get_counts ( dt , num = <NUM_LIT:1> ) : <EOL> """<STR_LIT>""" <EOL> if num == <NUM_LIT:1> : <EOL> rows = hbase . row ( row_name ( dt ) , [ '<STR_LIT>' ] ) <EOL> else : <EOL> rows = hbase . scanner ( row_name ( dt ) , [ '<STR_LIT>' ] ) . list ( num ) <EOL> return [ ( t . utctimetuple ( ) [ : <NUM_LIT:5> ] , row_sum ( row ) ) <EOL> for t , row in zip ( time_sequence ( dt , num ) , rows ) ] <EOL> def extend_counts ( counts ) : <EOL> for t , count in counts : <EOL> G [ '<STR_LIT>' ] += count <EOL> G [ '<STR_LIT>' ] . append ( ( t , G [ '<STR_LIT>' ] ) ) <EOL> G [ '<STR_LIT>' ] = G [ '<STR_LIT>' ] [ - <NUM_LIT> : ] <EOL> if len ( G [ '<STR_LIT>' ] ) == <NUM_LIT:1> : <EOL> t = datetime ( * G [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] ) <EOL> G [ '<STR_LIT>' ] . insert ( <NUM_LIT:0> , ( t . utctimetuple ( ) [ : <NUM_LIT:5> ] , <NUM_LIT:0> ) ) <EOL> def process_locations ( rows ) : <EOL> """<STR_LIT>""" <EOL> arc = G [ '<STR_LIT>' ] <EOL> continents , countries , regions = geo <EOL> rv = [ ] <EOL> total = <NUM_LIT:0> <EOL> alfred = <NUM_LIT:0> <EOL> for row in rows : <EOL> new = [ ] <EOL> for key , val in row . columns . iteritems ( ) : <EOL> total += val <EOL> country , region , city , lat , lon = key . split ( '<STR_LIT::>' ) [ - <NUM_LIT:5> : ] <EOL> if country in REDACTED : <EOL> continue <EOL> try : <EOL> if region . strip ( ) in ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> region = '<STR_LIT>' <EOL> log . debug ( '<STR_LIT>' % key ) <EOL> if float ( lat ) == float ( lon ) == <NUM_LIT:0> : <EOL> continue <EOL> if ( country , region , city ) == ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) : <EOL> alfred += <NUM_LIT:1> <EOL> continue <EOL> continent = continents [ country ] <EOL> arc [ continent ] [ country ] [ region ] [ city ] += val <EOL> new . append ( ( continent , country , region , city , <EOL> lat , lon , val ) ) <EOL> except ( KeyError , ValueError ) : <EOL> log . error ( '<STR_LIT>' % key , exc_info = True ) <EOL> pass <EOL> rv . append ( ( total , new ) ) <EOL> log . info ( '<STR_LIT>' % alfred ) <EOL> return rv <EOL> def _get_locations ( dt , num = <NUM_LIT:1> ) : <EOL> """<STR_LIT>""" <EOL> if num == <NUM_LIT:1> : <EOL> rows = hbase . row ( row_name ( dt ) , [ '<STR_LIT>' ] ) <EOL> else : <EOL> rows = hbase . scanner ( row_name ( dt ) , [ '<STR_LIT>' ] ) . list ( num ) <EOL> locs = process_locations ( rows ) <EOL> return [ ( t . utctimetuple ( ) [ : <NUM_LIT:5> ] , r ) <EOL> for t , r in zip ( time_sequence ( dt , num ) , locs ) ] <EOL> def get_map ( dt , num = <NUM_LIT:1> ) : <EOL> """<STR_LIT>""" <EOL> times = [ ( t , ( num , [ r [ - <NUM_LIT:3> : ] for r in rows ] ) ) <EOL> for t , ( num , rows ) in _get_locations ( dt , num ) ] <EOL> hits = [ row for t in times for row in t [ <NUM_LIT:1> ] [ <NUM_LIT:1> ] ] <EOL> return ( times [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] , len ( hits ) , hits ) <EOL> def get_arc ( ) : <EOL> """<STR_LIT>""" <EOL> def unpack ( dict_ ) : <EOL> """<STR_LIT>""" <EOL> return revsort ( ( a . strip ( ) , b , c ) for a , ( b , c ) in dict_ . iteritems ( ) ) <EOL> revsort = lambda xs : sorted ( xs , key = itemgetter ( <NUM_LIT:1> ) , reverse = True ) <EOL> continents , world_sum = { } , <NUM_LIT:0> <EOL> for continent , country_dict in G [ '<STR_LIT>' ] . iteritems ( ) : <EOL> countries , continent_sum = { } , <NUM_LIT:0> <EOL> for country , region_dict in country_dict . iteritems ( ) : <EOL> regions , country_sum = { } , <NUM_LIT:0> <EOL> for region , cities in region_dict . iteritems ( ) : <EOL> total = sum ( cities . itervalues ( ) ) <EOL> if total : <EOL> cs = [ ( k . strip ( ) , v ) for k , v in cities . iteritems ( ) ] <EOL> regions [ region ] = [ total , revsort ( cs ) ] <EOL> country_sum += total <EOL> if country_sum : <EOL> countries [ country ] = [ country_sum , unpack ( regions ) ] <EOL> continent_sum += country_sum <EOL> if continent_sum : <EOL> continents [ continent ] = [ continent_sum , unpack ( countries ) ] <EOL> world_sum += continent_sum <EOL> return ( None , world_sum , unpack ( continents ) ) <EOL> def makedirs ( d ) : <EOL> if not os . path . exists ( d ) : <EOL> log . info ( '<STR_LIT>' % d ) <EOL> os . makedirs ( d ) <EOL> def write_files ( dt , count_data = None , map_data = None , arc_data = None , <EOL> interval = <NUM_LIT> ) : <EOL> """<STR_LIT>""" <EOL> log . info ( '<STR_LIT>' % dt ) <EOL> xs = { '<STR_LIT:count>' : count_data , '<STR_LIT>' : map_data , '<STR_LIT>' : arc_data } <EOL> for name , data in xs . items ( ) : <EOL> if not data : <EOL> continue <EOL> fmt = '<STR_LIT>' . format ( name = name ) <EOL> path = os . path . join ( JSON_DIR , dt . strftime ( fmt ) ) <EOL> next = ( dt + timedelta ( seconds = interval ) ) . strftime ( fmt ) <EOL> makedirs ( os . path . dirname ( path ) ) <EOL> d = { '<STR_LIT>' : next , '<STR_LIT>' : interval , '<STR_LIT:data>' : data } <EOL> json . dump ( d , open ( path , '<STR_LIT:w>' ) , separators = ( '<STR_LIT:U+002C>' , '<STR_LIT::>' ) ) <EOL> def collect ( dt ) : <EOL> """<STR_LIT>""" <EOL> log . info ( '<STR_LIT>' % dt ) <EOL> extend_counts ( get_counts ( dt ) ) <EOL> write_files ( dt , G [ '<STR_LIT>' ] , get_map ( dt ) , get_arc ( ) ) <EOL> dump_state ( dt ) <EOL> def now ( ) : <EOL> return datetime . utcnow ( ) - timedelta ( minutes = <NUM_LIT:1> ) <EOL> def do_the_stuff_to_the_thing ( ) : <EOL> dt = now ( ) <EOL> next = dt + timedelta ( minutes = <NUM_LIT:1> ) <EOL> if dt . second < <NUM_LIT:15> : <EOL> log . info ( '<STR_LIT>' ) <EOL> time . sleep ( <NUM_LIT:15> - dt . second ) <EOL> collect ( dt ) <EOL> wait = next . replace ( second = <NUM_LIT:15> ) - now ( ) . replace ( microsecond = <NUM_LIT:0> ) <EOL> if wait . seconds <= <NUM_LIT> : <EOL> log . info ( '<STR_LIT>' % wait . seconds ) <EOL> time . sleep ( wait . seconds ) <EOL> else : <EOL> log . info ( '<STR_LIT>' ) <EOL> def main ( ) : <EOL> load_state ( ) <EOL> log . info ( '<STR_LIT>' ) <EOL> while <NUM_LIT:1> : <EOL> try : <EOL> do_the_stuff_to_the_thing ( ) <EOL> except hb . exceptions : <EOL> log . error ( '<STR_LIT>' , exc_info = True ) <EOL> hbase . recycle ( ) <EOL> def dump_state ( dt ) : <EOL> """<STR_LIT>""" <EOL> log . info ( '<STR_LIT>' % dt ) <EOL> if os . path . exists ( PICKLE ) : <EOL> shutil . copyfile ( PICKLE , BACKUP ) <EOL> d = { '<STR_LIT>' : G , '<STR_LIT>' : dt } <EOL> pickle . dump ( d , open ( PICKLE , '<STR_LIT:w>' ) ) <EOL> def load_state ( ) : <EOL> """<STR_LIT>""" <EOL> if not ( os . path . exists ( PICKLE ) or os . path . exists ( BACKUP ) ) : <EOL> return <EOL> log . info ( '<STR_LIT>' ) <EOL> try : <EOL> d = pickle . load ( open ( PICKLE ) ) <EOL> except Exception : <EOL> log . error ( '<STR_LIT>' , exc_info = True ) <EOL> if os . path . exists ( BACKUP ) : <EOL> log . info ( '<STR_LIT>' ) <EOL> d = pickle . load ( open ( BACKUP ) ) <EOL> if d [ '<STR_LIT>' ] . get ( '<STR_LIT:version>' ) == <NUM_LIT:7> : <EOL> upgrade_7to8 ( d [ '<STR_LIT>' ] ) <EOL> if d [ '<STR_LIT>' ] . get ( '<STR_LIT:version>' ) == G [ '<STR_LIT:version>' ] : <EOL> for k , v in d [ '<STR_LIT>' ] . items ( ) : <EOL> G [ k ] = v <EOL> else : <EOL> log . info ( '<STR_LIT>' % G [ '<STR_LIT:version>' ] ) <EOL> dt = now ( ) <EOL> delta = dt - d [ '<STR_LIT>' ] . replace ( second = <NUM_LIT:0> ) <EOL> if delta . seconds > <NUM_LIT> : <EOL> log . info ( '<STR_LIT>' % ( delta . seconds / <NUM_LIT> ) ) <EOL> for i in xrange ( <NUM_LIT:1> , delta . seconds / <NUM_LIT> ) : <EOL> collect ( d [ '<STR_LIT>' ] + timedelta ( minutes = i ) ) <EOL> if now ( ) . minute != dt . minute : <EOL> log . info ( '<STR_LIT>' ) <EOL> load_state ( ) <EOL> if now ( ) . minute == d [ '<STR_LIT>' ] . minute : <EOL> log . info ( '<STR_LIT>' ) <EOL> time . sleep ( <NUM_LIT> - now ( ) . second ) <EOL> def upgrade_7to8 ( d ) : <EOL> d [ '<STR_LIT:version>' ] = <NUM_LIT:8> <EOL> alfred = d [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> log . info ( '<STR_LIT>' % alfred ) <EOL> d [ '<STR_LIT>' ] = [ ( a , b - alfred ) for a , b in d [ '<STR_LIT>' ] ] <EOL> log . info ( '<STR_LIT>' % ( d [ '<STR_LIT>' ] , d [ '<STR_LIT>' ] - alfred ) ) <EOL> d [ '<STR_LIT>' ] -= alfred <EOL> d [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] = <NUM_LIT:0> <EOL> def cleanup ( ) : <EOL> d = ( now ( ) - timedelta ( days = <NUM_LIT:2> ) ) . strftime ( '<STR_LIT>' ) <EOL> path = os . path . join ( JSON_DIR , d ) <EOL> if os . path . exists ( path ) : <EOL> log . info ( '<STR_LIT>' % path ) <EOL> shutil . rmtree ( path ) <EOL> try : <EOL> os . rmdir ( os . path . dirname ( path ) ) <EOL> os . rmdir ( os . path . dirname ( os . path . dirname ( path ) ) ) <EOL> except OSError : <EOL> pass </s>
