def input_judgment(message):
print(Fore.YELLOW + message, end = '(y/n): ')
choice = input().lower()
if choice == 'y':
return True
else:
return False
python类YELLOW的实例源码
def quick_deploy_introduction():
user_message = Fore.YELLOW + "# Quick Deploy Script v1.0"
print(user_message)
def printHelp():
print('\nThis ' + Fore.BLUE + 'Intel' + Fore.WHITE
+ ' 8080 assembler was made for ' + Fore.BLUE + 'Project '
+ Fore.YELLOW + 'Week' + Fore.WHITE + ' at my school.')
print('It is written in ' + Fore.BLUE + 'Pyt' + Fore.YELLOW + 'hon'
+ Fore.WHITE)
print('Modules: ' + Fore.RED + 'Co' + Fore.BLUE + 'lo'
+ Fore.YELLOW + 'ra' + Fore.GREEN + 'ma' + Fore.WHITE)
print('\nPass a file path as an arguement.')
# Main function
def print_to_stdout(level, str_out):
""" The default debug function """
if level == NOTICE:
col = Fore.GREEN
elif level == WARNING:
col = Fore.RED
else:
col = Fore.YELLOW
if not is_py3:
str_out = str_out.encode(encoding, 'replace')
print(col + str_out + Fore.RESET)
# debug_function = print_to_stdout
def m_warn(self, m):
m = '[!] ' + m
if COLORAMA:
print Fore.YELLOW + m
else:
print m
def m_warn(self, m):
m = '[!] ' + m
if COLORAMA:
print Fore.YELLOW + m
else:
print m
def m_warn(self, m):
m = '[!] ' + m
if COLORAMA:
print Fore.YELLOW + m
else:
print m
def copyright():
copyright = """
=[ {0}smali-code-injector v1.1-dev{1} ]
+ -- --=[ Alexandre Teyar @Ares ]
+ -- --=[ Pentester at Ambersail Ltd. ]
+ -- --=[ GitHub: https://github.com/AresS31 ]
""".format(Fore.YELLOW, Fore.RESET)
print("{}".format(copyright))
def print_warnings(wn):
"""
Args:
wn (int): Counts of warnings
Returns:
str: Colorized information by using colorama
"""
if wn > 0:
return Fore.YELLOW + str(wn) + Fore.RESET
else:
return Fore.GREEN + str(wn) + Fore.RESET
def important(string):
print(Fore.YELLOW + string + Fore.RESET)
def reboot_loop(self):
while True:
self.ethers_manager.export_ethers()
print(Fore.YELLOW + "Reboot index %d" % self.reboot_string)
self.reboot_string += 1
gevent.sleep(self.reboot_delay)
def colorize(value, warning, error):
return wrap(value, Fore.RED if value >= error else YELLOW if value >= warning else Fore.RESET)
def _color_from_status(self, status):
mapping = {
'failed': Fore.RED,
'skipped': Fore.YELLOW,
'passed': Fore.GREEN,
'pipeline_error': Fore.RED
}
return mapping[status]
def _run_ssh_pexpect(cmd, password, using_bashc=False):
"""
Run a given command using pexpect.
"""
logger.debug(u'{}SSH Command: {}{}'.format(Style.DIM, cmd,
Style.RESET_ALL))
if using_bashc:
ssh_cli = pexpect.spawn('/bin/bash', ['-c', cmd])
else:
ssh_cli = pexpect.spawn(cmd)
i = ssh_cli.expect(['[Pp]assword: ', '\(yes/no\)\? '])
if i == 1:
ssh_cli.sendline('yes')
ssh_cli.expect('[Pp]assword: ')
ssh_cli.sendline(password)
time.sleep(1)
ssh_cli.expect(['Connection to [0-9\.a-z]+ is closed.', pexpect.EOF,
pexpect.TIMEOUT], timeout=5)
# Expected behavior is to get pexpect.EOF or closed connection, but due to
# a bug in docker we have to send an additional new line or Ctrl^C
out = str(ssh_cli.before) + str(ssh_cli.after)
logger.debug(u'Output:\n {}{}{}'.format(Fore.YELLOW, out, Style.RESET_ALL))
if ssh_cli.isalive():
ssh_cli.close()
return out
def _run_scp_command(cmd, user, host, password):
"""
Emulate user command line interation using SCP protocol
:param cmd: command to be executed
:param user: remote host user
:param host: remote host IP/hostname
:param password: passwrod for remote user on host
:returns None:
"""
logger.debug(u'{}Running SCP: {}{}'.format(
Style.DIM, cmd, Style.RESET_ALL))
scp = pexpect.spawn(cmd)
i = scp.expect(['\(yes/no\)\? ', '[Pp]assword: '])
if i == 0:
scp.sendline('yes')
scp.expect('[Pp]assword: ')
scp.sendline(password)
time.sleep(1)
try:
while True:
i = scp.expect([pexpect.EOF, '[0-9][0-9]:[0-9][0-9] '],
timeout=5)
if i == 0:
logger.debug(u'{}{}{}'.format(Fore.YELLOW, scp.before,
Style.RESET_ALL))
break
logger.debug(u'{}{}{}{}'.format(Fore.YELLOW, scp.before,
scp.after, Style.RESET_ALL))
time.sleep(.1)
except pexpect.TIMEOUT:
# A docker bug expecting an extra new line in the end. Ideally we
# will exit the loop getting pexpect.EOF, i.e. i==0
logger.debug(u'{}{}{}'.format(Fore.YELLOW, scp.before,
Style.RESET_ALL))
finally:
if scp.isalive():
scp.close()
def handle(self):
self.server.connection_list.append(self.client_address[0])
try:
# For TCP connections
data = self.request.recv(1024).strip()
except AttributeError:
# For UDP connections
data = self.request[0].strip()
LOG.debug('{}Client: {}\ndata: {}{}'.format(
Fore.YELLOW, self.client_address[0], data, Style.RESET_ALL))
def warn(message, detail=""):
level = Fore.YELLOW + "[WARN] " + Fore.RESET
print(prefix + level + Style.BRIGHT + message + Style.RESET_ALL + " " + detail + Style.RESET_ALL)
def _check_num_nodes(self):
if self.num_nodes < 1:
raise ValueError('Must provision at least one node.')
elif self.num_nodes < 3:
print(Fore.YELLOW + 'A Deis cluster must have 3 or more nodes, only continue if you adding to a current cluster.' + Fore.RESET)
print(Fore.YELLOW + 'Continue? (y/n)' + Fore.RESET)
accept = None
while True:
if accept == 'y':
return
elif accept == 'n':
raise StandardError('User canceled provisioning')
else:
accept = self._get_user_input('--> ').strip().lower()
def log_warning(message):
print(Fore.YELLOW + message + Fore.RESET)
def log_warning(message):
print(Fore.YELLOW + message + Fore.RESET)
