"""
   @Author      : hejiangming
   @Description : 店铺基础信息报表（数据源下沉版：ODS -> DIM/DWT）
                  把原来直接读 ODS 层的逻辑改为读 dim_fb_detail / dim_fb_asin_info / dwt_flow_asin，
                  不再做 rating 拆分、histogram 解析、metadata 覆盖、business_name/addr UDF 解析、
                  is_self_fb MySQL 读取、package_quantity/is_asin_new UDF 计算、parent_asin 变体表读取。
                  这些预处理已在 DIM 层完成，DWT 层只做跨维度聚合 + 功能1-6 计算 + 落表。
   @SourceTable :
                  1.dim_fb_detail
                  2.dim_fb_asin_info
                  3.ods_asin_detail_product
                  4.dwt_flow_asin
                  5.dim_asin_profit_rate_info
   @SinkTable   : dwt_fb_base_report
   @SinkTable   : dwt_fb_asin_info
   @CreateTime  : 2026/07/29 16:00
   @UpdateTime  : 2026/07/29 16:00
"""

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
from pyspark.sql.types import IntegerType
from utils.spark_util import SparkUtil
from pyspark.sql import functions as F
from yswg_utils.common_udf import udf_new_asin_flag       # top20 新品判断仍需 UDF


class DwtFbBaseReport(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_fb_base_report"
        self.partition_dict = {
            "site_name": site_name,
            "date_type": date_type,
            "date_info": date_info
        }
        self.hdfs_path = CommonUtil.build_hdfs_path(self.hive_tb, partition_dict=self.partition_dict)

        app_name = f"{self.__class__.__name__}:{site_name}:{date_info}"
        self.spark = SparkUtil.get_spark_session(app_name)

        # cal_date: 当前周期对应的具体日期（如月度 2023-08 -> 2023-08-31），用于 top20 新品判断
        self.cal_date = CommonUtil.get_calDay_by_dateInfo(self.spark, self.date_type, self.date_info)
        # last_month: 上个月（如 2023-08 -> 2023-07），用于计算 Feedback 环比变化率
        self.last_month = CommonUtil.get_month_offset(date_info, -1)

        # 全局 df 初始化
        self.df_fb_feedback = self.spark.sql(f"select 1+1;")     # dim_fb_detail 当月+上月环比
        self.df_fb_asin = self.spark.sql(f"select 1+1;")         # dim_fb_asin_info seller-asin 映射
        self.df_top20_asin = self.spark.sql(f"select 1+1;")      # ods_asin_detail_product top20 排名
        self.df_fb_agg = self.spark.sql(f"select 1+1;")          # 聚合结果
        self.df_fb_asin_detail = self.spark.sql(f"select 1+1;")  # seller->asin->flow 明细
        self.df_flow_asin = self.spark.sql(f"select 1+1;")       # dwt_flow_asin ASIN详情+功能1-6
        self.df_profit_rate = self.spark.sql(f"select 1+1;")     # 利润率

        # UDF 初始化
        self.udf_new_asin_flag = F.udf(udf_new_asin_flag, IntegerType())      # top20 新品判断仍需 UDF


    def read_data(self):
        # 读取 dim_fb_detail 当月数据
        # 替代: ods_seller_account_feedback + ods_seller_account_syn + MySQL selection.accounts
        # dim_fb_detail 已预处理: rating_*_num / fb_star_*_pct / business_name / business_addr /
        #   is_self_account / fb_web_asin_num(含 metadata 覆盖) / fb_crawl_date
        print("获取 dim_fb_detail（当月）")
        sql = f"""
        select
            seller_id, account_name, fb_country_name, fb_web_asin_num,
            is_self_account as is_self_fb,
            count_30_day_num, count_1_year_num, count_lifetime_num,
            business_name, business_addr,
            rating_30_day_num, rating_90_day_num, rating_1_year_num, rating_lifetime_num,
            fb_star_5_pct, fb_star_4_pct, fb_star_3_pct, fb_star_2_pct, fb_star_1_pct,
            fb_crawl_date
        from dim_fb_detail
        where site_name = '{self.site_name}'
          and date_type = '{self.date_type}'
          and date_info = '{self.date_info}'
        """
        df_cur = self.spark.sql(sqlQuery=sql)
        print(sql)

        # 上月 dim_fb_detail，只取 count 字段用于计算环比
        print("获取 dim_fb_detail（上月）")
        sql_last = f"""
        select
            seller_id,
            count_30_day_num as last_30_day_num,
            count_1_year_num as last_1_year_num,
            count_lifetime_num as last_lifetime_num
        from dim_fb_detail
        where site_name = '{self.site_name}'
          and date_type = '{self.date_type}'
          and date_info = '{self.last_month}'
        """
        df_last = self.spark.sql(sqlQuery=sql_last)
        print(sql_last)

        # 当月 LEFT JOIN 上月，计算三个 Feedback 环比变化率
        # 新店铺（上月无记录）left join 后 last_*_num = null -> rate = null（无法计算）
        self.df_fb_feedback = df_cur.join(df_last, on='seller_id', how='left')
        self.df_fb_feedback = self.df_fb_feedback \
            .withColumn('count_30_day_rate',
                F.round((F.col('count_30_day_num') - F.col('last_30_day_num')) / F.col('last_30_day_num'), 4)) \
            .withColumn('count_1_year_rate',
                F.round((F.col('count_1_year_num') - F.col('last_1_year_num')) / F.col('last_1_year_num'), 4)) \
            .withColumn('count_life_time_rate',
                F.round((F.col('count_lifetime_num') - F.col('last_lifetime_num')) / F.col('last_lifetime_num'), 4)) \
            .drop('last_30_day_num', 'last_1_year_num', 'last_lifetime_num') \
            .cache()

        # 读取 dim_fb_asin_info 店铺-ASIN 关系表（替代 ods_seller_asin_account）
        # 按月度分区读取，不再用 created_at 过滤
        print("获取 dim_fb_asin_info")
        sql = f"""
        select seller_id, account_name, asin
        from dim_fb_asin_info
        where site_name = '{self.site_name}'
          and date_type = '{self.date_type}'
          and date_info = '{self.date_info}'
        """
        self.df_fb_asin = self.spark.sql(sqlQuery=sql)
        self.df_fb_asin = self.df_fb_asin.drop_duplicates(['seller_id', 'asin'])
        print(sql)

        # 获取店铺 top20 ASIN 排名，只取 seller_id + asin + row_num
        # 价格/评分/评论数改从 flow 获取（handle_fb_top_20 中 join）
        print("获取 ods_asin_detail_product（top20 排名）")
        sql = f"""
        select seller_id, asin, row_num as rank
        from ods_asin_detail_product
        where site_name = '{self.site_name}'
          and date_type = '{self.date_type}'
          and date_info = '{self.date_info}'
          and row_num <= 20
        """
        self.df_top20_asin = self.spark.sql(sqlQuery=sql)
        self.df_top20_asin = self.df_top20_asin.drop_duplicates(['seller_id', 'asin'])
        print(sql)

        # 从 dwt_flow_asin 读取 ASIN 详情 + 功能1-6 所需字段
        # 合并了原 dim_cal_asin_history_detail（asin详情）+ dim_asin_variation_info（parent_asin）
        # asin_is_new: flow 已预计算，handle_fb_cal_agg 中直接使用（替代 udf_new_asin_flag）
        # package_quantity: flow 已预计算（替代 udf_get_package_quantity）
        # parent_asin: flow 已包含（替代 dim_asin_variation_info）
        # 不过滤 account_id is not null：asin 详情不依赖 account_id，功能1-6 在聚合前再过滤
        # 调度依赖: dwt_flow_asin 必须在 dwt_fb_base_report 之前完成
        print("获取 dwt_flow_asin（ASIN详情 + 功能1-6）")
        sql = f"""
        select account_id, asin, asin_bought_month,
               asin_buy_box_seller_type, first_category_rank,
               asin_is_new, nsr_last_seen_at, bsr_last_seen_at,
               asin_price, asin_title, asin_img_url,
               asin_rating, asin_total_comments,
               asin_weight, asin_volume,
               asin_launch_time, package_quantity, parent_asin
        from dwt_flow_asin
        where site_name = '{self.site_name}'
          and date_type = '{self.date_type}'
          and date_info = '{self.date_info}'
        """
        self.df_flow_asin = self.spark.sql(sqlQuery=sql)
        # cache（handle_fb_top_20 / handle_fb_cal_agg / profit 多处使用）
        self.df_flow_asin = self.df_flow_asin.cache()
        print(sql)

        # 功能6: 利润率数据（保持不变）
        print("获取 dim_asin_profit_rate_info 利润率数据")
        sql = f"""
        select asin, price as asin_price, ocean_profit, air_profit
        from dim_asin_profit_rate_info
        where site_name = '{self.site_name}'
        """
        self.df_profit_rate = self.spark.sql(sqlQuery=sql) \
            .dropDuplicates(['asin', 'asin_price'])
        print(sql)

    def handle_fb_top_20(self):
        # 基于每个卖家排名前20的 ASIN，计算均价、均评分、均评论数、新品数量
        print("处理asin_detail_product的top20指标")

        # 从 flow 获取 price/rating/total_comments/asin_launch_time
        self.df_top20_asin = self.df_top20_asin.join(
            self.df_flow_asin.select(
                'asin',
                F.col('asin_price').alias('price'),
                F.col('asin_rating').alias('rating'),
                F.col('asin_total_comments').alias('total_comments'),
                'asin_launch_time'
            ), on='asin', how='left')

        # udf_new_asin_flag: 上架时间距 cal_date <= 180 天为新品=1，否则=0
        self.df_top20_asin = self.df_top20_asin.withColumn("is_asin_new",
                                                           self.udf_new_asin_flag(F.col('asin_launch_time'),
                                                                                  F.lit(self.cal_date)))

        self.df_top20_agg = self.df_top20_asin.groupby('seller_id').agg(
            F.avg('price').alias('top_20_avg_price'),
            F.avg('rating').alias('top_20_avg_rating'),
            F.avg('total_comments').alias('top_20_avg_total_comments'),
            F.sum('is_asin_new').alias('top_20_new_asin_num')
        )

    def handle_fb_cal_agg(self):
        print("处理店铺基础报表聚合数据")

        # feedback.seller_id -> dim_fb_asin_info(ASIN映射) -> dwt_flow_asin(ASIN详情)
        df_fb_join = self.df_fb_feedback.select('seller_id').join(
            self.df_fb_asin, on='seller_id', how='left'
        )

        # 从 flow 获取 ASIN 详情（替代 dim_cal_asin_history_detail + dim_asin_variation_info）
        # is_asin_new / package_quantity 直接从 flow 取，不再用 UDF 计算
        df_fb_join = df_fb_join.join(
            self.df_flow_asin.select(
                'asin', 'asin_title', 'asin_img_url', 'asin_price',
                'asin_rating', 'asin_total_comments', 'asin_weight', 'asin_volume',
                'asin_launch_time',
                F.col('asin_is_new').cast(IntegerType()).alias('is_asin_new'),
                F.col('package_quantity').alias('asin_package_quantity'),
                'parent_asin'
            ),
            on='asin', how='left'
        )

        # 打包数量标记: >= 2 的标记为 1
        df_fb_join = df_fb_join.withColumn('is_pq_flag', F.when(F.col('asin_package_quantity') >= 2, F.lit(1)))

        # 补充 parent_asin: null 说明无变体或 flow 未匹配，设为 asin 本身
        df_fb_join = df_fb_join.withColumn('parent_asin', F.when(F.col('parent_asin').isNull(), F.col('asin'))
                                           .otherwise(F.col('parent_asin')))
        self.df_fb_asin_detail = df_fb_join.cache()

        # 计算多变体比率（逻辑不变）
        df_variant_radio = self.df_fb_asin_detail.groupby(['seller_id', 'parent_asin']).agg(
            F.count('asin').alias('asin_son_count')
        )
        # asin_son_count > 1 说明该父品下有多个子变体，标记为 1
        df_variant_radio = df_variant_radio.withColumn('is_variant_flag',
                                                       F.when(F.col('asin_son_count') > 1, F.lit(1)))
        df_variant_radio = df_variant_radio.groupby(['seller_id']).agg(
            F.sum('is_variant_flag').alias('fb_more_variant_num'),
            F.count('parent_asin').alias('fb_variant_asin_total')
        )
        df_variant_radio = df_variant_radio.withColumn('fb_variant_rate',
                                                       F.round(F.col('fb_more_variant_num') / F.col(
                                                           'fb_variant_asin_total'), 4))
        df_variant_radio = df_variant_radio.select('seller_id',
                                                   'fb_more_variant_num',
                                                   'fb_variant_asin_total',
                                                   'fb_variant_rate')

        # 老指标聚合（来源: dim_fb_asin_info -> flow，逻辑不变）
        df_seller_main_agg = self.df_fb_asin_detail.groupby('seller_id').agg(
            F.count('asin').alias('fb_asin_total'),
            F.sum('is_asin_new').alias('fb_new_asin_num'),
            F.sum('is_pq_flag').alias('fb_pq_num'),
        )
        df_seller_main_agg = df_seller_main_agg \
            .withColumn('fb_new_asin_rate',
                        F.round(F.col('fb_new_asin_num') / F.col('fb_asin_total'), 4)) \
            .withColumn('fb_pq_rate',
                        F.round(F.col('fb_pq_num') / F.col('fb_asin_total'), 4))

        # 功能1-5: 从 dwt_flow_asin 计算（逻辑不变）
        # 功能3/4: nsr_last_seen_at/bsr_last_seen_at 落在 [月初, 下月月初) 内视为当月上榜
        month_start      = f"{self.date_info}-01"
        next_month_start = f"{CommonUtil.get_month_offset(self.date_info, 1)}-01"

        # 功能1-6 只用 account_id 非空的行（不影响上面 asin 详情 join 的完整性）
        df_flow_for_agg = self.df_flow_asin.filter(F.col('account_id').isNotNull())

        df_flow_with_flags = df_flow_for_agg \
            .withColumn('is_nsr', F.when(
                (F.col('nsr_last_seen_at') >= month_start) &
                (F.col('nsr_last_seen_at') <  next_month_start),
                F.lit(1)
            )) \
            .withColumn('is_bs', F.when(
                (F.col('bsr_last_seen_at') >= month_start) &
                (F.col('bsr_last_seen_at') <  next_month_start),
                F.lit(1)
            ))

        df_flow_agg = df_flow_with_flags.groupby('account_id').agg(
            # fb_flow_asin_total: 流量选品 ASIN 总数，作为功能2/3/4 占比的分母
            F.count('asin').alias('fb_flow_asin_total'),
            # 功能2 FBM: asin_buy_box_seller_type=3 为 FBM，大类排名 <=500000 为有效 FBM
            F.sum(
                F.when(F.col('asin_buy_box_seller_type') == 3, 1).otherwise(0)
            ).alias('fb_fbm_asin_num'),
            F.sum(
                F.when(
                    (F.col('asin_buy_box_seller_type') == 3) & (F.col('first_category_rank') <= 500000),
                    1
                ).otherwise(0)
            ).alias('fb_fbm_valid_asin_num'),
            # 功能3 NSR / 功能4 BS
            F.coalesce(F.sum('is_nsr'), F.lit(0)).alias('fb_nsr_asin_num'),
            F.coalesce(F.sum('is_bs'),  F.lit(0)).alias('fb_bs_asin_num'),
            # 功能1 总销量
            F.sum('asin_bought_month').alias('fb_shop_total_sales_raw'),
            # 功能5 新品销量分子
            F.sum(
                F.when(F.col('asin_is_new') == 1,
                       F.coalesce(F.col('asin_bought_month'), F.lit(0)))
                 .otherwise(F.lit(0))
            ).alias('fb_new_asin_sales_raw'),
            # 功能5 新品数量占比分子
            F.sum(F.when(F.col('asin_is_new') == 1, F.lit(1)).otherwise(F.lit(0))).alias('fb_new_asin_count_raw'),
        ).withColumnRenamed('account_id', 'seller_id')

        # 功能1: 首页销量(top20 asin join flow 的 asin_bought_month)
        df_home_sales = self.df_top20_asin.join(
            self.df_flow_asin.select('asin', 'asin_bought_month'), on='asin', how='left'
        ).groupby('seller_id').agg(
            F.sum('asin_bought_month').alias('fb_shop_home_sales_raw'),
            F.max(
                F.when(F.col('asin_bought_month').isNull(), F.lit(1)).otherwise(F.lit(0))
            ).alias('fb_shop_home_null_flag_raw')
        )

        # 功能6: 新品/老品平均利润率（逻辑不变）
        # 以 flow 为基底，asin_price 与 dim_asin_profit_rate_info.price 对齐
        # 利润率为 null 的 ASIN 被 avg() 自动忽略；有效样本=0 时 avg()=null -> save_data 转 -1000
        df_profit_cal = df_flow_for_agg.join(
            self.df_profit_rate, on=['asin', 'asin_price'], how='left'
        ).filter(
            F.col('asin_bought_month') >= 50
        )

        df_new_profit = df_profit_cal.filter(F.col('asin_is_new') == 1) \
            .groupby('account_id').agg(
                F.round(F.avg('ocean_profit'), 4).alias('fb_new_ocean_profit_raw'),
                F.round(F.avg('air_profit'), 4).alias('fb_new_air_profit_raw')
            ).withColumnRenamed('account_id', 'seller_id')

        df_old_profit = df_profit_cal.filter(F.col('asin_is_new') == 0) \
            .groupby('account_id').agg(
                F.round(F.avg('ocean_profit'), 4).alias('fb_old_ocean_profit_raw'),
                F.round(F.avg('air_profit'), 4).alias('fb_old_air_profit_raw')
            ).withColumnRenamed('account_id', 'seller_id')

        # 合并所有计算结果到 df_fb_feedback（以 dim_fb_detail 卖家集合为基底）
        self.df_fb_agg = self.df_fb_feedback.join(
            df_seller_main_agg, on='seller_id', how='left'   # 老指标
        ).join(
            df_variant_radio,   on='seller_id', how='left'
        ).join(
            self.df_top20_agg, on='seller_id', how='left'
        ).join(
            df_flow_agg,        on='seller_id', how='left'   # 功能1总销量 + 功能2 FBM + 功能3 NSR + 功能4 BS + 功能5 新品销量
        ).join(
            df_home_sales,      on='seller_id', how='left'   # 功能1: 首页销量
        ).join(
            df_new_profit,      on='seller_id', how='left'   # 功能6: 新品利润率
        ).join(
            df_old_profit,      on='seller_id', how='left'   # 功能6: 老品利润率
        )

    # 输出数据集-report
    def save_data_report(self):
        df_save = self.df_fb_agg.filter(F.col('account_name').isNotNull())

        df_save = df_save.select(
            F.col('seller_id'),
            F.col('account_name'),
            F.col('fb_country_name'),
            F.col('fb_web_asin_num'),
            F.col('is_self_fb'),
            F.round('top_20_avg_price', 4).alias('top_20_avg_price'),
            F.round('top_20_avg_rating', 4).alias('top_20_avg_rating'),
            F.ceil('top_20_avg_total_comments').alias('top_20_avg_total_comments'),
            F.col('count_30_day_num'),
            F.col('count_1_year_num'),
            F.col('count_lifetime_num'),
            F.col('count_30_day_rate'),
            F.col('count_1_year_rate'),
            F.col('count_life_time_rate'),
            F.col('fb_new_asin_num'),
            F.col('fb_asin_total'),
            F.col('fb_new_asin_rate'),
            F.col('fb_variant_rate'),
            F.col('fb_more_variant_num'),
            F.col('fb_variant_asin_total'),
            F.col('fb_pq_rate'),
            F.col('fb_pq_num'),
            F.col('fb_crawl_date'),
            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('top_20_new_asin_num'),
            F.when(F.col('fb_country_name').isNull(), F.lit(0))
                .when(F.col('fb_country_name') == self.site_name.upper(), F.lit(1))
                .when(F.col('fb_country_name') == 'CN', F.lit(2))
                .otherwise(F.lit(3)).alias('fb_country_name_type'),
            F.when(F.col('fb_asin_total').isNull(), F.lit(0))
                .when(F.col('fb_asin_total') <= 300, F.lit(1))
                .when(F.col('fb_asin_total') <= 1000, F.lit(2))
                .otherwise(F.lit(3)).alias('fb_account_type'),
            F.when(F.col('fb_asin_total').isNull(), F.lit(0))
                .when(F.col('fb_asin_total') == 0, F.lit(1))
                .when(F.col('fb_asin_total') <= 50, F.lit(2))
                .when(F.col('fb_asin_total') <= 200, F.lit(3))
                .when(F.col('fb_asin_total') <= 500, F.lit(4))
                .when(F.col('fb_asin_total') <= 1000, F.lit(5))
                .when(F.col('fb_asin_total') <= 5000, F.lit(6))
                .when(F.col('fb_asin_total') <= 10000, F.lit(7))
                .otherwise(F.lit(8)).alias('fb_asin_total_type'),
            F.when(F.col('fb_new_asin_num').isNull(), F.lit(0))
                .when(F.col('fb_new_asin_num') == 0, F.lit(1))
                .when(F.col('fb_new_asin_num') <= 5, F.lit(2))
                .when(F.col('fb_new_asin_num') <= 10, F.lit(3))
                .when(F.col('fb_new_asin_num') <= 20, F.lit(4))
                .when(F.col('fb_new_asin_num') <= 30, F.lit(5))
                .when(F.col('fb_new_asin_num') <= 50, F.lit(6))
                .when(F.col('fb_new_asin_num') <= 100, F.lit(7))
                .otherwise(F.lit(8)).alias('fb_new_asin_num_type'),
            F.when(F.col('fb_new_asin_rate').isNull(), F.lit(0))
                .when(F.col('fb_new_asin_rate') == 0, F.lit(1))
                .when(F.col('fb_new_asin_rate') <= 0.05, F.lit(2))
                .when(F.col('fb_new_asin_rate') <= 0.1, F.lit(3))
                .when(F.col('fb_new_asin_rate') <= 0.2, F.lit(4))
                .when(F.col('fb_new_asin_rate') <= 0.5, F.lit(5))
                .otherwise(F.lit(6)).alias('fb_new_asin_rate_type'),
            F.lit(None).alias('usr_mask_type'),
            F.lit(None).alias('usr_mask_progress'),
            F.col('business_name'),
            F.col('business_addr'),
            F.col('rating_30_day_num'),
            F.col('rating_90_day_num'),
            F.col('rating_1_year_num'),
            F.col('rating_lifetime_num'),

            # 功能1: 销量展示
            # 总销量: 流量选品无该店铺数据时 sum=null -> -1（前端展示"--"）
            F.when(F.col('fb_shop_total_sales_raw').isNull(), F.lit(-1))
             .otherwise(F.col('fb_shop_total_sales_raw'))
             .alias('fb_shop_total_sales'),
            # 首页销量: 无产品时 null -> 0
            F.coalesce(F.col('fb_shop_home_sales_raw'), F.lit(0)).alias('fb_shop_home_sales'),
            # 首页缺数据标记
            F.coalesce(F.col('fb_shop_home_null_flag_raw'), F.lit(0)).alias('fb_shop_home_null_flag'),

            # 功能2: FBM 占比
            F.coalesce(F.col('fb_fbm_asin_num'), F.lit(-1)).alias('fb_fbm_asin_num'),
            F.coalesce(F.col('fb_fbm_valid_asin_num'), F.lit(-1)).alias('fb_fbm_valid_asin_num'),
            F.when(
                F.col('fb_flow_asin_total').isNull() | (F.col('fb_flow_asin_total') == 0), F.lit(-1.0)
            ).otherwise(
                F.round(F.col('fb_fbm_asin_num') / F.col('fb_flow_asin_total'), 4)
            ).alias('fb_fbm_rate'),
            F.when(
                F.col('fb_fbm_asin_num').isNull() | (F.col('fb_fbm_asin_num') == 0), F.lit(-1.0)
            ).otherwise(
                F.round(F.col('fb_fbm_valid_asin_num') / F.col('fb_fbm_asin_num'), 4)
            ).alias('fb_fbm_valid_rate'),

            # 功能3: Hot New 榜单
            F.coalesce(F.col('fb_nsr_asin_num'), F.lit(-1)).alias('fb_nsr_asin_num'),
            F.when(
                F.col('fb_flow_asin_total').isNull() | (F.col('fb_flow_asin_total') == 0), F.lit(-1.0)
            ).otherwise(
                F.round(F.col('fb_nsr_asin_num') / F.col('fb_flow_asin_total'), 4)
            ).alias('fb_nsr_rate'),

            # 功能4: Best Seller 榜单
            F.coalesce(F.col('fb_bs_asin_num'), F.lit(-1)).alias('fb_bs_asin_num'),
            F.when(
                F.col('fb_flow_asin_total').isNull() | (F.col('fb_flow_asin_total') == 0), F.lit(-1.0)
            ).otherwise(
                F.round(F.col('fb_bs_asin_num') / F.col('fb_flow_asin_total'), 4)
            ).alias('fb_bs_rate'),

            # 功能5: 新品销量占比
            F.when(
                F.col('fb_shop_total_sales_raw').isNull() | (F.col('fb_shop_total_sales_raw') == 0),
                F.lit(-1.0)
            ).otherwise(
                F.round(F.col('fb_new_asin_sales_raw') / F.col('fb_shop_total_sales_raw'), 4)
            ).alias('fb_new_asin_sales_rate'),

            # 功能6: 新品/老品平均利润率（-1000 占位，因为 -1=-100% 是真实亏损值）
            F.when(F.col('fb_new_ocean_profit_raw').isNull(), F.lit(-1000))
             .otherwise(F.col('fb_new_ocean_profit_raw'))
             .alias('fb_new_ocean_profit_rate'),
            F.when(F.col('fb_new_air_profit_raw').isNull(), F.lit(-1000))
             .otherwise(F.col('fb_new_air_profit_raw'))
             .alias('fb_new_air_profit_rate'),
            F.when(F.col('fb_old_ocean_profit_raw').isNull(), F.lit(-1000))
             .otherwise(F.col('fb_old_ocean_profit_raw'))
             .alias('fb_old_ocean_profit_rate'),
            F.when(F.col('fb_old_air_profit_raw').isNull(), F.lit(-1000))
             .otherwise(F.col('fb_old_air_profit_raw'))
             .alias('fb_old_air_profit_rate'),

            # 功能5: 新品数量占比（流量选品口径）
            F.when(F.col('fb_flow_asin_total').isNull() | (F.col('fb_flow_asin_total') == 0), F.lit(-1.0))
            .otherwise(F.round(F.col('fb_new_asin_count_raw') / F.col('fb_flow_asin_total'), 4))
            .alias('fb_flow_new_asin_rate'),

            # 功能7: 店铺综合评分（各星级占比）
            F.col('fb_star_5_pct'),
            F.col('fb_star_4_pct'),
            F.col('fb_star_3_pct'),
            F.col('fb_star_2_pct'),
            F.col('fb_star_1_pct'),
            F.lit(self.site_name).alias('site_name'),
            F.lit(self.date_type).alias('date_type'),
            F.lit(self.date_info).alias('date_info')
        )

        print(f"清除hdfs目录中:{self.hdfs_path}")
        HdfsUtils.delete_file_in_folder(self.hdfs_path)
        df_save = df_save.repartition(3)
        partition_by = ["site_name", "date_type", "date_info"]
        print(f"当前存储的表名为：{self.hive_tb},分区为{partition_by}", )
        df_save.write.saveAsTable(name=self.hive_tb, format='hive', mode='append', partitionBy=partition_by)
        df_save.unpersist()
        print("success")

    # 输出数据集-asin_info
    def save_data_asin_info(self):
        # 只保留新品的详情
        self.df_fb_asin_detail = self.df_fb_asin_detail.filter('is_asin_new = 1')
        df_save_asin = self.df_fb_asin_detail.filter(F.col('account_name').isNotNull())

        df_save_asin = df_save_asin.select(
            F.col('seller_id'),
            F.col('account_name'),
            F.col('asin'),
            F.col('asin_title'),
            F.col('asin_launch_time'),
            F.col('is_asin_new'),
            F.col('asin_package_quantity'),
            F.col('is_pq_flag'),
            F.col('parent_asin'),
            F.col('asin_img_url'),
            F.col('asin_price'),
            F.col('asin_rating'),
            F.col('asin_total_comments'),
            F.col('asin_weight'),
            F.col('asin_volume'),
            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.lit(self.site_name).alias('site_name'),
            F.lit(self.date_type).alias('date_type'),
            F.lit(self.date_info).alias('date_info')
        )

        hive_tb_name = 'dwt_fb_asin_info'
        hdfs_path_asin_info = CommonUtil.build_hdfs_path(hive_tb_name, partition_dict=self.partition_dict)
        print(f"清除hdfs目录中:{hdfs_path_asin_info}")
        HdfsUtils.delete_file_in_folder(hdfs_path_asin_info)

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

    def run(self):
        self.read_data()
        self.handle_fb_top_20()
        self.handle_fb_cal_agg()
        self.save_data_report()
        # report 写完后 df_fb_feedback 和 df_flow_asin 不再需要，立即释放
        self.df_fb_feedback.unpersist()
        self.df_flow_asin.unpersist()
        self.save_data_asin_info()
        # asin_info 写完后 df_fb_asin_detail 不再需要，立即释放
        self.df_fb_asin_detail.unpersist()


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)  # 参数3：年-周/年-月/年-季/年-月-日, 比如: 2022-1
    obj = DwtFbBaseReport(site_name=site_name, date_type=date_type, date_info=date_info)
    obj.run()
