/
opt
/
cloudlinux
/
venv
/
lib
/
python3.11
/
site-packages
/
lvestats
/
lib
/
commons
/
Upload Filee
HOME
# coding=utf-8 # # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2019 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT """ Extensions for validation and output formatting argparse """ import argparse import pwd import datetime from prettytable import PrettyTable from lvestats.lib.commons.dateutil import parse_date, parse_period2 class SmartFormatter(argparse.HelpFormatter): """ Format argparse help in many line, to format many lines use "R|" at start line >>>import argparse >>>parser = argparse.ArgumentParser(formatter_class=SmartFormatter) >>>parser.add_argument('-a', '--argument', help="R| Descriptions: \\n first new line\\n second new line") """ def _split_lines(self, text, width): if text.startswith('R|'): return text[2:].splitlines() return argparse.HelpFormatter._split_lines(self, text, width) # pylint: disable=protected-access def _format_args(self, action, default_metavar): if isinstance(action, ParseDatetime): get_metavar = self._metavar_formatter(action, default_metavar) return str(get_metavar(1)) return argparse.HelpFormatter._format_args(self, action, default_metavar) # pylint: disable=protected-access class ParseDatetime(argparse.Action): """ Interprets a string ([YY]YY-MM-DD[ HH:MM]) as a datetime for argument parsing. return parsed data as <type 'datetime.datetime'> """ def __call__(self, parser, namespace, values, option_string=None): datetime_str = ' '.join(values) if hasattr(values, '__iter__') else values try: result_datetime = parse_date(datetime_str) except ValueError as e: raise argparse.ArgumentError( self, f"{datetime_str} is an invalid; use [YY]YY-MM-DD[ HH:MM] format", ) from e setattr(namespace, self.dest, result_datetime) def check_percentages(value): """ Check if input argument between 0 and 100 :param str value: :return int: percent :except argparse.ArgumentTypeError: generate exception for command line warning message """ value = int(value) if value < 0 or value > 100: raise argparse.ArgumentTypeError(f"{value} is an invalid; this value can range from 0 to 100") return value def check_non_negative_int(value): """ Check if input argument greater than (or equal to) zero and not float :param str value: :return int: :except argparse.ArgumentTypeError: generate exception for command line warning message """ try: non_negative_int = int(value) except ValueError as e: raise argparse.ArgumentTypeError( f'"{value}" is an invalid; must be a number greater than (or equal to) zero' ) from e if '.' in value or non_negative_int < 0: raise argparse.ArgumentTypeError( f'"{value}" is an invalid; must be a number greater than (or equal to) zero' ) return non_negative_int def check_positive_int(value): """ Check if input argument is digest, greater than zero and not float :param str value: line to check :return int: positive integer :except argparse.ArgumentTypeError: generate exception for command line warning message """ try: positive_int = int(value) except ValueError as e: raise argparse.ArgumentTypeError( f'"{value}" is an invalid; must be a number greater than zero' ) from e if '.' in value or positive_int <= 0: raise argparse.ArgumentTypeError( f'"{value}" is an invalid; must be a number greater than zero' ) return positive_int def check_timestamp(value): """ Check if input argument is a correct timestamp 253370754000.0 == time.mktime(datetime.datetime(year=datetime.MAXYEAR, month=1, day=1).timetuple()) :param str value: line to check :return float: timestamp :except argparse.ArgumentTypeError: generate exception for command line warning message """ try: positive_float = float(value) except ValueError as e: raise argparse.ArgumentTypeError( f'"{value}" is an invalid; timestamp must be a number between 0 and 253370754000' ) from e if not 0 < positive_float < 253370754000: raise argparse.ArgumentTypeError( f'"{value}" is an invalid; timestamp must be a number between 0 and 253370754000' ) return positive_float def check_from_zero_to_max_int(value): """ Check if input argument less then maximum integer number PostgreSQL - 2147483647 MySQL - 2147483647 SQLite - 9223372036854775807 :param str value: :return int: :except argparse.ArgumentTypeError: generate exception for command line warning message """ max_int = 2147483647 try: non_negative_int = int(value) except ValueError as e: raise argparse.ArgumentTypeError( f'"{value}" is an invalid; must be a number greater than (or equal to) zero ' f'and less than {max_int}' ) from e if '.' in value or non_negative_int < 0 or non_negative_int > max_int: raise argparse.ArgumentTypeError( f'"{value}" is an invalid; must be a number greater than (or equal to) zero ' f'and less than {max_int}' ) return non_negative_int def check_user_name(value): """ Check if user present in system and convert user name to user id :param str value: user name (login) :return int: user id :except argparse.ArgumentTypeError: generate exception for command line warning message """ try: user_id = pwd.getpwnam(value).pw_uid except KeyError as e: raise argparse.ArgumentTypeError(f'user "{value}" not present in system') from e return user_id def period_type(value): """ Parse multiple period line, and validate input data :param str value: line to parsing :return int: period in seconds :except argparse.ArgumentTypeError: generate exception for command line warning message """ try: int_converted = int(value) if '.' in value or int_converted <= 0: raise argparse.ArgumentTypeError( f'"{value}" is an invalid; time period should be greater than (or equal to) ' '1 minute (60 seconds); not negative' ) value = str(int_converted) + 'm' # if only digest convert to minutes except ValueError: pass try: seconds = parse_period2(value) except ValueError as e: raise argparse.ArgumentTypeError(str(e)) from e if seconds < 60: raise argparse.ArgumentTypeError( f'"{value}" is an invalid; time period should be greater than (or equal to) ' '1 minute (60 seconds)' ) return seconds def period_type2(value, datetime_now=None, round_to_minutes=True): """ Extented period_type function Added support 'yesterday' and 'today' :param bool round_to_minutes: :param str value: line to parsing :return tuple: period from to (<type 'datetime.datetime'>, <type 'datetime.datetime'>) :except argparse.ArgumentTypeError: generate exception for command line warning message """ datetime_now = datetime_now or datetime.datetime.utcnow() value = value.lower() if value == 'yesterday': datetime_to = datetime_now.replace(hour=0, minute=0, second=0, microsecond=0) # today midnight datetime_from = datetime_to - datetime.timedelta(days=1) # yesterday midnight elif value == 'today': datetime_from = datetime_now.replace(hour=0, minute=0, second=0, microsecond=0) # today midnight datetime_to = datetime_now else: if round_to_minutes: datetime_from = datetime_now.replace(second=0, microsecond=0) # rounded to minutes datetime_to = datetime_now.replace(second=0, microsecond=0) # rounded to minutes else: datetime_from = datetime_now datetime_to = datetime_now datetime_from -= datetime.timedelta(seconds=period_type(value)) return datetime_from, datetime_to def format_aliases(description): """ Prepare aliases descriptions to table formatted :param tuple|list|iter description: massive of data :return str: ascii table """ fields = description[0] description = description[1:] table = PrettyTable(fields, padding_width=0, left_padding=2) for field in fields: table.align[field] = "l" list(map(table.add_row, description)) return table.get_string(border=False) BY_USAGE_PERCENT = 90 def alias_action(alias_dscr): alias_dscr = alias_dscr[1:] # first line is headers for descriptions in print tables aliases_func_dict = _alias_dscr_to_dict(alias_dscr) class AliasAction(argparse.Action): def __call__(self, parser, namespace, values, option_string=None): output_as_list = self.nargs in ('+', '*') if output_as_list: values_ = [] for val_ in values: # for using ',' as separator in options values_.extend(filter(bool, val_.lower().split(','))) # "filter(bool" for correct processing --show-columns=,ID, else: values_ = [values.strip('"').strip("'").lower()] # generate exception for not supported aliases not_supported_values = set(values_).difference(set(aliases_func_dict)) if not_supported_values: supported_values = tuple(aliases_func_dict) unsupported_aliases = '", "'.join(not_supported_values) raise argparse.ArgumentError( self, f'alias(es) "{unsupported_aliases}" not valid; ' f'choice from: {", ".join(supported_values)}' ) result = [aliases_func_dict[_] for _ in values_] # replace aliases to main aliases if not output_as_list: result = result[0] setattr(namespace, self.dest, result) # write result to namespace return AliasAction def _alias_dscr_to_dict(alias_dscr): """ preparation of the dictionary to extract the names of the aggregation function name (used in show_history) as aliases are used from the first to the penultimate elements alias_dscr as aggregation function name are used penultimate element >>> _alias_dscr_to_dict((('al_1', 'al_2', 'al_m1', 'description1'), ('al_3', 'al_4', 'al_m2', 'description1'))) {'al_2': 'al_m1', 'al_1': 'al_m1', 'al_m1': 'al_m1', 'al_3': 'al_m2', 'al_m2': 'al_m2', 'al_4': 'al_m2'} :param tuple|list alias_dscr: :type return: dict """ val_index_ = -2 aliases_func_pairs = [(_[val_index_].lower(), _[val_index_])for _ in alias_dscr] for key_index_ in range(len(alias_dscr[0]) - 1): aliases_func_pairs.extend([(_[key_index_], _[val_index_]) for _ in alias_dscr]) aliases_func_dict = dict(aliases_func_pairs) return aliases_func_dict def argparse_process_period(namespace): # prepare period period_from, period_to = getattr(namespace, 'from'), namespace.to if namespace.period: period_from, period_to = namespace.period setattr(namespace, 'from', period_from) namespace.to = period_to return namespace