dim_asin_detail.py

"""
   @Author      : HuangJian
   @Description : 关键词与Asin详情维表
   @SourceTable :
                  ①ods_asin_keep_date
                  ②ods_asin_variat
                  ③ods_asin_detail
                  ④dwd_bs_category_asin
   @SinkTable   : dim_asin_detail
   @CreateTime  : 2022/11/14 15:56
   @UpdateTime  : 2022/11/14 15:56
"""

import os
import sys

sys.path.append(os.path.dirname(sys.path[0]))  # 上级目录


from pyspark.sql.window import Window
from pyspark.sql import functions as F
from utils.spark_util import SparkUtil
from pyspark.sql.types import *
from yswg_utils.common_udf import udf_handle_string_null_value as myUDF
from yswg_utils.common_df import get_node_first_id_df, get_first_id_from_category_desc_df
from utils.common_util import CommonUtil, DateTypes
from yswg_utils.common_udf import udf_new_asin_flag
from utils.hdfs_utils import HdfsUtils
from utils.db_util import DBUtil
from datetime import datetime
from pyspark.storagelevel import StorageLevel
from utils.DorisHelper import DorisHelper
from yswg_utils.common_udf import udf_get_package_quantity_with_flag as udf_get_package_quantity, udf_parse_seller_json, udf_parse_amazon_orders


