globalutility.py

#!/usr/bin/env python
# -*- encoding: utf-8 -*-
"""
@Version: 1.0
@Python Version:3.6.6
@Author: ludq1
@Email: ludq1@chinaunicom.cn
@date: 2023/04/07 11:40:00
@Description:
"""

import logging
import re
import uuid
from datetime import datetime
from functools import reduce
from typing import Union, Optional

import urllib3
from urllib3.util import Url

from .globalconst import GlobalConst, DateTimeConst, RegexPatternConst
from .my_baseexception import create_base_exception
from .my_stringutils import MyStringUtils
from .my_utils import MyUtils


class Utility:
    """
    全局工具类
    """

    @classmethod
    def gen_new_domain_from_csm_domain(
            cls,
            csm_domain: str,
            replace_csm_console_to: str,
            csm_console: str = "csm.console",
    ) -> str:
        r"""
        根据 https://csm.console.tg.unicom.local/csm-microsrv 地址,
        将 csm.console 替换成指定内容, 生成类似 https://cke.console.tg.unicom.local 的地址

        :param csm_domain: 示例 https://csm.console.tg.unicom.local/csm-microsrv
        :param replace_csm_console_to: 示例 cke.console
        :param csm_console: 为了扩展该方法, 可以将其他域名进行替换, 该参数默认值为 csm.console
        """
        if not csm_domain:
            return ""

        parse_result: Url = urllib3.util.parse_url(csm_domain)
        scheme = parse_result.scheme
        host: str = parse_result.host
        port = parse_result.port
        host = host.replace(csm_console, replace_csm_console_to)
        if port is not None:
            result = f"{scheme}://{host}:{port}"
        else:
            result = f"{scheme}://{host}"

        return result

    @classmethod
    def dict2jsonstr(cls, a_dict: dict, value_when_obj_is_none: str = 'null') -> str:
        r"""
        将dict转成jsonstr

        Args:
            a_dict: 待转成json字符串的 dict
            value_when_obj_is_none: 当 a_dict 为 None 时返回的字符串,默认为null

        Returns:

        """
        return MyUtils.dict2jsonstr(a_dict, value_when_obj_is_none)

    @classmethod
    def list2jsonstr(cls, a_list: Union[list, tuple], value_when_obj_is_none: str = 'null') -> str:
        r"""
        将list or tuple 转成jsonstr

        Args:
            a_list: 待转成json字符串的 list or tuple
            value_when_obj_is_none: 当 a_list 为 None 时返回的字符串,默认为null

        Returns:

        """
        return MyUtils.list2jsonstr(a_list, value_when_obj_is_none)

    @classmethod
    def jsonstr2dict(cls, a_jsonstr: str, b_raise_error_when_illegal: bool = False) -> dict:
        r"""
        将 jsonstr 转成 dict

        Args:
            a_jsonstr: json格式的字符串
            b_raise_error_when_illegal: 当字符串不合法时抛出异常还是返回None
        Returns:

        """
        return MyUtils.jsonstr2dict(a_jsonstr, b_raise_error_when_illegal)

    @classmethod
    def jsonstr2list(cls, a_jsonarrstr: str, b_raise_error_when_illegal: bool = False) -> list:
        r"""
        将 jsonstr 转成 list

        Args:
            a_jsonarrstr: jsonarr格式的字符串
            b_raise_error_when_illegal: 当字符串不合法时抛出异常还是返回None
        Returns:

        """
        return MyUtils.jsonstr2list(a_jsonarrstr, b_raise_error_when_illegal)

    @classmethod
    def format_result(cls, ret_val: str, ret_code: str) -> str:
        r"""
        将ret_val和ret_code格式化成 ryy项目组使用的标准json 字符串,

        即 { "RetCode": ret_code , "RetVal": ret_val }

        Args:
            ret_val:string 返回结果
            ret_code:string 返回码

        Returns: json 返回json数据

        """
        dict_ret = {
            GlobalConst.RETKEY_RET_VAL: ret_val,
            GlobalConst.RETKEY_RET_CODE: ret_code
        }
        return cls.dict2jsonstr(dict_ret)

    @classmethod
    def format_error(cls, ret_val: str, ret_code: str = GlobalConst.RETCODE_COMMON_ERROR) -> str:
        r"""
        格式化错误信息

        Args:
            ret_val: string 输出结果
            ret_code:string 返回码

        Returns: json 返回错误结果

        """
        return cls.format_result(ret_val, ret_code)

    @classmethod
    def format_success(cls, ret_val: str, ret_code: str = GlobalConst.RETCODE_SUCESS_CODE) -> str:
        r"""
        格式化错误信息

        Args:
            ret_val: string
            ret_code:string 返回码

        Returns: json 返回正确结果

        """

        return cls.format_result(ret_val, ret_code)

    @classmethod
    def do_filter_xss(cls, to_filter_str: str) -> str:
        r"""
        防xss处理,将 < 替换成#《#，将 > 替换成 #》#

        Args:
            to_filter_str: string

        Returns:

        """
        return MyUtils.do_filter_xss(to_filter_str)

    @classmethod
    def now_day(cls, a_date: datetime = None, format_str: str = None) -> str:
        r"""
        以 %Y-%m-%d 格式化日期,默认为当前日期,时间为当前时区的时间字符串

        Args:
            a_date:
            format_str:

        Returns:

        """
        if a_date is None:
            a_date = datetime.now()

        return cls.format_date(a_date, format_str or DateTimeConst.COMMON_DATE_FORMAT)

    @classmethod
    def now_daytime(cls, a_date: datetime = None, format_str: str = None) -> str:
        r"""
        以 %Y-%m-%d %H:%M:%S 格式化日期,默认为当前日期,时间为当前时区的时间字符串

        Args:
            a_date:
            format_str:

        Returns:

        """
        if a_date is None:
            a_date = datetime.now()

        return cls.format_date(a_date, format_str or DateTimeConst.COMMON_DATETIME_FORMAT)

    @classmethod
    def now_daytime_utc(cls, a_date: datetime = None, format_str: str = None) -> str:
        r"""
        以 %Y-%m-%dT%H:%M:%SZ 格式化日期,默认为 datetime.utcnow(),,时间为标准时区的时间字符串

        Args:
            a_date:
            format_str:

        Returns:

        """
        if a_date is None:
            a_date = datetime.utcnow()

        return cls.format_date(a_date, format_str or DateTimeConst.RFC3339_DATETIME_FORMAT)

    @classmethod
    def format_date_to_utc(cls, a_date: datetime, format_str: str = None) -> str:
        r"""
        以 %Y-%m-%d %H:%M:%S 格式化日期,默认为当前日期,格式化后的字符串不会转换时区

        Args:
            a_date:
            format_str:

        Returns:

        """
        if not isinstance(a_date, datetime):
            raise ValueError("a_date 不是日期类型")

        return a_date.strftime(format_str or DateTimeConst.RFC3339_DATETIME_FORMAT)

    @classmethod
    def format_date(cls, a_date: datetime, format_str: str = None) -> str:
        r"""
        以 %Y-%m-%d %H:%M:%S 格式化日期,默认为当前日期,格式化后的字符串不会转换时区

        Args:
            a_date:
            format_str:

        Returns:

        """
        if not isinstance(a_date, datetime):
            raise ValueError("a_date 不是日期类型")

        return a_date.strftime(format_str or DateTimeConst.COMMON_DATETIME_FORMAT)

    @classmethod
    def parse_datestr(cls, a_datestr: str, pattern_str: str = None) -> Optional[datetime]:
        r"""
        解析日期字符串为日期, 解析失败返回None,解析的时间是当前时区的时间,
        如果要转换时区,请使用以下语句,先进行时区替换
        create_time = create_time.replace(tzinfo=pytz.timezone("UTC")).astimezone(pytz.timezone("Asia/Shanghai"))

        Args:
            a_datestr: 自动对 a_datestr 进行trim 操作
            pattern_str:

        Returns:

        """

        # 判断字符串是否是 blank,无需解析的情况
        if isinstance(a_datestr, datetime):
            return a_datestr

        if MyStringUtils.is_empty(a_datestr, trim=True):
            return None

        # 强制转成字符串
        a_datestr = MyStringUtils.to_str(a_datestr, trim=True)

        # 尝试使用提供的 pattern 解析字符串
        if not MyStringUtils.is_empty(pattern_str, trim=True):
            try:
                return datetime.strptime(a_datestr, MyStringUtils.to_str(pattern_str))
            except Exception:
                pass

        for tmp_pattern in DateTimeConst.ALL_DATEFORMAT_PATTERNS:
            try:
                return datetime.strptime(a_datestr, tmp_pattern)
            except Exception:
                pass

        return None

    @classmethod
    def check_param_equal(cls, to_check_param: str, base_value: str, err_msg: str) -> None:
        r"""
        检查 to_check_param 和 base_value 是否相等，不相等则抛出异常

        Args:
            to_check_param:
            base_value:
            err_msg:

        Returns:

        """
        if not MyStringUtils.equals(to_check_param, base_value):
            raise create_base_exception('参数检查错误:{}'.format(err_msg))

    @classmethod
    def check_param_bool(cls, to_check_param: bool, err_msg: str) -> None:
        r"""
        检查 to_check_param 是否为true，否则抛出异常

        Args:
            to_check_param:
            err_msg:

        Returns:

        """
        if bool is None or not isinstance(to_check_param, bool):
            raise create_base_exception('参数检查错误(参数类型不是bool):{}'.format(err_msg))

        if not to_check_param:
            raise create_base_exception('参数检查错误:{}'.format(err_msg))

    @classmethod
    def check_param_none_or_empty(cls, to_check_param: str, err_msg: str, trim: bool = False) -> None:
        r"""
        检查 to_check_param 是否为true，否则抛出异常

        Args:
            to_check_param:
            err_msg:
            trim:
        Returns:

        """
        if MyStringUtils.is_empty(MyStringUtils.to_str(to_check_param), trim=trim):
            raise create_base_exception('参数检查错误:{}'.format(err_msg))

    @classmethod
    def check_param_jsonstr(cls, to_check_param: str, err_msg: str) -> dict:
        r"""
        检查 to_check_param 是否是json格式的字符串,不是则抛出异常 , 是则返回dict

        Args:
            to_check_param:
            err_msg:

        Returns:

        """
        a_dict = cls.jsonstr2dict(to_check_param)
        if a_dict is None:
            raise create_base_exception('参数检查错误:{}'.format(err_msg))
        else:
            return a_dict

    @classmethod
    def do_filter_mysql_param(cls, to_filter_param: str, append_fix: str = "'") -> Optional[str]:
        r"""
        mysql防注入使用,用于过滤(替换)拼接字符串中的参数,当拼接`key`='value'时,对value中的特殊字符进行替换

        Args:
            to_filter_param:
            append_fix:

        Returns:

        """
        if to_filter_param is None:
            return None

        if isinstance(to_filter_param, int):
            to_filter_param = f"{to_filter_param}"
        elif isinstance(to_filter_param, dict):
            to_filter_param = MyUtils.dict2jsonstr(to_filter_param)
        elif isinstance(to_filter_param, list):
            to_filter_param = MyUtils.list2jsonstr(to_filter_param)
        elif hasattr(to_filter_param, "to_dict"):
            to_filter_param = MyUtils.dict2jsonstr(to_filter_param.to_dict())
        elif not isinstance(to_filter_param, str):
            raise create_base_exception(f'精确参数to_filter_param 不是字符串/数字/字典/列表/to_dict对象')
        elif not isinstance(append_fix, str):
            raise create_base_exception('append_fix 不是字符串')

        if not MyStringUtils.is_empty(to_filter_param):
            to_filter_param = to_filter_param.replace("\\", "\\\\").replace("'", "\\'")

        return '{0}{1}{0}'.format(append_fix, to_filter_param)

    @classmethod
    def do_filter_mysql_field(cls, to_filter_field: str, append_fix: str = "`") -> Optional[str]:
        r"""
        sql防注入使用,用于过滤(替换)拼接字符串中的参数,当拼接`key`='value'时,对key中的特殊字符进行替换

        Args:
            to_filter_field:
            append_fix: 默认为 `

        Returns:

        """
        if to_filter_field is None or to_filter_field == '':
            return to_filter_field

        if not isinstance(to_filter_field, str):
            raise create_base_exception('to_filter_field 不是字符串')
        elif not isinstance(append_fix, str):
            raise create_base_exception('append_fix 不是字符串')

        to_filter_field = to_filter_field.replace("\\", "\\\\")
        to_filter_field = to_filter_field.replace(append_fix, f"\\{append_fix}")

        return '{0}{1}{0}'.format(append_fix, to_filter_field)

    @classmethod
    def do_filter_mysql_param_with_single_quotes_for_fuzzy_search(
            cls, to_filter_param: str, append_fix: str = "'"
    ) -> Optional[str]:
        r"""
        mysql防注入使用,用于过滤(替换)拼接字符串中的参数,当拼接`key`='value'时,对value中的特殊字符进行替换

        Args:
            to_filter_param:
            append_fix: 默认为 '

        Returns:

        """
        if to_filter_param is None:
            return to_filter_param

        if not isinstance(to_filter_param, str):
            raise ValueError('模糊参数to_filter_param 不是字符串')
        elif not isinstance(append_fix, str):
            raise ValueError('append_fix 不是字符串')

        if MyStringUtils.is_empty(to_filter_param):
            return '{0}{1}{0}'.format(append_fix, '%')
        else:
            to_filter_param = to_filter_param.replace("\\", "\\\\")
            to_filter_param = to_filter_param.replace("%", "\\%")
            to_filter_param = to_filter_param.replace("_", "\\_")
            to_filter_param = to_filter_param.replace(append_fix, f"\\{append_fix}")

            return '{0}{1}{2}{1}{0}'.format(append_fix, '%', to_filter_param)

    @classmethod
    def is_hasxss(cls, to_check_str: str) -> str:
        r"""
        检查 是否包含 script label strictscript encodestrictscript 标签

        Args:
            to_check_str:

        Returns: 返回EMPTY 表示没有xss,其他内容为错误信息

        """

        to_check_str = MyStringUtils.to_str(to_check_str)

        if MyStringUtils.is_empty(to_check_str):
            return MyStringUtils.EMPTY

        # 定义要匹配的正则表达式
        match_dict = {
            'script': '.*<\\s*script\\s*>.*</\\s*script\\s*>.*',
            'label': '.*<\\s*.*\\s*>.*</\\s*.*\\s*>.*',
            'strictscript': '.*<\\s*.*=.*\\s*>.*',
            'encodestrictscript': '.*%3C\\s*.*%3D.*\\s*%3B.*'
        }

        for tmp_k, tmp_v in match_dict.items():
            tmp_pattern = re.compile(tmp_v, re.S | re.I)
            if re.search(tmp_pattern, to_check_str):
                return '存在违规数据:{}'.format(tmp_k)

        return MyStringUtils.EMPTY

    @classmethod
    def list_join_to_str(cls, to_join_list: Union[list, tuple], separator_str: str = ',', wrapper_str: str = None,
                         ignore_none: bool = True) -> str:
        r"""
        连接 list，使用 separator_str ,当值为 None 时可以忽略，也可以当做 null字符串 添加，但是不追加 wrapper_str

        Args:
            to_join_list:
            separator_str: 默认使用逗号
            wrapper_str:
            ignore_none:
        Returns:

        """
        if to_join_list is None:
            return MyStringUtils.EMPTY

        if not isinstance(to_join_list, (list, tuple)):
            raise ValueError("要连接的对象不是list或tuple")

        if not to_join_list:
            return MyStringUtils.EMPTY

        if wrapper_str is None:
            wrapper_str = MyStringUtils.EMPTY

        if separator_str is None:
            separator_str = ","

        separator_str = MyStringUtils.to_str(separator_str)
        wrapper_str = MyStringUtils.to_str(wrapper_str)

        def map_wrap_str(a_str):
            if a_str is None:
                return a_str
            else:
                return '{0}{1}{0}'.format(wrapper_str, a_str)

        tmp_r = map(map_wrap_str, to_join_list)

        def reduce_join_str(a_str, b_str) -> Optional[str]:
            if ignore_none and (a_str is None or b_str is None):
                return a_str or b_str
            else:
                if a_str is None:
                    a_str = 'null'

                if b_str is None:
                    b_str = 'null'

                return '{0}{2}{1}'.format(a_str, b_str, separator_str)

        return reduce(reduce_join_str, tmp_r)

    @classmethod
    def join_str(cls, *to_join_list: str, separator_str: str = '', wrapper_str: str = None,
                 ignore_none: bool = True) -> str:
        r"""
        连接 list，使用 separator_str ,当值为 None 时可以忽略，也可以当做 null字符串 添加，但是不追加 wrapper_str

        Args:
            to_join_list:
            separator_str: 默认使用空字符串
            wrapper_str:
            ignore_none:默认为True
        Returns:

        """
        return cls.list_join_to_str(to_join_list, separator_str=separator_str, wrapper_str=wrapper_str,
                                    ignore_none=ignore_none)

    @classmethod
    def join_str_with_none(cls, *to_join_list: str, separator_str: str = '', wrapper_str: str = None,
                           ignore_none: bool = False) -> str:
        r"""
        连接 list，使用 separator_str ,当值为 None 时可以忽略，也可以当做 null字符串 添加，但是不追加 wrapper_str

        Args:
            to_join_list:
            separator_str: 默认使用空字符串
            wrapper_str:
            ignore_none: 默认为False
        Returns:

        """
        return cls.list_join_to_str(to_join_list, separator_str=separator_str, wrapper_str=wrapper_str,
                                    ignore_none=ignore_none)

    @classmethod
    def create_uuid(cls, with_dash: bool = False) -> str:
        r"""
        生成随机字符串，例如"bc520134-cffc-46f4-8e47-7313b3b2485b" ,
        默认没有 - 字符

        Args:
            with_dash:

        Returns:

        """
        if with_dash:
            return MyStringUtils.to_str(uuid.uuid1())
        else:
            return MyStringUtils.to_str(uuid.uuid1()).replace('-', MyStringUtils.EMPTY)

    @classmethod
    def check_url_illegal(cls, url_str: str) -> bool:
        r"""
        判断URL是否合法 判断依据是网址以http://开头，不能以/结尾,至少有一个小数点 <br/>

        返回 true=成功 false=失败<br>

        判断前会进行trim

        Args:
            url_str:

        Returns:

        """
        url_str = MyStringUtils.to_str(url_str, trim=True)

        if MyStringUtils.is_empty(url_str):
            return False

        tmp_pattern = re.compile(RegexPatternConst.URL_PATTERN)
        if re.match(tmp_pattern, url_str):
            return True
        else:
            return False

    @classmethod
    def filter_utf8mb4(cls, to_filter_str: str) -> Optional[str]:
        r"""
        过滤字符串中的 4字节字符

        Args:
            to_filter_str:
        Returns:

        """
        if to_filter_str is None:
            return to_filter_str

        if MyStringUtils.is_empty(to_filter_str):
            return MyStringUtils.EMPTY

        def reduce_join_str(a_str: str, b_str: str) -> str:
            a_bytestr = a_str.encode()
            b_bytestr = b_str.encode()
            if len(a_str) == 1 and len(a_bytestr) >= 4:
                a_str = None

            if len(b_bytestr) >= 4:
                b_str = None

            return Utility.join_str(a_str, b_str, separator_str=MyStringUtils.EMPTY)

        return reduce(reduce_join_str, to_filter_str)

    @classmethod
    def find_utf8mb4(cls, a_str: str) -> str:
        r"""
        查找 字符串中第一个 4字节字符，如果返回空字符串，则说明字符串中不包含4字节字符

        Args:
            a_str:
        Returns:

        """

        if MyStringUtils.is_empty(a_str, trim=True):
            return MyStringUtils.EMPTY

        for tmp_char in MyStringUtils.to_str(a_str, trim=True):
            tmp_charbytestr = tmp_char.encode()
            if len(tmp_charbytestr) >= 4:
                return tmp_char

        return MyStringUtils.EMPTY

    @classmethod
    def do_check_and_put_param(cls, arg_dict: dict, to_check_param_dict: dict, param_type: int,
                               date_format_str: str = None) -> dict:
        r"""
        检查 to_check_param_dict 中的参数并put到 arg_dict 中,

        0=int型参数 1=number型参数 2=date型参数

        如果参数为None或空字符串则不进行检查和put,

        否则 检查其值是否合法,合法则put到map中, 不合法则抛出异常

        最后返回 arg_dict

        Args:
            arg_dict:
            to_check_param_dict: key为参数名称 value为参数值
            param_type: 0=int型参数 1=number型参数 2=date型参数
            date_format_str: 参数类型为日期时使用的格式化字符串,默认为使用默认的格式化字符串,目前是  %Y-%m-%d %H:%M:%S

        Returns:

        """

        if not isinstance(arg_dict, dict):
            raise ValueError('arg_dict 不是字典')

        if to_check_param_dict is None:
            return arg_dict

        if not isinstance(to_check_param_dict, dict):
            raise ValueError('to_check_param_dict 不是字典')

        for tmp_key, tmp_value in to_check_param_dict.items():
            param_name = MyStringUtils.to_str(tmp_key)
            param_value = MyStringUtils.to_str(tmp_value)

            if MyStringUtils.is_empty(param_value):
                continue

            tmp_error = create_base_exception(cls.join_str_with_none('参数', param_name, '不合法'),
                                              submitted_webarg=cls.join_str_with_none('提供的', param_name, '的值为',
                                                                                      param_value))
            if param_type == 0:
                # 检查 int型参数并put
                # 整型就是整型,防止是浮点或布尔
                if isinstance(param_value, (float, bool)):
                    raise tmp_error

                encountered_e = None
                try:
                    test_val = int(param_value)
                    arg_dict[param_name] = str(test_val)
                except Exception as e:
                    encountered_e = e
                if encountered_e:
                    raise tmp_error.with_traceback(encountered_e.__traceback__)
            elif param_type == 1:
                # 检查 number型参数并put
                encountered_e = None
                try:
                    test_val = float(param_value)
                    arg_dict[param_name] = str(test_val)
                except Exception as e:
                    encountered_e = e
                if encountered_e:
                    raise tmp_error.with_traceback(encountered_e.__traceback__)
            elif param_type == 2:
                # 检查 date型参数并put
                encountered_e = None
                try:
                    test_val = cls.parse_datestr(param_value)
                    arg_dict[param_name] = cls.format_date(test_val, format_str=date_format_str)
                except Exception as e:
                    encountered_e = e
                if encountered_e:
                    raise tmp_error.with_traceback(encountered_e.__traceback__)
            else:
                raise ValueError(f"param_type的值不支持,提供的值为{param_type}")

        return arg_dict

    @classmethod
    def do_check_and_put_param_for_not_none_or_empty(cls, arg_dict: dict, to_check_param_dict: dict,
                                                     ignore_none: bool = True, ignore_empty: bool = True) -> dict:
        r"""
        检查 to_check_param_dict 中的参数并put到 arg_dict 中,

        如果参数为None时 ignore_none 为True 则不进行put,否则put null值, 默认不put

        如果参数为EMPTY时 ignore_empty为True 则不进行put,否则put EMPTY值, 默认不put

        否则 put到 arg_dict 中

        最后返回 arg_dict

        Args:
            arg_dict:
            to_check_param_dict: key为参数名称 value为参数值
            ignore_none:
            ignore_empty:

        Returns:

        """

        if not isinstance(arg_dict, dict):
            raise ValueError('arg_dict 不是字典')

        if to_check_param_dict is None:
            return arg_dict

        if not isinstance(to_check_param_dict, dict):
            raise ValueError('to_check_param_dict 不是字典')

        formatted_dict = MyUtils.format_dict_to_str_dict(to_check_param_dict, ignore_none=ignore_none,
                                                         ignore_empty=ignore_empty)

        arg_dict.update(formatted_dict)
        return arg_dict

    @classmethod
    def get_env(cls, env_name: str) -> str:
        r"""
        获取系统环境变量的内容

        Args:
            env_name:
        Returns:

        """
        return MyUtils.get_env(env_name)

    @classmethod
    def get_common_logger(cls, logger: logging.Logger = None) -> logging.Logger:
        r"""
        获取一个通用的logger或者明确的返回app.logger

        Args:
            logger:

        Returns:

        """
        if logger is None:
            # 初始化一个通用的logger
            logger = logging.getLogger('common_logout.logger')
            logger.setLevel(logging.INFO)
            logger.propagate = True
            handler = logging.StreamHandler()
            handler.setLevel(logging.INFO)
            formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
            handler.setFormatter(formatter)
            logger.addHandler(handler)

        return logger