"""
   @Author      : HuangJian
   @Description : 关键词与Asin详情维表
   @SourceTable :
                  ①dwd_st_asin_measure
                  ②dwt_aba_st_analytics

   @SinkTable   : dwt_aba_last_change_rate
   @CreateTime  : 2022/03/13 14:55
   @UpdateTime  : 2022/03/13 14:55
"""
import os
import sys
sys.path.append(os.path.dirname(sys.path[0]))

from utils.hdfs_utils import HdfsUtils
from utils.common_util import CommonUtil, DateTypes
from utils.spark_util import SparkUtil
from pyspark.sql import functions as F


class DwtAbaLastChangeRate(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
        self.hive_tb = "dwt_aba_last_change_rate"
        app_name = f"{ self.hive_tb}:{site_name}:{date_type}:{date_info}"
        self.spark = SparkUtil.get_spark_session(app_name)
        self.partitions_num = CommonUtil.reset_partitions(site_name, 2)
        hdfs_path = f"/home/{SparkUtil.DEF_USE_DB}/dwt/{self.hive_tb}/site_name={self.site_name}/date_type={self.date_type}/date_info={self.date_info}"
        print(f"清除hdfs目录中数据:{hdfs_path}")
        HdfsUtils.delete_hdfs_file(hdfs_path)

        self.last_year_index = int
        # 计算环比日期
        self.last_date_info = self.handle_date_offset(0)
        # 计算同比日期
        self.last_year_date_info = self.handle_date_offset(1)

        # 初始化全局df
        self.df_aba_analytics = self.spark.sql(f"select 1+1;")
        self.df_aba_analytics_old = self.spark.sql(f"select 1+1;")
        self.df_st_base_data = self.spark.sql(f"select 1+1;")
        self.df_st_last_data = self.spark.sql(f"select 1+1;")
        self.df_st_last_year_data = self.spark.sql(f"select 1+1;")
        self.df_save = self.spark.sql(f"select 1+1;")
        # 需求1：近6月排名变化率/变化量（仅 month 类型生效）
        self.df_hist_rank = self.spark.sql(f"select 1+1;")  # M-1~M-6 历史 rank

    def handle_date_offset(self, handle_type: int):
        # handle_type = 0 代表计算环比日期，等于 1 代表计算同比日期
        handle_date = self.date_info
        if handle_type == 0:
            # 计算环比计算日期--通过检索表的各分区，取当前计算日上一周期
            if self.date_type == DateTypes.last365day.name:
                # 当date_type为['last365day']时，检索分区dwt_aba_last365
                date_df = CommonUtil.select_partitions_df(self.spark, "dwt_aba_last365")
                handle_date = date_df.filter(
                    f"site_name = '{self.site_name}' date_type = '{self.date_type}' and and date_info < '{self.date_info}' "
                ).selectExpr("max(date_info)").rdd.flatMap(lambda ele: ele).collect()[0]
            else:
                # 当date_type为['day','week','month','last30day']时，检索分区dwt_aba_st_analytics
                date_df = CommonUtil.select_partitions_df(self.spark, "dwt_aba_st_analytics")
                handle_date = date_df.filter(
                    f"site_name = '{self.site_name}' and date_type = '{self.date_type}' and date_info < '{self.date_info}' "
                ).selectExpr("max(date_info)").rdd.flatMap(lambda ele: ele).collect()[0]
        else:
            # 计算同比计算日期 (无论哪个日期类型，开头参数分割第一个参数均为年)
            year_int = int(CommonUtil.safeIndex(handle_date.split("-"), 0, None))
            last_year_int = year_int - 1
            self.last_year_index = last_year_int
            # 将当前年份，替换成同比年份
            handle_date = handle_date.replace(str(year_int), str(last_year_int))
        print("计算处理之后的日期:", handle_date)
        return handle_date

    def run(self):
        # aba ：365的取值表逻辑与其他时间区间（day,week,month,last30day）的取值逻辑不一致
        if self.date_type == DateTypes.last365day.name:
            self.handle_365_data()
        else:
            self.read_data()
            self.handle_base()
            self.handle_year_ratio()

        # 需求1：仅 month 类型计算近6月排名变化（12 字段），其他类型补 null 占位（确保 save_data 不报错）
        if self.date_type == DateTypes.month.name:
            self.handle_rank_rate_history()
        else:
            self.handle_rank_rate_padding()

        self.save_data()

    def read_data(self):
        sql1 = f"""
            select 
                id,
                search_term,
                rank,
                bsr_orders,
                asin_cn_count,
                asin_fbm_count,
                asin_amazon_count,
                search_volume
            from dwt_aba_st_analytics
            where site_name = '{self.site_name}'
            and date_type = '{self.date_type}'
            and date_info = '{self.date_info}'
        """
        self.df_aba_analytics = self.spark.sql(sql1).repartition(40, 'id').cache()
        self.df_aba_analytics.show(10, truncate=True)

        sql2 = f"""
            select 
                id,
                rank              as last_rank,
                bsr_orders        as last_bsr_orders,
                asin_cn_count     as last_asin_cn_count,
                asin_fbm_count    as last_asin_fbm_count,
                asin_amazon_count as last_asin_amazon_count
            from dwt_aba_st_analytics
            where site_name = '{self.site_name}'
            and date_type = '{self.date_type}'
            and date_info = '{self.last_date_info}'
        """
        self.df_aba_analytics_old = self.spark.sql(sql2).repartition(40, 'id').cache()
        self.df_aba_analytics_old.show(10, truncate=True)

        # 获取同比周期数据--（2022年的同比数据只能通过dim_st_detail取到搜索词排名）
        if self.last_year_date_info <= '2022-09':
            sql = f"""
                select 
                    search_term,
                    st_rank as last_year_rank,
                    null as last_year_bsr_orders,
                    null as last_year_asin_cn_count,
                    null as last_year_asin_fbm_count,
                    null as last_year_asin_amazon_count,
                    st_search_num as last_year_search_volume
                from dim_st_detail
                where site_name = '{self.site_name}'
                and date_type = '{self.date_type}'
                and date_info = '{self.last_year_date_info}'
            """
        else:
            sql = f""" 
                select
                    search_term,
                    rank as last_year_rank, 
                    bsr_orders as last_year_bsr_orders,
                    asin_cn_count as last_year_asin_cn_count,
                    asin_fbm_count as last_year_asin_fbm_count,
                    asin_amazon_count as last_year_asin_amazon_count,
                    search_volume as last_year_search_volume
                from dwt_aba_st_analytics
                where site_name = '{self.site_name}'
                and date_type = '{self.date_type}'
                and date_info = '{self.last_year_date_info}'
            """
        self.df_st_last_year_data = self.spark.sql(sql).repartition(40, 'search_term').cache()
        self.df_st_last_year_data.show(10, truncate=True)

        # 需求1：仅 month 类型读取历史数据用于近6月排名变化计算
        if self.date_type == DateTypes.month.name:
            # 计算 M-1 ~ M-6 月份字符串（如 2026-04 → ['2026-03', '2026-02', '2026-01', '2025-12', '2025-11', '2025-10']）
            month_list = [CommonUtil.get_month_offset(self.date_info, -i) for i in range(1, 7)]
            print(f"近6月历史月份列表: {month_list}")

            # 一次读 dwt_aba_st_analytics 的 M-1~M-6 历史分区，仅取 search_term + rank
            # 用于算 6 个变化量字段 + rank_rate_last_1_month
            sql_hist_rank = f"""
                select
                    search_term,
                    cast(rank as int) as rank,
                    date_info
                from dwt_aba_st_analytics
                where site_name = '{self.site_name}'
                  and date_type = '{self.date_type}'
                  and date_info in ({CommonUtil.list_to_insql(month_list)})
                  and rank > 0
            """
            self.df_hist_rank = self.spark.sql(sql_hist_rank).repartition(40, 'search_term').cache()
            print("self.df_hist_rank:")
            self.df_hist_rank.show(10, truncate=True)


    def handle_base(self):
        self.df_st_base_data = self.df_aba_analytics.join(
            self.df_aba_analytics_old, on='id', how='left'
        )
        self.df_st_base_data = self.df_st_base_data.withColumn(
            'rank_rate_of_change',
            F.round((F.col('rank') - F.col('last_rank')) / F.col('last_rank'), 4)
        ).withColumn(
            'bsr_orders_rate_of_change',
            F.round((F.col('bsr_orders') - F.col('last_bsr_orders')) / F.col('last_bsr_orders'), 4)
        ).withColumn(
            'cn_seller_rate_of_change',
            F.round((F.col('asin_cn_count') - F.col('last_asin_cn_count')) / F.col('last_asin_cn_count'), 4)
        ).withColumn(
            'fbm_rate_of_change',
            F.round((F.col('asin_fbm_count') - F.col('last_asin_fbm_count')) / F.col('last_asin_fbm_count'), 4)
        ).withColumn(
            'amazon_rate_of_change',
            F.round((F.col('asin_amazon_count') - F.col('last_asin_amazon_count')) / F.col('last_asin_amazon_count'), 4)
        ).select(
            'id', 'search_term', 'rank', 'bsr_orders', 'asin_cn_count', 'asin_fbm_count', 'asin_amazon_count',
            'rank_rate_of_change', 'bsr_orders_rate_of_change', 'cn_seller_rate_of_change',
            'fbm_rate_of_change', 'amazon_rate_of_change', 'search_volume'
        ).repartition(40, 'search_term').cache()
        self.df_aba_analytics.unpersist()
        self.df_aba_analytics_old.unpersist()

    def handle_year_ratio(self):
        # 计算同比逻辑
        df_year_ratio = self.df_st_base_data.join(
            self.df_st_last_year_data, on='search_term', how='left'
        )
        df_year_ratio = df_year_ratio.withColumn(
            "rank_change_rate",
            F.round(F.expr("(rank - last_year_rank) / last_year_rank"), 4)
        ).withColumn(
            "bsr_orders_change_rate",
            F.round(F.expr("(bsr_orders - last_year_bsr_orders) / last_year_bsr_orders"), 4)
        ).withColumn(
            "cn_seller_change_rate",
            F.round(F.expr("(asin_cn_count - last_year_asin_cn_count) / last_year_asin_cn_count"), 4)
        ).withColumn(
            "fbm_change_rate",
            F.round(F.expr("(asin_fbm_count - last_year_asin_fbm_count) / last_year_asin_fbm_count"), 4)
        ).withColumn(
            "amazon_change_rate",
            F.round(F.expr("(asin_amazon_count - last_year_asin_amazon_count) / last_year_asin_amazon_count"), 4)
        ).withColumn(
            "search_volume_change_rate",
            F.round(F.expr("(search_volume - last_year_search_volume) / last_year_search_volume"), 3)
        ).na.fill({
            # 默认值1000,本次有数据 同比(环比)没有数据 归为上升，排名负数代表上升
            "rank_change_rate": -1000.000,
            "rank_rate_of_change": -1000.000,
            "bsr_orders_change_rate": 1000.000,
            "bsr_orders_rate_of_change": 1000.000,
            "cn_seller_change_rate": 1000.000,
            "cn_seller_rate_of_change": 1000.000,
            "fbm_change_rate": 1000.000,
            "fbm_rate_of_change": 1000.000,
            "amazon_change_rate": 1000.000,
            "amazon_rate_of_change": 1000.000,
            "search_volume_change_rate": 1000.000
        })
        self.df_save = df_year_ratio

    # 需求1：近6月排名变化率/变化量（仅 month 类型调用）
    def handle_rank_rate_history(self):
        """
        基于 self.df_save（已含同比/环比所有字段）+ M-1~M-6 历史 rank，
        计算 12 个新字段并 join 到 self.df_save。

        字段计算逻辑：
        - rank_change_last_1_month: 当月 rank - 上1月 rank
        - rank_change_1~5_month_ago: 上 N 月 rank - 上 N+1 月 rank
        - rank_rate_last_1_month: 当月 rank vs 上1月 rank
        - rank_rate_1~5_month_ago: 上 N 月 rank vs 上 N+1 月 rank（直接从 df_hist_rank 相邻月算，不再读历史 change_rate 分区）

        下榜/新进榜区分（月约定：负数=改善/排名上升，正数=变差/排名下降）：
        - 当期 rank 为 null（下榜）:  变化量 +10000000，变化率 +1000.0
        - 上期 rank 为 null（新进榜）: 变化量 -10000000，变化率 -1000.0
        - 两端均 null:               变化量 0，        变化率 0.0
        """
        month_list = [CommonUtil.get_month_offset(self.date_info, -i) for i in range(1, 7)]

        # 把 df_hist_rank 按 date_info 拆成 6 份 DF，分别对应 rank_last_1_month ~ rank_last_6_month
        df_rank_dfs = {}
        for i in range(1, 7):
            df_rank_dfs[i] = self.df_hist_rank.filter(
                F.col('date_info') == month_list[i - 1]
            ).drop('date_info').withColumnRenamed('rank', f'rank_last_{i}_month')

        # 依次 left join 6 份历史 rank 到 self.df_save
        df = self.df_save
        for i in range(1, 7):
            df = df.join(df_rank_dfs[i], on='search_term', how='left')

        # 变化量：区分下榜(+10000000) / 新进榜(-10000000) / 两端无数据(0) / 正常计算
        def _rank_change(cur, prev):
            return (
                F.when(F.col(cur).isNull() & F.col(prev).isNull(), F.lit(0))
                 .when(F.col(cur).isNull(), F.lit(10000000))
                 .when(F.col(prev).isNull(), F.lit(-10000000))
                 .otherwise(F.col(cur) - F.col(prev))
            )

        # 变化率：月约定负数=改善（rank数字变小=排名上升），区分下榜(+1000) / 新进榜(-1000) / 两端无数据(0) / 正常计算
        def _rank_rate(cur, prev):
            return (
                F.when(F.col(cur).isNull() & F.col(prev).isNull(), F.lit(0.0))
                 .when(F.col(cur).isNull(), F.lit(1000.0))
                 .when(F.col(prev).isNull(), F.lit(-1000.0))
                 .otherwise(F.round((F.col(cur) - F.col(prev)) / F.col(prev), 4))
            )

        # 6 个变化量
        df = df.withColumn('rank_change_last_1_month', _rank_change('rank',             'rank_last_1_month')) \
               .withColumn('rank_change_1_month_ago',  _rank_change('rank_last_1_month', 'rank_last_2_month')) \
               .withColumn('rank_change_2_month_ago',  _rank_change('rank_last_2_month', 'rank_last_3_month')) \
               .withColumn('rank_change_3_month_ago',  _rank_change('rank_last_3_month', 'rank_last_4_month')) \
               .withColumn('rank_change_4_month_ago',  _rank_change('rank_last_4_month', 'rank_last_5_month')) \
               .withColumn('rank_change_5_month_ago',  _rank_change('rank_last_5_month', 'rank_last_6_month'))

        # 6 个变化率（全部统一用 _rank_rate，含下榜/新进榜/正常三种情况）
        df = df.withColumn('rank_rate_last_1_month', _rank_rate('rank',             'rank_last_1_month')) \
               .withColumn('rank_rate_1_month_ago',  _rank_rate('rank_last_1_month', 'rank_last_2_month')) \
               .withColumn('rank_rate_2_month_ago',  _rank_rate('rank_last_2_month', 'rank_last_3_month')) \
               .withColumn('rank_rate_3_month_ago',  _rank_rate('rank_last_3_month', 'rank_last_4_month')) \
               .withColumn('rank_rate_4_month_ago',  _rank_rate('rank_last_4_month', 'rank_last_5_month')) \
               .withColumn('rank_rate_5_month_ago',  _rank_rate('rank_last_5_month', 'rank_last_6_month'))

        self.df_save = df
        self.df_hist_rank.unpersist()

    # 需求1：非 month 类型（day/week/last30day/last365day）补 12 字段为 null
    # 原因：Hive 表加了这 12 列后，所有分区 schema 都得有这些字段，否则 save_data 的 select 报错
    def handle_rank_rate_padding(self):
        new_cols = [
            'rank_rate_last_1_month', 'rank_rate_1_month_ago', 'rank_rate_2_month_ago',
            'rank_rate_3_month_ago', 'rank_rate_4_month_ago', 'rank_rate_5_month_ago',
            'rank_change_last_1_month', 'rank_change_1_month_ago', 'rank_change_2_month_ago',
            'rank_change_3_month_ago', 'rank_change_4_month_ago', 'rank_change_5_month_ago',
        ]
        for col_name in new_cols:
            self.df_save = self.df_save.withColumn(col_name, F.lit(None))

    def handle_365_data(self):
        sql = f"""
            with base_data as (
                select 
                    id,
                    search_term,
                    rank,
                    bsr_orders
                from dwt_aba_last365
                where site_name = '{self.site_name}'
                and date_type = '{self.date_type}'
                and date_info = '{self.date_info}'
            ),
            chain_ratio_data as (
                select 
                    id,
                    rank              as last_rank,
                    bsr_orders        as last_bsr_orders
                from dwt_aba_last365
                where site_name = '{self.site_name}'
                and date_type = '{self.date_type}'
                and date_info = '{self.last_date_info}'
            ),
            year_ratio_data as (
                select 
                    id,
                    rank              as last_year_rank,
                    bsr_orders        as last_year_bsr_orders
                from dwt_aba_last365
                where site_name = '{self.site_name}'
                and date_type = '{self.date_type}'
                and date_info = '{self.last_year_date_info}'
            )

            select 
                base.id,
                base.search_term,
                base.rank,
                base.bsr_orders,
                round((base.rank - chain.last_rank)/chain.last_rank,4) as rank_rate_of_change,
                round((base.bsr_orders - chain.last_bsr_orders)/chain.last_bsr_orders,4) as bsr_orders_rate_of_change,
                round((base.rank - year.last_year_rank)/year.last_year_rank,4) as rank_change_rate,
                round((base.bsr_orders - year.last_year_bsr_orders)/year.last_year_bsr_orders,4) as bsr_orders_change_rate
            from base_data base left join chain_ratio_data chain
            on base.id = chain.id
            left join year_ratio_data year
            on base.id = year.id
        """
        print("动态365的同比、环比计算sql语句：", sql)
        self.df_save = self.spark.sql(sqlQuery=sql)
        # 字段补全
        self.df_save = self.df_save.withColumn(
            "cn_seller_rate_of_change", F.lit(None)
        ).withColumn(
            "cn_seller_change_rate", F.lit(None)
        ).withColumn(
            "fbm_rate_of_change", F.lit(None)
        ).withColumn(
            "fbm_change_rate", F.lit(None)
        ).withColumn(
            "amazon_rate_of_change", F.lit(None)
        ).withColumn(
            "amazon_change_rate", F.lit(None)
        )

    def save_data(self):
        self.df_save = self.df_save.select(
            F.col("id").alias("search_term_id"),
            F.col("search_term"),
            F.col("rank_rate_of_change"),
            F.col("rank_change_rate"),
            F.col("bsr_orders_rate_of_change"),
            F.col("bsr_orders_change_rate"),
            F.col("cn_seller_rate_of_change"),
            F.col("cn_seller_change_rate"),
            F.col("fbm_rate_of_change"),
            F.col("fbm_change_rate"),
            F.col("amazon_rate_of_change"),
            F.col("amazon_change_rate"),
            F.date_format(F.current_timestamp(), 'yyyy-MM-dd HH:mm:SS').alias("created_time"),
            F.date_format(F.current_timestamp(), 'yyyy-MM-dd HH:mm:SS').alias("updated_time"),
            F.col("search_volume_change_rate"),
            F.lit(self.site_name).alias("site_name"),
            F.lit(self.date_type).alias("date_type"),
            F.lit(self.date_info).alias("date_info"),
            # 需求1：12 个新字段，顺序与 Hive ALTER ADD COLUMNS、PG ADD COLUMN、sqoop_export 严格一致
            F.col("rank_rate_last_1_month"),
            F.col("rank_rate_1_month_ago"),
            F.col("rank_rate_2_month_ago"),
            F.col("rank_rate_3_month_ago"),
            F.col("rank_rate_4_month_ago"),
            F.col("rank_rate_5_month_ago"),
            F.col("rank_change_last_1_month"),
            F.col("rank_change_1_month_ago"),
            F.col("rank_change_2_month_ago"),
            F.col("rank_change_3_month_ago"),
            F.col("rank_change_4_month_ago"),
            F.col("rank_change_5_month_ago")
        )

        # 类型转换
        self.df_save = CommonUtil.auto_transfer_type(self.spark, self.df_save, self.hive_tb)

        self.df_save = self.df_save.repartition(self.partitions_num)
        partition_by = ["site_name", "date_type", "date_info"]
        print(f"当前存储的表名为：{self.hive_tb},分区为{partition_by}", )
        self.df_save.write.saveAsTable(name=self.hive_tb, format='hive', mode='append', partitionBy=partition_by)
        print("success")


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)
    obj = DwtAbaLastChangeRate(site_name, date_type, date_info)
    obj.run()
