Python 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']), *args, **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, 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 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, testcases, 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