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__':
    site_name = 'us'
    hive_table = f"ods_merchantwords_asin_detail"
    partition_dict = {
        "site_name": site_name,
        "date_type": 'day',
        "date_info": '2024-04-24-test'
    }
    # 落表路径校验
    hdfs_path = CommonUtil.build_hdfs_path(hive_table, partition_dict=partition_dict)
    print(f"hdfs_path is {hdfs_path}")
    import_table = "us_merchantwords_search_term_asin_detail_2024_04_24"
    sql_query = f"""
        select 
         id,
         asin,
         title,
         img,
         price,
         rating,
         reviews,
         created_time,
         updated_time
        from {import_table} 
        where 1=1
        and \$CONDITIONS
    """
    # 生成导出脚本
    import_sh = f"""
    /opt/module/sqoop-1.4.6/bin/sqoop yswg_import -D mapred.job.queue.name=default -D mapred.task.timeout=0  --append \\
    --connect "jdbc:mysql://192.168.10.151:19030/test"  \\
    --username "chenyuanjie"  \\
    --password "chenyuanjie12345" \\
    --target-dir {hdfs_path}  \\
    --mapreduce-job-name "sqoop_task: sr_to_hive_by_sqoop_test"  \\
    --query "{sql_query}"  \\
    --fields-terminated-by '\\t'  \\
    --hive-drop-import-delims  \\
    --null-string '\\\\N'  \\
    --null-non-string '\\\\N'  \\
    --compress \\
    -m 50 \\
    --split-by "id" \\
    --compression-codec lzop  \\
    --outdir "/tmp/sqoop/"
    """
    HdfsUtils.delete_hdfs_file(hdfs_path)
    client = SSHUtil.get_ssh_client()
    SSHUtil.exec_command_async(client, import_sh, ignore_err=False)
    CommonUtil.after_import(hdfs_path=hdfs_path, hive_tb=hive_table)
    client.close()