"""
   @Author      : HuangJian
   @Description : 上架日期补充表
   @SourceTable :
                  ①dim_asin_launchtime_info
   @SinkTable   : dim_asin_label
   @CreateTime  : 2023/12/12 11:20
   @UpdateTime  : 2022/12/12 11:20
"""
import os
import sys
import re

sys.path.append(os.path.dirname(sys.path[0]))
from utils.common_util import CommonUtil, DateTypes
from utils.hdfs_utils import HdfsUtils
from utils.spark_util import SparkUtil
from pyspark.sql.window import Window
from pyspark.sql import functions as F
from pyspark.sql.types import StringType, IntegerType,DoubleType
from yswg_utils.common_udf import udf_handle_string_null_value as NullUDF
from utils.redis_utils import RedisUtils



class DimAsinLaunchtimeInfo(object):

    def __init__(self, site_name, date_type, date_info):
        self.site_name = site_name
        self.date_type = date_type
        self.date_info = date_info
        app_name = f"{self.__class__.__name__}:{site_name}:{date_type}:{date_info}"
        self.spark = SparkUtil.get_spark_session(app_name)
        self.hive_table = "dim_asin_launchtime_info"
        self.partition_dict = {
            "site_name": site_name
        }
        # 落表路径校验
        self.hdfs_path = CommonUtil.build_hdfs_path(self.hive_table, partition_dict=self.partition_dict)
        self.partitions_num = CommonUtil.reset_partitions(site_name, 50)

        # 自定义全局df
        self.df_asin_detail = self.spark.sql(f"select 1+1;")
        self.df_history_launchtime = self.spark.sql(f"select 1+1;")
        self.df_asin_keepa_date = self.spark.sql(f"select 1+1;")
        self.df_asin_handle_launchtime = self.spark.sql(f"select 1+1;")


        # udf函数
        self.handle_string_num_value = F.udf(NullUDF, StringType())
        self.spark.udf.register('u_handle_string_num_value', NullUDF,
                                StringType())

    def read_data(self):
        # 读取asin_detail数据
        if self.date_type == 'all':
            # 说明全局数据量执行
            print("==================执行全量数据整合=================")
            sql = f"""
                with st_asin as(
                select
                    asin,
                    u_handle_string_num_value(launch_time) as crawl_asin_launch_time,
                    date_format(updated_at, '{CommonUtil._date_time_format}') as updated_time,
                    concat(site_name,',',date_type,',',date_info) as period_label
                    from ods_asin_detail
                    where site_name = '{self.site_name}'
                    ),
                bsr_asin as(
                select
                    asin,
                    u_handle_string_num_value(launch_time) as crawl_asin_launch_time,
                    date_format(created_at, '{CommonUtil._date_time_format}') as updated_time,
                    concat(site_name,',',date_type,',',date_info) as period_label
                    from ods_self_asin_detail
                    where site_name = '{self.site_name}'
                )
                select asin, crawl_asin_launch_time,updated_time,period_label from st_asin 
                union 
                select asin, crawl_asin_launch_time,updated_time,period_label from bsr_asin
            """
            print(sql)
            self.df_asin_detail = self.spark.sql(sql)
        else:
            print("==================执行分区数据整合=================")
            # 按分区检测是否有新增的asin
            if self.date_type in (DateTypes.week.name,DateTypes.month.name,DateTypes.month_week.name, 'month_aba_me'):
                # 取st维度的st下的asin数据
                sql = f"""
                            select 
                                asin,
                                u_handle_string_num_value(launch_time) as crawl_asin_launch_time,
                                date_format(updated_at, '{CommonUtil._date_time_format}') as updated_time,
                                concat(site_name,',',date_type,',',date_info) as period_label
                            from ods_asin_detail
                            where site_name = '{self.site_name}'
                            and date_type = '{self.date_type}'
                            and date_info = '{self.date_info}'
                            """
            elif self.date_type in (DateTypes.day.name):
                # 取bsr日维度的asin
                sql = f"""
                    select
                        asin,
                        u_handle_string_num_value(launch_time) as crawl_asin_launch_time,
                        date_format(created_at, '{CommonUtil._date_time_format}') as updated_time,
                        concat(site_name,',',date_type,',',date_info) as period_label
                    from ods_self_asin_detail
                    where site_name = '{self.site_name}'
                    and date_type = '{self.date_type}'
                    and date_info = '{self.date_info}'
                """
            print(sql)
            # 如果是分区执行可进行缓存数据
            self.df_asin_detail = self.spark.sql(sql).cache()

        # 读取keepa数据
        if self.site_name == 'us':
            sql = f"""select asin, 
                            launch_time as keepa_asin_launch_time, 
                            updated_time as updated_time,
                            1 as keepa_crawl_flag 
                        from ods_asin_keepa_date_tmp
                          where site_name='{self.site_name}'
                          and state = 1 """
        else:
            sql = f"""select asin, 
                            launch_time as keepa_asin_launch_time, 
                            updated_at as updated_time,
                            1 as keepa_crawl_flag  
                        from ods_asin_keep_date 
                        where site_name='{self.site_name}'
                        and state = 3 """
        print(sql)
        self.df_asin_keepa_date = self.spark.sql(sqlQuery=sql).cache()
        self.df_asin_keepa_date = self.df_asin_keepa_date.orderBy(self.df_asin_keepa_date.updated_time.desc_nulls_last())
        self.df_asin_keepa_date = self.df_asin_keepa_date.drop_duplicates(['asin'])

        # 读取历史已经整合好的上架日期数据
        sql = f"""
                    select asin,
                           crawl_asin_launch_time,
                           '9999-12-31 23:59:59' as updated_time,
                           appear_period_label as period_label 
                    from dim_asin_launchtime_info 
                    where site_name='{self.site_name}'
                """
        print(sql)
        self.df_history_launchtime = self.spark.sql(sqlQuery=sql)


    def handle_launchtime_data(self):
        if self.date_type != 'all':
            # 如果不是走all逻辑，则说明走日常调度逻辑；可以将我们已有的历史上架日期放入逻辑中去重
            self.df_asin_detail = self.df_asin_detail.unionByName(self.df_history_launchtime).cache()
        df_all_asin = self.df_asin_detail.orderBy(self.df_asin_detail.updated_time.desc_nulls_last())
        # 去重，保留所有的asin
        df_all_asin = df_all_asin.drop_duplicates(['asin'])
        df_all_asin = df_all_asin.select(
            F.col("asin"),F.col("crawl_asin_launch_time").alias("crawl_asin_launch_time_left"), 'period_label')

        # 过滤找出抓取的上架日期不为空的数据
        df_not_null_asin = self.df_asin_detail.filter(" crawl_asin_launch_time is not null")
        df_not_null_asin = df_not_null_asin.orderBy(df_not_null_asin.updated_time.desc_nulls_last())
        df_not_null_asin = df_not_null_asin.drop_duplicates(['asin'])
        df_not_null_asin = df_not_null_asin.select(
            F.col("asin"),
            F.col("crawl_asin_launch_time").alias("crawl_asin_launch_time_right"),
            F.col("period_label").alias("period_label_right")
        )
        self.df_asin_handle_launchtime = df_all_asin.join(
            df_not_null_asin, on='asin', how='left'
        )

        self.df_asin_handle_launchtime = self.df_asin_handle_launchtime.select(
            F.col("asin"),
            F.when((F.col("crawl_asin_launch_time_left").isNull()) & (F.col("crawl_asin_launch_time_right").isNotNull()), F.col("crawl_asin_launch_time_right"))
            .otherwise(F.col("crawl_asin_launch_time_left")).alias("crawl_asin_launch_time"),
            F.when((F.col("crawl_asin_launch_time_left").isNull()) & (F.col("crawl_asin_launch_time_right").isNotNull()), F.col("period_label_right"))
            .otherwise(F.col("period_label")).alias("period_label")
        )

        # 跟keepa_date进行关联，补充launch_time数据
        self.df_asin_handle_launchtime = self.df_asin_handle_launchtime.join(
            self.df_asin_keepa_date, on='asin', how='left'
        )

        self.df_asin_handle_launchtime = self.df_asin_handle_launchtime.select(
            F.col("asin"),
            F.when(F.col("crawl_asin_launch_time").isNull(), F.col("keepa_asin_launch_time"))
                .otherwise(F.col("crawl_asin_launch_time")).alias("asin_launch_time"),
            F.col("crawl_asin_launch_time"),
            F.col("keepa_asin_launch_time"),
            F.col("period_label").alias("appear_period_label"),
            F.when(F.col("keepa_crawl_flag") == 1, F.lit(1)).otherwise(F.lit(0)).alias("keepa_crawl_flag"),
            F.lit(self.site_name).alias("site_name")
        )

    def save_data(self):
        # print(f"清除hdfs目录中:{self.hdfs_path}")
        # HdfsUtils.delete_file_in_folder(self.hdfs_path)
        df_save = self.df_asin_handle_launchtime.repartition(self.partitions_num)
        CommonUtil.save_or_update_table(spark_session=self.spark,hive_tb_name=self.hive_table,partition_dict=self.partition_dict,df_save=df_save,drop_exist_tmp_flag=False)
        print("success")

    def run(self):
        self.read_data()
        self.handle_launchtime_data()
        self.save_data()


if __name__ == '__main__':
    site_name = CommonUtil.get_sys_arg(1, None)
    date_type = CommonUtil.get_sys_arg(2, None)
    date_info = CommonUtil.get_sys_arg(3, None)
    lock_name = "dim_asin_launchtime_info"
    if date_type == "all":
        # 如果执行数据为all的情况，非自然解锁情况，则需锁定该表2h
        lock_flag = RedisUtils.acquire_redis_lock(lock_name, expire_time=120 * 60, retry_flag=True, retry_time=10*60)
    else:
        lock_flag = RedisUtils.acquire_redis_lock(lock_name, expire_time=30 * 60, retry_flag=True, retry_time=10 * 60)
    if lock_flag:
        try:
            obj = DimAsinLaunchtimeInfo(site_name, date_type, date_info)
            obj.run()
        finally:
            # 执行完成后释放锁
            RedisUtils.release_redis_lock(lock_name)
