utils_base.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 base64
import hashlib
import json
import logging
import os
import pickle
import platform
import random
import re
import string
import subprocess
import tempfile
import time
import traceback
import urllib.parse
from datetime import datetime
from typing import Union, Optional, Dict, List

import yaml
from flask import Request
from jinja2 import Environment, FileSystemLoader, StrictUndefined
from jinja2.defaults import VARIABLE_START_STRING, VARIABLE_END_STRING

from .app_exception import AppRuntimeException, AppException
from .app_response import AppResponse
from .base_const import ConstResponseCode
from .common_app_config import CommonAppConfig
from .logined_user import LoginedUserInfo
from .send_notices_utils import SendNoticeUtils
from ..globalutility import Utility
from ..my_stringutils import MyStringUtils

try:
    import thread
except ImportError:
    import _thread as thread


class CrnParseResult:
    product_code: Optional[str] = None
    region_code: Optional[str] = None
    acct_id: Optional[str] = None
    first_id_type: Optional[str] = None
    first_id: Optional[str] = None
    second_id_type: Optional[str] = None
    second_id: Optional[str] = None
    third_id_type: Optional[str] = None
    third_id: Optional[str] = None
    other_id_types: Optional[List[str]] = None
    other_ids: Optional[List[str]] = None
    first_crn: Optional[str] = None
    second_crn: Optional[str] = None
    third_crn: Optional[str] = None
    other_crns: Optional[List[str]] = None

    def __init__(self, **kwargs):
        self.product_code = kwargs.get("product_code")
        self.region_code = kwargs.get("region_code")
        self.acct_id = kwargs.get("acct_id")
        self.first_id_type = kwargs.get("first_id_type")
        self.first_id = kwargs.get("first_id")
        self.second_id_type = kwargs.get("second_id_type")
        self.second_id = kwargs.get("second_id")
        self.third_id_type = kwargs.get("third_id_type")
        self.third_id = kwargs.get("third_id")
        self.other_id_types = kwargs.get("other_id_types")
        self.other_ids = kwargs.get("other_ids")
        self.first_crn = kwargs.get("first_crn")
        self.second_crn = kwargs.get("second_crn")
        self.third_crn = kwargs.get("third_crn")
        self.other_crns = kwargs.get("other_crns")

    def __str__(self):
        return Utility.dict2jsonstr(self.__dict__)


