<s> import os <EOL> import sys <EOL> import atexit <EOL> import json <EOL> from . connection import Connection , Pager <EOL> from . exceptions import ResponseError <EOL> from functools import wraps <EOL> if "<STR_LIT>" not in sys . argv : <EOL> cache_file = "<STR_LIT>" <EOL> cache = { } <EOL> if os . path . isfile ( cache_file ) : <EOL> with open ( cache_file , "<STR_LIT:r>" ) as f : <EOL> try : <EOL> cache = json . load ( f ) <EOL> except ValueError : <EOL> pass <EOL> def save_cache ( ) : <EOL> with open ( cache_file , '<STR_LIT:w>' ) as f : <EOL> json . dump ( cache , f ) <EOL> atexit . register ( save_cache ) <EOL> def memoise ( wrapped ) : <EOL> @ wraps ( wrapped ) <EOL> def wrapper ( * args , ** kwargs ) : <EOL> key = args [ <NUM_LIT:0> ] <EOL> if key not in cache : <EOL> cache [ key ] = wrapped ( * args , ** kwargs ) <EOL> return cache [ key ] <EOL> return wrapper <EOL> else : <EOL> def memoise ( wrapped ) : <EOL> @ wraps ( wrapped ) <EOL> def wrapper ( * args , ** kwargs ) : <EOL> return wrapped ( * args , ** kwargs ) <EOL> return wrapper <EOL> token = os . environ . get ( "<STR_LIT>" ) <EOL> debug = os . environ . get ( "<STR_LIT>" ) <EOL> if token is None : <EOL> print ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> else : <EOL> if debug : <EOL> print ( "<STR_LIT>" % len ( token ) ) <EOL> conn = Connection ( token ) <EOL> def unique ( array ) : <EOL> array = [ x for x in array if x is not None ] <EOL> return list ( { v [ '<STR_LIT>' ] : v for v in array } . values ( ) ) <EOL> def flatten ( array ) : <EOL> return [ item for sublist in array for item in sublist ] <EOL> @ memoise <EOL> def get_paged_json ( uri ) : <EOL> json = [ ] <EOL> try : <EOL> pager = Pager ( conn , uri , params = { } , max_pages = <NUM_LIT:0> ) <EOL> for response in pager : <EOL> progress_advance ( ) <EOL> json += response . json ( ) <EOL> except ResponseError : <EOL> pass <EOL> return json <EOL> def get_page_int_json ( uri ) : <EOL> json = [ ] <EOL> page = <NUM_LIT:1> <EOL> while True : <EOL> progress_advance ( ) <EOL> response = get_data ( "<STR_LIT>" % ( uri , page ) ) <EOL> if len ( response ) == <NUM_LIT:0> : <EOL> break <EOL> else : <EOL> page = page + <NUM_LIT:1> <EOL> json += response <EOL> return json <EOL> def get_code_contributors ( repo_name ) : <EOL> progress ( "<STR_LIT>" ) <EOL> users = [ ] <EOL> response = get_page_int_json ( "<STR_LIT>" % repo_name ) <EOL> for entry in response : <EOL> users . append ( get_user_data ( entry ) ) <EOL> progress_complete ( ) <EOL> return unique ( users ) <EOL> def get_code_commentors ( repo_name , limit ) : <EOL> progress ( "<STR_LIT>" ) <EOL> pri_count = get_pri_count ( repo_name ) <EOL> if limit == <NUM_LIT:0> : <EOL> minimum = <NUM_LIT:1> <EOL> else : <EOL> minimum = max ( <NUM_LIT:1> , pri_count - limit ) <EOL> users = [ ] <EOL> for index in range ( minimum , pri_count + <NUM_LIT:1> ) : <EOL> users . append ( get_user ( "<STR_LIT>" % ( repo_name , index ) ) ) <EOL> users . append ( get_user ( "<STR_LIT>" % ( repo_name , index ) ) ) <EOL> for entry in get_paged_json ( "<STR_LIT>" % <EOL> ( repo_name , index ) ) : <EOL> users . append ( get_user_data ( entry ) ) <EOL> for entry in get_paged_json ( "<STR_LIT>" % <EOL> ( repo_name , index ) ) : <EOL> users . append ( get_user_data ( entry ) ) <EOL> progress_complete ( ) <EOL> return unique ( users ) <EOL> @ memoise <EOL> def get_data ( uri ) : <EOL> try : <EOL> resp = conn . send ( "<STR_LIT:GET>" , uri ) <EOL> return resp . json ( ) <EOL> except ResponseError : <EOL> return None <EOL> def get_pri_count ( repo_name ) : <EOL> prs = get_data ( "<STR_LIT>" % repo_name ) <EOL> issues = get_data ( "<STR_LIT>" % repo_name ) <EOL> if not prs : <EOL> pr_count = <NUM_LIT:0> <EOL> else : <EOL> pr_count = prs [ <NUM_LIT:0> ] [ "<STR_LIT>" ] <EOL> if not issues : <EOL> issue_count = <NUM_LIT:0> <EOL> else : <EOL> issue_count = issues [ <NUM_LIT:0> ] [ "<STR_LIT>" ] <EOL> return max ( pr_count , issue_count ) <EOL> def get_user_data ( entry ) : <EOL> if "<STR_LIT:user>" in entry . keys ( ) : <EOL> return { "<STR_LIT>" : entry [ "<STR_LIT:user>" ] [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : "<STR_LIT>" % entry [ "<STR_LIT:user>" ] [ "<STR_LIT>" ] , <EOL> "<STR_LIT:name>" : get_user_name ( entry [ "<STR_LIT:user>" ] [ "<STR_LIT>" ] ) } <EOL> else : <EOL> return { "<STR_LIT>" : entry [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : "<STR_LIT>" % entry [ "<STR_LIT>" ] , <EOL> "<STR_LIT:name>" : get_user_name ( entry [ "<STR_LIT>" ] ) } <EOL> @ memoise <EOL> def get_user ( uri ) : <EOL> progress_advance ( ) <EOL> entry = get_data ( uri ) <EOL> if entry is not None : <EOL> return get_user_data ( entry ) <EOL> @ memoise <EOL> def repo_exists ( repo_name ) : <EOL> try : <EOL> conn . send ( "<STR_LIT:GET>" , "<STR_LIT>" % repo_name ) <EOL> return True <EOL> except ResponseError : <EOL> return False <EOL> def progress ( message ) : <EOL> sys . stdout . write ( "<STR_LIT>" % message ) <EOL> sys . stdout . flush ( ) <EOL> def progress_advance ( ) : <EOL> sys . stdout . write ( "<STR_LIT:.>" ) <EOL> sys . stdout . flush ( ) <EOL> def progress_complete ( ) : <EOL> sys . stdout . write ( "<STR_LIT:\n>" ) <EOL> def get_user_name ( login ) : <EOL> user = get_data ( "<STR_LIT>" % login ) <EOL> if user [ "<STR_LIT:name>" ] is None : <EOL> user [ "<STR_LIT:name>" ] = login <EOL> return user [ "<STR_LIT:name>" ] <EOL> def consolidate ( contributors , commentors ) : <EOL> non_code_contributors = [ ] <EOL> for user in commentors : <EOL> user_name , avatar , name = user <EOL> if user not in contributors : <EOL> non_code_contributors . append ( user ) <EOL> return non_code_contributors <EOL> def display_users ( user_list , title , array = False ) : <EOL> print ( "<STR_LIT>" % ( title , len ( user_list ) ) ) <EOL> if array : <EOL> print ( "<STR_LIT:\n>" . join ( user_list ) ) <EOL> else : <EOL> for user in sorted ( user_list , key = lambda k : k [ '<STR_LIT>' ] . lower ( ) ) : <EOL> if user [ "<STR_LIT:name>" ] != user [ "<STR_LIT>" ] : <EOL> print ( "<STR_LIT>" % ( user [ "<STR_LIT>" ] , user [ '<STR_LIT:name>' ] ) ) <EOL> else : <EOL> print ( user [ "<STR_LIT>" ] ) </s>
