1#@!#!123s

: / proc / self / root / opt / cloudlinux / venv / lib64 / python3.11 / site-packages / cllimits / lib /
Filename : utils.py
back
# 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

import json
import os
import sys
import time

from clcommon.utils import run_command, ExternalProgramFailed

VALUES_STR = 'Available values for option'


def replace_params(data):
    """
    Replacing params in data for show error message
    :param data: error's data for show message
    :return:
    """
    if data.startswith("--"):
        param, text = data.split(" ", 1)
        return {"result": "%(param)s " + text, "context": {"param": param}}

    if data.startswith(VALUES_STR):
        text, param = data.split(":", 1)
        return {"result": text + ": %(available_options)s",
                "context": {"available_options": param.strip()}}

    return {"result": data}


def _is_string_number(s_val):
    """
    Checks is string contains a number (integer or float)
    :param s_val: String to check
    :return: True - string is number, False - not number
    """
    try:
        float(s_val)
        return True
    except ValueError:
        return False


def convert_mem_value_to_bytes(value):
    """
    Convert value in Gbytes,Mbytes to bytes

    :param value: value of mem limit
    :return: value in bytes
    """
    value = str(value).lower()
    if value.endswith('k'):
        power = 1
    elif value.endswith('m'):
        power = 2
    elif value.endswith('g'):
        power = 3
    elif _is_string_number(value):
        power = 1
        value = f'{value}k'
    else:
        raise ValueError('Wrong memory value')
    return int(1024 ** power * float(value[:-1]))


def _convert_memory_value_to_adaptive_format(value, convert=True):
    """
    Convert memory value to adaptive value in GB, TB, etc

    :param value: memory value in MB or KB
    :param convert: if True - convert value, False - not convert
    :return: adaptive value in GB, TB, etc

    """

    if not convert:
        return value

    value = str(value).lower()

    units = ['K', 'M', 'G', 'T', 'P']
    if value.endswith('m'):
        del units[0]
    value = str(value).lower().replace('m', '').replace('k', '')  # remove unit symbol

    if value.startswith('*'):
        result = '*'
        value = value.replace('*', '')
    else:
        result = ''
    value = float(value)
    for unit in units:
        if value // 1024 > 0:
            value /= 1024
        elif value == 0:
            result = f'{result}0K'
            break
        else:
            result = f'{result}{value:.2f}{unit}'
            break

    return result


def print_dictionary(data_dict, is_json=False):
    """
    Print specified dictionary
    :param data_dict: data dictionary to print
    :param is_json: True - print in JSON, False - in text
    :return: None
    """
    # Print as JSON
    # print json.dumps(data_dict, indent=2, sort_keys=True)
    if is_json:
        # Print as JSON
        print(json.dumps(data_dict, sort_keys=True))
    else:
        # Print as text
        print(data_dict)


def print_error_and_exit(message, is_json=False):
    """
    Prints to stdout
    :param: is_json - True if print error in json format, False - text
    """
    data = {"timestamp": time.time(), "result": str(message)}
    print_dictionary(data, is_json)
    sys.exit(1)


def is_quota_supported(cl_quota_path, repquota_path):
    """
    Detect quota is supported
    :return: True/False - quotas supported/not supported
    """
    # Check if all needed utilities present
    if not os.path.isfile(cl_quota_path) or not os.path.isfile(repquota_path):
        return False
    return True


def is_quota_active(cl_quota_path, repquota_path):
    """
    Detect quota is activated
    :return: True/False - quotas activated/not activated
    """
    # If quotas not supported they are not activated
    if not is_quota_supported(cl_quota_path, repquota_path):
        return False
    # Check is quota activated
    cmd = [repquota_path, '-nva']
    try:
        stdout = run_command(cmd)
    except ExternalProgramFailed:
        return False
    # quotas not supported if repqouta returns nothing
    if not stdout:
        return False
    return True