Python colorama.Fore 模块，GREEN 实例源码
我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用colorama.Fore.GREEN。
def read_chat_history():
choice = select_a_friend()
if choice is not None:
print (Fore.BLUE + "Messages sent are shown in blue color \n" + Fore.GREEN + " Received Messages and Read Messages are shown in green color:"+Fore.RESET)
chats = friends[choice].chats
for chat in chats:
if chat.sent_by_me:
print (Fore.RED + str(chat['time']) + " " + Fore.BLUE + friends[choice]['name'] + " " + Fore.RESET + chat['message'])
else:
print (Fore.RED + str(chat['time']) + " " + Fore.GREEN + friends[choice]['name'] + " " + Fore.RESET + chat['message'])
else:
print "Wrong choice"
def ex_print (type, msg, ret):
colorama.init()
# Define color and style constants
c_error = Fore.RED
c_action = Fore.YELLOW
c_ok = Fore.GREEN
c_white = Fore.WHITE
s_br = Style.BRIGHT
s_reset = Style.RESET_ALL
message = {
"error": c_error + s_br,
"action": c_action,
"positive": c_ok + s_br,
"info": c_white + s_br,
"reset": s_reset
}
style = message.get(type, s_reset)
if ret == 0:
print(style + msg, end = "")
else:
print(style + msg)
def tenant_report(tenant_config):
from drift.tenant import get_connection_string
conn_string = get_connection_string(tenant_config)
print "Tenant configuration for '{}' on tier '{}':" \
.format(tenant_config["name"], get_tier_name())
for k in sorted(tenant_config.keys()):
print " {} = {}".format(k, tenant_config[k])
print "Connection string:\n {}".format(conn_string)
print "Database check... "
db_error = db_check(tenant_config)
if db_error:
if "does not exist" in db_error:
print Fore.RED + " FAIL! DB does not exist"
print " You can create this database by running this " \
"command again with the action 'create'"
else:
print Fore.RED + " {}".format(db_error)
else:
print Fore.GREEN + " OK! Database is online and reachable"
def tenants_report():
print "The following tenants are registered in config on tier '{}':".format(get_tier_name())
config = load_config()
for tenant_config in config.get("tenants", []):
name = tenant_config["name"]
# Todo: Get rid of this
if name == "*":
continue
sys.stdout.write(" {}... ".format(name))
db_error = db_check(tenant_config)
if not db_error:
print Fore.GREEN + "OK"
else:
if "does not exist" in db_error:
print Fore.RED + "FAIL! DB does not exist"
else:
print Fore.RED + "Error: %s" % db_error
print "To view more @R_89_4045@ion about each tenant run this command again with the tenant name"
def instant_giveaway(prize_name):
global won_giveaways, lost_giveaways
giveaway_Box = chromedriver.find_element_by_id(instant_Box)
giveaway_Box.click()
time.sleep(10)
get_result = chromedriver.find_element_by_id('title')
time.sleep(5)
if "you didn't win" in get_result.text:
lost_giveaways += 1
print(Fore.YELLOW + Style.BRIGHT + '\n **** You did not win: %s' % prize_name)
time.sleep(5)
elif "you're a winner!" in get_result.text:
won_giveaways += 1
print(Fore.GREEN + Style.BRIGHT + '\n **** Winner Winner! Chicken Dinner!: %s' % prize_name)
time.sleep(1)
playsound('.\sounds\\btyswt.mp3')
time.sleep(5)
else:
print(Fore.RED + Style.BRIGHT + '\n ---- UNRECOGNIZED RESPONSE FOR: %s' % prize_name)
# function to process the 'None' requirement giveaways.
def load_url(url, timeout):
# Build URL query to email signup page
urlquery = "http://" + url + "/m-users-a-email_list-job-add-email-" + targetEmail + "-source-2.htm"
print_out(Style.BRIGHT + Fore.WHITE + "Sending request to: " + url)
# Build the request
req = urllib.request.Request(
urlquery,
data=None,
headers={
'User-Agent': random.choice(useragents),
'Host': url
}
)
# Send
try:
f = urllib.request.urlopen(req)
print_out(Style.BRIGHT + Fore.GREEN + "Successfully sent!")
f.close()
except urllib.error.URLError as e:
print_out(Style.BRIGHT + Fore.RED + e.reason)
def watch(self, raw_args):
def tail(_file):
_file.seek(0, 2) # Go to the end of the file
while True:
line = _file.readline()
if not line:
time.sleep(0.1)
continue
yield line
_file = open(_log_path, 'r')
print (Fore.GREEN + '[bitmask] ' +
Fore.RESET + 'Watching log file %s' % _log_path)
for line in _file.readlines():
print line,
for line in tail(_file):
print line,
def tick(game, template, is_json, no_color):
if not game:
raise click.BadParameter('Missing required parameter "game"')
matches = download_history(game)
if is_json:
click.echo(json.dumps(list(matches), indent=2, sort_keys=True))
return
template = template if template else DEFAULT_TEMPLATE_RECAP
template = Template(template)
for m in matches:
if no_color or not COLOR_ENABLED: # if color is disabled just stdout
print_match(m, template)
continue
if m['t1_score'] > m['t2_score']:
m['t1'] = Fore.GREEN + m['t1'] + Fore.RESET
m['t2'] = Fore.RED + m['t2'] + Fore.RESET
else:
m['t2'] = Fore.GREEN + m['t2'] + Fore.RESET
m['t1'] = Fore.RED + m['t1'] + Fore.RESET
print_match(m, template)
def authorize_concept(self, concept):
if '2' not in concept.sf:
raise ValueError('No vocabulary code (2) given!')
if concept.sf['2'] in self.vocabularies:
vocab = self.vocabularies[concept.sf['2']]
else:
log.info(Fore.RED + '?' + Style.RESET_ALL + ' Could not authorize: %s', concept)
return
response = vocab.authorize_term(concept.term, concept.tag)
if response.get('id') is not None:
identifier = response.get('id')
if concept.sf.get('0'):
if concept.sf.get('0') == ANY_VALUE:
pass # ignore ANY_VALUE
elif identifier != concept.sf['0']:
identifier = pick_one('The $$0 value does not match the authority record id. ' +
'Please select which to use',
[concept.sf['0'], identifier])
concept.sf['0'] = identifier
log.info(Fore.GREEN + '?' + Style.RESET_ALL + ' Authorized: %s', concept)
else:
log.info(Fore.RED + '?' + Style.RESET_ALL + ' Could not authorize: %s', concept)
def fetch_missing_ids(self):
#Twitternames wihout kNown id's
swoid = [s.name for s in self.sources if s.id_str == None]
if len(swoid) == 0: return
user_lookups = self.api.lookup_users(screen_names=swoid)
# print('user_lookups response:\n{}'.format(user_lookups))
user_ids = dict([(u.screen_name, u.id_str) for u in user_lookups])
for k,v in user_ids.items():
sdb_id = [s.id for s in self.sources if s.name == k.lower()]
loggit('\nSource Id: {}'.format(sdb_id))
sdb = Source.query.get(sdb_id)
sdb.id_str = v
loggit('\nUpdated: {} with twitter_id: {}{}{}'.format(k, Fore.GREEN, v, Fore.RESET).encode("utf8"))
#Store to DB
db.flush()
#update twitter_user_ids array
# Refresh id's and screen_names globally
db.expire_all()
self.sources = Source.query.all()
global twitter_user_ids
global twitter_screen_names
twitter_user_ids = [s.id_str for s in self.sources if s.id_str != None]
twitter_screen_names = [s.name.lower() for s in self.sources if s.id_str != None]
def facebook_login(driver, username, password):
print("\n\n\nLogin to Facebook...."),
sys.stdout.flush()
url = "http://www.facebook.com"
driver.get(url)
elem = driver.find_element_by_id("email")
elem.send_keys(username)
elem = driver.find_element_by_id("pass")
elem.send_keys(password)
elem.send_keys(Keys.RETURN)
time.sleep(1)
html_source = driver.page_source
if "Please re-enter your password" in html_source or "Incorrect Email" in html_source:
print(Fore.RED + "Incorrect Username or Password")
driver.close()
exit()
else:
print(Fore.GREEN + "Success\n")
return driver.get_cookies()
def defrag(self):
download_queue = self.handle.get_download_queue()
downloading = [piece['piece_index'] for piece in download_queue]
numerales = ""
pieces = self.status.pieces
for i, piece in enumerate(pieces):
numeral = Fore.GREEN + "#" if piece else Fore.RED + "#"
if i in downloading:
numeral = Fore.YELLOW + "v"
elif self._served_blocks is not None and self._served_blocks[i]:
numeral = Fore.BLUE + ">"
numeral += str(self.handle.piece_priority(i))
numerales += numeral
if numerales != "":
numerales = term.magenta("\nPieces download state:\n" + numerales)
return "%s\n" % numerales
def echo_llamalist_stats(dev_names, badges=False, proof=True):
stats = {}
for dev_name in dev_names:
counts = get_llama_stats(dev_name) if not badges else get_badges_stats(dev_name)
dev_name = counts.get('Name')
if not dev_name: continue
stats[dev_name] = counts
print('@{} {:,} badges sent,{:,} badges received'.format(dev_name, counts['Given'], counts['Received']))
print(Fore.GREEN + '---' + Style.RESET_ALL)
num = 1
#+k[1]['Received']
for dev_name, counts in sorted(stats.items(), key=lambda k: k[1]['Given'], reverse=True):
if proof:
print('{}. @{} {:,} badges received <a href="https://{}.deviantart.com/badges/{}">[proof]</a><br>'
.format(num, dev_name, counts['Received'], '' if badges else 'llama/'))
else:
print('{}. @{} {:,} badges received<br>'
.format(num, counts['Received']))
num += 1
def TOR_PROC_CHECK():
isTorRunnin = False
TOR_INFO = {}
TOR_PROC = None
for proc in psutil.process_iter():
try:
pinfo = proc.as_dict(attrs=['pid', 'name'])
except psutil.NoSuchProcess:
pass
else:
if pinfo['name'] == "tor":
isTorRunnin = True
TOR_INFO['pid'] = pinfo['pid']
TOR_INFO['name'] = pinfo['name']
break
if isTorRunnin == True:
print ("[" + Fore.GREEN + Style.BRIGHT + "+" + Style.RESET_ALL + "]" + Fore.GREEN + Style.BRIGHT + " Tor is running." + Style.RESET_ALL)
TOR_PROC = psutil.Process(int(TOR_INFO['pid']))
return TOR_PROC
else:
print ("[" + Fore.RED + Style.BRIGHT + "-" + Style.RESET_ALL + "]" + Fore.RED + Style.BRIGHT + " Tor is not running." + Style.RESET_ALL)
exit()
def trains(self):
for raw_train in self.available_trains:
raw_train = raw_train['queryLeftNewDTO']
train_no = raw_train['station_train_code']
initial = train_no[0].lower()
if not self.options or initial in self.options:
train = [
train_no,
'\n'.join([Fore.GREEN + raw_train['from_station_name'] + Fore.RESET,
Fore.RED + raw_train['to_station_name'] + Fore.RESET]),
'\n'.join([Fore.GREEN + raw_train['start_time'] + Fore.RESET,
Fore.RED + raw_train['arrive_time'] + Fore.RESET]),
self._get_duration(raw_train),
raw_train['zy_num'],
raw_train['ze_num'],
raw_train['rw_num'],
raw_train['yw_num'],
raw_train['yz_num'],
raw_train['wz_num'],
]
yield train
def Brute_Thread(ip,username,passwd):
ssh=paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
global n,flag,flag1
n=n+1
try:
ssh.connect(ip,username=username,password=passwd)
except paramiko.AuthenticationException:
print Fore.RED+"[-]Username: %s\tPassword: %s Failed."%(username,passwd) + Fore.RESET
else:
print Fore.GREEN+"\n********************************************************"
print "[#]Username: %s\tPassword: %s Found........!!!"%(username,passwd)
print "********************************************************"+Fore.RESET
flag=1
flag1=1
print Fore.RED+"\nFound correct password after %s attempts..." %n +Fore.RESET
return
ssh.close()
return
def Gen_Dict():
ch=str(raw_input( Fore.CYAN +"Want to enter custom charset??(Enter y or n): "+Fore.RESET))
if ch == 'y':
charset=str(raw_input( Fore.CYAN +"Enter custom charset: "+Fore.RESET))
elif ch == 'n':
charset=string.letters[0:26]
min_length=int(input( Fore.CYAN +"Enter min passwd length: "+Fore.RESET))
max_length=int(input( Fore.CYAN +"Enter max passwd length: "+Fore.RESET))
f=open("tempwlist","w")
count=0
for wordlen in range(min_length,max_length+1):
for word in listwords(charset,wordlen):
f.write(word+'\n')
count+=1
print Fore.GREEN+"\nDictionary created with %s words....\n" %count + Fore.RESET
f.close()
def show_help(self):
help = 'dashboard section:\n'
help += ' show interfaces : Print all interfaces found by cmanager\n'
help += ' wizard wifi : Go to wifi wizard section\n'
help += ' exit : Exit cmanager\n\n'
help += 'wifi-wizard section:\n'
help += ' show profile : List profile that saved by cmanager\n'
help += ' show options : List available options used to create a profile\n'
help += ' set [options] [value] : Set value to available options before save the profile\n'
help += ' save profile : Save profile after options data\'s been filled\n'
help += ' del profile [profile] : Del profile by profile\'s name\n'
help += ' use [wireless_intarface] : Use this command BEFORE scanning available network or connecting a profile\n'
help += ' scan : Scanning available networks\n'
help += ' connect [profile] : Connecting wireless interface to a wifi network using specified profile\'s name\n'
help += ' back : Back to dashboard section'
print (Fore.GREEN + Style.BRIGHT + help)
def process_ele(self, reddit_element):
""" Accepts a RedditElement of Post/Comment details,then runs through the Handlers loaded from the other directory,attempting to download the url. """
print('%i/%i: ' % (self.loader.count_completed()+1, self.loader.count_total() ), end='')
stringutil.print_color(Fore.LIGHTYELLOW_EX, stringutil.out("[%s](%s): %s" % (reddit_element.type, reddit_element.subreddit, reddit_element.title), False))
for url in reddit_element.get_urls():
print('\tURL: %s' % url)
file = self.loader.url_exists(url)
if file:
stringutil.print_color(Fore.GREEN, "\t\t+URL already taken care of.")
reddit_element.add_file(url, file)
continue
if self.manifest:
skip, file = self.manifest.url_completed(url)
if skip and (file is None or os.path.exists(file)):
stringutil.print_color(Fore.GREEN, "\t\t+URL already handled in prevIoUs run.")
reddit_element.add_file(url, file)
if file is not None:
hashjar.add_hash(file) # Add the existing file hash so we can deduplicate against it.
continue
base_file, file_info = self.build_file_info(reddit_element)# Build the file @R_89_4045@ion array using this RedditElement's @R_89_4045@ion
file_path = self.process_url(url, file_info)
reddit_element.add_file(url, self.check_duplicates(file_path))
#
def process_url(self, url, info):
""" Accepts a URL and the array of file info generated for it by this class,and then attempts to download it using any possible handler.
Returns whatever the handlers do,which should be a path to the file itself or the contianing directory for an album.
+Also returns False or None if no appropriate handler was found,or if the handler told us not to download anything.
"""
ret_val = False # Default to 'False',meaning no file was located by a handler.
for h in self.handlers:
print("\tChecking handler: %s" % h.tag)
ret = h.handle(url, info)
if ret is None:
# None is returned when the handler specifically wants this URL to be "finished",but not added to the files list.
stringutil.print_color(Fore.GREEN, "\t+Handler '%s' completed correctly,but returned no files!" % h.tag )
ret_val = None
break
if ret:
# The handler will return a file/directory name if it worked properly.
ret_val = stringutil.normalize_file(ret)
stringutil.out("%s\t+Handler '%s' completed correctly! %s%s" % (Fore.GREEN, h.tag, stringutil.fit(ret_val, 75), Style.RESET_ALL) )
break
#
#
if ret_val is False:
stringutil.error("\t!No handlers were able to accept this URL." )
return ret_val
#
def log(self, prefix, text, line=False):
Now = datetime.Now()
message = ""
if prefix == '?':
c = Fore.CYAN
elif prefix == '+':
c = Fore.GREEN
elif prefix == '-':
c = Fore.RED
elif prefix == '!':
c = Fore.YELLOW
c = Style.BRIGHT + c
e = Style.RESET_ALL + Fore.RESET
if line:
print c+"["+Now.strftime("%Y-%m-%d %H:%M:%s")+"]["+prefix+"] "+text+e
else :
print "["+Now.strftime("%Y-%m-%d %H:%M:%s")+"]["+c+prefix+e+"] "+text
def log(prefix, line=False):
Now = datetime.Now()
message = ""
if prefix == '?':
c = Fore.CYAN
elif prefix == '+':
c = Fore.GREEN
elif prefix == '-':
c = Fore.RED
elif prefix == '!':
c = Fore.YELLOW
c = Style.BRIGHT + c
e = Style.RESET_ALL + Fore.RESET
if line:
print c+"["+Now.strftime("%Y-%m-%d %H:%M")+"]["+prefix+"] "+text+e
else :
print "["+Now.strftime("%Y-%m-%d %H:%M")+"]["+c+prefix+e+"] "+text
def colorprint(text_color, bail_result=False):
'''
Colorprint text into the console. Call it as a curried function.
greenPrinter = colorprinter('GREEN')
greenPrinter('this will be green')
OR:
colorprint('GREEN')('this will be green')
'''
def printer(text):
''' This actually does the printing part. Allows for reusable color printers. '''
color = Fore.GREEN if text_color == 'GREEN' else Fore.RED
if not bail_result:
print(color + text)
print(Style.RESET_ALL)
else:
return color + text
return printer
def status_loop(self):
x = self.targets
status = ""
for i in range(0, len(x), 3):
a1 = "%d. %s (%s)" % (i+1, x[i]['team'], x[i]['ip'])
if i + 1 < len(x):
a2 = "%d. %s (%s)" % (i+2, x[i+1]['team'], x[i+1]['ip'])
else:
a2 = ""
if i + 2 < len(x):
a3 = "%d. %s (%s)" % (i+3, x[i+2]['team'], x[i+2]['ip'])
else:
a3 = ""
a1f = Fore.GREEN if x[i]['compromised'] else Fore.RED
if i + 1 < len(x):
a2f = Fore.GREEN if x[i+1]['compromised'] else Fore.RED
if i + 2 < len(x):
a3f = Fore.GREEN if x[i+2]['compromised'] else Fore.RED
a1 = a1f + a1.ljust(45, " ") + Style.RESET_ALL
a2 = a2f + a2.ljust(45, " ") + Style.RESET_ALL
a3 = a3f + a3.ljust(20, " ") + Style.RESET_ALL
status += ("%s%s%s\n" % (a1, a2, a3))
open("status", 'w').write(status)
self.loop.call_later(self.detect_interval, self.status_work)
def trains(self):
for raw_train in self.available_trains:
raw_train = raw_train['queryLeftNewDTO']
train_no = raw_train['station_train_code']
initial = train_no[0].lower()
if not self.options or initial in self.options:
train = [
train_no,
]
yield train
def getparam(ser, p_name, p_data):
"""Decode a specific parameter from the serial port."""
printc(Back.RED + p_name, prepend='\n')
for typ in p_data:
schr = ser.read()
if typ == 'h':
printc(c2hex(schr), Fore.GREEN)
elif typ == 'H':
if len(HIGHLIGHT):
HIGHLIGHT.popleft()
HIGHLIGHT.append(c2hex(schr))
printc(c2hex(schr), Fore.BLUE)
elif typ == 'a':
printc(c2ascii(schr) or c2hex(schr), Fore.WHITE, prepend='')
else:
printc(c2hex(schr), Fore.WHITE)
printc(Style.RESET_ALL, prepend='')
def setup_method(self, test_method):
init()
self.stack_status_colourer = StackStatusColourer()
self.statuses = {
"CREATE_COMPLETE": Fore.GREEN,
"CREATE_Failed": Fore.RED,
"CREATE_IN_PROGRESS": Fore.YELLOW,
"DELETE_COMPLETE": Fore.GREEN,
"DELETE_Failed": Fore.RED,
"DELETE_IN_PROGRESS": Fore.YELLOW,
"ROLLBACK_COMPLETE": Fore.RED,
"ROLLBACK_Failed": Fore.RED,
"ROLLBACK_IN_PROGRESS": Fore.YELLOW,
"UPDATE_COMPLETE": Fore.GREEN,
"UPDATE_COMPLETE_CLEANUP_IN_PROGRESS": Fore.YELLOW,
"UPDATE_Failed": Fore.RED,
"UPDATE_IN_PROGRESS": Fore.YELLOW,
"UPDATE_ROLLBACK_COMPLETE": Fore.GREEN,
"UPDATE_ROLLBACK_COMPLETE_CLEANUP_IN_PROGRESS": Fore.YELLOW,
"UPDATE_ROLLBACK_Failed": Fore.RED,
"UPDATE_ROLLBACK_IN_PROGRESS": Fore.YELLOW
}
def main():
"""Handles command-line params and runs the respective core function."""
colorama.init(autoreset=True)
args = docopt(__doc__, version="Foliant %s (Python)" % foliant_version)
if args["build"] or args["make"]:
result = builder.build(args["<target>"], args["--path"])
elif args["upload"] or args["up"]:
result = uploader.upload(args["<document>"])
print("---")
print(Fore.GREEN + "Result: %s" % result)
colorama.deinit()
def production_config(project_name: str):
"""
Create production.cfg file
"""
print(Fore.GREEN + "[qoqa] Creating production.cfg file")
config = configparser.RawConfigParser()
config['STATUS'] = {
'Production': True
}
config['SECRET_KEY'] = {
'value': ''.join(generate_key())
}
db_conf = INIT_PROJECT_CONfig['PROD_DB']
config['PRODUCTION_DATABASE'] = {k: v for k, v in db_conf.items()}
config['ALLOWED_HOSTS'] = {'Hosts': '127.0.0.1,localhost'}
with open('production.cfg', 'w') as production_file:
config.write(production_file)
print(Fore.GREEN + "[qoqa] Configuration file for production created")
def development_config(project_name: str):
"""
Create configuration file for django project
"""
print(Fore.GREEN + "[qoqa] Creating configuration file")
config = configparser.RawConfigParser()
config['STATUS'] = {
'Production': False
}
config['SECRET_KEY'] = {
'value': ''.join(generate_key())
}
dev = INIT_PROJECT_CONfig['DEV_DB']
config['DEVELOPMENT_DATABASE'] = {k: v for k, v in dev.items()}
config['ALLOWED_HOSTS'] = {'Hosts': '127.0.0.1,localhost'}
with open(project_name+'.cfg', 'w') as project_file:
config.write(project_file)
print(Fore.GREEN + '[qoqa] Configuration file has been created')
def create(project_name: str):
"""
Launch interactive console and setup new project
"""
if project_name in ['new', 'build', 'release']:
print(Fore.RED + '[qoqa] invalid project name: {}'.format(project_name))
exit()
print("[qoqa] Configuring New Django Project")
if os.path.isdir(project_name):
print(Fore.RED + "[qoqa] project directory already exists")
exit()
db.setup()
os.mkdir(project_name)
os.chdir(project_name)
virtualenv.create(project_name)
development_config(project_name)
production_config(project_name)
print(Fore.GREEN + "[qoqa] Project {} has been setup".format(project_name))
def _startproject(self, context):
"""
Create a new django project
"""
dj_admin_script = os.path.join(context.bin_path, 'django-admin')
print(Fore.GREEN + "[qoqa] initializing django project")
subprocess.run([
dj_admin_script,
'startproject',
'--template='+template_zip,
self._project_name])
print(Fore.GREEN + "[qoqa] django project directory created")
os.chdir(self._project_name)
with open('__init__.py', 'a'):
pass
os.chmod('manage.py', stat.S_IRWXU)
def dpkg(version):
"""
Start the build process
"""
try:
subprocess.run([
'dch',
'-r',
version
], check=True)
except subprocess.CalledProcessError as error:
print(Fore.RED + "[qoqa] unable to release project")
print(Fore.RED + "[qoqa] {}".format(error))
exit()
try:
subprocess.run([
'dpkg-buildpackage',
'-us',
'-uc'
], check=True)
except subprocess.CalledProcessError as error:
print(Fore.RED + "[qoqa] unable to releaseproject")
print(Fore.RED + '[qoqa] {}'.format(error))
exit()
print(Fore.GREEN + "[qoqa] django project built")
def development_postgresql():
"""
Setup development configurations for postgresql database
"""
host = input(Fore.GREEN + 'Database Host: ')
name = input(Fore.GREEN + 'Database Name: ')
port = input(Fore.GREEN + 'Database Port: ')
user = input(Fore.GREEN + 'Database User: ')
password = input(Fore.GREEN + "Database Password: ")
if '' in [host, name, user, password, port]:
print(Fore.RED + "Please enter all database details")
development_postgresql()
else:
qoqa.INIT_PROJECT_CONfig['DEV_DB'] = {
'DATABASE': 'postgresql',
'HOST': host,
'NAME': name,
'USER': user,
'PASSWORD': password,
'PORT': port,
}
def create_playlist_m3u(self, tracks):
args = self.args
ripper = self.ripper
name = self.get_playlist_name()
if name is not None and args.playlist_m3u:
name = sanitize_playlist_name(to_ascii(name))
_base_dir = base_dir()
playlist_path = to_ascii(
os.path.join(_base_dir, name + '.m3u'))
print(Fore.GREEN + "Creating playlist m3u file " +
playlist_path + Fore.RESET)
encoding = "ascii" if args.ascii else "utf-8"
with codecs.open(enc_str(playlist_path), 'w', encoding) as playlist:
for idx, track in enumerate(tracks):
track.load()
if track.is_local:
continue
_file = ripper.format_track_path(idx, track)
if path_exists(_file):
playlist.write(os.path.relpath(_file, _base_dir) +
"\n")
def print_green(string):
if windows_client(): reinit()
print (Fore.GREEN + Style.BRIGHT + string + Style.RESET_ALL)
if windows_client(): deinit()
def get_colored_header():
return Fore.GREEN + get_header() + Fore.RESET
def compile_file (filename, language = None, args = ''):
print (args)
ret = compile.compile(filename, language = language, args = args)
if ret['status'] == compile.COMPILATION_CODE.ERROR:
print ('Compile: ' + Fore.RED + 'ERROR\n')
print (ret['stderr'])
exit(1)
else:
print ('Compile: ' + Fore.GREEN + 'OK\n')
def print_output(obj):
# to-do:
# Error while executing code
if obj['status'] != 0:
print ( '~ Test #{} - '.format(obj['testcase']) + Fore.RED + 'ERROR')
# In some cases the process spawn by cftool returns SIGSEGV (-11)
# and process.stderr is empty
if obj['stderr'] == '' and obj['status'] == -11:
print ('Process exited with SIGSEGV,probably because of a segmentation fault')
else:
print (obj['stderr'])
return
# split time between numbers and letters
m = re.split(r'(\d+)', obj['time'])
if int(m[1]) >= 5 and m[2] in ['s', 'm', 'h']:
print ( '~ Test #{} - '.format(obj['testcase']) + Fore.RED + '{}'.format(obj['time']) )
else:
print ( '~ Test #{} - {}'.format(obj['testcase'], obj['time']) )
stdout = obj['stdout']
expected = obj['expected']
if compare_outputs(stdout, expected):
print (Fore.GREEN + stdout)
print ('')
else:
print (Fore.RED + stdout)
print ('')
print (Fore.LIGHTBLACK_EX + 'Expected:')
print (Fore.LIGHTBLACK_EX + expected)
print ('')
def _compile_table(self, name):
# for readability
metrics = self.metrics[name]
fn = metrics['fn']; fp = metrics['fp']; tp = metrics['tp']; tn = metrics['tn']
pt = prettytable([Fore.GREEN + name + Style.RESET_ALL,'Predicted NO','Predicted YES','Total'])
pt.add_row(['Actual NO',tn,fp,tn+fp])
pt.add_row(['Actual YES',fn,tp,fn+tp])
pt.add_row(['Total',tn+fn,fp+tp,''])
pt.hrules = ALL
rows = ['' for i in range(6)]
rows[0] = pt.get_string(padding_width=5)
rows[1] = "Accuracy: {:^1}{:<.2f}".format("", metrics['accuracy'])
rows[2] = "Precision: {:^}{:<.2f}".format("", metrics['precision'])
rows[3] = "Recall: {:^3}{:<.2f}".format("", metrics['recall'])
rows[4] = "F1-score: {:^1}{:<.2f}".format("", metrics['f1_score'])
rows[5] = ""
self.rows.extend(rows)
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 process_classify(options):
if os.path.exists(options.files[0]):
filelist = options.files
if os.path.isdir(options.files[0]):
filelist = get_filelist(filelist[0])
result = BINOBJ.classify_files(
filelist, upload_missing=options.u, status_callback=my_callback)
else:
result = BINOBJ.classify_hashes(options.files)
if 'error' in result or result['status'] != 'done':
print(Style.BRIGHT + Fore.RED + "Request Failed")
else:
print("Classification Results:")
reqid = result.get('results', None)
if reqid is None:
# the request Failed before any files Could be analyzed
print(Style.BRIGHT + Fore.RED +
"Fail reason: {0} (error code={1})".format(
result['error']['message'], result['error']['code']))
return
classify_data = []
for key, value in result['results'].iteritems():
status = Style.RESET_ALL + Fore.GREEN + "OK" + Style.RESET_ALL
if 'error' in value:
status = Fore.RED + value['error']['message'] + Style.RESET_ALL
row = {'SHA1': key, 'label': value.get('label', '.'), 'family': value.get('family', 'Status': status}
classify_data.append(row)
if options.pretty_print:
show_results(classify_data, pretty_print=options.pretty_print)
else:
print("-" * 100)
for row in classify_data:
show_row(row)
return
def test_usertags():
user_tags = {
'info': F.GREEN + S.BRIGHT,
'info1': p('<g><b>'),
'call': lambda: F.BLUE + B.RED
}
am = AnsiMarkup(tags=user_tags)
assert am.parse('<info>1</info>') == F.GREEN + S.BRIGHT + '1' + S.RESET_ALL
assert am.parse('<info>1</info>') == am.parse('<info1>1</info1>')
assert am.parse('<call>1</call>') == F.BLUE + B.RED + '1' + S.RESET_ALL
assert am.strip('<info1>1</info1>') == '1'
def m_ok(self, m):
m = '[OK] ' + m
if COLORAMA:
print Fore.GREEN + m
else:
print m
def m_ok(self, m):
m = '[OK] ' + m
if COLORAMA:
print Fore.GREEN + m
else:
print m
def m_ok(self, m):
m = '[OK] ' + m
if COLORAMA:
print Fore.GREEN + m
else:
print m
def tor_test():
# Get Tor IP from wtfismyip
try:
with urllib.request.urlopen('https://wtfismyip.com/text') as response:
html = response.read()
print_out(Style.BRIGHT + Fore.GREEN + "Your Tor IP is: " + html.decode('utf-8'))
except HTTPError as e:
# do something
print_out(Style.BRIGHT + Fore.RED + "Error code: " + str(e.code))
exit(1)
except URLError as e:
# do something
print_out(Style.BRIGHT + Fore.RED + "Reason: " + str(e.reason))
exit(1)
def print_major_alarms(mj):
"""
Args:
mj (int): Counts of major alarms
Returns:
str: Colorized @R_89_4045@ion by using colorama
"""
if mj > 0:
return Back.RED + Fore.WHITE + str(mj) + Fore.RESET + Back.RESET
else:
return Fore.GREEN + str(mj) + Fore.RESET
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 [email protected] 举报，一经查实，本站将立刻删除。