Python colorama.Fore 模块，LIGHTRED_EX 实例源码
我们从Python开源项目中，提取了以下11个代码示例，用于说明如何使用colorama.Fore.LIGHTRED_EX。
def source_path(dir_name):
for root in os.walk("/home"):
if dir_name == root[0].split('/')[-1]:
all_paths.append(root[0])
for i in range(len(all_paths)):
print()
print("{}. {}".format(i+1,all_paths[i]))
if len(all_paths) == 0:
print(Fore.LIGHTRED_EX + 'No directory found')
exit()
choice = int(input('\nEnter the option number: '))
if choice < 1 or choice > len(all_paths):
print(Fore.LIGHTRED_EX +'Wrong choice entered')
exit()
else:
path = all_paths[choice-1]
return path
def filter(self, record):
record.spent = record.relativeCreated // 1000
if iswindows:
record.fg = ''
elif record.levelname == 'DEBG':
record.fg = Fore.LIGHTBLACK_EX
elif record.levelname == 'INFO':
record.fg = Fore.LIGHTWHITE_EX
elif record.levelname == 'WARN':
record.fg = Fore.LIGHTYELLOW_EX
elif record.levelname == 'ERR ':
record.fg = Fore.LIGHTRED_EX
elif record.levelname == 'CRIT':
record.fg = Fore.RED
else:
record.fg = Fore.LIGHTWHITE_EX
return True
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']), **kwargs)
def source_path(dir_name):
all_paths = []
for root in os.walk("/home"):
if dir_name == root[0].split('/')[-1]:
all_paths.append(root[0])
for i, path_info in enumerate(all_paths):
print()
print("{}. {}".format(i + 1, path_info))
if len(all_paths) == 0:
print(Fore.LIGHTRED_EX + 'No directory found')
exit()
if six.PY2:
choice = int(raw_input('\nEnter the option number: '))
else:
choice = int(input('\nEnter the option number: '))
if choice < 1 or choice > len(all_paths):
path = ''
print(Fore.LIGHTRED_EX + 'Wrong choice entered')
exit()
else:
path = all_paths[choice - 1]
return path
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, new_text)
elif print_type == 'alert':
return '{0}{1}!{2}{3} {4}'.format(lb, Fore.LIGHTRED_EX, new_text)
elif print_type == 'sleep':
return '{0}-{1} {2}'.format(lb, 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","label":"Last day"},{"from":"Now-7d","label":"Last week"},{"from":"Now-30d","label":"Last 30 days"},{"from":"Now-90d","label":"Last 90 days"}]},{"name":"status","type":"terms","field":"status"},{"name":"severity","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":"@R_51_4045@ional","id":0}]},{"name":"scenario","size":0,"field":"behaviors.scenario"},{"name":"assigned_to_uid","min_doc_count":1,"field":"assigned_to_uid","missing":"Unassigned"},{"name":"host","field":"device.hostname.raw","missing":"UnkNown"},{"name":"triggering_file","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 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))
def publish_results(api, testcases, run_id=0, plan_id=0, version='', publish_blocked=True):
# pylint: disable=too-many-arguments,too-many-branches
""" Update testcases with provided Test Run or Test Plan
:param api: Client to TestRail API
:param testcases: List of testcases with status,returned by `get_testcases`
:param run_id: TestRail ID of Test Run to update
:param plan_id: TestRail ID of Test Plan to update
:param version: Version to indicate in Test Case result
:param publish_blocked: If False,results of "blocked" Test cases in TestRail are not published
:return: True if publishing was done. False in case of error.
"""
if run_id:
if api.is_testrun_available(run_id):
count = 0
if publish_blocked is False:
logging.info('Option "Don\'t publish blocked testcases" activated')
blocked_tests_list = [
test.get('case_id') for test in api.get_tests(run_id) if test.get('status_id') == 2
]
logging.info('Blocked testcases excluded: %s', ','.join(str(elt) for elt in blocked_tests_list))
testcases = [
testcase for testcase in testcases
if api.extract_testcase_id(testcase.get('id')) not in blocked_tests_list
]
for testcase in testcases:
if version:
testcase['version'] = version
try:
api.add_result(run_id, testcase)
count += 1
pretty_print_testcase(testcase)
print()
except testrail.APIError as error:
if 'No (active) test found for the run/case combination' not in str(error):
pretty_print_testcase(testcase, str(error))
print()
time.sleep(0.5)
logging.info('%d result(s) published in Test Run #%d.', count, run_id)
else:
logging.error('Test Run #%d is is not available', run_id)
return False
elif plan_id:
if api.is_testplan_available(plan_id):
for _run_id in api.get_available_testruns(plan_id):
publish_results(api, run_id=_run_id, version=version, publish_blocked=publish_blocked)
else:
logging.error('Test Plan #%d is is not available', plan_id)
return False
else:
logging.error("You have to indicate a Test Run or a Test Plan ID")
print(Fore.LIGHTRED_EX + 'ERROR')
return False
return True
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 [email protected] 举报，一经查实，本站将立刻删除。