common_util.py

import calendar
import json
import os
import sys
from datetime import timedelta
from enum import Enum
from typing import Dict

import requests
from textblob import Word

sys.path.append(os.path.dirname(sys.path[0]))
from pyspark.sql import functions as F
from pyspark.sql.dataframe import DataFrame
from pyspark.sql.session import SparkSession
from pyspark.sql.types import *
from pyspark.sql import Window

from utils.db_util import DBUtil
from utils.hdfs_utils import HdfsUtils
from utils.spark_util import SparkUtil
from utils.ssh_util import SSHUtil
from utils.es_util import EsUtils
from utils.DolphinschedulerHelper import DolphinschedulerHelper
from datetime import datetime
from yswg_utils.common_udf import udf_parse_amazon_orders
from utils.StarRocksHelper import StarRocksHelper

class DateTypes(Enum):
    """
    日期格式枚举
    """
    day = "day"
    last30day = "last30day"
    week = "week"
    month = "month"
    month_week = "month_week"
    month_old = "month_old"
    last365day = "last365day"
    year = "year"


class CommonUtil(object):
    __SITE_SET__ = {'us', 'uk', 'de', 'fr', 'es', 'it', 'au', 'ca'}

    __sqoop_home__ = "/opt/module/sqoop-1.4.6/bin/sqoop"
    __sqoop_1_4_7_home__ = "/mnt/opt/module/sqoop-1.4.7/bin/sqoop"
    __python_home__ = "/opt/module/anaconda3/envs/pyspark/bin/python3.8"
    __desploy_home__ = "/opt/module/spark/demo/py_demo"

    # __hive_home__ = "/opt/module/hive/bin/hive"
    __hive_home__ = "/opt/datasophon/hive-3.1.0/bin/hive"
    __hadoop_home__ = "/opt/module/hadoop/bin/hadoop"

    __msg_usr__ = ['wujicang', 'huangjian', 'fangxingjun', 'chenjianyun', 'wangrui4']

    _date_time_format = "yyyy-MM-dd HH:mm:ss"

    _py_date_time_format = '%Y-%m-%d %H:%M:%S'

    _py_date_format = '%Y-%m-%d'

    __start_time_of_morning_str__ = "08:00"
    __end_time_of_morning_str__ = "12:30"
    __start_time_of_afternoon_str__ = "14:00"
    __end_time_of_afternoon_str__ = "19:00"

    __export_process_map__ = {
        "新ABA流程": ['dwt_aba_st_analytics.py', 'dwd_st_volume_fba.py', 'dwt_aba_st_analytics_report_pg.py',
                   'dwt_aba_last_change_rate.py', 'dwt_st_market_pg.py'],
        "反查搜索词": ['dwt_st_asin_reverse.py'],
        "店铺流程": ['dwt_fb_asin_info.py', 'dwt_fb_base_report.py', 'dwt_fb_category_report.py', 'dwt_fb_top20_asin_info.py'],
        "流量选品": ['es_flow_asin.py']
    }
    u_parse_amazon_orders = F.udf(udf_parse_amazon_orders, IntegerType())

    """
    一般工具类
    """

    @classmethod
    def to_int(cls, obj, defval=None):
        """
        安全转为 int
        :param obj:
        :param defval: 默认值
        :return:
        """
        if CommonUtil.notBlank(obj):
            return int(obj)
        return defval

    @classmethod
    def to_float(cls, obj, defval=None):
        """
        安全转为 float
        :param obj:
        :param defval:默认值
        :return:
        """
        if CommonUtil.notBlank(obj):
            return float(obj)
        return defval

    @classmethod
    def to_str(cls, obj, defval=None):
        """
        安全转为 str
        :param obj:
        :param defval:默认值
        :return:
        """
        if CommonUtil.notBlank(obj):
            return str(obj)
        return defval

    @staticmethod
    def reset_partitions(site_name, partitions_num=10):
        """
        按不同站点划分分区数量
        :param site_name: 站点名称
        :param partitions_num: 自定义分区数量
        :return: partitions_num
        """
        print("重置分区数")
        if site_name in ['us']:
            partitions_num = partitions_num
        elif site_name in ['uk', 'de']:
            partitions_num = partitions_num // 2 if partitions_num // 2 > 0 else 1
        elif site_name in ['es', 'fr', 'it']:
            partitions_num = partitions_num // 4 if partitions_num // 4 > 0 else 1
        return partitions_num

    @staticmethod
    def split_month_week_date(date_type, date_info):
        """
        对month类型和week类型date_info进行拆解
        :param date_type: 分区类型
        :param date_info: 入参时间
        :return: d1 d2
        """
        if date_type == DateTypes.week.name:
            year, week = date_info.split('-')
            return int(year), int(week)
        elif date_type == DateTypes.month.name:
            year, month = date_info.split('-')
            return int(year), int(month)
        elif date_type == DateTypes.month_week.name:
            year, month = date_info.split('-')
            return int(year), int(month)

    @staticmethod
    def safeIndex(list: list, index: int, default: object = None):
        """
        安全获取list的索引对应的值
        :param list: 列表
        :param index: 索引
        :param default: 默认值
        :return:
        """
        if (index <= len(list) - 1):
            return list[index]
        return default

    @staticmethod
    def get_calDay_by_dateInfo(spark_session: SparkSession, date_type: str, date_info: str):
        """
        根据不同日期维度，获取当前维度下的最后一天
        :param spark_session: sparksession对象
        :param date_type: 日期类型
        :param date_info: 日期值
        :return: cal_date:根据不同日期维度，获取当前维度下的最后一天
        """
        assert date_type is not None, "date_type不能为空！"
        assert date_info is not None, "date_info不能为空！"
        df_date = spark_session.sql(f"select * from dim_date_20_to_30;")
        df = df_date.toPandas()
        if date_type in [DateTypes.day.name, DateTypes.last30day.name]:
            cal_day = date_info
        # 如果为 周、月则取该周、月的最后一日，作为新品计算基准日
        elif date_type in [DateTypes.week.name, DateTypes.month.name]:
            sorted_df = df.loc[df[f'year_{date_type}'] == f"{date_info}"].sort_values('date', ascending=False)
            cal_day = sorted_df.head(1)['date'].iloc[0]
        elif date_type == '4_week':
            sorted_df = df.loc[(df.year_week == f"{date_info}")].sort_values('date', ascending=False)
            cal_day = sorted_df.head(1)['date'].iloc[0]
        elif date_type == DateTypes.month_week.name:
            current_date = datetime.now().date()
            cal_day = current_date.strftime("%Y-%m-%d")
        else:
            return None
        print("cal_day:", str(cal_day))
        return str(cal_day)

    @staticmethod
    def get_rel_exception_info():
        import sys
        exc_type, exc_value, exc_traceback = sys.exc_info()
        print(exc_traceback)
        return exc_type, f"""{exc_value} in {exc_traceback.tb_frame.f_code.co_filename}  {exc_traceback.tb_lineno} line"""

    @staticmethod
    def get_sys_arg(index: int, defVal: object):
        """
        获取main系统输入参数脚标从1开始
        :param index: 索引
        :param defVal: 默认值
        :return:
        """
        return CommonUtil.safeIndex(sys.argv, index, defVal)

    @staticmethod
    def listNotNone(listVal: list = None):
        """
        判断是否是空数组
        """
        return listVal is not None or len(listVal) > 0

    @staticmethod
    def notNone(obj: object = None):
        """
        判断是否是None
        """
        return obj is not None

    @staticmethod
    def notBlank(strVal: str = None):
        """
        判断是否是空字符串
        """
        return strVal is not None and strVal != ''

    @staticmethod
    def get_day_offset(day: str, offset: int):
        """
        获取日期偏移值
        :param day: 类似 2022-11-01
        :param offset: 偏移值
        :return: 过去或将来的时间
        """
        pattern = "%Y-%m-%d"
        rel_day = datetime.strptime(day, pattern)
        d = rel_day + timedelta(days=offset)
        return d.strftime(pattern)

    @staticmethod
    def get_month_offset(month: str, offset: int):
        """
        获取月份偏移值
        :param month: 类似 2022-11
        :param offset: 偏移值
        :return: 过去或将来的月份
        """
        year_int = int(CommonUtil.safeIndex(month.split("-"), 0, None))
        month_int = int(CommonUtil.safeIndex(month.split("-"), 1, None))

        if offset > 0:
            for i in range(0, offset):
                year_int, month_int = calendar._nextmonth(year_int, month_int)

        if offset < 0:
            for i in range(0, abs(offset)):
                year_int, month_int = calendar._prevmonth(year_int, month_int)

        return datetime(year_int, month_int, 1).strftime("%Y-%m")

    @staticmethod
    def reformat_date(date_str: str, from_format: str, to_format: str):
        """
        重新格式化日期
        :param date_str:
        :param from_format:
        :param to_format:
        :return:
        """
        return datetime.strptime(date_str, from_format).strftime(to_format)

    @staticmethod
    def format_now(from_format: str):
        now = datetime.now()
        return datetime.strftime(now, from_format)

    @staticmethod
    def format_timestamp(timestamp: int, format: str = _py_date_time_format):
        """
        格式化毫秒级别时间戳
        :param timestamp:
        :param format:
        :return:
        """
        from datetime import datetime
        return datetime.strftime(datetime.fromtimestamp(timestamp / 1000), format)

    @staticmethod
    def calculate_date_offset(date1, date2):
        """
        计算日期偏移量
        :param date1: 日期1 格式：%Y-%m-%d
        :param date2: 日期2 格式：%Y-%m-%d
        :return: 日期差值
        """
        if date1 is None or date2 is None:
            return None

        date_format = "%Y-%m-%d"
        try:
            # 将日期字符串转换为 datetime 对象
            datetime1 = datetime.strptime(date1, date_format)
            datetime2 = datetime.strptime(date2, date_format)

            # 计算日期的偏移量
            offset = abs((datetime2 - datetime1).days)
            return offset
        except ValueError:
            # 日期字符串格式不正确
            return None

    @staticmethod
    def list_to_insql(arr: list):
        """
        数组转为in中的sql
        :param arr:
        :return:
        """
        return str.join(",", list(map(lambda item: f"'{item}'", arr)))

    @staticmethod
    def arr_to_spark_col(arr: list):
        """
        python数组转为df中的数组
        """
        return F.array(list(map(lambda item: F.lit(item), arr)))

    @staticmethod
    def build_es_option(site_name="us"):
        """
        构建spark导出用 es参数
        """
        site_port = {
            "us": 9200,
            "uk": 9201,
            "de": 9201,
        }
        return {
            "es.nodes": "120.79.147.190",
            "es.port": site_port[site_name],
            "es.net.http.auth.user": "elastic",
            "es.net.http.auth.pass": "selection2021.+",
            "es.nodes.wan.only": True,
            "es.index.auto.create": True
        }

    @staticmethod
    def str_compress(strLines: str):
        """
        多行字符串压缩
        :param strLines:
        :return:
        """
        strArr = []
        splitArr = strLines.splitlines()
        for s in splitArr:
            strArr.append(s.strip())
        return ' '.join(strArr)

    @staticmethod
    def build_export_sh(site_name: str,
                        db_type: str,
                        hive_tb: str,
                        export_tb: str,
                        col: list,
                        partition_dict: dict,
                        num_mappers=20
                        ):

        conn_info = DBUtil.get_connection_info(db_type, site_name)
        cols = str.join(",", col)
        if len(partition_dict.keys()) > 0:
            p_keys = str.join(",", partition_dict.keys())
            p_values = str.join(",", partition_dict.values())
            return f"""
    {CommonUtil.__sqoop_home__} export -D mapred.job.queue.name=default -D mapred.task.timeout=0 \\
    --connect {conn_info['url']}  \\
    --username {conn_info['username']}  \\
    --password {conn_info['pwd']} \\
    --table {export_tb}  \\
    --input-fields-terminated-by '\\001'  \\
    --hcatalog-database big_data_selection  \\
    --hcatalog-table {hive_tb}  \\
    --hcatalog-partition-keys {p_keys}  \\
    --hcatalog-partition-values {p_values}  \\
    --input-null-string '\\\\N'  \\
    --input-null-non-string '\\\\N'  \\
    --num-mappers {num_mappers} \\
    --columns {cols}  \\
    --outdir "/tmp/sqoop/"
    """
        return f"""
        {CommonUtil.__sqoop_home__} export -D mapred.job.queue.name=default -D mapred.task.timeout=0 \\
    --connect {conn_info['url']}  \\
    --username {conn_info['username']}  \\
    --password {conn_info['pwd']} \\
    --table {export_tb}  \\
    --input-fields-terminated-by '\\001'  \\
    --hcatalog-database big_data_selection  \\
    --hcatalog-table {hive_tb}  \\
    --input-null-string '\\\\N'  \\
    --input-null-non-string '\\\\N'  \\
    --num-mappers {num_mappers} \\
    --columns {cols}  \\
    --outdir "/tmp/sqoop/"
"""

    @staticmethod
    def build_import_sh_tmp_inner(conn_info: Dict,
                                  query: str,
                                  hive_tb_name: str,
                                  map_num: int = 1,
                                  split_by: str = None
                                  ):
        """
        直接导入到临时内部临时表用于一次性计算用
        :param conn_info:
        :param query:
        :param hive_tb_name:
        :return:
        """
        default_db = 'big_data_selection'
        cmd = f"""
    {CommonUtil.__sqoop_home__} yswg_import -D mapred.job.queue.name=default -D mapred.task.timeout=0 \\
    --connect {conn_info['url']}  \\
    --username {conn_info['username']}  \\
    --password {conn_info['pwd']} \\
    --query "{query}"  \\
    --mapreduce-job-name f"sqoop_task{hive_tb_name}"  \\
    --hcatalog-database {default_db} \\
    --create-hcatalog-table \\
    --hcatalog-table {hive_tb_name} \\
    --fields-terminated-by '\\t'  \\
    --hive-drop-import-delims  \\
    --null-string '\\\\N'  \\
    --null-non-string '\\\\N'  \\
    --m {map_num} \\
    --split-by {split_by} \\
    --outdir "/tmp/sqoop/"
    """
        return cmd

    @staticmethod
    def build_hive_import_sh(site_name: str,
                             db_type: str,
                             query: str,
                             hive_table: str,
                             partition_dict: dict
                             ):
        """
        导入到 hive 内部表指定分区 注意使用 orcfile 格式进行压缩
        """
        default_db = 'big_data_selection'
        conn_info = DBUtil.get_connection_info(db_type, site_name)
        #  对query中的特殊字符自动转义
        query = query.strip()
        query = query.replace("`", r"\`")
        keys = ",".join(partition_dict.keys())
        values = ",".join(partition_dict.values())

        return f"""
    {CommonUtil.__sqoop_home__} yswg_import -D mapred.job.queue.name=default -D mapred.task.timeout=0 \\
    --connect {conn_info['url']}  \\
    --username {conn_info['username']}  \\
    --password {conn_info['pwd']} \\
    --query "{query}"  \\
    --mapreduce-job-name f"sqoop_task{hive_table}"  \\
    --hcatalog-database {default_db} \\
    --hcatalog-table {hive_table} \\
    --hcatalog-partition-keys {keys} \\
    --hcatalog-partition-values {values} \\
    --hcatalog-storage-stanza "stored as orcfile" \\
    --m 1 \\
    --outdir "/tmp/sqoop/"
    """

    @staticmethod
    def build_import_sh(site_name: str,
                        db_type: str,
                        query: str,
                        hdfs_path: str,
                        map_num: int = 1,
                        key: str = None
                        ):
        """
        导入到hdfs外部表
        :param site_name:
        :param db_type:
        :param query:
        :param hdfs_path:
        :param map_num:
        :param key:
        :return:
        """
        conn_info = DBUtil.get_connection_info(db_type, site_name)
        #  对query中的特殊字符自动转义
        query = query.strip()
        query = query.replace("`", r"\`")
        start_name = CommonUtil.get_start_name_from_hdfs_path(hdfs_path)
        if start_name:
            start_name = "sqoop_task:"+start_name
        else:
            start_name = "sqoop_task"
        return f"""
    {CommonUtil.__sqoop_home__} yswg_import -D mapred.job.queue.name=default -D mapred.task.timeout=0  --append \\
    --connect {conn_info['url']}  \\
    --username {conn_info['username']}  \\
    --password {conn_info['pwd']} \\
    --target-dir {hdfs_path}  \\
    --mapreduce-job-name "{start_name}"  \\
    --query "{query}"  \\
    --fields-terminated-by '\\t'  \\
    --hive-drop-import-delims  \\
    --null-string '\\\\N'  \\
    --null-non-string '\\\\N'  \\
    --compress \\
    -m {map_num} \\
    --split-by {key} \\
    --compression-codec lzop  \\
    --outdir "/tmp/sqoop/"
    """

    @staticmethod
    def after_import(hdfs_path: str, hive_tb: str):
        """
        导入hdfs后对hive表进行压缩和分区修复
        :param hdfs_path:
        :param hive_tb:
        :return:
        """
        cmd = rf"""
    {CommonUtil.__hadoop_home__} jar  \
    /opt/module/hadoop/share/hadoop/common/hadoop-lzo-0.4.20.jar  \
    com.hadoop.compression.lzo.DistributedLzoIndexer -Dmapreduce.job.queuename=default -Dmapreduce.framework.name=local\
    {hdfs_path}
    """
        print("lzo 压缩中")
        print(cmd)
        client = SSHUtil.get_ssh_client()
        SSHUtil.exec_command_async(client, cmd, ignore_err=False)
        print(f"修复表{hive_tb}中")
        cmd = rf"""{CommonUtil.__hive_home__} -e "set hive.msck.path.validation=ignore; msck repair table big_data_selection.{hive_tb};" """
        print(cmd)
        SSHUtil.exec_command_async(client, cmd, ignore_err=False)
        client.close()
        pass

    @staticmethod
    def hive_cmd_exec(cmd: str):
        """
        使用命令行直接执行执行hive命令
        """
        import os
        hive_cmd = rf"""{CommonUtil.__hive_home__} -e '{cmd}' """
        print(f"执行hive命令中{hive_cmd}")
        os.system(hive_cmd)
        # client = SSHUtil.get_ssh_client()
        # SSHUtil.exec_command_async(client, hive_cmd, ignore_err=False)
        # client.close()
        pass

    @staticmethod
    def orctable_concatenate(hive_table: str,
                             partition_dict: Dict,
                             innerFlag: bool = False,
                             min_part_num: int = 5,
                             max_retry_time: int = 10):
        # 查看有多少分区小文件
        path = CommonUtil.build_hdfs_path(hive_table, partition_dict, innerFlag)
        part_list = HdfsUtils.read_list(path)
        if part_list is None:
            return
        retry_time = 0
        partition = []
        for key in partition_dict.keys():
            partition.append(f""" {key}="{partition_dict.get(key)}" """)
        default_db = 'big_data_selection'
        partition_str = ",".join(partition)
        while len(part_list) > min_part_num and retry_time <= max_retry_time:
            # 先进行修复
            # CommonUtil.hive_cmd_exec(f"""msck repair table {default_db}.{hive_table};""")
            if len(partition_dict) == 0:
                # 表进行小文件合并
                CommonUtil.hive_cmd_exec(f"""alter table {default_db}.{hive_table} concatenate;""")
            else:
                # 分区进行小文件合并
                CommonUtil.hive_cmd_exec(f"""alter table {default_db}.{hive_table} partition ({partition_str}) concatenate;""")
            part_list = HdfsUtils.read_list(path)
            pass
        pass

    @staticmethod
    def check_schema(spark_session: SparkSession, df_schema: DataFrame, save_tb_name: str, filter_cols: list = None):
        """
                schema验证，可验证数仓中save_table与传入的df的schema的差异
                :param spark_session: spark任务对象
                :param df_schema: 需要比较的df
                :param save_tb_name: 存储表
                :param filter_cols: 不参与比较的字段过滤，不想参与比较的字段可以写在该list中；
                :return:DataFrame：返回有差异的字段数据的DataFrame
        """
        # 基础不比较的过滤字段
        base_filter_cols = ['site_name', 'date_type', 'date_info']
        sql = f"select * from {save_tb_name} limit 0"
        tb_schema = spark_session.sql(sql).schema
        # filter_cols 用于维护不参与对比的字段
        if filter_cols is None:
            filter_cols = base_filter_cols
        else:
            filter_cols = base_filter_cols.extend(filter_cols)
        list1 = []
        list2 = []
        for item in tb_schema.fields:
            if item.name not in filter_cols:
                list1.append((item.name, item.dataType.simpleString()))

        for item in df_schema.schema.fields:
            if item.name not in filter_cols:
                list2.append((item.name, item.dataType.simpleString()))

        df1 = spark_session.createDataFrame(list1, ('name', 'type'))
        df2 = spark_session.createDataFrame(list2, ('name', 'type'))

        show_df = df1.join(df2, "name", how="outer").select(
            df1.name.alias("hive_column"),
            df1.type.alias("hive_column_type"),
            df2.name.alias("df_column"),
            df2.type.alias("df_column_type"),
        ).cache()
        show_df.show(n=300, truncate=False)

        # 筛选出两表不一致字段展示
        show_df_diff = show_df.filter('hive_column is null or df_column is null')
        show_df_diff.show(n=300, truncate=False)
        # 如果为空说明没有不一致字段，则为true，否则有不一致字段为false
        schema_flag = show_df_diff.count() == 0
        return schema_flag

    @staticmethod
    def check_ods_sync_schema(spark_session: SparkSession, import_table: str, db_type: str, site_name: str,
                              hive_table: str, msg_usr: list = __msg_usr__):
        """
        校验ods层schema是否变动--检查的是ods与hive的schema
        :param spark_session: spark任务对象
        :param import_table:  ods层对应导入表
        :param db_type: ods导入链接类型 mysql / pgsql
        :param site_name: 站点
        :param hive_table: 对应导入的hive ods表
        :param msg_usr: 通知人list--不填写则默认群发
        """
        schema_sql = f"select * from {import_table} limit 0"
        conn_info = DBUtil.get_connection_info(db_type, site_name)
        df_schema = SparkUtil.read_jdbc_query(
            session=spark_session,
            url=conn_info["url"],
            pwd=conn_info["pwd"],
            username=conn_info["username"],
            query=schema_sql
        )
        schema_flag = CommonUtil.check_schema(spark_session, df_schema, hive_table)
        # Todo 这里发送消息提醒的验证需要进一步确认
        if not schema_flag:
            msg = f"{hive_table} 与 {import_table} 数据得schema不一致，请查看日志！！  "
            CommonUtil.send_wx_msg(msg_usr, f"\u26A0 {hive_table}同步schema校验异常 \u26A0", msg)
        pass

    @staticmethod
    def check_tb_schema_same(spark_session: SparkSession, tb1_name: str, tb2_name: str):
        """
        检查两个表表结构是不是一样的
        :param tb1_name: 表1
        :param tb2_name: 表2
        :return:
        """
        tb_schema1 = spark_session.sql(f"select * from {tb1_name} limit 0")
        tb_schema2 = spark_session.sql(f"select * from {tb2_name} limit 0")
        list1 = []
        list2 = []
        for i, item in enumerate(tb_schema1.schema.fields):
            list1.append((item.name, item.dataType.simpleString(), i))

        for i, item in enumerate(tb_schema2.schema.fields):
            list2.append((item.name, item.dataType.simpleString(), i))

        df1 = spark_session.createDataFrame(list1, ('name', 'type', "index"))
        df2 = spark_session.createDataFrame(list2, ('name', 'type', "index"))

        show_df = df2.join(df1, "index", how="left").select(
            df2['index'].alias("index"),
            df2.name.alias(f"表2{tb2_name}字段"),
            df2.type.alias(f"表2{tb2_name}类型"),
            df1.name.alias(f"表1{tb1_name}字段"),
            df1.type.alias(f"表1{tb1_name}类型"),
            F.when(df1['name'] == df2['name'], F.lit(1)).otherwise(0).alias("是否一致")
        )

        # 如果最小值返回的为0，则为false：说明有不一致的；如果最小值为1，则为true：说明没有不一致
        schema_flag = bool(show_df.select(F.min("是否一致").alias("result")).first().asDict()['result'])
        if not schema_flag:
            show_df.show(n=300, truncate=False)
        return schema_flag

    @staticmethod
    def check_schema_before_import(db_type: str,
                                   site_name: str,
                                   query: str,
                                   hive_tb_name: str,
                                   msg_usr: list = __msg_usr__,
                                   partition_dict: Dict = None):
        """
        导入前进行原始表数据检查，以及导入query顺序检查
        :param db_type: 原始表db链接类型
        :param site_name: 站点
        :param query: 导入时查询语句
        :param hive_tb_name: 导入的hive表名称
        :param msg_usr: 异常消息通知人
        :param partition_dict: 同步条件dict
        :return: empty_flag、schema_flag
        """
        if partition_dict is not None:
            msg_params = ""
            for key, value in partition_dict.items():
                if value is not None:
                    msg_params += f"{value} "
        else:
            msg_params = ""
        spark_session = SparkUtil.get_spark_session("check_schema")
        rel_query = query.strip()
        rel_query = rel_query.replace(f"and \$CONDITIONS", "")
        if "limit" in rel_query:
            rel_query = rel_query[:rel_query.find("limit")]
        rel_query = f"""{rel_query} limit 1"""

        conn_info = DBUtil.get_connection_info(db_type, site_name)
        import_tb_schema = SparkUtil.read_jdbc_query(
            session=spark_session,
            url=conn_info["url"],
            pwd=conn_info["pwd"],
            username=conn_info["username"],
            query=rel_query
        )

        # 如果为空则为true,否则false
        empty_flag = import_tb_schema.count() == 0
        if empty_flag:
            person_in_charge = ",".join(msg_usr)
            msg = f"任务信息：{hive_tb_name} {msg_params}\n负责人：{person_in_charge}"
            msg_usr = msg_usr + ['chenyuanjie', 'chenjianyun', 'leichao', 'chenbo']
            CommonUtil.send_wx_msg(msg_usr, "\u26A0 数据同步异常", msg)
            spark_session.stop()
            raise Exception(msg)

        sql = f"select * from {hive_tb_name} limit 0"
        tb_schema = spark_session.sql(sql)

        list1 = []
        list2 = []
        for i, item in enumerate(tb_schema.schema.fields):
            list1.append((item.name, item.dataType.simpleString(), i))

        for i, item in enumerate(import_tb_schema.schema.fields):
            list2.append((item.name, item.dataType.simpleString(), i))

        df1 = spark_session.createDataFrame(list1, ('name', 'type', "index"))
        df2 = spark_session.createDataFrame(list2, ('name', 'type', "index"))

        show_df = df2.join(df1, "index", how="left").select(
            df2['index'].alias("index"),
            df2.name.alias(f"导入表字段"),
            df2.type.alias(f"导入表类型"),
            df1.name.alias(f"hive表{hive_tb_name}字段"),
            df1.type.alias(f"hive表{hive_tb_name}类型"),
            F.when(df1['name'] == df2['name'], F.lit(1)).otherwise(0).alias("是否一致")
        )
        show_df.show(n=300, truncate=False)

        # 如果最小值返回的为0，则为false：说明有不一致的；如果最小值为1，则为true：说明没有不一致
        schema_flag = bool(show_df.select(F.min("是否一致").alias("result")).first().asDict()['result'])
        if not schema_flag:
            person_in_charge = ",".join(msg_usr)
            msg = f"任务信息：{hive_tb_name} {msg_params}\n负责人：{person_in_charge}"
            msg_usr = msg_usr + ['chenyuanjie', 'chenjianyun', 'leichao', 'chenbo']
            CommonUtil.send_wx_msg(msg_usr, "\u26A0 数据同步异常", msg)
            spark_session.stop()
            raise Exception(msg)
        spark_session.stop()
        return empty_flag, schema_flag

    @staticmethod
    def check_import_sync_num(db_type: str,
                              partition_dict: Dict,
                              import_query: str,
                              hive_tb_name: str,
                              msg_usr: list = __msg_usr__):
        """
        导入前进行原始表数据检查，以及导入query顺序检查
        :param db_type: 原始表db链接类型
        :param partition_dict: 入参dict
        :param import_query: 导入的原始表查询query
        :param hive_tb_name: 导入的hive表名称
        :param msg_usr: 异常消息通知人
        :return:
        """
        spark_session = SparkUtil.get_spark_sessionV3("check_sync_num")
        site_name = partition_dict.get("site_name")
        conn_info = DBUtil.get_connection_info(db_type, site_name)
        # 根据query解析可以获取真实where 条件
        import_query = import_query.replace(f"and \$CONDITIONS", "").strip()
        table_where_query = import_query.split("from")[1]
        select_count_query = "select count(1) as import_total_num from"
        import_count_sql = select_count_query + table_where_query
        print(import_count_sql)
        import_tb_df = SparkUtil.read_jdbc_query(
            session=spark_session,
            url=conn_info["url"],
            pwd=conn_info["pwd"],
            username=conn_info["username"],
            query=import_count_sql
        )
        import_tb_count = import_tb_df.collect()[0]['import_total_num']

        # 解析partition_dict获取分区查询条件
        partition_conditions = []
        msg_params = ""
        for key, value in partition_dict.items():
            if value is not None:
                partition_conditions.append(f"{key} = '{value}'")
                msg_params += f"{value} "

        # 拼接分区查询语句
        partition_query = f"SELECT count(1) as hive_total_num  FROM {hive_tb_name}"
        if partition_conditions:
            partition_query += f" WHERE {' AND '.join(partition_conditions)}"

        hive_tb_count = spark_session.sql(partition_query).collect()[0]['hive_total_num']

        # 判断两者数量是否一致
        total_num_flag = bool(import_tb_count == hive_tb_count)
        print(f"import_total_num:{import_tb_count}")
        print(f"{hive_tb_name} total_num:{hive_tb_count}")

        if not total_num_flag:
            person_in_charge = ",".join(msg_usr)
            msg = f"任务信息：{hive_tb_name} {msg_params}\n负责人：{person_in_charge}"
            msg_usr = msg_usr + ['chenyuanjie', 'chenjianyun', 'leichao', 'chenbo']
            CommonUtil.send_wx_msg(msg_usr, "\u26A0 数据同步异常", msg)
            spark_session.stop()
            raise Exception(msg)
        spark_session.stop()

    @staticmethod
    def check_fields_and_warning(hive_tb_name: str, partition_dict: Dict):
        """
        对配置表(hive_field_verify_config) 配置的相应表和相应字段进行校验
        :param hive_tb_name:校验表的表名
        :param partition_dict:校验表的分区条件
        :param msg_usr:异常消息通知人
        :return:
        """
        # 获取计算分区
        msg_params = ""
        for key, value in partition_dict.items():
            if value is not None:
                msg_params += f"{value} "
        base_msg = f"{hive_tb_name} {msg_params} "
        site_name = partition_dict.get("site_name")
        date_type = partition_dict.get("date_type")
        spark_session = SparkUtil.get_spark_sessionV3("check_fields_rule")
        # 获取维护的字段验证配置表数据
        config_table_query = f"""select * from hive_field_verify_config 
                                    where table_name ='{hive_tb_name}' 
                                    and site_name = '{site_name}'
                                    and use_flag = 1 """
        conn_info = DBUtil.get_connection_info('mysql', 'us')
        check_field_df = SparkUtil.read_jdbc_query(
            session=spark_session,
            url=conn_info["url"],
            pwd=conn_info["pwd"],
            username=conn_info["username"],
            query=config_table_query
        )
        check_field_list = check_field_df.select('field_name', 'verify_desc', 'verify_type', 'config_json',
                                                 'msg_usr_list').collect()
        if not check_field_list:
            print("============================无验证匹配条件跳过验证===================================")
            return
        for row in check_field_list:
            field_name = row['field_name']
            verify_type = row['verify_type']
            config_json = json.loads(row['config_json'])
            msg_usr = row['msg_usr_list']
            msg_usr_list = [user.strip() for user in msg_usr.split(",")] if msg_usr else []
            if verify_type == "空值率验证":
                query = CommonUtil.generate_null_ratio_query(table_name=hive_tb_name,
                                                             field_name=field_name,
                                                             partition_dict=partition_dict)
                ratio_df = spark_session.sql(query).cache()
                ratio_num = float(ratio_df.collect()[0]['null_ratio'])
                waring_max = float(config_json['max'])
                ratio_df = ratio_df.select(
                    F.col('field_name').alias('校验字段'),
                    F.lit(verify_type).alias('校验类型'),
                    F.col('null_ratio').alias('校验字段空值率'),
                    F.lit(waring_max).alias('空值率阈值'),
                    F.when((F.col('null_ratio') < waring_max), 1).otherwise(0).alias('是否验证通过')
                )
                ratio_df.show(10, truncate=False)

                if ratio_num >= waring_max:
                    # 进行微信推送
                    msg = f"{base_msg} 字段：{field_name}的{verify_type}不通过，请注意该字段的使用！！"
                    CommonUtil.send_wx_msg(msg_usr_list, f"\u26A0 {hive_tb_name} {msg_params}数据{verify_type}异常",
                                           msg)
            elif verify_type == "最大最小值验证":
                query = CommonUtil.generate_min_max_query(table_name=hive_tb_name,
                                                          field_name=field_name,
                                                          partition_dict=partition_dict)
                ratio_df = spark_session.sql(query).cache()
                field_max_vale = float(ratio_df.collect()[0]['max_value'])
                field_min_vale = float(ratio_df.collect()[0]['min_value'])
                waring_max = float(config_json['max'])
                waring_min = float(config_json['min'])
                ratio_df = ratio_df.select(
                    F.col('field_name').alias('校验字段'),
                    F.lit(verify_type).alias('校验类型'),
                    F.col('max_value').alias('校验字段最大值'),
                    F.col('min_value').alias('校验字段最小值'),
                    F.lit(waring_max).alias('最大值上限'),
                    F.lit(waring_min).alias('最小值下限'),
                    F.when((F.col('max_value') <= waring_max) | (F.col('min_value') >= waring_min), 1).otherwise(
                        0).alias('是否验证通过')
                )
                ratio_df.show(10, truncate=False)
                if field_max_vale > waring_max:
                    # 进行微信推送
                    msg = f"{base_msg} 字段：{field_name}的最大值上限验证不通过，请注意该字段的使用！！"
                    CommonUtil.send_wx_msg(msg_usr_list, f"\u26A0 {hive_tb_name} {msg_params}数据{verify_type}异常",
                                           msg)

                if field_min_vale < waring_min:
                    # 进行微信推送
                    msg = f"{base_msg} 字段：{field_name}的最小值下限验证不通过，请注意该字段的使用！！"
                    CommonUtil.send_wx_msg(msg_usr_list, f"\u26A0 {hive_tb_name} {msg_params}数据{verify_type}异常",
                                           msg)
            # elif verify_type == "数据量合法验证":
            #     sql_condition = config_json['sql_condition']
            #     partition_conf_list = config_json['partition_conf']
            #     for conf in partition_conf_list:
            #         conf_site_name = conf["site_name"]
            #         conf_date_type = conf["date_type"]
            #
            #         if site_name == conf_site_name and date_type == conf_date_type:
            #             base_count = conf["base_count"]
            #             break
            #     assert base_count is not None, f"未配置{field_name}验证周期{date_type}的基准值，请检查！"
            #
            #     query = CommonUtil.generate_total_cal_query(table_name=hive_tb_name,
            #                                                 field_name=field_name,
            #                                                 partition_dict=partition_dict,
            #                                                 sql_condition=sql_condition)
            #     ratio_df = spark_session.sql(query).cache()
            #     verify_total_count = int(ratio_df.collect()[0]['verify_total_count'])
            #     waring_max = int(base_count * config_json['max_rate'])
            #     waring_min = int(base_count * config_json['min_rate'])
            #     ratio_df = ratio_df.select(
            #         F.lit(row['verify_desc']).alias('验证描述'),
            #         F.lit(verify_type).alias('验证类型'),
            #         F.col('field_name').alias('校验字段'),
            #         F.col('verify_total_count').alias('校验字段统计值'),
            #         F.lit(waring_max).alias('最大临界值上限'),
            #         F.lit(waring_min).alias('最小临界值下限'),
            #         F.when((F.col('verify_total_count') <= waring_max) | (F.col('verify_total_count') >= waring_min),
            #                F.lit(1)).otherwise(F.lit(0)).alias('是否验证通过')
            #     )
            #
            #     ratio_df.show(10, truncate=False)
            #     if verify_total_count > waring_max:
            #         # 进行微信推送
            #         msg = f"{base_msg} 字段：{field_name}的值{verify_total_count}超出限定最大值：{waring_max}，请注意该字段的使用！！"
            #         CommonUtil.send_wx_msg(msg_usr_list, f"\u26A0 {hive_tb_name} {msg_params}数据{verify_type}异常",
            #                                msg)
            #     if verify_total_count < waring_min:
            #         # 进行微信推送
            #         msg = f"{base_msg} 字段：{field_name}的值：{verify_total_count}低于限定最小值：{waring_min}，请注意该字段的使用！！"
            #         CommonUtil.send_wx_msg(msg_usr_list, f"\u26A0 {hive_tb_name} {msg_params}数据{verify_type}异常",
            #                                msg)
        pass

    @staticmethod
    def format_df_with_template(spark_session: SparkSession, save_df: DataFrame, save_tb_name: str,
                                roundDouble: bool = False):
        """
        insert into 之前对data_frame 进行自动对齐 及 schema检查
        :param spark_session:
        :param save_df:
        :param save_tb_name:
        :param roundDouble: 是否对double字段进行round截取
        :return:
        """
        sql = f"select * from {save_tb_name} limit 0"
        template_df = spark_session.sql(sql)
        if roundDouble:
            round_val = 4
            for field in save_df.schema.fields:
                if field.dataType == DoubleType():
                    col_name = field.name
                    print(f"{col_name}从{field.dataType}保留小数位数为{round_val}中...")
                    save_df = save_df.withColumn(col_name, F.round(F.col(col_name), round_val))

        return template_df.unionByName(save_df, allowMissingColumns=False)

    @staticmethod
    def auto_transfer_type(spark_session: SparkSession, save_df: DataFrame, hive_tb: str, transfer_dict: Dict = None):
        """
        自动进行类型转换 默认对和hive字段类型不同的进行转换，如果是Double类型则自动转为 DecimalType(10, 3)；
        需要特殊处理的传入transfer_dict
        :param spark_session:
        :param save_df:
        :param hive_tb:
        :param transfer_dict:
        :return:
        """
        sql = f"select * from {hive_tb} limit 0"
        tmp_dict = transfer_dict or {}
        tb_schema = spark_session.sql(sql).schema
        for field1 in save_df.schema.fields:
            for field2 in tb_schema.fields:
                col_name = field1.name
                hive_col_name = field2.name
                if col_name == hive_col_name:
                    transfer_flag = (field1.dataType != field2.dataType)
                    transfer_type = field2.dataType
                    if field2.dataType == DoubleType():
                        transfer_type = tmp_dict.get(col_name) or DecimalType(10, 3)
                        transfer_flag = True
                    if transfer_flag:
                        print(f"{col_name}从{field1.dataType}转化为{transfer_type}")
                        save_df = save_df.withColumn(col_name, F.col(col_name).cast(transfer_type))

        return save_df

    @staticmethod
    def select_partitions_df(spark_session: SparkSession, tb_name: str):
        """
        获取表分区df
        """
        df = spark_session.sql(f"show partitions {tb_name}")
        partitions = df.select("partition").rdd.flatMap(lambda x: x).collect()

        values = []
        for index in range(0, len(partitions)):
            item = partitions[index]
            obj = {}
            for sp in item.split("/"):
                val = sp.split("=")
                obj[val[0]] = val[1]
            values.append(obj)

        return spark_session.createDataFrame(values)

    @staticmethod
    def select_col_all(df: DataFrame):
        """
        选择df的所有的列
        """
        return [df[col_name].alias(col_name) for col_name in df.columns]

    @staticmethod
    def df_export_csv(spark_session: SparkSession, export_df: DataFrame, csv_name: str, limit: int = 20 * 10000):
        # output 不进行压缩
        compress_flag = spark_session.conf.get("mapred.output.compress")
        spark_session.sql("set mapred.output.compress=false")
        csv_path = f"/tmp/csv/{csv_name}"
        # 最多导出20w行
        tmp_export_df = export_df.limit(limit)
        tmp_export_df.repartition(1).write.mode("overwrite").option("header", True).csv(csv_path)
        # 合并为一个文件
        client = HdfsUtils.get_hdfs_cilent()
        src_path = list(filter(lambda path: str(path).endswith("csv"), client.list(csv_path)))[0]
        rel_path = f"{csv_path}.csv"
        client.delete(rel_path, True)
        client.rename(f"{csv_path}/{src_path}", rel_path)
        client.delete(csv_path, True)
        print("======================csv生成hdfs文件路径如下======================")
        print(rel_path)
        spark_session.sql(f"set mapred.output.compress={compress_flag}")
        return rel_path

    @classmethod
    def transform_week_tuple(cls, spark_session: SparkSession, date_type: str, date_info: str):
        """
        对周流程进行日期转换，返回日期元祖：如传入month，则返回该月下所有的周
        周流程的week元祖获取
        :param spark_session: spark对象
        :param date_type: 日期类型date_type
        :param date_info: 具体日期date_info
        :return: complete_date_info_tuple: 周数据元祖
        """
        complete_date_info_tuple = tuple()
        df_date = spark_session.sql(f"select * from dim_date_20_to_30 ;")
        df = df_date.toPandas()
        if date_type == 'week':
            complete_date_info_tuple = f"('{date_info}')"
        elif date_type == '4_week':
            print(date_info)
            df_loc = df.loc[(df.year_week == f"{date_info}") & (df.week_day == 1)]
            cur_id = list(df_loc.id)[0]
            df_loc = df.loc[df.id == int(cur_id)]
            week1 = list(df_loc.year_week)[0]
            df_loc = df.loc[df.id == int(cur_id) - 7]
            week2 = list(df_loc.year_week)[0]
            df_loc = df.loc[df.id == int(cur_id) - 14]
            week3 = list(df_loc.year_week)[0]
            df_loc = df.loc[df.id == int(cur_id) - 21]
            week4 = list(df_loc.year_week)[0]
            complete_date_info_tuple = (week1, week2, week3, week4)
        elif date_type == 'month':
            df_loc = df.loc[(df.year_month == f"{date_info}") & (df.week_day == 1)]
            complete_date_info_tuple = tuple(df_loc.year_week)
        print("complete_date_info_tuple:", complete_date_info_tuple)
        return complete_date_info_tuple

    @classmethod
    def create_tmp_tb(cls, spark_session: SparkSession, ddl: str, tb_name: str, drop_exist: bool = False):
        # 默认执行ddl创建表会生成 spark.sql.sources.schema.numParts 语句需要删除
        if drop_exist:
            print(f"drop table {tb_name}")
            spark_session.sql(f"drop table if exists  {tb_name}")
        print(f"创建临时表中：ddl sql 为")
        print(ddl)
        spark_session.sql(ddl)

        sql = f'show tblproperties {tb_name};'
        tblproperties_df = spark_session.sql(sql)
        print(tblproperties_df)
        keys = tblproperties_df.select("key").rdd.flatMap(lambda ele: ele).collect()
        del_key = []
        for key in keys:
            if str(key).startswith("spark.sql.create.version") or str(key).startswith("spark.sql.sources.schema"):
                del_key.append(f"'{key}'")
        if len(del_key) > 0:
            del_sql = f"""alter table {tb_name} unset tblproperties ({",".join(del_key)});"""
            spark_session.sql(del_sql)
        return True

    @classmethod
    def save_or_update_table(cls, spark_session: SparkSession,
                             hive_tb_name: str,
                             partition_dict: Dict,
                             df_save: DataFrame,
                             drop_exist_tmp_flag=True
                             ):
        """
        插入或更新表的分区
        :param spark_session:
        :param hive_tb_name:实际保存表名
        :param partition_dict:
        :param df_save:
        :param drop_exist_tmp_flag:  是否创建表前先删除临时表 如果不删除则在备份表插入分区数据
        :return:
        """
        partition_by = list(partition_dict.keys())
        if HdfsUtils.path_exist(CommonUtil.build_hdfs_path(hive_tb_name, partition_dict)):
            table_copy = f"{hive_tb_name}_copy"
            CommonUtil.create_tmp_tb(
                spark_session,
                ddl=f"""create table if not exists {table_copy} like {hive_tb_name}""",
                tb_name=table_copy,
                drop_exist=drop_exist_tmp_flag
            )

            print(f"当前存储的临时表名为：{table_copy},分区为{partition_by}", )

            if not drop_exist_tmp_flag:
                flag = CommonUtil.check_tb_schema_same(spark_session, tb1_name=hive_tb_name, tb2_name=table_copy)
                if not flag:
                    raise Exception(f"{table_copy}表结构同{hive_tb_name}不一致，交换分区后可能存在错位现象，请检查！！")

            # 插入前先删除copy表的数据再save到cp表
            path = CommonUtil.build_hdfs_path(hive_tb=table_copy, partition_dict=partition_dict, innerFlag=True)
            if HdfsUtils.path_exist(path):
                HdfsUtils.delete_hdfs_file(path)

            df_save.write.saveAsTable(name=table_copy, format='hive', mode='append', partitionBy=partition_by)
            # 交换表名
            CommonUtil.exchange_partition_data(
                spark_session=spark_session,
                tb_src=hive_tb_name,
                partition_dict_src=partition_dict,
                tb_target=table_copy,
                partition_dict_target=partition_dict
            )
        else:
            # 不存在则直接插入
            df_save.write.saveAsTable(name=hive_tb_name, format='hive', mode='append', partitionBy=partition_by)
        print("success")
        pass

    @classmethod
    def exchange_partition_data(cls, spark_session: SparkSession,
                                tb_src: str,
                                partition_dict_src: Dict,
                                tb_target: str,
                                partition_dict_target: Dict,
                                ):
        """
        交换两个分区表数据
        :param spark_session: spark_session
        :param tb_src: 分区表A
        :param partition_dict_src: 分区dict
        :param tb_target:分区表B
        :param partition_dict_target:分区dict
        :return:
        """

        location1: str = spark_session.sql(
            f"""describe formatted {tb_src};"""
        ).where("col_name = 'Location' ").first().asDict().get("data_type").replace("hdfs://nameservice1:8020", "")

        location2: str = spark_session.sql(
            f"""describe formatted {tb_target};"""
        ).where("col_name = 'Location' ").first().asDict().get("data_type").replace("hdfs://nameservice1:8020", "")

        for key in partition_dict_src.keys():
            location1 = location1 + f"/{key}={partition_dict_src.get(key)}"

        for key in partition_dict_target.keys():
            location2 = location2 + f"/{key}={partition_dict_target.get(key)}"

        assert HdfsUtils.path_exist(location1), f"分区【{location1}】不存在！"
        assert HdfsUtils.path_exist(location2), f"分区【{location2}】不存在！"

        HdfsUtils.exchange_path(path_one=location1, path_two=location2)
        return True

    @classmethod
    def get_next_val(cls, date_type: str, date_info: str):
        """
        根据时间类型获取下一个值
        :param date_type:
        :param date_info:
        :return:
        """
        if date_type == DateTypes.day.name:
            result = cls.get_day_offset(date_info, 1)

        elif date_type == DateTypes.week.name:
            engine = DBUtil.get_db_engine("mysql", "us")
            with engine.connect() as connection:
                sql = f"""
                select year_week
                from date_20_to_30
                where year_week > '{date_info}'
                order by year_week
                limit 1  """
                print("================================执行sql================================")
                print(sql)
                result = connection.execute(sql)
                next_week = result.cursor.fetchone()[0]
                result = next_week


        elif date_type == DateTypes.month.name:
            result = cls.get_month_offset(date_info, 1)

        else:
            raise Exception(f"时间类型{date_type}不支持")

        return result

    @classmethod
    def build_ddl_form_df(cls, df: DataFrame, partition_list: list, tb_name: str):
        """
        df 生成 ddl sql
        :param df:
        :param partition_list:
        :param tb_name:
        :return:
        """

        df.schema.fieldNames()
        part = partition_list

        type_dict = {
            DoubleType(): "double",
            DecimalType(): "double",
            StringType(): "string",
            LongType(): "int",
            IntegerType(): "int",
        }
        line1 = []
        line2 = []
        for field in df.schema.fields:
            type = type_dict.get(field.dataType)

            if field.name in part:
                line2.append(f"{field.name}         {type}")
            else:
                line1.append(f"{field.name}         {type}")

        str1 = ",\n".join(line1)
        str2 = ",\n".join(line2)

        ddl = f"""
create table {tb_name}
(
{str1}
)
partitioned by
(
{str2} 
)
row format serde 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'
stored as
inputformat 'com.hadoop.mapred.DeprecatedLzoTextInputFormat'
outputformat 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
    """
        return ddl

    @classmethod
    def get_rel_date_type(cls, tb_name: str, date_type: str):
        assert tb_name is not None, "表名不能为空！"
        # 需要特殊处理的就算这些表支持 month_old 其他的不支持
        tmp_list = [
            'dim_st_detail',
            'dwd_st_measure',
            'dwd_st_asin_measure',
            'dwd_asin_measure',
            'dwd_st_volume_fba',
            'dwt_st_market',
            'dws_st_num_stats',
            'dwt_aba_st_analytics'
        ]
        # if date_type in ['month_old'] and date_info < '2023-10':
        if date_type in ['month_old'] and tb_name not in tmp_list:
            return 'month'
        return date_type

    @staticmethod
    def build_hdfs_path(hive_tb: str, partition_dict: Dict = None, innerFlag: bool = False):
        """
        构建对应的表名称
        :param hive_tb:
        :param partition_dict:
        :param innerFlag:
        :return:
        """
        suffix = ""
        if partition_dict is not None:
            tmp = []
            for key in partition_dict.keys():
                tmp.append(f"{key}={partition_dict.get(key)}")
            suffix += "/".join(tmp)

        hdfs_path = None

        if innerFlag:
            if partition_dict is not None:
                hdfs_path = f"/user/hive/warehouse/big_data_selection.db/{hive_tb}/{suffix}"
            else:
                hdfs_path = f"/user/hive/warehouse/big_data_selection.db/{hive_tb}"
        else:
            allow = ['ods', 'dim', 'dwd', 'dws', 'dwt', 'tmp']
            prefix = None
            for tmp in allow:
                if hive_tb.startswith(tmp):
                    prefix = tmp
                pass
            assert prefix is not None, f"{hive_tb}表名不合规，请检查！"
            if partition_dict is not None:
                hdfs_path = f"/home/big_data_selection/{prefix}/{hive_tb}/{suffix}"
            else:
                hdfs_path = f"/home/big_data_selection/{prefix}/{hive_tb}"

        return hdfs_path

    @staticmethod
    def send_wx_msg(users: list, title: str, content: str, msgtype: str = "textcard" ):
        """
        通过选品wx消息推送接口，推送消息到oa
        :param users: 填写需要推送的微信用户名list
        :param title: 推送的标题（如果msgtype采用markdown形式，则不附带标题）
        :param content: 推送的主体内容
        :param msgtype: 推送的消息类型（textcard：默认卡片类型；markdown：markdaown结构）
        """
        if users is not None:
            accounts = ",".join(users)
            # 排除users_list=[''] 无需发送
            if bool(accounts):
                host = "http://120.79.147.190:8080"
                url = f'{host}/soundasia_selection/dolphinScheduler/sendMessage'
                data = {
                    'account': accounts,
                    'title': title,
                    'content': content,
                    'msgtype': msgtype
                }
                try:
                    requests.post(url=url, data=data, timeout=15)
                except:
                    pass
        return True

    @classmethod
    def print_hive_ddl(cls,
                       db_type: str,
                       site_name: str,
                       from_tb: str,
                       hive_tb: str,
                       partition_dict: Dict
                       ):

        engine = DBUtil.get_db_engine(db_type, site_name)

        hdfs_path = cls.build_hdfs_path(hive_tb, partition_dict)
        cols = []
        with engine.connect() as connection:
            sql = f"""
                        select a.attname                                                                           col_name,
                        d.description                                                                              col_desc,
                        concat_ws('', t.typname, SUBSTRING(format_type(a.atttypid, a.atttypmod) from '\(.*\)')) as col_type
                        from pg_class c
                        left join pg_attribute a on a.attrelid = c.oid
                        left join pg_type t on t.oid = a.atttypid
                        left join pg_description d on d.objoid = a.attrelid and d.objsubid = a.attnum
                        where 1 = 1
                        and a.attnum > 0
                        and c.relname in (select tablename from pg_tables where schemaname = 'public')
                        and c.relname = '{from_tb}'
                        and t.typname is not null
                        order by c.relname, a.attnum;
                        """
            for row in list(connection.execute(sql)):
                col_name = row['col_name']
                col_desc = row['col_desc']
                col_type = row['col_type']

                if "int" in col_type:
                    hive_col_type = 'int'
                elif "varchar" in col_type or "text" in col_type:
                    hive_col_type = 'string'
                elif "numeric" in col_type:
                    hive_col_type = 'double'
                elif "float8" in col_type:
                    hive_col_type = 'double'
                elif "date" in col_type:
                    hive_col_type = 'string'
                elif "timestamp" in col_type:
                    hive_col_type = 'string'
                else:
                    hive_col_type = 'string'
                cols.append(f"{col_name}\t{hive_col_type}\tcomment\t'{col_desc}'")
            print("================================执行sql================================")

        partitioned_by = []
        for key in partition_dict.keys():
            partitioned_by.append(f"{key} string comment 'you comment' ")

        col_str = ",\n".join(cols)

        ddl = f"""
        create external table {hive_tb}
        (
            {col_str}
        )
            partitioned by ({",".join(partitioned_by)})
            row format serde 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'
            stored as
                inputformat 'com.hadoop.mapred.DeprecatedLzoTextInputFormat'
                outputformat 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
            location 'hdfs://nameservice1:8020{hdfs_path}';
        
        alter table {hive_tb}
            set tblproperties ('comment' = '{hive_tb}表注释');
"""
        print(ddl)
        return ddl

    @staticmethod
    def drop_part(hive_tb: str, partition_dict: Dict):
        """
        删除hive分区数据默认是外部表 不仅仅是删除数据还是删除hive的分区
        :param hive_tb:
        :param partition_dict:
        :return:
        """
        tmparr = []
        for key in partition_dict.keys():
            tmparr.append(f"{key} = '{partition_dict.get(key)}'")
        part_str = ','.join(tmparr)
        hive_ddl = f"""alter table big_data_selection.{hive_tb} drop if exists partition ({part_str});"""
        cmd = rf"""{CommonUtil.__hive_home__} -e "{hive_ddl}" """
        print(f"=============================删除分区中==============================")
        print(cmd)
        client = SSHUtil.get_ssh_client()
        SSHUtil.exec_command_async(client, cmd, ignore_err=False)
        client.close()
        path = CommonUtil.build_hdfs_path(hive_tb=hive_tb, partition_dict=partition_dict)
        print(f"=============================删除分区数据中==============================")
        print(path)
        HdfsUtils.delete_hdfs_file(path)
        pass

    @staticmethod
    def generate_null_ratio_query(table_name: str, field_name: str, partition_dict: Dict):
        """
        构建空值率计算query
        :param table_name: hive表名称
        :param field_name: 需要验证空值率字段
        :param partition_dict:校验表的分区条件
        :return: query: 计算空值率的query
        """

        # 计算空值率sql
        query = f"""SELECT '{field_name}' AS field_name, 
                    COUNT(1) AS total_count, 
                    COUNT(CASE WHEN {field_name} IS NULL THEN 1 WHEN {field_name} = -1 THEN 1 END) AS null_count, 
                    COUNT(CASE WHEN {field_name} IS NOT NULL THEN 1 WHEN {field_name} != -1 THEN 1 END) AS not_null_count, 
                    ROUND(COUNT(CASE WHEN {field_name} IS NULL THEN 1 WHEN {field_name} = -1 THEN 1 END)  / COUNT(1), 4) AS null_ratio 
                    FROM {table_name} """

        # 解析partition_dict获取分区查询条件
        partition_conditions = []
        for key, value in partition_dict.items():
            if value is not None:
                partition_conditions.append(f"{key} = '{value}'")
        # 拼接where条件
        if partition_conditions:
            query += f" WHERE {' AND '.join(partition_conditions)}"
        return query

    @staticmethod
    def generate_min_max_query(table_name: str, field_name: str, partition_dict: Dict):
        """
        构建最大最小值计算query
        :param table_name: hive表名称
        :param field_name: 需要验证空值率字段
        :param partition_dict:校验表的分区条件
        :return: query: 计算空值率的query
        """
        query = f"SELECT '{field_name}' AS field_name, " \
                f"MIN({field_name}) AS min_value, " \
                f"MAX({field_name}) AS max_value " \
                f"FROM {table_name}"

        # 解析partition_dict获取分区查询条件
        partition_conditions = []
        for key, value in partition_dict.items():
            if value is not None:
                partition_conditions.append(f"{key} = '{value}'")
        # 拼接where条件
        if partition_conditions:
            query += f" WHERE {' AND '.join(partition_conditions)}"
        return query

    @staticmethod
    def generate_total_cal_query(table_name: str, field_name: str, partition_dict: Dict, sql_condition: str):
        """
        计算带条件判断的单字段总数query
        :param table_name: hive表名称
        :param field_name: 需要验证空值率字段
        :param partition_dict:校验表的分区条件
        :param sql_condition:其他过滤条件补充
        :return: query: 计算返回使用query
        """
        query = f"SELECT '{field_name}' AS field_name, " \
                f"count({field_name}) AS verify_total_count " \
                f"FROM {table_name}"

        # 解析partition_dict获取分区查询条件
        partition_conditions = []
        for key, value in partition_dict.items():
            if value is not None:
                partition_conditions.append(f"{key} = '{value}'")
        # 拼接where条件
        if partition_conditions:
            query += f" WHERE {' AND '.join(partition_conditions)}"

        # 拼接外部查询条件
        if sql_condition:
            query += f" AND {sql_condition} "

        return query

    @staticmethod
    def judge_is_work_hours(site_name: str = 'us', date_type: str = None, date_info: str = None,
                            principal: str = "wangrui4, huangjian", priority: int = 1, export_tools_type: int = 1,
                            belonging_to_process: str = None):
        """
          导出任务时间约束：控制数据导出任务在非上班时间段进行
        :param site_name: 站点
        :param date_type: 调度类型
        :param date_info: 调度周期
        :param principal: 流程维护人员（与企业微信对应）
        :param priority:  优先级（耗时短的给小的数字，耗时长的给大的数字，非反查搜索词≤3）
        :param export_tools_type: 导出工具（1：sqoop, 2:elasticsearch）
        :param belonging_to_process: 所属流程
        :return:
        """
        exec_env = "/opt/module/anaconda3/envs/pyspark/bin/python3.8"
        # 获取流程的id，看该任务是否流程启动
        process_id = None
        ds_result = DolphinschedulerHelper.get_running_process_task()
        if CommonUtil.notNone(ds_result):
            process_id = str(ds_result.get('process_df_id'))
        #  获取最后一个参数判断是否使用测试导入
        test_flag = CommonUtil.get_sys_arg(len(sys.argv) - 1, None)
        if test_flag == 'test' or process_id is None:
            print("为测试导出或者本地导出，无需监控！")
            return
        # 获取脚本名称
        script_name = sys.argv[0].split("/")[-1]
        print("当前执行的脚本是：" + script_name)
        script_path = ''
        if export_tools_type == 1:
            script_path = "/opt/module/spark/demo/py_demo/sqoop_export/" + script_name
        elif export_tools_type == 2:
            script_path = "/opt/module/spark/demo/py_demo/export_es/" + script_name
        arguments = sys.argv[1:]
        arguments_str = ''
        if len(arguments) > 0:
            arguments_str = ' '.join(arguments)
        # 获取执行命令
        commands = exec_env + " " + script_path + " " + arguments_str
        # 拼接流程命名，进行后期检查导出任务分组
        process_name = process_id if belonging_to_process is None else process_id + "|" + belonging_to_process
        print("执行脚本命令的语句是：" + commands)
        exec_sql = f"""
                    INSERT INTO export_command_records
                    (site_name, date_type, date_info, script_name, commands, status, principal, priority, export_tools_type, belonging_to_process)
                    VALUES
                      ('{site_name}', '{date_type}', '{date_info}', '{script_name}', '{commands}', 1, '{principal}', {priority}, {export_tools_type}, '{process_name}')
                    ON CONFLICT (commands) DO UPDATE
                    SET 
                        site_name = excluded.site_name,
                        date_type = excluded.date_type,
                        date_info = excluded.date_info,
                        script_name = excluded.script_name,
                        status = excluded.status,
                        principal = excluded.principal,
                        priority = excluded.priority,
                        export_tools_type = excluded.export_tools_type,
                        belonging_to_process = excluded.belonging_to_process;
                      """
        print("exec_sql:" + exec_sql)
        DBUtil.exec_sql("postgresql_cluster", "us", exec_sql)
        sys.exit(0)

    @staticmethod
    def modify_export_workflow_status(update_workflow_sql: str, site_name: str = 'us',
                                      date_type: str = None,
                                      date_info: str = None):
        """
        根据流程名称检查导出流程是否完成，更改workflow工作流工具类
        :param update_workflow_sql: 更改workflow工作流的更新语句
        :param site_name: 站点名称
        :param date_type: 日期维度类型
        :param date_info: 日期
        :return:
        """

        #  获取最后一个参数判断是否使用测试导入
        test_flag = CommonUtil.get_sys_arg(len(sys.argv) - 1, None)
        if test_flag == 'test':
            print("测试导出，无需监控！")
            return
        mysql_engine = DBUtil.get_db_engine('mysql', 'us')
        pg_engine = DBUtil.get_db_engine('postgresql_cluster', 'us')
        # 获取脚本名称
        script_name = sys.argv[0].split("/")[-1]
        get_process_sql = f"""select belonging_to_process from export_command_records 
                                        where site_name='{site_name}' 
                                        and date_type = '{date_type}' 
                                        and date_info = '{date_info}' 
                                        and script_name = '{script_name}'
                                        and status != 3
                                          """
        with pg_engine.connect() as connection:
            result = connection.execute(get_process_sql).first()
            belonging_to_process = result['belonging_to_process'] if result else None
        if belonging_to_process is None:
            print("export_command_records库中未记录该流程，无需监控！")
            return
        exec_sql = f"""update export_command_records set status = 3 
                        where script_name='{script_name}' 
                        and belonging_to_process = '{belonging_to_process}'
                                          """
        DBUtil.engine_exec_sql(pg_engine, exec_sql)
        # 检查导出脚本是否都已完成
        check_process_sql = f"""select count(script_name) as uncompleted_num from export_command_records 
                                  where belonging_to_process = '{belonging_to_process}'
                                  and status != 3
                                  """
        with pg_engine.connect() as connection:
            uncompleted_num = connection.execute(check_process_sql).scalar()

        # 看是否导出都已经完成，如果不为3(成功的数量)都为0了，则说明全部导完
        if int(uncompleted_num) == 0:
            print("执行流程更改:" + exec_sql)
            assert update_workflow_sql is not None, "流程更新语句不能为空！！请检查！"
            DBUtil.engine_exec_sql(mysql_engine, update_workflow_sql)
        else:
            print("当前流程下仍有脚本未执行完成！暂未更改流程状态！")
        # 关闭连接
        mysql_engine.dispose()
        pg_engine.dispose()

    @staticmethod
    def judge_not_working_hour():
        """
        判断当前时间是不是用户上班时间
        :return:
        """
        from datetime import datetime
        now = datetime.now()
        hour_minute = CommonUtil.format_now("%H:%M")
        if now.weekday() + 1 in [1, 2, 3, 4, 5]:
            if ('08:20' <= hour_minute <= '12:35') or ('13:30' <= hour_minute <= '18:50'):
                return False
            return True
        else:
            return True

    @staticmethod
    def convert_singular_plural(word):
        """
        单词单复数转换，与java一致的转换词库textblob库支持
        :param word: 需要进行单复数转换的词
        :return:convert_word 根据传入的单词word解析单复数词性后，返回转换词
        """
        if not word:
            return None

        word_object = Word(word)
        singular_form = word_object.singularize()
        plural_form = word_object.pluralize()
        # 判断word到底原始词性是单数还是复数，决定返回转换后的值
        convert_word = plural_form if word == singular_form else singular_form
        return convert_word

    @staticmethod
    def list_build_sqlin_condition(word_list):
        """
        将list转换成 where condition in (a,x,c) 子条件语句
        :param word_list: 需要组装成in语句的列表
        :return: condition: 组装好后的 in的子条件语句如 (a,b,x)
        """
        # 检查列表长度
        if len(word_list) == 1:
            condition = f"'({word_list[0]})'"
        else:
            condition = tuple(word_list)
        return condition

    @staticmethod
    def get_start_name_from_hdfs_path(hdfs_path: str):
        locate_index = ['ods', 'dim', 'dwd', 'dws', 'dwt', 'tmp']
        hidden_param = ['site_name', 'date_type', 'date_info']
        for locate_word in locate_index:
            word_index = hdfs_path.find(locate_word)
            if word_index != -1:
                break
        if word_index != -1:
            content_extract = hdfs_path[word_index + 4:]  # 4 是 "xxx/" 的长度
            content_extract = content_extract.replace("/", ":")
            for hidden_word in hidden_param:
                content_extract = content_extract.replace(hidden_word + "=", "")
            return content_extract
        else:
            return None

    @classmethod
    def get_asin_variant_attribute(cls, df_asin_detail: DataFrame, df_asin_measure: DataFrame, partition_num: int=80, use_type: int=0):
        """
        Param df_asin_detail: asin详情DataFrame(
            字段要求：
                必须有asin，
                asin_vartion_list(Kafka中有，schema参照：StructField("asin_vartion_list", ArrayType(ArrayType(StringType()), True), True)),
                buy_sales(kafka中有，schema参照： StructField("buy_sales", StringType(), True))

        );
        Param df_asin_measure: asin度量信息DataFrame(
            字段要求：
                必须有asin、asin_zr_counts, asin_adv_counts， asin_st_counts, asin_amazon_orders,
                asin_zr_flow_proportion, asin_ao_val
        );
        Param partition_num: 运行并行度（根据脚本运行资源设置）
        Param use_type: 使用类型（0：默认，插件； 1：流量选品）
        return :
                1. dwd_asin_measure必须携带的：
                    asin、asin_zr_counts, asin_adv_counts， asin_st_counts, asin_amazon_orders, asin_zr_flow_proportion, asin_ao_val
                2. 处理得到的：
                    matrix_ao_val, matrix_flow_proportion, asin_amazon_orders, variant_info(变体asin列表)
                3. 流量选品特定的： color, size, style
                4. dwd_asin_measure自行携带的字段
        """
        # 1.关联获取ao、各类型数量、流量占比信息、月销信息等
        df_asin_detail = df_asin_detail.repartition(partition_num)
        df_asin_measure = df_asin_measure.repartition(partition_num)
        df_asin_detail = df_asin_detail.join(
            df_asin_measure, on=['asin'], how='left'
        )
        # 2.解析亚马逊月销信息
        df_asin_detail = df_asin_detail.withColumn(
            "bought_month",
            F.when(F.col("buy_sales").isNotNull(), cls.u_parse_amazon_orders(F.col("buy_sales"))).otherwise(F.lit(None))
        )
        df_asin_detail = df_asin_detail.withColumn("asin_amazon_orders", F.coalesce(F.col("bought_month"), F.col("asin_amazon_orders"))).drop("bought_month")
        # 3.统计母体ao和流量占比
        df_with_variant_attribute = df_asin_detail.filter(F.expr("size(asin_vartion_list) > 0"))
        df_explode_variant_attribute = df_with_variant_attribute.select(
            "asin", F.explode("asin_vartion_list").alias("variant_attribute")
        ).select(
            "asin", F.col("variant_attribute")[0].alias("variant_asin"), F.col("variant_attribute")[1].alias("color"),
            F.col("variant_attribute")[3].alias("size"), F.col("variant_attribute")[5].alias("style")
        )
        df_variant_asin_detail = df_asin_measure.select(F.col("asin").alias("variant_asin"), "asin_zr_counts", "asin_adv_counts", "asin_st_counts")
        df_explode_variant_attribute = df_explode_variant_attribute.repartition(partition_num)
        df_explode_variant_attribute_detail = df_explode_variant_attribute.join(
            df_variant_asin_detail, on=["variant_asin"], how="inner"
        )
        df_explode_variant_attribute_agg = df_explode_variant_attribute_detail.groupby(['asin']).agg(
            F.sum("asin_zr_counts").alias("sum_zr_counts"),
            F.sum("asin_adv_counts").alias("sum_adv_counts"),
            F.sum("asin_st_counts").alias("sum_st_counts"),
            F.collect_set(F.col("variant_asin")).alias("variant_info")
        )
        df_explode_variant_attribute_agg = df_explode_variant_attribute_agg.repartition(partition_num)
        df_explode_variant_attribute_agg = df_explode_variant_attribute_agg.withColumn(
            "matrix_flow_proportion",
            F.when(F.col("sum_st_counts").isNotNull(), F.round(F.col("sum_zr_counts") / F.col("sum_st_counts"), 4))
        ).withColumn(
            "matrix_ao_val",
            F.when(F.col("sum_zr_counts").isNotNull(), F.round(F.col("sum_adv_counts") / F.col("sum_zr_counts"), 3))
        ).drop("sum_zr_counts", "sum_adv_counts", "sum_st_counts")
        df_asin_detail = df_asin_detail.join(
            df_explode_variant_attribute_agg, on=['asin'], how='left'
        )
        df_asin_detail = df_asin_detail.withColumn(
            "matrix_ao_val", F.coalesce(F.col("matrix_ao_val"), F.col("asin_ao_val"))
        ).withColumn(
            "matrix_flow_proportion", F.coalesce(F.col("matrix_flow_proportion"), F.col("asin_zr_flow_proportion"))
        )
        # 4.解析变体属性信息（颜色、 尺寸、 风格等）
        if use_type == 1:
            df_asin_attribute = df_explode_variant_attribute.filter(F.col("asin") == F.col("variant_asin")).drop("variant_asin")
            df_asin_detail = df_asin_detail.join(
                df_asin_attribute, on=['asin'], how='left'
            )
        return df_asin_detail

    @staticmethod
    def unified_variant_asin_basic_detail(df_asin_detail: DataFrame, columns_list: list, partition_num: int=80, use_type: int=0):
        """
            Param: df_asin_detail   每批次ASIN详情数据;
            Param: columns_list     变体ASIN间共用属性字段（根据场景传入指定的字段）;
            Param: partition_num    分区数（根据任务运行申请的资源配置）
            Parm:  use_type         使用场景：0：插件，1：流量选品
            Return: df_asin_detail  ASIN详情数据
                    df_latest_asin_detail_with_parent   每一批ASIN详情数据中最新的变体ASIN公用属性信息
        """
        if use_type == 0:
            df_asin_detail = df_asin_detail.withColumnRenamed("parentAsin", "parent_asin")
        df_asin_detail = df_asin_detail.repartition(partition_num)
        # 将公用属性字段切换名称，防止冲突
        renamed_columns = [F.col(c).alias(f"new_{c}") for c in columns_list]
        df_with_parent_asin = df_asin_detail.filter("parent_asin is not null").select("parent_asin", "asinUpdateTime", *renamed_columns)
        # 获取每一批ASIN详情数据中有parent_asin信息且最新爬取的ASIN详情作为共用属性
        parent_asin_window = Window.partitionBy("parent_asin").orderBy(F.desc_nulls_last("asinUpdateTime"))
        df_with_parent_asin = df_with_parent_asin.withColumn("ct_rank", F.row_number().over(window=parent_asin_window))
        df_with_parent_asin = df_with_parent_asin.repartition(partition_num)
        df_latest_asin_detail_with_parent = df_with_parent_asin.filter("ct_rank = 1").drop("ct_rank", "asinUpdateTime")
        # 将每一批ASIN详情数据中相同变体ASIN之间属性统一
        df_asin_detail = df_asin_detail.join(df_latest_asin_detail_with_parent, on=['parent_asin'], how='left')
        # 丢弃原有属性字段，使用统一后的属性
        for column in columns_list:
            df_asin_detail = df_asin_detail.withColumn(column, F.coalesce(F.col(f"new_{column}"), F.col(column))).drop(f"new_{column}")
            df_latest_asin_detail_with_parent = df_latest_asin_detail_with_parent.withColumnRenamed(f"new_{column}", f"{column}")
        return df_asin_detail, df_latest_asin_detail_with_parent