class DimAsinDetail(object):

    def __init__(self, site_name, date_type, date_info):
        super().__init__()
        self.site_name = site_name
        self.date_type = date_type
        self.date_info = date_info
        self.hive_tb = f'dim_asin_detail'
        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)
        # 创建spark_session对象相关
        app_name = f"{self.__class__.__name__}:{site_name}:{date_info}"
        self.spark = SparkUtil.get_spark_session(app_name)
        self.complete_date_info_tuple = self.get_complete_week_tuple()
        self.date_sql = self.date_sql_padding()
        self.launch_time_upper_limit = self.get_effective_launch_time()
        self.launch_time_lower_limit = "1995-01-01"
        self.cal_date = CommonUtil.get_calDay_by_dateInfo(self.spark, self.date_type, self.date_info)
        self.partitions_by = ['site_name', 'date_type', 'date_info']
        self.partition_num = CommonUtil.reset_partitions(self.site_name, partitions_num=80)
        # Doris相关参数
        self.doris_db = "selection"
        self.parent_asin_latest_detail_table = f"{self.site_name}_parent_asin_latest_detail"
        # 自定义df相关
        self.df_asin_keep_date = self.spark.sql(f"select 1+1;")
        self.df_asin_detail = self.spark.sql(f"select 1+1;")
        self.df_asin_label = self.spark.sql("select 1+1;")
        self.df_asin_weight_and_volume = self.spark.sql("select 1+1;")
        self.df_asin_new_cate = self.spark.sql("select 1+1;")
        self.df_user_package_num = self.spark.sql(f"select 1+1;")
        self.df_self_asin = self.spark.sql(f"select 1+1;")
        self.df_asin_category = self.spark.sql(f"select 1+1;")
        self.df_asin_variat = self.spark.sql(f"select 1+1;")
        # 调用公用udf函数
        self.udf_new_asin_flag = F.udf(udf_new_asin_flag, IntegerType())
        self.handle_string_num_value = F.udf(myUDF, StringType())
        seller_schema = StructType([
            StructField("buy_box_seller_type", IntegerType(), True),
            StructField("account_name", StringType(), True),
            StructField("account_id", StringType(), True)
        ])
        self.u_parse_seller_info = self.spark.udf.register('u_parse_seller_info', udf_parse_seller_json, seller_schema)
        package_schema = StructType([
            StructField("parse_package_quantity", IntegerType(), True),
            StructField("is_package_quantity_abnormal", IntegerType(), True),
        ])
        self.u_judge_package_quantity = self.spark.udf.register('u_judge_package_quantity', udf_get_package_quantity, package_schema)
        self.u_parse_amazon_orders = F.udf(udf_parse_amazon_orders, IntegerType())

    def get_complete_week_tuple(self):
        complete_date_info_tuple = None
        df_date = self.spark.sql(f"select * from dim_date_20_to_30 ;")
        df = df_date.toPandas()
        if self.date_type == 'week':
            complete_date_info_tuple = f"('{self.date_info}')"
        elif self.date_type == '4_week':
            print(self.date_info)
            df_loc = df.loc[(df.year_week == f"{self.date_info}") & (df.week_day == 1)]
            cur_id = list(df_loc.id)[0]
            df_loc = df.loc[df.id == int(cur_id)]
            week1 = list(df_loc.year_week)[0]
            df_loc = df.loc[df.id == int(cur_id) - 7]
            week2 = list(df_loc.year_week)[0]
            df_loc = df.loc[df.id == int(cur_id) - 14]
            week3 = list(df_loc.year_week)[0]
            df_loc = df.loc[df.id == int(cur_id) - 21]
            week4 = list(df_loc.year_week)[0]
            complete_date_info_tuple = (week1, week2, week3, week4)
        elif self.date_type in ['month', 'month_week']:
            df_loc = df.loc[(df.year_month == f"{self.date_info}") & (df.week_day == 1)]
            complete_date_info_tuple = tuple(df_loc.year_week)
        print("self.complete_date_info_tuple:", complete_date_info_tuple)
        return complete_date_info_tuple

    def date_sql_padding(self):
        if 'us' == self.site_name:
            if self.date_type == DateTypes.month_week.name:
                date_sql = f" and date_type='{self.date_type}' and date_info = '{self.date_info}'"
            elif self.date_type == DateTypes.month.name and self.date_info >= '2023-10':
                date_sql = f" and date_type='{self.date_type}' and date_info = '{self.date_info}'"
            else:
                date_sql = f"and date_type='week' and date_info in {self.complete_date_info_tuple}"
        elif self.site_name in ['uk', 'de']:
            if self.date_type in [DateTypes.month.name, DateTypes.month_week.name] and self.date_info >= '2024-05':
                date_sql = f" and date_type='{self.date_type}' and date_info='{self.date_info}'"
            else:
                date_sql = f" and date_type='week' and date_info in {self.complete_date_info_tuple}"
        print(date_sql)
        return date_sql

    def get_effective_launch_time(self):
        year_upper_limit = datetime.now().year + 1
        launch_time_upper_limit = str(year_upper_limit) + "-01-01"
        return launch_time_upper_limit

    # 读取数据
    def read_data(self):
        # 获取dim层的dim_asin_launchtime_info
        print("1. 获取dim层的dim_asin_launchtime_info填充的上架日期")
        sql = f"""select asin, asin_launch_time as populated_asin_launch_time from dim_asin_launchtime_info where site_name='{self.site_name}'  and asin_launch_time is not null"""
        print(sql)
        self.df_asin_keep_date = self.spark.sql(sqlQuery=sql)
        self.df_asin_keep_date = self.df_asin_keep_date.repartition(100).persist(StorageLevel.DISK_ONLY)
        self.df_asin_keep_date.show(10, truncate=False)
        print("2. 获取ods_asin_detail")
        sql = f"""
            select asin, img_url as asin_img_url, lower(title) as asin_title, title_len as asin_title_len, 
            price as asin_price, rating as asin_rating, total_comments as asin_total_comments, 
            page_inventory as asin_page_inventory, category as asin_category_desc, 
            launch_time as crawl_asin_launch_time, img_num as asin_img_num, img_type as asin_img_type, 
            category_state as asin_category_state, material as asin_material, lower(brand) as asin_brand_name, 
            activity_type as asin_activity_type, one_two_val as act_one_two_val, three_four_val as act_three_four_val, 
            five_six_val as act_five_six_val, eight_val as act_eight_val, one_star, two_star, three_star, four_star, 
            five_star, low_star, together_asin, ac_name, node_id, data_type as asin_data_type, variat_list, 
            `describe` as asin_describe, follow_sellers as asin_follow_sellers, product_description, 
            image_view as asin_image_view, spider_int as asin_spider_num, buy_sales, lob_asin_json as asin_lob_info, 
            REGEXP_REPLACE(seller_json, chr(10), '') as seller_json, buy_box_seller_type as asin_buy_box_seller_type, customer_reviews_json, parent_asin, img_list, 
            created_at as created_time, updated_at as updated_time, updated_at as dt, variat_num as variation_num 
            from ods_asin_detail where site_name='{self.site_name}' {self.date_sql}"""
        print(sql)
        self.df_asin_detail = self.spark.sql(sqlQuery=sql)
        self.df_asin_detail = self.df_asin_detail.repartition(100).persist(StorageLevel.DISK_ONLY)
        self.df_asin_detail.show(10, truncate=False)
        print("3. 读取asin体积重量相关信息")
        sql = f"""
                SELECT asin, asin_volume, asin_volume_type, asin_length_sorted as asin_length, 
                asin_width_sorted as asin_width, asin_height_sorted as asin_height, 
                asin_weight, asin_weight_str, asin_weight_type FROM dim_asin_stable_info WHERE site_name='{self.site_name}'"""
        print(sql)
        self.df_asin_weight_and_volume = self.spark.sql(sqlQuery=sql)
        self.df_asin_weight_and_volume = self.df_asin_weight_and_volume.repartition(100).persist(StorageLevel.DISK_ONLY)
        self.df_asin_weight_and_volume.show(10, truncate=False)
        print("4. 获取用户修改打包数量信息")
        pg_con_info = DBUtil.get_connection_info("postgresql", "us")
        sql = f"""
               WITH ranked_edit_logs AS (SELECT edit_key_id, lower(val_related_info) AS val_related_info, val_after, 
               ROW_NUMBER() OVER (PARTITION BY edit_key_id ORDER BY create_time DESC) AS rn FROM sys_edit_log 
               WHERE module = '流量选品' AND filed = 'package_quantity' and site_name='{self.site_name}')
               SELECT edit_key_id AS asin, val_related_info AS asin_title, cast(val_after as int) AS user_package_num, 
               0 AS user_is_package_quantity_abnormal FROM ranked_edit_logs WHERE rn = 1"""
        print(sql)
        if pg_con_info is not None:
            df_user_package_num = SparkUtil.read_jdbc_query(session=self.spark, url=pg_con_info['url'], pwd=pg_con_info['pwd'], username=pg_con_info['username'], query=sql)
            self.df_user_package_num = F.broadcast(df_user_package_num)
            self.df_user_package_num.show(10, truncate=False)
        print("5.读取ods_self_asin,获得公司内部asin信息")
        sql = f"""select asin, 1 as asin_is_self from ods_self_asin where site_name='{self.site_name}' group by asin"""
        print("sql:" + sql)
        df_self_asin = self.spark.sql(sqlQuery=sql)
        self.df_self_asin = F.broadcast(df_self_asin)
        self.df_self_asin.show(10, truncate=False)
        print("6. node_id对应的头部分类信息")
        self.df_asin_new_cate = get_node_first_id_df(self.site_name, self.spark)
        self.df_asin_new_cate = self.df_asin_new_cate.filter('node_id is not null').persist(StorageLevel.DISK_ONLY)
        self.df_asin_new_cate.show(10, truncate=False)
        print("7. 获取asin的标签信息")
        sql = f"""
            select asin, asin_label_list, asin_label_type as asin_is_movie from dim_asin_label 
            where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info='{self.date_info}'
        """
        print("sql:" + sql)
        self.df_asin_label = self.spark.sql(sqlQuery=sql)
        self.df_asin_label = self.df_asin_label.repartition(100).persist(StorageLevel.DISK_ONLY)
        self.df_asin_label.show(10, truncate=False)
        print("8. 获取分类id和分类名称的对应关系")
        self.df_asin_category = get_first_id_from_category_desc_df(self.site_name, self.spark)
        self.df_asin_category = self.df_asin_category.withColumn(
            "category_first_name", F.lower("category_first_name")
        ).repartition(100).persist(StorageLevel.DISK_ONLY)
        self.df_asin_category.show(10, truncate=False)
        if self.date_type in ['month', 'month_week'] and self.date_info < '2024-06':
            sql = f"""
            SELECT asin, parent_asin, color as asin_color, `size` as asin_size, style as asin_style, 
            CASE WHEN state = 1 THEN 1 WHEN state = 2 THEN 0 ElSE NULL END as asin_is_sale, updated_time 
            FROM dim_asin_variation_info where site_name='{self.site_name}' and created_date >='2024-05-21'
            """
            print("sql:" + sql)
            self.df_asin_variat = self.spark.sql(sqlQuery=sql)
            self.df_asin_variat = self.df_asin_variat.repartition(100).persist(StorageLevel.DISK_ONLY)
            self.df_asin_variat.show(10, truncate=False)
        else:
            pass

    # asin详情数据去重
    def handle_df_duplicated(self):
        # 12月数据标记多个data_type用于区分改asin由什么场景抓取
        df_asin_datatype_handle = self.df_asin_detail.select("asin", "asin_data_type")
        df_asin_datatype_handle_agg = df_asin_datatype_handle.groupby(["asin"]).agg(
            F.concat_ws(",", F.collect_set("asin_data_type")).alias("asin_data_type"))
        # asin窗口内排序，按照dt降序
        window = Window.partitionBy(['asin']).orderBy(self.df_asin_detail.dt.desc_nulls_last())
        self.df_asin_detail = self.df_asin_detail.withColumn("dt_rank", F.row_number().over(window=window))
        # 取按asin分组的组内第一条，就是去重后的最新asin_detail
        self.df_asin_detail = self.df_asin_detail.filter("dt_rank=1").drop("dt", "dt_rank", "asin_data_type")
        self.df_asin_detail = self.df_asin_detail.repartition(100)
        df_asin_datatype_handle_agg = df_asin_datatype_handle_agg.repartition(100)
        self.df_asin_detail = self.df_asin_detail.join(df_asin_datatype_handle_agg, on=['asin'], how='left').persist(StorageLevel.MEMORY_ONLY)
        print("asin详情去重后的数量为： ", self.df_asin_detail.count())
        if self.date_type == 'month' and self.date_info == '2024-03':
            sql = f"""
                select asin from dwd_asin_measure where site_name='{self.site_name}' and date_type='{self.date_type}'
                and date_info='{self.date_info}' group by asin"""
            df_effective_asin = self.spark.sql(sql)
            self.df_asin_detail = self.df_asin_detail.join(df_effective_asin, on=['asin'], how='inner')

    # 处理asin的亚马逊月销信息
    def handle_asin_bought_month(self):
        self.df_asin_detail = self.df_asin_detail.withColumn("asin_bought_month", self.u_parse_amazon_orders(self.df_asin_detail.buy_sales))

    # 处理asin的lob_info信息
    def handle_asin_lob_info(self):
        self.df_asin_detail = self.df_asin_detail.withColumn(
            "is_contains_lob_info", F.when(F.col("asin_lob_info").isNotNull(), F.lit(1)).otherwise(F.lit(0)))
        df_asin_detail_parsed = self.df_asin_detail.withColumn(
            "parse_asin_lob", F.when(F.col("is_contains_lob_info") == 1, F.from_json("asin_lob_info", "array<struct<lob_asin:string>>")))
        df_asin_detail_result = df_asin_detail_parsed.withColumn("asin_lob_info", F.expr("transform(parse_asin_lob, x -> x.lob_asin)"))
        self.df_asin_detail = df_asin_detail_result.withColumn(
            "asin_lob_info", F.regexp_replace(F.concat_ws(",", "asin_lob_info"), "[{}]", "")).drop("parse_asin_lob")

    # 处理asin的变体信息
    def handle_asin_variation_attribute(self):
        if self.date_type in ['month', 'month_week'] and self.date_info >= '2024-06':
            print("执行新版的变体信息整合")
            variat_schema = ArrayType(ArrayType(StringType()))
            self.df_asin_detail = self.df_asin_detail.withColumn("variat_list_change", F.from_json(F.col("variat_list"), variat_schema))
            df_asin_with_variation = self.df_asin_detail.filter(F.size("variat_list_change") > 0).\
                select("asin", F.explode("variat_list_change").alias("variant_attribute")).\
                select("asin", F.col("variant_attribute")[0].alias("son_asin"),
                       F.col("variant_attribute")[1].alias("asin_color"),
                       F.col("variant_attribute")[3].alias("asin_size"),
                       F.col("variant_attribute")[4].alias("asin_is_sale"),
                       F.col("variant_attribute")[5].alias("asin_style"))
            df_asin_with_variation = df_asin_with_variation.filter(F.col("asin") == F.col("son_asin")).drop("son_asin", "variant_attribute")
            df_asin_with_variation = df_asin_with_variation.withColumn(
                "asin_is_sale",
                F.when(F.col("asin_is_sale")==1, F.lit(1)).when(F.col("asin_is_sale")==2, F.lit(0)).otherwise(F.lit(None))
            )
            self.df_asin_detail = self.df_asin_detail.join(
                df_asin_with_variation, on=['asin'], how='left'
            )
        elif self.date_type in ['month', 'month_week'] and self.date_info < '2024-06':
            print("执行历史数据的变体信息整合")
            window = Window.partitionBy(self.df_asin_variat.asin).orderBy(
                self.df_asin_variat.updated_time.desc_nulls_last())
            self.df_asin_variat = self.df_asin_variat.withColumn("t_rank", F.row_number().over(window=window))
            self.df_asin_variat = self.df_asin_variat.filter("t_rank = 1").drop("updated_time", "t_rank")
            self.df_asin_variat = self.df_asin_variat.repartition(100)
            self.df_asin_detail = self.df_asin_detail.drop("variation_num", "parent_asin")
            df_asin_variat = self.df_asin_variat.filter("parent_asin is not null").select("asin", "parent_asin")
            df_parent_asin_agg = df_asin_variat.groupby(['parent_asin']).agg(F.count("asin").alias("variation_num"))
            df_parent_asin_agg = df_parent_asin_agg.repartition(100)
            self.df_asin_variat = self.df_asin_variat.join(df_parent_asin_agg, on=['parent_asin'], how='left')
            self.df_asin_detail = self.df_asin_detail.join(self.df_asin_variat, on=['asin'], how='left').cache()
            self.df_asin_detail.show(10, truncate=False)
            self.df_asin_variat.unpersist()
        else:
            pass

    # 处理asin的配送方式信息
    def handle_asin_buy_box_seller_type(self):
        if (self.date_type in ['month', 'month_week'] and self.date_info >= '2024-05') \
                or (self.date_type == '4_week' and self.date_info >= '2024-21'):
            self.df_asin_detail = self.df_asin_detail.withColumn(
                'seller_json',
                F.when(
                    F.length(F.trim(F.regexp_replace('seller_json', chr(10), ''))) == 0, F.lit('')
                       ).otherwise(F.regexp_replace('seller_json', chr(10), '')))
            self.df_asin_detail = self.df_asin_detail.drop("asin_buy_box_seller_type")
            self.df_asin_detail = self.df_asin_detail.withColumn("seller_json_parsed", self.u_parse_seller_info(self.df_asin_detail.seller_json))
            self.df_asin_detail = self.df_asin_detail.withColumn(
                "asin_buy_box_seller_type", self.df_asin_detail.seller_json_parsed.buy_box_seller_type).withColumn(
                "account_name", self.df_asin_detail.seller_json_parsed.account_name).withColumn(
                "account_id", self.df_asin_detail.seller_json_parsed.account_id).drop("seller_json_parsed")
        else:
            self.df_asin_detail = self.df_asin_detail.withColumn("account_id", F.lit(None)).\
                withColumn("seller_json", F.lit(None)).withColumn("account_name", F.lit(None))

    # 处理asin体积重量信息
    def handle_asin_basic_attribute(self):
        self.df_asin_detail = self.df_asin_detail.repartition(100)
        df_weight = self.df_asin_weight_and_volume.filter("asin_weight is not null").select("asin", "asin_weight", "asin_weight_str", "asin_weight_type")
        self.df_asin_detail = self.df_asin_detail.join(df_weight, on=['asin'], how='left')
        if self.site_name == 'us':
            df_volume = self.df_asin_weight_and_volume.filter("asin_volume_type is not null and asin_volume_type not in ('cm', 'none')").select("asin", "asin_volume", "asin_length", "asin_width", "asin_height")
        else:
            df_volume = self.df_asin_weight_and_volume.filter("asin_volume_type is not null and asin_volume_type not in ('inches', 'none')").select("asin", "asin_volume", "asin_length", "asin_width", "asin_height")
        self.df_asin_detail = self.df_asin_detail.join(df_volume, on=['asin'], how='left').drop("asin_volume_type")
        self.df_asin_weight_and_volume.unpersist()

    # 打包数量解析
    def get_package_quantity(self):
        self.df_asin_detail = self.df_asin_detail.repartition(100)
        self.df_user_package_num = self.df_user_package_num.repartition(100)
        self.df_asin_detail = self.df_asin_detail.withColumn("variat_attribute", F.concat_ws("&&&%", F.col("asin_color"), F.col("asin_style"), F.col("asin_size"), F.col("asin_material")))
        self.df_asin_detail = self.df_asin_detail.withColumn(
            "title_parse", self.u_judge_package_quantity(self.df_asin_detail.asin_title)).withColumn(
            "variat_parse", self.u_judge_package_quantity(self.df_asin_detail.variat_attribute))
        self.df_asin_detail = self.df_asin_detail.withColumn(
            "title_package_quantity", self.df_asin_detail.title_parse.getField("parse_package_quantity")).withColumn(
            "variat_package_quantity", self.df_asin_detail.variat_parse.getField("parse_package_quantity")).withColumn(
            "title_package_quantity_is_abnormal", self.df_asin_detail.title_parse.getField("is_package_quantity_abnormal")
        ).withColumn(
            "variat_package_quantity_is_abnormal", self.df_asin_detail.variat_parse.getField("is_package_quantity_abnormal")
        ).drop("title_parse", "variat_parse", "variat_attribute")
        self.df_asin_detail = self.df_asin_detail.withColumn(
            "package_quantity", F.expr(""" CASE
                     WHEN title_package_quantity is null and variat_package_quantity is not null THEN variat_package_quantity
                     WHEN title_package_quantity is not null THEN title_package_quantity
                     ELSE 1 END""")).withColumn(
            "is_package_quantity_abnormal", F.expr("""CASE
                     WHEN title_package_quantity is null  and variat_package_quantity is not null THEN variat_package_quantity_is_abnormal
                     WHEN title_package_quantity is not null THEN title_package_quantity_is_abnormal
                     ELSE 2 END""")).drop("title_package_quantity", "variat_package_quantity", "title_package_quantity_is_abnormal", "variat_package_quantity_is_abnormal")
        self.df_asin_detail = self.df_asin_detail.join(self.df_user_package_num, on=['asin', 'asin_title'], how='left')
        self.df_asin_detail = self.df_asin_detail.withColumn(
            "package_quantity", F.coalesce(F.col("user_package_num"), F.col("package_quantity"))).withColumn(
            "is_package_quantity_abnormal", F.coalesce(F.col("user_is_package_quantity_abnormal"), F.col("is_package_quantity_abnormal"))).\
            drop("user_package_num", "user_is_package_quantity_abnormal").persist(StorageLevel.MEMORY_ONLY)
        print("打包数量解析完毕")
        self.df_asin_detail.select("asin", "asin_title", "package_quantity", "is_package_quantity_abnormal").show(10, truncate=False)
        self.df_user_package_num.unpersist()

    # 补充asin的一级分类ID(通过node_id以及分类描述补充)
    def handle_asin_top_category(self):
        self.df_asin_detail = self.df_asin_detail.join(self.df_asin_new_cate, on=['node_id'], how='left')
        self.df_asin_detail = self.df_asin_detail.withColumn("category_id", F.col("node_id"))
        df_asin_with_category_desc = self.df_asin_detail.filter("category_first_id is null and asin_category_desc is not null").select("asin", "asin_category_desc")
        df_asin_with_category_desc = df_asin_with_category_desc.withColumn(
            "asin_category_split", F.split(F.col("asin_category_desc"), "›")
        ).withColumn("category_first_name", F.lower(F.col("asin_category_split").getItem(0))).drop("asin_category_split", "asin_category_desc")
        df_asin_with_category_desc = df_asin_with_category_desc.join(self.df_asin_category, on=['category_first_name'], how='inner')
        df_asin_with_category_desc = df_asin_with_category_desc.withColumnRenamed("category_first_id", "category_first_id_with_name").drop("category_first_name")
        self.df_asin_detail = self.df_asin_detail.join(df_asin_with_category_desc, on=['asin'], how='left')
        self.df_asin_detail = self.df_asin_detail.withColumn("category_first_id", F.coalesce(F.col("category_first_id"), F.col("category_first_id_with_name"))).drop("category_first_id_with_name")
        self.df_asin_new_cate.unpersist()
        self.df_asin_category.unpersist()

    # 处理asin上架时间信息
    def handle_asin_launch_time(self):
        # 根据asin，且launch_time为空的，去找keep_date补全launch_time
        self.df_asin_detail = self.df_asin_detail.repartition(100)
        self.df_asin_detail = self.df_asin_detail.join(self.df_asin_keep_date, on='asin', how='left')
        self.df_asin_detail = self.df_asin_detail.withColumn(
            "crawl_asin_launch_time", F.when(
                (F.col("crawl_asin_launch_time") <= self.launch_time_upper_limit) &
                (F.col("crawl_asin_launch_time") >= self.launch_time_lower_limit),
                F.col("crawl_asin_launch_time")
            ).otherwise(F.lit(None))
        )
        self.df_asin_detail = self.df_asin_detail.withColumn(
            "asin_launch_time", F.when((F.isnull("crawl_asin_launch_time")) | (F.col("crawl_asin_launch_time") == 'null'), F.col("populated_asin_launch_time")).otherwise(F.col("crawl_asin_launch_time")))
        self.df_asin_detail = self.df_asin_detail.withColumn(
            "asin_launch_time", F.when(
                (F.col("asin_launch_time") <= self.launch_time_upper_limit) &
                (F.col("asin_launch_time") >= self.launch_time_lower_limit),
                F.col("asin_launch_time")
            ).otherwise(F.lit(None))
        )
        self.df_asin_keep_date.unpersist()

    # 处理asin各类型信息
    def handle_asin_flag(self):
        # 生成is_asin_new字段（是否asin新品标记）
        self.df_asin_detail = self.df_asin_detail.withColumn(
            "asin_is_new", self.udf_new_asin_flag(F.col('asin_launch_time'), F.lit(self.cal_date)))\
            .withColumn("asin_is_aadd", F.expr(f"""CASE WHEN INSTR(asin_img_type, '3') > 0 THEN 1 ELSE 0 END"""))\
            .withColumn("asin_is_video", F.expr(f"""CASE WHEN INSTR(asin_img_type, '2') > 0 THEN 1 ELSE 0 END"""))\
            .withColumn("asin_is_picture", F.expr(f"""CASE WHEN INSTR(asin_img_type, '1') > 0 THEN 1 ELSE 0 END"""))\
            .withColumn("asin_is_amazon", F.expr(f"""CASE WHEN asin_buy_box_seller_type == 1 THEN 1 ELSE 0 END"""))\
            .withColumn("asin_is_fba", F.expr(f"""CASE WHEN asin_buy_box_seller_type == 2 THEN 1 ELSE 0 END"""))\
            .withColumn("asin_is_fbm", F.expr(f"""CASE WHEN asin_buy_box_seller_type == 3 THEN 1 ELSE 0 END"""))\
            .withColumn("asin_is_other", F.expr(f"""CASE WHEN asin_buy_box_seller_type == 4 THEN 1 ELSE 0 END"""))\
            .withColumn("asin_is_brand", F.when((F.col("asin_brand_name").cast("string") != 'null') &
                                                (F.col("asin_brand_name").cast("string") != 'none'), 1).otherwise(F.lit(0)))\
            .withColumn("asin_quantity_variation_type", F.when((F.lower(F.col("asin_size").cast("string")) != 'null') &
                               (F.lower(F.col("asin_size").cast("string")) != 'none') &
                               (F.lower(F.col("asin_size").cast("string")).contains('quantity')), 1).otherwise(F.lit(0)))
        if self.site_name == 'us':
            pg_sql = f"""
            select asin_brand_name, 1 as asin_is_alarm from (select lower(trim(brand_name)) as asin_brand_name from brand_alert_erp where brand_name is not null) t group by asin_brand_name"""
            db_type = "postgresql_cluster"
            con_info = DBUtil.get_connection_info(db_type=db_type, site_name=self.site_name)
            if con_info is not None:
                df_alarm_brand = SparkUtil.read_jdbc_query(
                    session=self.spark, url=con_info['url'], pwd=con_info['pwd'], username=con_info['username'], query=pg_sql)
                df_alarm_brand = df_alarm_brand.repartition(100)
                self.df_asin_detail = self.df_asin_detail.join(df_alarm_brand, on=['asin_brand_name'], how='left')
        else:
            self.df_asin_detail = self.df_asin_detail.withColumn("asin_is_alarm", F.lit(0))
        self.df_asin_detail = self.df_asin_detail.na.fill({"asin_is_alarm": 0})
        # 处理是否内部asin信息
        self.df_asin_detail = self.df_asin_detail.join(self.df_self_asin, on=['asin'], how='left')
        self.df_asin_detail = self.df_asin_detail.na.fill({"asin_is_self": 0})
        self.df_self_asin.unpersist()

    # 处理影视标签字段
    def handle_asin_label(self):
        self.df_asin_detail = self.df_asin_detail.join(self.df_asin_label, on=['asin'], how='left')
        self.df_asin_label.unpersist()

    # 处理asin小图信息
    def handle_asin_img_info(self):
        if self.date_type in ['month', 'month_week'] and self.date_info >= '2024-06':
            img_schema = ArrayType(ArrayType(StringType()))
            df_asin_with_img = self.df_asin_detail.withColumn("img_list", F.from_json(F.col("img_list"), img_schema)).filter(F.size("img_list") > 0).\
                select("asin", F.explode("img_list").alias("img_attributes")).\
                select("asin", F.col("img_attributes")[1].alias("img_url"), F.col("img_attributes")[2].alias("img_order_by"),
                       F.col("img_attributes")[3].alias("data_type"))
            df_asin_with_img_agg = df_asin_with_img.groupby("asin").agg(
                F.to_json(F.collect_list(F.struct(F.col("img_url"), F.col("img_order_by"), F.col("data_type")))).alias("img_list")
            )
            self.df_asin_detail = self.df_asin_detail.drop("img_list")
            self.df_asin_detail = self.df_asin_detail.join(df_asin_with_img_agg, on=['asin'], how='left')
        else:
            pass

    # 处理parent_asin下最新变体信息
    def handle_latest_variation_info(self):
        if self.date_type in ['month', 'month_week'] and self.date_info >= '2024-06':
            max_report_sql = f"""
                SELECT MAX(date_info) as table_date_info FROM {self.doris_db}.{self.parent_asin_latest_detail_table}
            """
            df_date_info = DorisHelper.spark_import_with_sql(self.spark, query=max_report_sql)
            table_date_info = df_date_info.take(1)[0]['table_date_info']
            print("doris中记录最新的日期为：", table_date_info)
            if self.date_info >= table_date_info:
                df_asin_variat = self.df_asin_detail.filter("parent_asin is not null").select("parent_asin", "variat_list_change", "created_time")
                latest_asin_window = Window.partitionBy('parent_asin').orderBy(
                    F.desc_nulls_last("created_time")
                )
                df_asin_variat = df_asin_variat.withColumn("p_rank", F.row_number().over(window=latest_asin_window))
                df_asin_variat = df_asin_variat.filter("p_rank = 1").drop("p_rank")
                df_asin_variat =df_asin_variat.filter(F.size("variat_list_change") > 0). \
                    select("parent_asin", "created_time", F.explode("variat_list_change").alias("variant_attribute")). \
                    select("parent_asin", "created_time", F.col("variant_attribute")[0].alias("asin"),
                           F.col("variant_attribute")[1].alias("color"), F.col("variant_attribute")[3].alias("size"),
                           F.col("variant_attribute")[5].alias("style"))
                df_asin_variat_agg = df_asin_variat.groupby(['parent_asin']).agg(
                    F.first("created_time").alias("asin_crawl_date"),
                    F.concat_ws(',', F.collect_list("asin")).alias("variation_info"),
                    F.to_json(F.collect_list(F.struct(F.col("color"), F.col("size"), F.col("style")))).alias("attr_info")
                )
                print("导出父ASIN最新变体信息到doris：")
                df_doris = df_asin_variat_agg.select(
                    "parent_asin", F.lit(self.date_info).alias("date_info"), "asin_crawl_date", "variation_info", "attr_info")
                table_columns="parent_asin, date_info, asin_crawl_date, variation_info, attr_info"
                DorisHelper.spark_export_with_columns(df_save=df_doris, db_name=self.doris_db, table_name=self.parent_asin_latest_detail_table, table_columns=table_columns)
            else:
                print("不用导出旧数据到doris中")
                pass
        else:
            pass

    # 字段标准化及存储
    def df_save(self):
        df_save = self.df_asin_detail \
            .select("asin",
                    self.handle_string_num_value('asin_title').alias('asin_title'),
                    "asin_title_len",
                    F.when(F.col('asin_price') < 0, F.lit(None)).otherwise(F.col('asin_price')).alias('asin_price'),
                    "asin_rating", "asin_total_comments", "asin_buy_box_seller_type", "asin_page_inventory",
                    self.handle_string_num_value('asin_category_desc').alias('asin_category_desc'),
                    self.handle_string_num_value('asin_volume').alias('asin_volume'),
                    "asin_weight", "asin_color", "asin_size", "asin_style", "asin_is_sale",
                    F.lit(None).alias("asin_rank"),
                    self.handle_string_num_value('asin_launch_time').alias('asin_launch_time'),
                    "asin_is_new", "asin_img_num",
                    self.handle_string_num_value('asin_img_type').alias('asin_img_type'),
                    "asin_category_state",
                    self.handle_string_num_value('asin_material').alias('asin_material'),
                    self.handle_string_num_value('asin_brand_name').alias('asin_brand_name'),
                    F.lit(None).alias("bsr_cate_1_id"),
                    F.lit(None).alias("bsr_cate_current_id"),
                    self.handle_string_num_value('asin_activity_type').alias('asin_activity_type'),
                    "act_one_two_val", "act_three_four_val", "act_five_six_val", "act_eight_val",
                    F.lit(None).alias("qa_num"),
                    "one_star", "two_star", "three_star", "four_star", "five_star", "low_star",
                    self.handle_string_num_value('together_asin').alias('together_asin'),
                    self.handle_string_num_value('ac_name').alias('ac_name'),
                    self.handle_string_num_value('node_id').alias('node_id'),
                    "asin_data_type",
                    F.lit(None).alias('sp_num'),
                    self.handle_string_num_value('asin_describe').alias('asin_describe'),
                    "asin_is_amazon", "asin_is_fba", "asin_is_fbm", "asin_is_other", "asin_is_picture", "asin_is_video",
                    "asin_is_aadd",
                    self.handle_string_num_value('asin_img_url').alias('asin_img_url'),
                    "account_id", "account_name", "variation_num", "asin_is_brand", "asin_is_alarm", "created_time",
                    "updated_time", "parent_asin", "asin_is_movie",
                    self.handle_string_num_value('asin_label_list').alias('asin_label_list'),
                    self.handle_string_num_value('asin_weight_type').alias('asin_weight_type'),
                    self.handle_string_num_value('asin_weight_str').alias('asin_weight_str'),
                    F.lit(None).alias('asin_package_quantity'),
                    F.lit(None).alias('asin_pattern_name'),
                    "category_id", "category_first_id",
                    F.lit(None).alias("buy_data_bought_month"),
                    F.lit(None).alias("buy_data_bought_week"),
                    F.lit(None).alias("buy_data_viewed_month"),
                    F.lit(None).alias("buy_data_viewed_week"),
                    "crawl_asin_launch_time", "populated_asin_launch_time", "asin_follow_sellers", "asin_image_view",
                    "product_description", "asin_spider_num", "asin_lob_info", "is_contains_lob_info",
                    "package_quantity", "is_package_quantity_abnormal", "asin_quantity_variation_type", "seller_json",
                    "asin_bought_month", "asin_length", "asin_width", "asin_height", "asin_is_self",
                    "customer_reviews_json", "img_list", "variat_list",
                    F.lit(self.site_name).alias('site_name'),
                    F.lit(self.date_type).alias('date_type'),
                    F.lit(self.date_info).alias('date_info')).persist(StorageLevel.MEMORY_ONLY)
        print("dim_asin_detail处理完毕, 最后的数据量为： ", df_save.count())
        df_save = df_save.filter(F.length(F.col("asin")) <= 10)
        df_save = df_save.repartition(100)
        df_save.show(10, truncate=False)
        print(f"清除hdfs目录中:{self.hdfs_path}")
        HdfsUtils.delete_file_in_folder(self.hdfs_path)
        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=self.partitions_by)
        print("success")

    def run(self):
        # 读取数据
        self.read_data()
        # asin详情数据去重
        self.handle_df_duplicated()
        # 处理asin的亚马逊月销信息
        self.handle_asin_bought_month()
        # 处理asin的lob_info信息
        self.handle_asin_lob_info()
        # 处理asin的变体信息
        self.handle_asin_variation_attribute()
        # 处理asin的配送方式信息
        self.handle_asin_buy_box_seller_type()
        # 处理asin体积重量信息
        self.handle_asin_basic_attribute()
        # 打包数量解析
        self.get_package_quantity()
        # 获取node_id对应的分类信息(头部分类)
        self.handle_asin_top_category()
        # 处理asin上架时间信息
        self.handle_asin_launch_time()
        # 处理asin各类型信息
        self.handle_asin_flag()
        # 处理影视标签字段
        self.handle_asin_label()
        # 处理asin小图信息
        self.handle_asin_img_info()
        # 处理parent_asin下最新变体信息
        self.handle_latest_variation_info()
        # 字段标准化及存储
        self.df_save()


if __name__ == '__main__':
    site_name = sys.argv[1]  # 参数1：站点
    date_type = sys.argv[2]  # 参数2：类型：week/4_week/month/quarter/day
    date_info = sys.argv[3]  # 参数3：年-周/年-月/年-季/年-月-日, 比如: 2022-1
    handle_obj = DimAsinDetail(site_name=site_name, date_type=date_type, date_info=date_info)
    handle_obj.run()