def _update_ui():
global timeout_count, result_codes, connection_error_count
print '\r',
for k, v in result_codes.iteritems():
print "%s:" % k,
if k == '200 OK':
print(Fore.LIGHTGREEN_EX),
else:
print(Fore.RED),
print "%s " % v,
print(Style.RESET_ALL),
if timeout_count > 0:
print('Timeouts: '+Fore.YELLOW + str(timeout_count) + Style.RESET_ALL) + ' ',
if connection_error_count >0:
print('Connection Errors: '+Fore.RED + str(connection_error_count) + Style.RESET_ALL),
sys.stdout.flush()
python类LIGHTGREEN_EX的实例源码
def trains(self):
for raw_train in self.available_trains:
raw_train_list = raw_train.split('|')
train_no = raw_train_list[3]
initial = train_no[0].lower()
duration = raw_train_list[10]
if initial in self.options:
train = [
train_no,
'\n'.join([Fore.LIGHTGREEN_EX + self.available_place[raw_train_list[6]] + Fore.RESET,
Fore.LIGHTRED_EX + self.available_place[raw_train_list[7]] + Fore.RESET]),
'\n'.join([Fore.LIGHTGREEN_EX + raw_train_list[8] + Fore.RESET,
Fore.LIGHTRED_EX + raw_train_list[9] + Fore.RESET]),
duration,
raw_train_list[-4] if raw_train_list[-4] else '--',
raw_train_list[-5] if raw_train_list[-5] else '--',
raw_train_list[-14] if raw_train_list[-14] else '--',
raw_train_list[-12] if raw_train_list[-12] else '--',
raw_train_list[-7] if raw_train_list[-7] else '--',
raw_train_list[-6] if raw_train_list[-6] else '--',
raw_train_list[-9] if raw_train_list[-9] else '--',
]
yield train
def diffs(self, diffs, *args, **kwargs):
def __coloring(elem, color):
return tuple(['%s%s%s' % (color, item, Style.RESET_ALL)
for item in list(elem)])
rows = [__coloring(item, Fore.LIGHTGREEN_EX if item[0] == '+' else Fore.LIGHTRED_EX) for item in diffs]
self.info(tabulate(rows,
tablefmt="psql", headers=['', 'File1', 'File2', 'Key', 'Value']), *args, **kwargs)
def formatEntry(e:Entry):
R = _S.RESET_ALL
ID = R + _B.BLUE
P = R + _F.LIGHTGREEN_EX
TAG = R + _F.RED
NAME = R + _B.RED
Bd = R + _F.CYAN
PATH = R + _F.YELLOW
TITLE = R + _F.LIGHTMAGENTA_EX
AUTHORS = R + _F.MAGENTA
prefix = Bd + '| ' + R
comment = ( s + '\n' + prefix for s in e.comment.split('\n') ) if e.comment != '' else ()
return ''.join( (
Bd, '--------------------------------------------------------------------------------', R, '\n',
prefix, ID, 'ID : ', '{:>5}'.format(e.ID or ''), R, ' '*47, P, '{:>20}'.format(e.priority), R, '\n',
prefix, NAME, e.name, R, '\n',
prefix, PATH, e.pathstr(), R, '\n',
*( (
prefix, TITLE, e.bibtex.get('title', ''), R, '\n',
prefix, AUTHORS, e.bibtex.get('author', ''), R, '\n',
) if e.bibtex else (
prefix, TITLE, '<No Bibtex>', R, '\n')
),
prefix, (R + ' ').join(''.join((TAG, '#', t)) for t in e.tags), '\n',
prefix, R, *comment , '\n',
Bd, '--------------------------------------------------------------------------------', R, '\n',
))
def info_format(print_type, text):
# info, prompt, alert, sleep
lb = '{0}[{1}'.format(Fore.LIGHTGREEN_EX, Style.RESET_ALL)
rb = '{0}]{1}'.format(Fore.LIGHTGREEN_EX, Style.RESET_ALL)
new_text = Fore.LIGHTWHITE_EX + text + Style.RESET_ALL
if print_type == 'info':
return '{0}{1}*{2}{3} {4}'.format(lb, Fore.LIGHTGREEN_EX, Style.RESET_ALL, rb, new_text)
elif print_type == 'prompt':
return '{0}{1}${2}{3} {4}'.format(lb, Fore.LIGHTYELLOW_EX, Style.RESET_ALL, rb, new_text)
elif print_type == 'alert':
return '{0}{1}!{2}{3} {4}'.format(lb, Fore.LIGHTRED_EX, Style.RESET_ALL, rb, new_text)
elif print_type == 'sleep':
return '{0}-{1} {2}'.format(lb, rb, new_text)
def get_alerts(customer_name, status, quiet=False):
""" gets alerts """
# There are 3 other v1 posts passed per customer with varying payloads.The dictionary below is required to return
# the necessary data; modifying it can break the request (needs more testing). I know it is not pep8 (too long)
data_dict = {"name":"time","min_doc_count":0,"size":5,"type":"date_range","field":"last_behavior","date_ranges":[{"from":"now-1h","to":"now","label":"Last hour"},{"from":"now-24h","to":"now","label":"Last day"},{"from":"now-7d","to":"now","label":"Last week"},{"from":"now-30d","to":"now","label":"Last 30 days"},{"from":"now-90d","to":"now","label":"Last 90 days"}]},{"name":"status","min_doc_count":0,"size":5,"type":"terms","field":"status"},{"name":"severity","min_doc_count":0,"size":5,"type":"range","field":"max_severity","ranges":[{"from":80,"to":101,"label":"Critical","id":4},{"from":60,"to":80,"label":"High","id":3},{"from":40,"to":60,"label":"Medium","id":2},{"from":20,"to":40,"label":"Low","id":1},{"from":0,"to":20,"label":"Informational","id":0}]},{"name":"scenario","min_doc_count":0,"size":0,"type":"terms","field":"behaviors.scenario"},{"name":"assigned_to_uid","min_doc_count":1,"size":5,"type":"terms","field":"assigned_to_uid","missing":"Unassigned"},{"name":"host","min_doc_count":1,"size":5,"type":"terms","field":"device.hostname.raw","missing":"Unknown"},{"name":"triggering_file","min_doc_count":1,"size":5,"type":"terms","field":"behaviors.filename.raw"}
s10 = falcon.post('https://falcon.crowdstrike.com/api2/detects/aggregates/detects/GET/v1', headers=header,
data=json.dumps(data_dict))
try:
if len(s10.json()['resources']) > 0:
# print(json.dumps(s10.json(), indent=4)) # full json data set!
cust_data = s10.json()
for bucket in cust_data['resources']:
if bucket['name'] == 'status':
for value in bucket['buckets']:
if value['label'] in status:
if 'count' in value and value['count'] > 0:
alert_str = info_format('alert', '{0}{1}{2} alert(s) detected!\n'.format(
Fore.LIGHTRED_EX, value['count'], Fore.LIGHTWHITE_EX))
alert_str += '----> {0}{1}{2}'.format(Fore.LIGHTGREEN_EX, customer_name, Style.RESET_ALL)
# print(json.dumps(bucket['buckets'], indent=4)) # for testing!
return alert_str
except KeyError:
if not quiet:
return info_format('alert', 'There was an issue retrieving alerts for {0}. Skipping...'.format(customer_name))
else:
return None
def parse_alert_short(raw_alert, quiet, customer_name=None, color=True):
""""""
if color is True:
yellow = Fore.LIGHTYELLOW_EX
green = Fore.LIGHTGREEN_EX
red = Fore.LIGHTRED_EX
reset = Style.RESET_ALL
else:
yellow = ''
green = ''
red = ''
reset = ''
try:
alert = enum_alert(raw_alert)
alert_str = ''
alert_cust_name = alert.get('customer_name')
alert_host = alert.get('hostname')
alert_severity = alert.get('max_severity_displayname')
alert_reason = alert.get('scenario')
alert_time = alert.get('timestamp')
alert_status = alert.get('status')
#alert_str += '{8} {4}{9}{6} - {0}{1}{6} alert on {2}{3}{6} for {4}{5}{6} ({7})!'.format(
# Fore.LIGHTYELLOW_EX, alert_severity, Fore.LIGHTGREEN_EX, alert_host, Fore.LIGHTRED_EX, alert_reason,
# Style.RESET_ALL, alert_time, alert_status.upper().replace('_', '-'), alert_cust_name)
alert_str += '{8} {4}{9}{6} - {0}{1}{6} alert on {2}{3}{6} for {4}{5}{6} ({7})!'.format(
yellow, alert_severity, green, alert_host, red, alert_reason,
reset, alert_time, alert_status.upper().replace('_', '-'), alert_cust_name)
return alert_str
except Exception: # KeyError:
if not quiet:
return info_format('alert',
'There was an issue retrieving alerts for {0}. Skipping...'.format(customer_name))
def pretty_print_testcase(testcase, error=''):
""" Pretty print a testcase """
if error:
msg_template = Style.BRIGHT + '{id}' + Style.RESET_ALL + '\t' + \
Fore.MAGENTA + '{status}' + Fore.RESET + '\t' + \
'{name}\t=> ' + str(error)
elif testcase['status'] == 'PASS':
msg_template = Style.BRIGHT + '{id}' + Style.RESET_ALL + '\t' + \
Fore.LIGHTGREEN_EX + '{status}' + Fore.RESET + '\t' + \
'{name}\t'
else:
msg_template = Style.BRIGHT + '{id}' + Style.RESET_ALL + '\t' + \
Fore.LIGHTRED_EX + '{status}' + Fore.RESET + '\t' + \
'{name}\t'
print(msg_template.format(**testcase), end=Style.RESET_ALL)
def _report_created(self):
# set up the report data
rows = []
ips = []
data_center = self._get_data_center().get('ABBR')
plan = self._get_plan().get('RAM')
for linode in self._created_linodes:
rows.append((
linode['hostname'],
linode['public'],
linode['private'],
linode['gateway'],
data_center,
plan
))
ips.append(linode['public'])
firewall_command = './apply-firewall.py --private-key /path/to/key/deis --hosts ' + string.join(ips, ' ')
# set up the report constants
divider = Style.BRIGHT + Fore.MAGENTA + ('=' * 109) + Fore.RESET + Style.RESET_ALL
column_format = " {:<20} {:<20} {:<20} {:<20} {:<12} {:>8}"
formatted_header = column_format.format(*('HOSTNAME', 'PUBLIC IP', 'PRIVATE IP', 'GATEWAY', 'DC', 'PLAN'))
# display the report
print('')
print(divider)
print(divider)
print('')
print(Style.BRIGHT + Fore.LIGHTGREEN_EX + ' Successfully provisioned ' + str(self.num_nodes) + ' nodes!' + Fore.RESET + Style.RESET_ALL)
print('')
print(Style.BRIGHT + Fore.CYAN + formatted_header + Fore.RESET + Style.RESET_ALL)
for row in rows:
print(Fore.CYAN + column_format.format(*row) + Fore.RESET)
print('')
print('')
print(Fore.LIGHTYELLOW_EX + ' Finish up your installation by securing your cluster with the following command:' + Fore.RESET)
print('')
print(' ' + firewall_command)
print('')
print(divider)
print(divider)
print('')
def _report_created(self):
# set up the report data
rows = []
ips = []
data_center = self._get_data_center().get('ABBR')
plan = self._get_plan().get('RAM')
for linode in self._created_linodes:
rows.append((
linode['hostname'],
linode['public'],
linode['private'],
linode['gateway'],
data_center,
plan
))
ips.append(linode['public'])
firewall_command = './apply-firewall.py --private-key /path/to/key/deis --hosts ' + string.join(ips, ' ')
# set up the report constants
divider = Style.BRIGHT + Fore.MAGENTA + ('=' * 109) + Fore.RESET + Style.RESET_ALL
column_format = " {:<20} {:<20} {:<20} {:<20} {:<12} {:>8}"
formatted_header = column_format.format(*('HOSTNAME', 'PUBLIC IP', 'PRIVATE IP', 'GATEWAY', 'DC', 'PLAN'))
# display the report
print('')
print(divider)
print(divider)
print('')
print(Style.BRIGHT + Fore.LIGHTGREEN_EX + ' Successfully provisioned ' + str(self.num_nodes) + ' nodes!' + Fore.RESET + Style.RESET_ALL)
print('')
print(Style.BRIGHT + Fore.CYAN + formatted_header + Fore.RESET + Style.RESET_ALL)
for row in rows:
print(Fore.CYAN + column_format.format(*row) + Fore.RESET)
print('')
print('')
print(Fore.LIGHTYELLOW_EX + ' Finish up your installation by securing your cluster with the following command:' + Fore.RESET)
print('')
print(' ' + firewall_command)
print('')
print(divider)
print(divider)
print('')
def print_errors(matches, api_url, version, print_color=True):
def colored(text, color):
if print_color:
init_colors()
return color + text + Fore.RESET
else:
return text
tick = colored(u"\u2713", Fore.LIGHTGREEN_EX) + " "
cross = colored(u"\u2717", Fore.LIGHTRED_EX) + " "
for error in matches:
context_object = error["context"]
context = context_object["text"]
length = context_object["length"]
offset = context_object["offset"]
endpostion = offset + length
print(error["message"])
print(
indention[:2] +
cross +
colored(context[:offset], Fore.LIGHTBLACK_EX) +
colored(context[offset:endpostion], Fore.LIGHTRED_EX) +
colored(context[endpostion:], Fore.LIGHTBLACK_EX)
)
print(
indention +
offset * " " +
colored(length * "^", Fore.LIGHTRED_EX)
)
if error["replacements"]:
# only print first 5 replacements
for replacement in error["replacements"][:5]:
print(
indention[:2] +
tick +
colored(context[:offset], Fore.LIGHTBLACK_EX) +
colored(replacement["value"], Fore.LIGHTGREEN_EX) +
colored(context[endpostion:], Fore.LIGHTBLACK_EX)
)
print()
print(colored("Text checked by {url} ({version})".format(url=api_url, version=version), Fore.LIGHTBLACK_EX))
