<s> """<STR_LIT>""" <EOL> import sys <EOL> import os <EOL> import logging <EOL> from pip import backwardcompat <EOL> from pip . _vendor import colorama , pkg_resources <EOL> def _color_wrap ( * colors ) : <EOL> def wrapped ( inp ) : <EOL> return "<STR_LIT>" . join ( list ( colors ) + [ inp , colorama . Style . RESET_ALL ] ) <EOL> return wrapped <EOL> def should_color ( consumer , environ , std = ( sys . stdout , sys . stderr ) ) : <EOL> real_consumer = ( consumer if not isinstance ( consumer , colorama . AnsiToWin32 ) <EOL> else consumer . wrapped ) <EOL> if real_consumer not in std : <EOL> return False <EOL> if hasattr ( real_consumer , "<STR_LIT>" ) and real_consumer . isatty ( ) : <EOL> return True <EOL> if environ . get ( "<STR_LIT>" ) == "<STR_LIT>" : <EOL> return True <EOL> return False <EOL> def should_warn ( current_version , removal_version ) : <EOL> current_version = "<STR_LIT:.>" . join ( current_version . split ( "<STR_LIT:.>" ) [ : <NUM_LIT:2> ] ) <EOL> removal_version = "<STR_LIT:.>" . join ( removal_version . split ( "<STR_LIT:.>" ) [ : <NUM_LIT:2> ] ) <EOL> major , minor = removal_version . split ( "<STR_LIT:.>" ) <EOL> minor = str ( int ( minor ) - <NUM_LIT:1> ) <EOL> warn_version = "<STR_LIT:.>" . join ( [ major , minor ] ) <EOL> return ( pkg_resources . parse_version ( current_version ) <EOL> < pkg_resources . parse_version ( warn_version ) ) <EOL> class Logger ( object ) : <EOL> """<STR_LIT>""" <EOL> VERBOSE_DEBUG = logging . DEBUG - <NUM_LIT:1> <EOL> DEBUG = logging . DEBUG <EOL> INFO = logging . INFO <EOL> NOTIFY = ( logging . INFO + logging . WARN ) / <NUM_LIT:2> <EOL> WARN = WARNING = logging . WARN <EOL> ERROR = logging . ERROR <EOL> FATAL = logging . FATAL <EOL> LEVELS = [ VERBOSE_DEBUG , DEBUG , INFO , NOTIFY , WARN , ERROR , FATAL ] <EOL> COLORS = { <EOL> WARN : _color_wrap ( colorama . Fore . YELLOW ) , <EOL> ERROR : _color_wrap ( colorama . Fore . RED ) , <EOL> FATAL : _color_wrap ( colorama . Fore . RED ) , <EOL> } <EOL> def __init__ ( self ) : <EOL> self . consumers = [ ] <EOL> self . indent = <NUM_LIT:0> <EOL> self . explicit_levels = False <EOL> self . in_progress = None <EOL> self . in_progress_hanging = False <EOL> def add_consumers ( self , * consumers ) : <EOL> if sys . platform . startswith ( "<STR_LIT>" ) : <EOL> for level , consumer in consumers : <EOL> if hasattr ( consumer , "<STR_LIT>" ) : <EOL> self . consumers . append ( <EOL> ( level , colorama . AnsiToWin32 ( consumer ) ) , <EOL> ) <EOL> else : <EOL> self . consumers . append ( ( level , consumer ) ) <EOL> else : <EOL> self . consumers . extend ( consumers ) <EOL> def debug ( self , msg , * args , ** kw ) : <EOL> self . log ( self . DEBUG , msg , * args , ** kw ) <EOL> def info ( self , msg , * args , ** kw ) : <EOL> self . log ( self . INFO , msg , * args , ** kw ) <EOL> def notify ( self , msg , * args , ** kw ) : <EOL> self . log ( self . NOTIFY , msg , * args , ** kw ) <EOL> def warn ( self , msg , * args , ** kw ) : <EOL> self . log ( self . WARN , msg , * args , ** kw ) <EOL> def error ( self , msg , * args , ** kw ) : <EOL> self . log ( self . ERROR , msg , * args , ** kw ) <EOL> def fatal ( self , msg , * args , ** kw ) : <EOL> self . log ( self . FATAL , msg , * args , ** kw ) <EOL> def deprecated ( self , removal_version , msg , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> from pip import __version__ <EOL> if should_warn ( __version__ , removal_version ) : <EOL> self . warn ( msg , * args , ** kwargs ) <EOL> else : <EOL> self . error ( msg , * args , ** kwargs ) <EOL> def log ( self , level , msg , * args , ** kw ) : <EOL> if args : <EOL> if kw : <EOL> raise TypeError ( <EOL> "<STR_LIT>" ) <EOL> args = args or kw <EOL> if args : <EOL> rendered = msg % args <EOL> else : <EOL> rendered = msg <EOL> rendered = '<STR_LIT:U+0020>' * self . indent + rendered <EOL> if self . explicit_levels : <EOL> rendered = '<STR_LIT>' % ( level , rendered ) <EOL> for consumer_level , consumer in self . consumers : <EOL> if self . level_matches ( level , consumer_level ) : <EOL> if ( self . in_progress_hanging <EOL> and consumer in ( sys . stdout , sys . stderr ) ) : <EOL> self . in_progress_hanging = False <EOL> sys . stdout . write ( '<STR_LIT:\n>' ) <EOL> sys . stdout . flush ( ) <EOL> if hasattr ( consumer , '<STR_LIT>' ) : <EOL> write_content = rendered + '<STR_LIT:\n>' <EOL> if should_color ( consumer , os . environ ) : <EOL> colorizer = self . COLORS . get ( level , lambda x : x ) <EOL> write_content = colorizer ( write_content ) <EOL> consumer . write ( write_content ) <EOL> if hasattr ( consumer , '<STR_LIT>' ) : <EOL> consumer . flush ( ) <EOL> else : <EOL> consumer ( rendered ) <EOL> def _show_progress ( self ) : <EOL> """<STR_LIT>""" <EOL> return ( self . stdout_level_matches ( self . NOTIFY ) and sys . stdout . isatty ( ) ) <EOL> def start_progress ( self , msg ) : <EOL> assert not self . in_progress , ( <EOL> "<STR_LIT>" <EOL> % ( msg , self . in_progress ) ) <EOL> if self . _show_progress ( ) : <EOL> sys . stdout . write ( '<STR_LIT:U+0020>' * self . indent + msg ) <EOL> sys . stdout . flush ( ) <EOL> self . in_progress_hanging = True <EOL> else : <EOL> self . in_progress_hanging = False <EOL> self . in_progress = msg <EOL> self . last_message = None <EOL> def end_progress ( self , msg = '<STR_LIT>' ) : <EOL> assert self . in_progress , ( <EOL> "<STR_LIT>" ) <EOL> if self . _show_progress ( ) : <EOL> if not self . in_progress_hanging : <EOL> sys . stdout . write ( '<STR_LIT>' + self . in_progress + msg + '<STR_LIT:\n>' ) <EOL> sys . stdout . flush ( ) <EOL> else : <EOL> logger . show_progress ( '<STR_LIT>' ) <EOL> logger . show_progress ( '<STR_LIT>' ) <EOL> sys . stdout . write ( msg + '<STR_LIT:\n>' ) <EOL> sys . stdout . flush ( ) <EOL> self . in_progress = None <EOL> self . in_progress_hanging = False <EOL> def show_progress ( self , message = None ) : <EOL> """<STR_LIT>""" <EOL> if self . in_progress_hanging : <EOL> if message is None : <EOL> sys . stdout . write ( '<STR_LIT:.>' ) <EOL> sys . stdout . flush ( ) <EOL> else : <EOL> if self . last_message : <EOL> padding = '<STR_LIT:U+0020>' * max ( <NUM_LIT:0> , len ( self . last_message ) - len ( message ) ) <EOL> else : <EOL> padding = '<STR_LIT>' <EOL> sys . stdout . write ( '<STR_LIT>' % <EOL> ( '<STR_LIT:U+0020>' * self . indent , self . in_progress , message , padding ) ) <EOL> sys . stdout . flush ( ) <EOL> self . last_message = message <EOL> def stdout_level_matches ( self , level ) : <EOL> """<STR_LIT>""" <EOL> return self . level_matches ( level , self . _stdout_level ( ) ) <EOL> def _stdout_level ( self ) : <EOL> """<STR_LIT>""" <EOL> for level , consumer in self . consumers : <EOL> if consumer is sys . stdout : <EOL> return level <EOL> return self . FATAL <EOL> def level_matches ( self , level , consumer_level ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( level , slice ) : <EOL> start , stop = level . start , level . stop <EOL> if start is not None and start > consumer_level : <EOL> return False <EOL> if stop is not None or stop <= consumer_level : <EOL> return False <EOL> return True <EOL> else : <EOL> return level >= consumer_level <EOL> @ classmethod <EOL> def level_for_integer ( cls , level ) : <EOL> levels = cls . LEVELS <EOL> if level < <NUM_LIT:0> : <EOL> return levels [ <NUM_LIT:0> ] <EOL> if level >= len ( levels ) : <EOL> return levels [ - <NUM_LIT:1> ] <EOL> return levels [ level ] <EOL> def move_stdout_to_stderr ( self ) : <EOL> to_remove = [ ] <EOL> to_add = [ ] <EOL> for consumer_level , consumer in self . consumers : <EOL> if consumer == sys . stdout : <EOL> to_remove . append ( ( consumer_level , consumer ) ) <EOL> to_add . append ( ( consumer_level , sys . stderr ) ) <EOL> for item in to_remove : <EOL> self . consumers . remove ( item ) <EOL> self . consumers . extend ( to_add ) <EOL> logger = Logger ( ) </s>
