import os
import sys

sys.path.append(os.path.dirname(sys.path[0]))
from utils.ssh_util import SSHUtil
from utils.common_util import CommonUtil
from utils.hdfs_utils import HdfsUtils

if __name__ == '__main__':
    st_type_list = ['zr', 'sb', 'sp', 'ac', 'bs', 'hr']
    for st_type in st_type_list:
        hive_tb = f"ods_merchantwords_search_term_{st_type}"
        partition_dict = {
            "site_name": 'us',
            "date_type": 'day',
            "date_info": '2024-05-06'
        }
        if st_type in ["zr", "sp"]:
            cols = "search_term,asin,page,page_row,id,created_time,updated_time"
        elif st_type in ["sb"]:
            cols = "search_term,asin,page,data_type,id,created_time,updated_time"
        else:
            cols = "search_term,asin,page,created_time,updated_time"
        db_type = 'postgresql_16'
        import_tb = f"us_merchantwords_search_term_rank_{st_type}"
        query = f"""
            select 
                {cols}
            from 
                {import_tb}
            where 1 = 1
            and \$CONDITIONS
        """
        hdfs_path = CommonUtil.build_hdfs_path(hive_tb, partition_dict=partition_dict)
        print(f"hdfs_path is {hdfs_path}")
        empty_flag, check_flag = CommonUtil.check_schema_before_import(db_type=db_type,
                                                                       site_name='us',
                                                                       query=query,
                                                                       hive_tb_name=hive_tb,
                                                                       msg_usr=['chenyuanjie'],
                                                                       partition_dict=partition_dict
                                                                       )
        assert check_flag, f"导入hive表{hive_tb}表结构检查失败！请检查query是否异常！！"
        if not empty_flag:
            if st_type in ["zr", "sp", "sb"]:
                if st_type == "zr":
                    map_num = 50
                elif st_type == "sp":
                    map_num = 25
                else:
                    map_num = 15
                sh = CommonUtil.build_import_sh(site_name='us',
                                                db_type=db_type,
                                                query=query,
                                                hdfs_path=hdfs_path,
                                                map_num=map_num,
                                                key="id")
            else:
                sh = CommonUtil.build_import_sh(site_name='us',
                                                db_type=db_type,
                                                query=query,
                                                hdfs_path=hdfs_path)
            HdfsUtils.delete_hdfs_file(hdfs_path)
            client = SSHUtil.get_ssh_client()
            SSHUtil.exec_command_async(client, sh, ignore_err=False)
            CommonUtil.after_import(hdfs_path=hdfs_path, hive_tb=hive_tb)
            client.close()
        pass
    pass