class UtilityBaseV2(Utility):
    r"""
    便捷功能类,不依赖任何类的应用的便捷功能类
    """

    def __init__(self, logger: logging.Logger = None):
        self.logger = logger or CommonAppConfig().common_logger

    @classmethod
    def gen_logger(
            cls,
            logger_name: str = "not_named_logger",
            log_level: int = None,
    ):
        r"""
        生成logger
        :param logger_name: 通常可以用 self.__class__.__name__ 来提供
        :param log_level: 默认为 logging.WARNING , 默认为 CommonAppConfig().log_level
        """
        logger = logging.getLogger(logger_name)
        if log_level is None:
            log_level = CommonAppConfig().log_level
        logger.setLevel(log_level)
        if not logger.handlers:
            ch = logging.StreamHandler()
            ch.setLevel(logging.DEBUG)
            formatter = logging.Formatter(
                '%(asctime)s %(name)s - %(pathname)s - func:%(funcName)s - lineno:%(lineno)s - %(levelname)s - %(message)s',
                datefmt='%Y-%m-%d %H:%M:%S'
            )
            ch.setFormatter(formatter)
            logger.addHandler(ch)
        return logger

    @classmethod
    def dict2jsonstr(cls, a_dict: dict, value_when_obj_is_none: str = 'null', sort_keys: bool = False,
                     remove_key_of_none_value: bool = False, attr_name_when_value_is_obj: str = '__dict__') -> str:
        r"""
        复用父类的 dict2jsonstr 的基础上，增加按排序key的方式生成json字符串，并且生成的字符串为最紧密的字符串

        :param a_dict:
        :param value_when_obj_is_none:
        :param sort_keys
        :param remove_key_of_none_value
        :param attr_name_when_value_is_obj
        :return:
        """
        if a_dict is None:
            return value_when_obj_is_none
        elif not isinstance(a_dict, dict):
            raise ValueError('参数类型不是字典')

        if remove_key_of_none_value:
            # 遍历dict的所有key，并递归删除value为None的key
            a_dict = cls.dict_after_remove_key_of_none_value(a_dict, attr_name_when_value_is_obj)

        sort_keys = True if sort_keys else False

        return json.dumps(a_dict, ensure_ascii=False,
                          default=lambda o: cls.gen_dict_for_a_obj(o, attr_name_when_value_is_obj),
                          sort_keys=sort_keys, separators=(',', ':'))

    @classmethod
    def gen_dict_for_a_obj(cls, a_obj, attr_name):
        r"""
        主要用于json序列化时，value为对象时如何序列化，a_obj指定的属性或方法必须返回一个dict

        :param a_obj:
        :param attr_name:
        :return:
        """
        if not attr_name:
            return getattr(a_obj, "__dict__", None)

        if isinstance(attr_name, str):
            if hasattr(a_obj, attr_name):
                attr_obj = getattr(a_obj, attr_name, None)
                if '__call__' in dir(attr_obj):
                    return attr_obj()
                else:
                    return attr_obj
            else:
                return getattr(a_obj, "__dict__", None)
        else:
            if '__call__' in dir(attr_name):
                return attr_name(a_obj)
            else:
                return getattr(a_obj, "__dict__", None)

    @classmethod
    def dict_after_remove_key_of_none_value(
            cls,
            a_dict: dict, attr_name_when_value_is_obj: str = '__dict__') -> Optional[dict]:
        r"""
        遍历dict的所有key，并递归删除value为None的key

        :param a_dict:
        :param attr_name_when_value_is_obj
        :return:
        """
        if a_dict is None:
            return None
        elif not isinstance(a_dict, dict):
            raise ValueError('参数类型不是字典')

        result = dict()
        for tmp_key, tmp_value in a_dict.items():
            if tmp_value is None:
                continue
            elif isinstance(tmp_value, dict):
                result[tmp_key] = cls.dict_after_remove_key_of_none_value(tmp_value)
            elif isinstance(tmp_value, list):
                result[tmp_key] = cls.list_after_remove_key_of_none_value(tmp_value)
            elif isinstance(tmp_value, str) or isinstance(tmp_value, int) or isinstance(tmp_value, float) or isinstance(
                    tmp_value, bool):
                result[tmp_key] = tmp_value
            else:
                tmp_value = cls.dict_after_remove_key_of_none_value(
                    cls.gen_dict_for_a_obj(tmp_value, attr_name_when_value_is_obj))
                if tmp_value is not None:
                    result[tmp_key] = tmp_value

        return result

    @classmethod
    def list_after_remove_key_of_none_value(cls, a_list: Union[list, tuple],
                                            attr_name_when_value_is_obj: str = '__dict__') -> Optional[list]:
        r"""
        遍历list中每个dict的所有key，并递归删除value为None的key，list中的

        :param a_list:
        :param attr_name_when_value_is_obj
        :return:
        """
        if a_list is None:
            return None
        elif not isinstance(a_list, (list, tuple)):
            raise ValueError('参数类型不是list或元组')

        result = list()
        for tmp_value in a_list:
            if tmp_value is None \
                    or isinstance(tmp_value, str) \
                    or isinstance(tmp_value, int) \
                    or isinstance(tmp_value, float) \
                    or isinstance(tmp_value, bool):
                result.append(tmp_value)
            elif isinstance(tmp_value, dict):
                result.append(cls.dict_after_remove_key_of_none_value(tmp_value))
            elif isinstance(tmp_value, list):
                result.append(cls.list_after_remove_key_of_none_value(tmp_value))
            else:
                tmp_value = cls.dict_after_remove_key_of_none_value(
                    cls.gen_dict_for_a_obj(tmp_value, attr_name_when_value_is_obj))
                result.append(tmp_value)

        return result

    @classmethod
    def list2jsonstr(cls, a_list: Union[list, tuple], value_when_obj_is_none: str = 'null',
                     sort_keys: bool = False, remove_key_of_none_value: bool = False,
                     attr_name_when_value_is_obj: str = '__dict__') -> str:
        r"""

        :param a_list:
        :param value_when_obj_is_none:
        :param sort_keys:
        :param remove_key_of_none_value
        :param attr_name_when_value_is_obj
        :return:
        """

        if a_list is None:
            return value_when_obj_is_none
        elif not isinstance(a_list, (list, tuple)):
            raise ValueError('参数类型不是list或元组')

        if remove_key_of_none_value:
            # 遍历dict的所有key，并递归删除value为None的key
            a_list = cls.list_after_remove_key_of_none_value(a_list, attr_name_when_value_is_obj)

        sort_keys = True if sort_keys else False
        return json.dumps(a_list, ensure_ascii=False,
                          default=lambda o: cls.gen_dict_for_a_obj(o, attr_name_when_value_is_obj),
                          sort_keys=sort_keys,
                          separators=(',', ':'))

    def generate_id_16(self) -> str:
        r"""
        生成一个 16 位的唯一数字编码

        :return:
        """
        #  产生10000 - 99999 的随机数
        r1 = 10000 + random.choice(range(90000))
        tmp_time = str(time.time()).replace(".", "")
        length_of_tmp_time = len(tmp_time)
        while length_of_tmp_time < 13:
            tmp_time = str(time.time()).replace(".", "")
            length_of_tmp_time = len(tmp_time)

        return self.join_str(r1, tmp_time[length_of_tmp_time - 13:length_of_tmp_time - 2])

    def generate_id_12(self) -> str:
        r"""
        生成一个 12 位的唯一数字编码

        :return:
        """
        #  产生10000 - 99999 的随机数
        r1 = 10000 + random.choice(range(90000))
        tmp_time = str(time.time()).replace(".", "")
        length_of_tmp_time = len(tmp_time)
        while length_of_tmp_time < 9:
            tmp_time = str(time.time()).replace(".", "")
            length_of_tmp_time = len(tmp_time)

        return self.join_str(r1, tmp_time[length_of_tmp_time - 9:length_of_tmp_time - 2])

    def get_cookies_dict_from_request(self, request, decoding: bool = True) -> dict:
        r"""
        使用比较原始的方法将request.headers.get('Cookie')中的内容放到一个dict[string,list[string]]结构中

        :param request:
        :param decoding:
        :return:
        """
        cookies_str: str = request.headers.get('Cookie')
        if not cookies_str:
            return dict()

        all_cookies_list: list = cookies_str.split(';')
        result_dict = dict()

        for tmp_cookie_str in all_cookies_list:
            tmp_cookie_str_to_list = tmp_cookie_str.split('=')
            tmp_cookie_key = tmp_cookie_str_to_list[0].lstrip()
            tmp_cookie_value = Utility.list_join_to_str(tmp_cookie_str_to_list[1:], separator_str='=')
            if decoding:
                tmp_cookie_key = urllib.parse.unquote(tmp_cookie_key, encoding='utf-8')
                tmp_cookie_value = urllib.parse.unquote(tmp_cookie_value, encoding='utf-8')

            tmp_original_value_list = result_dict.get(tmp_cookie_key)
            if tmp_original_value_list is None:
                tmp_original_value_list = list()
                result_dict[tmp_cookie_key] = tmp_original_value_list

            tmp_original_value_list.append(tmp_cookie_value)

        return result_dict

    def get_bearer_str_from_request(self, request):
        r"""
        从request的Header Bearer中获取bearer字符串

        :param request:
        :return:
        """
        bearer_str: str = request.headers.get('authorization')
        if not bearer_str:
            return None
        else:
            start_str = 'Bearer '
            if not bearer_str.startswith(start_str):
                return None
            else:
                return bearer_str[len(start_str):]

    def is_mobile_number(self, mobile_str: str) -> bool:
        r"""
        检查字符串是否是手机格式

        :param mobile_str
        """
        return re.match('^1\\d{10}$', MyStringUtils.to_str(mobile_str)) is not None

    def is_email_address(self, email_str: str) -> bool:
        r"""
        检查字符串是否是邮箱格式

        :param email_str
        """
        return re.match(
            '^([a-zA-Z0-9_\\-\\.]+)@((\\[[0-9]{1,3}\\.[0-9]{1,3}\\.[0-9]{1,3}\\.)|(([a-zA-Z0-9\\-]+\\.)+))([a-zA-Z]{2,4}|[0-9]{1,3})(\\]?)$',
            MyStringUtils.to_str(email_str)) is not None

    def deep_dict_copy(self, a_dict: dict) -> dict:
        r"""

        :param a_dict:
        :return:
        """
        result: dict = a_dict.copy()
        for tmp_key, tmp_value in result.items():
            if isinstance(tmp_value, dict):
                result[tmp_key] = self.deep_dict_copy(tmp_value)
            if isinstance(tmp_value, list):
                result[tmp_key] = self.deep_list_copy(tmp_value)

        return result

    def deep_list_copy(self, a_list: list) -> list:
        r"""

        :param a_list:
        :return:
        """
        result: list = a_list.copy()
        for tmp_index, tmp_value in enumerate(result):
            if isinstance(tmp_value, dict):
                result[tmp_index] = self.deep_dict_copy(tmp_value)
            if isinstance(tmp_value, list):
                result[tmp_index] = self.deep_list_copy(tmp_value)

        return result

    def do_notice_admin(self, **kwargs) -> bool:
        r"""
        通知管理员

        异步的方式发送,总是返回true

        :key subject 标题，默认为None
        :key notice_content 内容，不能为空
        :key admin_list 接收人，使用app_config.admin_list 提供
        :key runtime_env 使用app_config.runtime_env 提供
        :key send_email_config 使用app_config.send_email_config 提供

        """

        admin_list = kwargs.get("admin_list")
        if not admin_list:
            return True

        runtime_env = kwargs.get("runtime_env")
        send_email_config = kwargs.get("send_email_config")

        subject = kwargs.get("subject")

        notice_content = kwargs.get("notice_content")

        def run():
            real_subject = self.join_str(subject, '(运行时环境:', runtime_env, ')')
            send_notice_utils = SendNoticeUtils(send_email_config=send_email_config)

            send_notice_utils.do_send_email(to_mails=admin_list, subject=real_subject,
                                            notice_content=notice_content)

        thread.start_new_thread(run, ())
        return True

    def common_operation(self, retry_times_when_exception, operation_func, on_operation_failed, on_operation_completed,
                         *args,
                         **kwargs):
        r"""
        操作失败后重试的通用流程

        :key retry_times_when_exception
        :key operation_func
        :key on_operation_completed
        :key on_operation_failed
        :return:
        """
        has_attempt_times = 0

        encountered_exception: Optional[BaseException] = None
        result: Optional[bool] = None

        while has_attempt_times <= retry_times_when_exception:
            # 按需重新初始化jedisPool
            # 即上次循环中如果发生了异常，则在此次尝试执行操作前先重新初始化
            if encountered_exception is not None:
                try:
                    on_operation_failed()
                except BaseException as e:
                    # 如果重新初始化失败，直接结束并抛出异常
                    encountered_exception = e
                    break

            # 尝试执行操作
            try:
                result = operation_func(*args, **kwargs)
            except AppRuntimeException as e:
                encountered_exception = e
                break
            except BaseException as e:
                encountered_exception = e

            # 如果没有遇到异常，直接结束循环
            if not encountered_exception:
                break

            # 尝试次数+1
            has_attempt_times += 1

        on_operation_completed()

        if not encountered_exception:
            return result
        else:
            raise encountered_exception

    def get_container_name_from_container_image(self, container_image: str):
        r"""
        从形如 abc.com/istio/tgdevops/tgdevopspyservice:prod-1.0 的uri中提取 tgdevopspyservice

        :param container_image:
        :return:
        """
        sections_of_image = container_image.split(":")
        if len(sections_of_image) == 1:
            str_contains_name: str = container_image
        else:
            str_contains_name: str = self.list_join_to_str(sections_of_image[0:len(sections_of_image) - 1],
                                                           separator_str="/")

        sections_of_image = str_contains_name.split("/")
        return sections_of_image[len(sections_of_image) - 1]

    def dict_update(self, to_update_dict, update_dict):
        r"""
        按深度更新字典，尽量保留原字典的内容

        列表中不想更新的位置要跳过去需要提供占位符，使用字符串 __placeholder_for_update__ 表示

        :param to_update_dict:  被更新的字典
        :param update_dict: 提供更新内容的字典
        :return:
        """

        for key, value in update_dict.items():
            original_value = to_update_dict[key]
            # 如果更新字典的值不是list，也不是字典，则直接设置更新字典中的内容
            # 或者如果原字典中不包含这个key，则直接设置更新字典中的内容
            # 或者如果原字典中包含这个key，但是值不是list，也不是字典，则直接设置更新字典中的内容
            # 或者原字典中的值的类型和更新字典中的值的类型不一致，则直接设置更新字典中的内容
            if (not isinstance(value, dict) and not isinstance(value, list)) \
                    or key not in to_update_dict \
                    or (not isinstance(original_value, dict) and not isinstance(original_value, list)) \
                    or type(original_value) != type(value):
                to_update_dict[key] = value
            elif isinstance(value, dict):
                self.dict_update(original_value, value)
            else:
                self.list_update(original_value, value)

    def list_update(self, to_update_list, update_list):
        r"""
        按深度更新字典，尽量保留原字典的内容，

        列表更新和字典不一样，不想更新的位置要跳过去需要提供占位符，使用字符串 __placeholder_for_update__ 表示

        :param to_update_list:  被更新的列表
        :param update_list: 提供更新内容的列表，不想更新的index的值使用 __placeholder_for_update__ 表示
        :return:
        """

        len_of_to_update_list = len(to_update_list)
        for index, value in enumerate(update_list):
            if index >= len_of_to_update_list:
                return

            if value == "__placeholder_for_update__":
                continue

            original_value = to_update_list[index]
            # 如果更新列表的值不是list，也不是字典，则直接设置更新字典中的内容
            # 或者如果原字典中不包含这个key，则直接设置更新字典中的内容
            # 或者如果原字典中包含这个key，但是值不是list，也不是字典，则直接设置更新字典中的内容
            # 或者原字典中的值的类型和更新字典中的值的类型不一致，则直接设置更新字典中的内容
            if (not isinstance(value, dict) and not isinstance(value, list)) \
                    or (not isinstance(original_value, dict) and not isinstance(original_value, list)) \
                    or type(original_value) != type(value):
                to_update_list[index] = value
            elif isinstance(value, dict):
                self.dict_update(original_value, value)
            else:
                self.list_update(original_value, value)

    def replace_placeholder_in_str(self, a_str: str, **kwargs):
        r"""
        替换字符串中的形如正则表达式${abc}的值为kwargs中的值，如果kwargs中不包含abc，则不进行替换

        :param a_str:
        :return:
        """

        def _get_value_for_matched_key(matched):
            key_in_kwargs = matched.group(1)
            if key_in_kwargs in kwargs:
                result2 = kwargs[key_in_kwargs]
                if kwargs is not None:
                    result2 = str(result2)
            else:
                result2 = matched.group()

            return result2

        result = re.sub("\"\\${{int:(.+?)}}\"", _get_value_for_matched_key, a_str)
        return re.sub("\\${{(.+?)}}", _get_value_for_matched_key, result)

    def generate_random_str(self, randomlength: int = 16) -> str:
        """
        生成一个指定长度的随机字符串，其中

        string.digits=0123456789

        string.ascii_letters=abcdefghigklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ

        """
        str_list = [random.choice(string.digits + string.ascii_letters + r"""!@#$%^&*()""") for _
                    in range(randomlength)]
        random_str = ''.join(str_list)
        return random_str

    def check_resp_status_code_and_content(
            self, action, detail_info, resp, is_dict_for_resp_text: bool = True,
            expect_http_status_code: int = 200,
            **kwargs
    ) -> Union[Dict, List]:
        r"""
        通用的检查http请求返回的resp的状态码是否是200，且返回内容是否是json格式，并最终返回resp.text表示的字典

        :key action
        :key detail_info
        :key resp
        :key is_dict_for_resp_text 默认为True
        :key expect_http_status_code 默认为200
        :key other_expect_http_status_codes 其他期望的status_code值
        :key action_desc
        :return:
        """
        # 判断状态码
        other_expect_http_status_codes = kwargs.get("other_expect_http_status_codes")
        if other_expect_http_status_codes is None:
            other_expect_http_status_codes = list()

        other_expect_http_status_codes.append(expect_http_status_code)

        action_desc = kwargs.get("action_desc")
        if not action_desc:
            action_desc = action

        if resp.status_code not in other_expect_http_status_codes:
            raise AppRuntimeException(Utility.join_str(
                action_desc,
                f"失败,返回的状态码不是{other_expect_http_status_codes}"),
                Utility.join_str(
                    action,
                    f"返回的状态码不是{other_expect_http_status_codes},statusCode=",
                    resp.status_code, " ",
                    detail_info))
        # 判断返回内容
        result_str = resp.text
        if is_dict_for_resp_text:
            resp_obj = Utility.jsonstr2dict(result_str)
        else:
            resp_obj = Utility.jsonstr2list(result_str)
        if resp_obj is None:
            raise AppRuntimeException(
                Utility.join_str(action_desc, "失败,返回的内容不是json格式"),
                Utility.join_str(
                    action,
                    f"返回的内容不是json格式,content={result_str}",
                    " ",
                    detail_info
                )
            )

        return resp_obj

    def check_resp_status_code_and_code(
            self,
            action,
            detail_info,
            resp,
            expect_code_value: str = ConstResponseCode.CODE_OK,
            other_expect_code_values: list = None,
            b_check_data_in_app_response: bool = True,
            **kwargs
    ):
        r"""
        通用的检查http请求返回的resp的状态码是否是200，且返回内容为iam标准response，且其中的code为OK，且其中的data字段不为空,

        并最终返回data字段

        :key action:
        :key detail_info:
        :key resp:
        :key expect_code_value:
        :key other_expect_code_values:
        :key b_check_data_in_app_response:
        :key key_for_data:
        :key key_for_message:
        :key key_for_code:
        :key action_desc:
        :return:
        """
        return self.check_resp_status_code_and_code_and_return_app_response(
            action,
            detail_info,
            resp,
            expect_code_value=expect_code_value,
            other_expect_code_values=other_expect_code_values,
            b_check_data_in_app_response=b_check_data_in_app_response,
            **kwargs
        ).data

    def check_resp_status_code_and_code_and_return_app_response(
            self,
            action,
            detail_info,
            resp,
            expect_code_value: str = ConstResponseCode.CODE_OK,
            other_expect_code_values: list = None,
            b_check_data_in_app_response: bool = True,
            **kwargs
    ) -> AppResponse:
        r"""
        通用的检查http请求返回的resp的状态码是否是200，且返回内容为iam标准response，且其中的code为OK，且其中的data字段不为空,

        并最终返回data字段

        :key action:
        :key detail_info:
        :key resp:
        :key expect_code_value:
        :key other_expect_code_values:
        :key b_check_data_in_app_response:
        :key key_for_data:
        :key key_for_message:
        :key key_for_code:
        :key action_desc:
        :return:
        """
        action_desc = kwargs.get("action_desc")
        if not action_desc:
            action_desc = action

        resp_dict = self.check_resp_status_code_and_content(action, detail_info, resp, action_desc=action_desc)

        # 转换成标准的 AppResponse
        key_for_code = kwargs.get("key_for_code")
        key_for_data = kwargs.get("key_for_data")
        key_for_message = kwargs.get("key_for_message")
        app_response = AppResponse.from_dict(resp_dict, key_for_data=key_for_data, key_for_msg=key_for_message,
                                             key_for_code=key_for_code)

        if other_expect_code_values is None:
            other_expect_code_values = list()
        other_expect_code_values.append(expect_code_value)

        if app_response.code not in other_expect_code_values:
            raise AppRuntimeException(
                Utility.join_str(action_desc, "失败,返回的code=", app_response.code, " message=",
                                 app_response.message),
                Utility.join_str(action, "返回的code=", app_response.code, " message=",
                                 app_response.message, " ", detail_info))

        iam_response_data = app_response.data

        # 如果未获取到权限验证结果信息，直接抛出异常
        if not iam_response_data and b_check_data_in_app_response:
            raise AppRuntimeException(Utility.join_str(action_desc, "失败，返回的内容不包含data数据"),
                                      Utility.join_str(action, "返回的内容不包含data数据,content=",
                                                       Utility.dict2jsonstr(resp_dict), " ", detail_info))

        return app_response

    def parse_crn(self, crn: str, *crn_list) -> CrnParseResult:
        r"""
        解析格式为 crn:ucs::{product_code}:{region_code}:{account_id}:first/{first_id}/second/{second_id} 的crn表达式

        从中解析出 product_code,region_code,acct_id,first_id,second_id,third_id

        :param crn:
        :param crn_list: 多个crn,解析第一个不为空的crn
        """
        if not crn:
            for tmp_crn in crn_list:
                if tmp_crn:
                    crn = tmp_crn
                    break

        if not crn:
            raise ValueError(f"crn为空")

        colon_parts = crn.split(":")
        if len(colon_parts) < 7:
            raise ValueError(f"{crn}不是有效的crn表达式")

        product_code = colon_parts[3]
        region_code = colon_parts[4]
        acct_id = colon_parts[5]
        base_crn = ":".join(colon_parts[0:6])

        res_part = colon_parts[6]
        res_parts = res_part.split("/")
        len_parts = len(res_parts)
        first_id_type = None
        first_id = None
        second_id_type = None
        second_id = None
        third_id_type = None
        third_id = None
        other_id_types = list()
        other_ids = list()
        first_crn = None
        second_crn = None
        third_crn = None
        other_crns = list()
        if len_parts >= 2:
            first_id_type = res_parts[0]
            first_id = res_parts[1]
            tmp_res_part = "/".join(res_parts[0:2])
            first_crn = ":".join([base_crn, tmp_res_part])
        if len_parts >= 4:
            second_id_type = res_parts[2]
            second_id = res_parts[3]
            tmp_res_part = "/".join(res_parts[0:4])
            second_crn = ":".join([base_crn, tmp_res_part])
        if len_parts >= 6:
            third_id_type = res_parts[4]
            third_id = res_parts[5]
            tmp_res_part = "/".join(res_parts[0:6])
            third_crn = ":".join([base_crn, tmp_res_part])

        if len_parts >= 8:
            for index in range(7, len_parts):
                if index % 2 != 0:
                    other_ids.append(res_parts[index])
                    tmp_res_part = "/".join(res_parts[0:index + 1])
                    other_crns.append(":".join([base_crn, tmp_res_part]))
                else:
                    other_id_types.append(res_parts[index])

        return CrnParseResult(
            product_code=product_code,
            region_code=region_code,
            acct_id=acct_id,
            first_id_type=first_id_type,
            first_id=first_id,
            second_id_type=second_id_type,
            second_id=second_id,
            third_id_type=third_id_type,
            third_id=third_id,
            other_id_types=other_id_types,
            other_ids=other_ids,
            first_crn=first_crn,
            second_crn=second_crn,
            third_crn=third_crn,
            other_crns=other_crns,
        )

    def gen_crn_expression(self, product_code: str, region_code: str, account_id: str, res_part: str) -> str:
        r"""
        生成crn表达式, 示例 crn:ucs::cfc:gz-tst:12345:func/helloworld

        :param product_code: 例如 cfc
        :param region_code: 例如 gz-tst
        :param account_id:
        :param res_part:资源部分的描述字符串,例如 func/helloworld
        """
        region_code = region_code or ""
        return f"crn:ucs::{product_code}:{region_code}:{account_id}:{res_part}"

    def rmdir_or_file(self, file_path: str):
        r"""
        根据操作系统执行删除文件夹/文件

        :param file_path:
        """
        if not os.path.exists(file_path):
            return

        if os.path.isdir(file_path):
            # 执行删除文件夹操作
            if platform.system() == 'Windows':
                subprocess.call(f"rmdir /S /Q {file_path}", shell=True)
            else:
                subprocess.call(f"rm -rf {file_path}", shell=True)
        elif os.path.isfile(file_path):
            # 执行删除文件操作
            os.remove(file_path)
        else:
            # 特殊文件,不执行操作
            self.logger.warning(
                f"执行删除操作时发现{file_path} is a special file(socket,FIFO,device file),不执行删除操作")

    def read_file_content(self, file_path) -> str:
        """
        读取文件内容

        :param file_path: 文件路径
        :return: str -> file-content
        """
        with open(file_path, 'r') as f:  # 打开一个文件，必须是'rb'模式打开
            return f.read()

    def yaml_load(self, yaml_str: str) -> Union[dict, str]:
        r"""
        无论是 yaml 格式 还是 json 格式,都可以使用 yaml_load 读取,但读取的结果可能是 字典或字符串

        """
        dict_list = self.yaml_load_list(cfg_content=yaml_str)
        if dict_list:
            return dict_list[0]
        else:
            return dict()

    def yaml_load_list(self, path: str = None, cfg_content: str = None) -> List[Union[Dict, str]]:
        """
        将yaml格式文件转换为dict值，该yaml文件可以包含多块yaml数据，每个dict放到list中返回
        无论是 yaml 格式 还是 json 格式,都可以使用 yaml_load 读取,但读取的结果可能是 字典或字符串
        :param path: 文件路径
        :param cfg_content:
        :return: list -> []
        """
        if path:
            cfg = self.read_file_content(file_path=path)
        else:
            cfg = cfg_content or ""

        yaml_generator = yaml.safe_load_all(cfg)  # 将yaml格式文件转换为dict值，该yaml文件可以包含多块yaml数据
        if yaml_generator is None:
            return list()

        # 转成dict并保存到list中
        yaml_list = list()
        for one_yaml_generator in yaml_generator:
            if one_yaml_generator is None:
                continue
            yaml_list.append(json.loads(json.dumps(one_yaml_generator)))
        return yaml_list

    def yaml_dump_list_to_str(self, source_list: List[Dict]) -> str:
        r"""
        将字典列表序列化成为yaml字符串

        """
        return yaml.safe_dump_all(documents=source_list, encoding='utf-8', allow_unicode=True).decode('utf-8')

    def yaml_dump_list_to_file(self, source_list: List[Dict], file_path: str):
        r"""
        将字典列表序列化成为yaml字符串

        """
        with open(file_path, 'w', encoding='utf-8') as f:  # 打开一个文件，必须是'rb'模式打开
            yaml.safe_dump_all(documents=source_list, stream=f, encoding='utf-8', allow_unicode=True)

    def dict_pop_key(self, source_dict: dict, key_name: str):
        r"""
        改进的移除dict的key,如果key不存在,则返回None,否则返回原始的 dict.pop(key)

        :param source_dict:
        :param key_name:
        """
        return source_dict.pop(key_name) if key_name in source_dict else None

    def gen_user_info(self, login_user: LoginedUserInfo, output_type: str = "str") -> Union[str, dict]:
        r"""
        生成 {"user_id":"","user_name":""}格式的字典或字符串

        :param login_user:
        :param output_type: str or dict, 输出格式,默认为 str
        """
        output_type = output_type or "str"
        result_dict = {
            "user_id": str(login_user.user_id),
            "user_name": login_user.user_name
        }
        if output_type == "str":
            return self.dict2jsonstr(result_dict)
        else:
            return result_dict

    def get_content_after_render(
            self,
            dir_path: str = "./resource/",
            encoding: str = "utf-8",
            template_name: str = None,
            render_dict: dict = None,
            **kwargs
    ) -> str:
        r"""
        获取jinja2 template渲染后的文件内容

        :param dir_path: 被渲染文件所在目录,默认为 ./resource/
        :param encoding: 获取文件内容编码,默认为 utf-8
        :param template_name: 模板文件名称
        :param render_dict: 渲染模板用的字典
        :key variable_start_string:
        :key variable_end_string:
        """
        dir_path = dir_path or "./resource/"
        encoding = encoding or "utf-8"
        render_dict = render_dict or dict()
        variable_start_string = kwargs.get("variable_start_string") or VARIABLE_START_STRING
        variable_end_string = kwargs.get("variable_end_string") or VARIABLE_END_STRING
        env = Environment(
            loader=FileSystemLoader(dir_path, encoding=encoding),
            variable_start_string=variable_start_string,
            variable_end_string=variable_end_string,
            undefined=StrictUndefined,
        )  # 创建文件加载器对象
        template = env.get_template(template_name)  # 获取一个模板文件
        return template.render(render_dict)  # 渲染

    def gen_ip_port_str(self, schema: Optional[str] = "http", num_list: list = None, port: Optional[int] = None):
        r"""
        生成url格式字符串,防止sonar检查

        :param schema: 默认为http,None时不计入字符串内容
        :param num_list: IP地址数字
        :param port: 端口数字
        """
        result_str = f"{schema}://" if schema else ""
        str_list = [str(num) for num in num_list]
        result_str += ".".join(str_list)
        if port is not None:
            result_str += f":{port}"
        return result_str

    def get_md5_of_string(self, src: str) -> str:
        """
        获取字符串的md5值

        :param src:
        :return:
        """
        md1 = hashlib.md5()  # 创建一个md5算法对象
        md1.update(src.encode('UTF-8'))
        return md1.hexdigest()

    def get_md5_of_file(self, filepath: str, read_byte_once: int = 8096) -> str:
        """
        获取文件的md5值

        :param filepath:
        :param read_byte_once:
        :return:
        """
        if not os.path.isfile(filepath):
            return ""
        myhash = hashlib.md5()  # 创建一个md5算法对象
        with open(filepath, 'rb') as f:  # 打开一个文件，必须是'rb'模式打开
            while True:
                b = f.read(read_byte_once)  # 由于是一个文件，每次只读取固定字节
                if not b:
                    break
                # 当读取内容不为空时对读取内容进行update
                myhash.update(b)
        return myhash.hexdigest()

    def get_md5_of_dir(self, dirpath: str) -> str:
        """
        获取文件夹的md5值

        :param dirpath:
        :return:
        """
        if not os.path.isdir(dirpath):
            return ""
        dir_path_for_md5_file = tempfile.mkdtemp()
        md5_file = os.path.join(dir_path_for_md5_file, "tmp.md5")
        with open(md5_file, 'w') as outfile:
            for root, subdirs, files in os.walk(dir_path_for_md5_file):
                for file in files:
                    filefullpath = os.path.join(root, file)
                    md5 = self.get_md5_of_file(filefullpath)
                    outfile.write(md5)
        val = self.get_md5_of_file(md5_file)
        self.rmdir_or_file(dir_path_for_md5_file)
        return val

    def get_md5_of_file_or_dir(self, filepath: str) -> str:
        r"""
        获取文件/文件夹的md5值

        :param filepath:文件/文件夹的路径
        """
        if os.path.isfile(filepath):
            return self.get_md5_of_file(filepath)
        elif os.path.isdir(filepath):
            return self.get_md5_of_dir(filepath)
        else:
            return ""

    def base64_decode_with_padding(self, payload: Union[str, bytes]) -> str:
        r"""
        base64解码,如果payload字节数不是4的倍数,会使用=补足

        :param payload: 字符串或bytes
        """
        if isinstance(payload, str):
            byte_payload = payload.encode("utf-8")
        else:
            byte_payload = payload

        missing_padding = 4 - len(byte_payload) % 4
        if missing_padding:
            byte_payload += b'=' * missing_padding
        b_result = base64.b64decode(byte_payload)
        return str(b_result, encoding="utf-8")

    def parse_jwt_str(self, jwt_str: str) -> str:
        r"""
        :param jwt_str: jwt字符串
        """
        payload = jwt_str.split(".")[1]
        return self.base64_decode_with_padding(payload)

    def attemp_parse_login_user_info(self, jwt_str: str = None, request: Request = None) -> Optional[LoginedUserInfo]:
        r"""
        尝试从jwt字符串中解析出登陆用户信息,解析失败返回None

        :param jwt_str: jwt字符串
        :param request: flask.Request对象
        """

        if not jwt_str and request:
            jwt_str = request.cookies.get("accessToken") or ""

        try:
            payload = self.parse_jwt_str(jwt_str) if jwt_str else None
            if not payload:
                raise AppRuntimeException(
                    message="解析jwt字符串后获取的payload内容为空",
                    detail=f"jwt_str= {jwt_str}",
                )

            payload_dict = self.jsonstr2dict(payload)
            if not payload_dict:
                raise AppRuntimeException(
                    message="jwt的payload为空或者不是有效的json字符串",
                    detail=f"payload= {payload}",
                )
            return LoginedUserInfo.from_dict(payload_dict, jwt_str)
        except BaseException as e:
            self.logger.error(e)
            exception_tracback = traceback.format_exc()
            self.logger.error(exception_tracback)
            return None

    def retry_operation(
            self, operation_func, is_ok_function, is_ok_function_for_exception=None,
            operation_func_desc: str = None,
            max_execute_times: int = 240, wait_seconds_for_retry: float = 1
    ):
        r"""
        通用的重试动作

        :param operation_func: 操作函数,请使用 functools.partial 保证传入的函数在执行时不用再传入任何参数
        :param is_ok_function: 对函数执行结果的判断函数,
            该函数应返回一个Tuple,第一个元素为 bool 类型,表示对操作结果是否认为成功,第二个参数为真正的函数返回结果
        :param is_ok_function_for_exception: 如果提供了该函数,则函数执行期间发生异常时使用该函数对异常进行判断,
            该函数应返回一个Tuple,第一个元素为 bool 类型,表示发生异常时是否认为操作成功,第二个参数为认为成功时的函数返回结果
        :param operation_func_desc: 操作函数描述,默认为 字面值"operation_func"
        :param max_execute_times: 默认 240次
        :param wait_seconds_for_retry: 默认 1 秒
        """
        operation_func_desc = operation_func_desc or "operation_func"
        has_executed_times = 0
        while True:
            # 尝试执行,并获得结果或捕获异常
            tmp_result = None
            encountered_exception = None
            try:
                tmp_result = operation_func()
                is_ok, tmp_result = is_ok_function(tmp_result)
                if is_ok:
                    return tmp_result
            except BaseException as e:
                if is_ok_function_for_exception:
                    is_ok, tmp_result2 = is_ok_function_for_exception(e)
                    if is_ok:
                        return tmp_result2
                    else:
                        self.logger.error(f"执行{operation_func_desc}时发生异常:{e}")
                        self.logger.error(traceback.format_exc())
                        encountered_exception = e
                else:
                    self.logger.error(f"执行{operation_func_desc}时发生异常:{e}")
                    self.logger.error(traceback.format_exc())
                    encountered_exception = e

            # 记录执行次数+1
            has_executed_times += 1
            # 判断是否已经达到最大执行次数
            if has_executed_times >= max_execute_times > 0:
                # 返回最后一次执行的结果或抛出异常
                if encountered_exception:
                    raise encountered_exception
                else:
                    return tmp_result
            # 睡眠间隔
            time.sleep(wait_seconds_for_retry)

    def get_dict_from_dict_and_set_when_none(self, source_dict: dict, key_name: str) -> dict:
        r"""
        从字典中获取字典,如果指定的key不存在,则设置key对应的对象为一个空的字典

        :param source_dict:
        :param key_name:
        """
        result_dict = source_dict.get(key_name)
        if result_dict is None:
            result_dict = dict()
            source_dict[key_name] = result_dict
        return result_dict

    def get_list_from_dict_and_set_when_none(self, source_dict: dict, key_name: str) -> list:
        r"""
        从字典中获取字典,如果指定的key不存在,则设置key对应的对象为一个空的字典

        :param source_dict:
        :param key_name:
        """
        result_list = source_dict.get(key_name)
        if result_list is None:
            result_list = list()
            source_dict[key_name] = result_list
        return result_list

    def serialized(self, source_obj) -> str:
        r"""
        序列化为字符串,使用 pickle.dump 然后 base64编码, 然后以 utf-8 解码为字符串

        """
        obj_src_bytes = pickle.dumps(source_obj)
        return base64.b64encode(obj_src_bytes).decode("utf-8")

    def reverse_serialized(self, source_str: str):
        r"""
        反序列化,将字符串以utf-8编码为bytes,然后base64解码,然后 pickle.loads为对象

        """
        obj_src_bytes = base64.b64decode(source_str.encode("utf-8"))
        return pickle.loads(obj_src_bytes)

    def parse_cpu_value_to_m(
            self,
            source_str: str,
            parse_empty_to_0: bool = False,
    ) -> int:
        """
        根据 10m 或 10 这样的字符串计算 cpu 有多少 m
        """
        source_str_desc = source_str or "空字符串"
        if not source_str and parse_empty_to_0:
            return 0
        elif source_str.endswith("m"):
            return int(source_str[0:-1])
        elif source_str.isdigit():  # 全部是数字,则转换为m需要乘1000
            return int(float(source_str) * 1000)
        else:
            failed_reason = f"无法将{source_str_desc}解析为单位为m的CPU核数"
            raise AppException(
                code=ConstResponseCode.CODE_MISSING_PARAMETER,
                message=failed_reason,
            )

    def parse_mem_value_to_b(
            self,
            source_str: str,
            parse_empty_to_0: bool = False,
    ) -> int:
        """
        根据 100Mi 或 10Gi 这样的字符串计算 mem 有多少字节
        """
        source_str_desc = source_str or "空字符串"
        if not source_str and parse_empty_to_0:
            return 0
        elif source_str.endswith("Ki"):
            return int(float(source_str[0:-2]) * 1024 ** 1)
        elif source_str.endswith("Mi"):
            return int(float(source_str[0:-2]) * 1024 ** 2)
        elif source_str.endswith("Gi"):
            return int(float(source_str[0:-2]) * 1024 ** 3)
        elif source_str.endswith("Ti"):
            return int(float(source_str[0:-2]) * 1024 ** 4)
        elif source_str.endswith("Pi"):
            return int(float(source_str[0:-2]) * 1024 ** 5)
        elif source_str.endswith("Ei"):
            return int(float(source_str[0:-2]) * 1024 ** 6)
        elif source_str.endswith("k"):
            return int(float(source_str[0:-1]) * 1000 ** 1)
        elif source_str.endswith("M"):
            return int(float(source_str[0:-1]) * 1000 ** 2)
        elif source_str.endswith("G"):
            return int(float(source_str[0:-1]) * 1000 ** 3)
        elif source_str.endswith("T"):
            return int(float(source_str[0:-1]) * 1000 ** 4)
        elif source_str.endswith("P"):
            return int(float(source_str[0:-1]) * 1000 ** 5)
        elif source_str.endswith("E"):
            return int(float(source_str[0:-1]) * 1000 ** 6)
        elif source_str.isdigit():  # 全部是数字则单位为B
            return int(source_str)
        else:
            failed_reason = f"无法将{source_str_desc}解析为内存字节数量"
            raise AppException(
                code=ConstResponseCode.CODE_MISSING_PARAMETER,
                message=failed_reason,
            )

    def format_time_to_utc_str(self, param_time: Union[int, float, str, datetime]) -> Optional[str]:
        r"""
        将时间格式化成UTC字符串,格式为标准时区的 %Y-%m-%dT%H:%M:%SZ
        :param param_time: 可以是unix时间戳(秒数),也可以是本地日期的字符串,或者是datetime类型的日期
        """
        if param_time is None:
            return None
        if isinstance(param_time, (int, float)):
            parsed_utctime = datetime.utcfromtimestamp(param_time)
            return parsed_utctime.strftime('%Y-%m-%dT%H:%M:%SZ')
        elif isinstance(param_time, datetime):
            parsed_utctime = datetime.utcfromtimestamp(param_time.timestamp())
            return parsed_utctime.strftime('%Y-%m-%dT%H:%M:%SZ')
        elif isinstance(param_time, str):
            if not param_time:
                return None
            parsed_time = self.parse_datestr(param_time)
            if parsed_time is None:
                raise ValueError(f"param_time不是有效的时间字符串:{param_time}")
            parsed_utctime = datetime.utcfromtimestamp(parsed_time.timestamp())
            return parsed_utctime.strftime('%Y-%m-%dT%H:%M:%SZ')
        else:
            raise ValueError(f"param_time不能解析为时间:不支持的类型:{param_time.__class__}")

    def format_time_to_unix_timestamp_seconds(self, param_time: Union[int, float, str, datetime]) -> Optional[float]:
        r"""
        将时间格式化成unix时间戳(秒数)
        :param param_time: 可以是unix时间戳(秒数),也可以是本地日期的字符串,或者是datetime类型的日期
        """
        if param_time is None:
            return None
        if isinstance(param_time, (int, float)):
            return param_time
        elif isinstance(param_time, datetime):
            return param_time.timestamp()
        elif isinstance(param_time, str):
            if not param_time:
                return None
            parsed_time = self.parse_datestr(param_time)
            if parsed_time is None:
                raise ValueError(f"param_time不是有效的时间字符串:{param_time}")
            return parsed_time.timestamp()
        else:
            raise ValueError(f"param_time不能解析为unix时间戳:不支持的类型:{param_time.__class__}")

    def get_object_from_obj_by_paths(self, source_obj, key_paths: List[str] = None, key_path_str: str = None):
        r"""
        按照key的路径依次获取指定路径序列的key对应的字典或列表,如果中途key miss,或index超出范围,则返回None,
        :param source_obj: 列表或字典
        :param key_paths: 优先级比 key_path_str 高
        :param key_path_str: 格式为 components.ingressGateways.0.k8s.resources
        """
        if not key_paths:
            key_path_str = key_path_str or ""
            key_paths = [int(x) if x.isdigit() else x for x in key_path_str.strip(".").split(".")]

        result_obj = source_obj
        for tmp_key in key_paths:
            if isinstance(tmp_key, int):
                if len(result_obj) > tmp_key:
                    result_obj = result_obj[tmp_key]
                else:
                    result_obj = None
            else:
                result_obj = result_obj.get(tmp_key)

            if result_obj is None:
                return result_obj

        return result_obj

    def load_file_as_dict(self, filepath: str) -> dict:
        r"""
        判断文件后缀名, 如果是 .yaml 或 .yml 则按 yaml 方式读取,
        文件内容不合规,则抛出异常
        """
        with open(filepath, 'r', encoding="utf-8") as dict_file:
            conf_content = dict_file.read()
            # 无论是 yaml 格式 还是 json 格式,都可以使用 yaml_load 读取,但读取的结果可能是 字典或字符串
            dict_content = self.yaml_load(conf_content)
            if isinstance(dict_content, str):
                raise ValueError(f"{filepath}的内容不是有效的json或yaml字符串")

        return dict_content