<s> import sys <EOL> import os <EOL> import subprocess <EOL> import platform <EOL> from . helper import Helper <EOL> from . . questions import Question , ChoiceQuestion <EOL> from . . outputs import ConsoleOutput <EOL> from . . formatters import OutputFormatterStyle <EOL> from . . validators import Validator , Callable <EOL> from . . _compat import decode <EOL> class QuestionHelper ( Helper ) : <EOL> name = '<STR_LIT>' <EOL> _input_stream = None <EOL> stty = None <EOL> def ask ( self , input_ , output , question ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( output , ConsoleOutput ) : <EOL> output = output . get_error_output ( ) <EOL> if not input_ . is_interactive ( ) : <EOL> return question . default <EOL> if not question . validator : <EOL> return self . _do_ask ( output , question ) <EOL> interviewer = lambda : self . _do_ask ( output , question ) <EOL> return self . _validate_attempts ( interviewer , output , question ) <EOL> @ property <EOL> def input_stream ( self ) : <EOL> return self . _input_stream or sys . stdin <EOL> @ input_stream . setter <EOL> def input_stream ( self , stream ) : <EOL> self . _input_stream = stream <EOL> def _do_ask ( self , output , question ) : <EOL> """<STR_LIT>""" <EOL> self . _write_prompt ( output , question ) <EOL> input_stream = self . input_stream <EOL> autocomplete = question . autocompleter_values <EOL> if autocomplete is None or not self . _has_stty_available ( ) : <EOL> ret = False <EOL> if question . hidden : <EOL> try : <EOL> ret = self . _get_hidden_response ( output , input_stream ) <EOL> except RuntimeError : <EOL> if not question . hidden_fallback : <EOL> raise <EOL> if not ret : <EOL> ret = self . _read_from_input ( input_stream ) <EOL> else : <EOL> ret = self . _autocomplete ( output , question , input_stream ) <EOL> if len ( ret ) <= <NUM_LIT:0> : <EOL> ret = question . default <EOL> if question . normalizer : <EOL> return question . normalizer ( ret ) <EOL> return ret <EOL> def _write_prompt ( self , output , question ) : <EOL> """<STR_LIT>""" <EOL> message = question . question <EOL> if isinstance ( question , ChoiceQuestion ) : <EOL> width = max ( * map ( self . len , [ str ( k ) for k , _ in enumerate ( question . choices ) ] ) ) <EOL> messages = [ message ] <EOL> for key , value in enumerate ( question . choices ) : <EOL> messages . append ( '<STR_LIT>' % ( width , key , value ) ) <EOL> output . writeln ( messages ) <EOL> message = question . prompt <EOL> output . write ( message ) <EOL> def _write_error ( self , output , error ) : <EOL> """<STR_LIT>""" <EOL> if self . helper_set is not None and self . helper_set . has ( '<STR_LIT>' ) : <EOL> message = self . helper_set . get ( '<STR_LIT>' ) . format_block ( decode ( str ( error ) ) , '<STR_LIT:error>' ) <EOL> else : <EOL> message = '<STR_LIT>' % decode ( str ( error ) ) <EOL> output . writeln ( message ) <EOL> def _autocomplete ( self , output , question , input_stream ) : <EOL> """<STR_LIT>""" <EOL> autocomplete = question . autocompleter_values <EOL> ret = '<STR_LIT>' <EOL> i = <NUM_LIT:0> <EOL> ofs = - <NUM_LIT:1> <EOL> matches = [ x for x in autocomplete ] <EOL> num_matches = len ( matches ) <EOL> stty_mode = decode ( subprocess . check_output ( [ '<STR_LIT>' , '<STR_LIT>' ] ) ) . rstrip ( '<STR_LIT:\n>' ) <EOL> subprocess . check_output ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> output . get_formatter ( ) . set_style ( '<STR_LIT>' , OutputFormatterStyle ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> while True : <EOL> c = input_stream . read ( <NUM_LIT:1> ) <EOL> if c == '<STR_LIT>' : <EOL> if num_matches == <NUM_LIT:0> and i != <NUM_LIT:0> : <EOL> i -= <NUM_LIT:1> <EOL> output . write ( '<STR_LIT>' ) <EOL> if i == <NUM_LIT:0> : <EOL> ofs = - <NUM_LIT:1> <EOL> matches = [ x for x in autocomplete ] <EOL> num_matches = len ( matches ) <EOL> else : <EOL> num_matches = <NUM_LIT:0> <EOL> ret = ret [ : i ] <EOL> elif c == '<STR_LIT>' : <EOL> c += input_stream . read ( <NUM_LIT:2> ) <EOL> if c [ <NUM_LIT:2> ] == '<STR_LIT:A>' or c [ <NUM_LIT:2> ] == '<STR_LIT:B>' : <EOL> if c [ <NUM_LIT:2> ] == '<STR_LIT:A>' and ofs == - <NUM_LIT:1> : <EOL> ofs = <NUM_LIT:0> <EOL> if num_matches == <NUM_LIT:0> : <EOL> continue <EOL> ofs += - <NUM_LIT:1> if c [ <NUM_LIT:2> ] == '<STR_LIT:A>' else <NUM_LIT:1> <EOL> ofs = ( num_matches + ofs ) % num_matches <EOL> elif ord ( c ) < <NUM_LIT:32> : <EOL> if c == '<STR_LIT:\t>' or c == '<STR_LIT:\n>' : <EOL> if num_matches > <NUM_LIT:0> and ofs != - <NUM_LIT:1> : <EOL> ret = matches [ ofs ] <EOL> output . write ( ret [ i : ] ) <EOL> i = len ( ret ) <EOL> if c == '<STR_LIT:\n>' : <EOL> output . write ( c ) <EOL> break <EOL> num_matches = <NUM_LIT:0> <EOL> continue <EOL> else : <EOL> output . write ( c ) <EOL> ret += c <EOL> i += <NUM_LIT:1> <EOL> num_matches = <NUM_LIT:0> <EOL> ofs = <NUM_LIT:0> <EOL> for value in autocomplete : <EOL> if value . startswith ( ret ) and i != len ( value ) : <EOL> num_matches += <NUM_LIT:1> <EOL> matches [ num_matches - <NUM_LIT:1> ] = value <EOL> output . write ( '<STR_LIT>' ) <EOL> if num_matches > <NUM_LIT:0> and ofs != - <NUM_LIT:1> : <EOL> output . write ( '<STR_LIT>' ) <EOL> output . write ( '<STR_LIT>' + matches [ ofs ] [ i : ] + '<STR_LIT>' ) <EOL> output . write ( '<STR_LIT>' ) <EOL> subprocess . call ( [ '<STR_LIT>' , '<STR_LIT:%s>' % decode ( stty_mode ) ] ) <EOL> return ret <EOL> def _get_hidden_response ( self , output , input_stream ) : <EOL> """<STR_LIT>""" <EOL> if platform . system ( ) . lower ( ) == '<STR_LIT>' : <EOL> exe = os . path . join ( <EOL> os . path . dirname ( os . path . realpath ( __file__ ) ) , <EOL> '<STR_LIT:..>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' <EOL> ) <EOL> value = decode ( os . popen ( os . path . realpath ( exe ) ) . read ( ) ) . strip ( ) <EOL> output . writeln ( '<STR_LIT>' ) <EOL> return value <EOL> if self . _has_stty_available ( ) : <EOL> stty_mode = decode ( subprocess . check_output ( [ '<STR_LIT>' , '<STR_LIT>' ] ) ) . rstrip ( '<STR_LIT:\n>' ) <EOL> subprocess . check_output ( [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> value = decode ( input_stream . readline ( ) ) <EOL> subprocess . check_output ( [ '<STR_LIT>' , '<STR_LIT:%s>' % stty_mode ] ) <EOL> if not value : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> value = value . strip ( ) <EOL> output . writeln ( '<STR_LIT>' ) <EOL> return value <EOL> raise Exception ( '<STR_LIT>' ) <EOL> def _validate_attempts ( self , interviewer , output , question ) : <EOL> """<STR_LIT>""" <EOL> error = None <EOL> attempts = question . max_attempts <EOL> if not isinstance ( question . validator , Validator ) : <EOL> validator = Callable ( question . validator ) <EOL> else : <EOL> validator = question . validator <EOL> while attempts is None or attempts : <EOL> if error is not None : <EOL> self . _write_error ( output , error ) <EOL> try : <EOL> return validator . validate ( interviewer ( ) ) <EOL> except Exception as e : <EOL> error = e <EOL> if attempts is not None : <EOL> attempts -= <NUM_LIT:1> <EOL> raise error <EOL> def _read_from_input ( self , stream ) : <EOL> """<STR_LIT>""" <EOL> if stream == sys . stdin : <EOL> ret = stream . readline ( ) <EOL> else : <EOL> ret = stream . readline ( <NUM_LIT> ) <EOL> if not ret : <EOL> raise RuntimeError ( '<STR_LIT>' ) <EOL> return decode ( ret . strip ( ) ) <EOL> def _has_stty_available ( self ) : <EOL> if self . stty is not None : <EOL> return self . stty <EOL> devnull = open ( os . devnull , '<STR_LIT:w>' ) <EOL> try : <EOL> exit_code = subprocess . call ( [ '<STR_LIT>' ] , stdout = devnull , stderr = devnull ) <EOL> except Exception : <EOL> exit_code = <NUM_LIT:2> <EOL> self . stty = exit_code == <NUM_LIT:0> <EOL> return self . stty <EOL> def get_name ( self ) : <EOL> return self . name </s>
