dim_asin_stable_info.py

import os
import re
import sys

os.environ["PYARROW_IGNORE_TIMEZONE"] = "1"
sys.path.append(os.path.dirname(sys.path[0]))  # 上级目录
from utils.templates import Templates
# from ..utils.templates import Templates
from pyspark.sql.types import StringType, BooleanType, StructType, StructField, DoubleType, FloatType

# 分组排序的udf窗口函数
from pyspark.sql.window import Window
from pyspark.sql import functions as F
from yswg_utils.common_udf import parse_weight_str


class DimAsinStableInfo(Templates):

    def __init__(self, site_name='us'):
        super().__init__()
        self.site_name = site_name
        self.db_save = f'dim_asin_stable_info'
        self.spark = self.create_spark_object(app_name=f"{self.db_save}: {self.site_name}")
        self.df_asin_detail = self.spark.sql(f"select 1+1;")
        self.df_theme = self.spark.sql(f"select 1+1;")
        self.df_asin_img_url = self.spark.sql(f"select 1+1;")
        self.df_asin_title = self.spark.sql(f"select 1+1;")
        self.df_asin_weight = self.spark.sql(f"select 1+1;")
        self.df_asin_weight_new = self.spark.sql(f"select 1+1;")
        self.df_asin_weight_old = self.spark.sql(f"select 1+1;")
        self.df_asin_volume = self.spark.sql(f"select 1+1;")
        self.df_save = self.spark.sql(f"select 1+1;")
        self.df_save_std = self.spark.sql(f"select * from ods_search_term_zr limit 0;")
        self.partitions_by = ['site_name']
        self.reset_partitions(100)
        self.window = Window.partitionBy(['asin']).orderBy(F.desc("date_info"))  # 按照 date_info 列进行分区，并按照 date 列进行排序
        # self.window = Window.partitionBy(['asin']).orderBy(F.desc("created_time"))  # 按照 date_info 列进行分区，并按照 date 列进行排序

        schema = StructType([
            StructField('weight', FloatType(), True),
            StructField('weight_type', StringType(), True),
        ])
        self.u_get_weight = F.udf(parse_weight_str, schema)
        self.weight_type = 'pounds' if site_name == 'us' else 'grams'

        self.db_save_vertical = f'dim_asin_title_info_vertical'  # 主题竖表

        # 注册自定义函数 (UDF)
        self.u_contains_digit_udf = F.udf(self.udf_contains_digit, BooleanType())
        self.u_asin_volume_type = F.udf(self.udf_asin_volume_type, StringType())
        # 定义 UDF 的返回类型，即一个包含三个 DoubleType 字段的 StructType
        schema = StructType([
            StructField('length', DoubleType(), True),
            StructField('width', DoubleType(), True),
            StructField('height', DoubleType(), True)
        ])
        self.u_extract_dimensions = F.udf(self.udf_extract_dimensions, schema)
        self.u_extract_dimensions_others = F.udf(self.udf_extract_dimensions_others, schema)
        schema = StructType([
            StructField('asin_length', StringType(), True),
            StructField('asin_width', StringType(), True),
            StructField('asin_height', StringType(), True)
        ])
        self.u_volume_sorted = F.udf(self.udf_volume_sorted, schema)

        # 注册自定义函数 (UDF)
        self.u_theme_pattern = F.udf(self.udf_theme_pattern, StringType())
        # 其他变量
        # self.pattern = str()  # 正则匹配
        self.theme_list_str = str()  # 正则匹配

    @staticmethod
    def udf_theme_pattern(title, theme_list_str):
        found_themes = [theme.strip() for theme in eval(theme_list_str) if theme in title]
        if found_themes:
            return ','.join(set(found_themes))
        else:
            return None

    # 定义一个函数，检查字符串中是否包含数字
    @staticmethod
    def udf_contains_digit(s):
        # return any(char.isdigit() for char in s)

        if s is None:
            return False
        return any(char.isdigit() for char in s)

    # 定义一个函数，将asin_volume进行分类
    @staticmethod
    def udf_asin_volume_type(x):
        # pattern = r'\b\w+\b'
        pattern = r'[a-z]+'
        matches = re.findall(pattern, x)

        # 使用集合存储匹配的单词
        type_set = set()
        for word in matches:
            if word in ['inches', 'inch']:
                type_set.add('inches')
            elif word in ['cm', 'centímetros', 'centimetres']:
                type_set.add('cm')
            elif word in ['milímetros', 'millimeter', 'mm']:
                type_set.add('mm')
            elif word in ['metros']:
                type_set.add('m')

        # 根据集合的长度返回结果
        if len(type_set) == 1:
            return list(type_set)[0]
        elif len(type_set) >= 2:
            return ','.join(type_set)
        else:
            return 'none'

    @staticmethod
    def udf_extract_dimensions(volume_str, asin_volume_type):
        length, width, height = None, None, None
        dimensions = []
        if asin_volume_type == 'cm,inches':
            num_inches = volume_str.find('inch')
            num_cm = volume_str.find('cm')
            volume_str = volume_str[:num_inches] if num_cm > num_inches else volume_str[num_cm:num_inches]
        dimensions = re.findall(r"(\d+(\.\d+)?)", volume_str)
        dimensions = [float(dim[0]) for dim in dimensions]
        # if asin_volume_type == 'inches':
        #     dimensions = volume_str.split(' x ')
        #     dimensions = [dim.split()[0] for dim in dimensions]
        #     dimensions = [float(dim) if dim.replace('.', '', 1).isdigit() else None for dim in dimensions]
        # else:
        #     if asin_volume_type == 'cm,inches':
        #         # 保留inches
        #         num_inches = volume_str.find('inch')
        #         num_cm = volume_str.find('cm')
        #         volume_str = volume_str[:num_inches] if num_cm > num_inches else volume_str[num_cm:num_inches]
        #
        #     dimensions = re.findall(r"(\d+(\.\d+)?)", volume_str)
        #     dimensions = [float(dim[0]) for dim in dimensions]

        if len(dimensions) == 1:
            length = dimensions[0]
        elif len(dimensions) == 2:
            if asin_volume_type == 'none':
                if "l" in volume_str and "w" in volume_str:
                    length, width = dimensions
                elif "w" in volume_str and "h" in volume_str:
                    width, height = dimensions
                elif "l" in volume_str and "h" in volume_str:
                    length, height = dimensions
                elif "d" in volume_str and "w" in volume_str:
                    length, width = dimensions
                elif "d" in volume_str and "h" in volume_str:
                    length, height = dimensions
            else:
                length, width = dimensions
        elif len(dimensions) == 3:
            length, width, height = dimensions
        elif len(dimensions) >= 4:
            length, width, height = dimensions[:3]
        return (length, width, height)

    @staticmethod
    def udf_extract_dimensions_others(volume_str, asin_volume_type):
        length, width, height = None, None, None
        if asin_volume_type == 'cm':
            dimensions = re.findall(r"(\d+(\.\d+)?)", volume_str)
            dimensions = [float(dim[0]) for dim in dimensions]
            if len(dimensions) == 1:
                length = dimensions[0]
            elif len(dimensions) == 2:
                length = dimensions[0]
                width = dimensions[1]
            elif len(dimensions) >= 3:
                length, width, height = dimensions[:3]
        return (length, width, height)

    @staticmethod
    def udf_volume_sorted(asin_length, asin_weight, asin_height):
        # 如果输入数据中有None，替换为0
        dimensions = [0 if x is None else x for x in [asin_length, asin_weight, asin_height]]
        # 对数据进行排序
        dimensions.sort(reverse=True)
        return tuple(dimensions)

    def sort_by_latest(self, df):
        df = df.withColumn('row_number', F.row_number().over(self.window))  # 使用窗口函数为每个分区的行编号
        df = df.filter(df.row_number == 1).drop('row_number')  # 只保留每个分区中 row_number 最大的行，并删除 row_number 列
        return df

    def read_data(self):
        if self.site_name == 'us':
            params = f" and (date_type='week' or (date_type='month' and date_info='2023-10') or (date_type in ('month_week', 'month') and date_info>='2023-11'))"
        else:
            params = f" and (date_type='week' or (date_type in ('month_week', 'month') and date_info>='2023-05'))"

        sql = f"select asin, img_url as asin_img_url, title as asin_title, weight, weight_str, volume as asin_volume, date_type, date_info, site_name, created_at as created_time " \
              f"from ods_asin_detail where site_name='{self.site_name}' {params}"  #  and date_info='2023-27' limit 100000
        print("sql:", sql)
        self.df_asin_detail = self.spark.sql(sql).cache()
        self.df_asin_detail.show(10, truncate=False)
        params = "" if self.site_name == 'us' else " limit 10"
        sql = f"select id as theme_id, theme_type_en, theme_en, theme_en_lower, theme_ch from ods_theme where site_name='us' {params};"
        print("sql:", sql)
        self.df_theme = self.spark.sql(sql).cache()
        self.df_theme.show(10, truncate=False)

    def handle_data(self):
        # 根据created_time时间来去重保留最新
        window = Window.partitionBy(['asin']).orderBy(F.desc("created_time"))  # 按照 date_info 列进行分区，并按照 date 列进行排序
        self.df_asin_detail = self.df_asin_detail.withColumn('row_number', F.row_number().over(window))  # 使用窗口函数为每个分区的行编号
        self.df_asin_detail = self.df_asin_detail.filter(self.df_asin_detail.row_number == 1).drop('row_number', 'created_time')  # 只保留每个分区中 row_number 最大的行，并删除 row_number 列

        self.handle_img_url()
        if self.site_name == 'usx':
            self.handle_title()
        else:
            sql = f"select asin, asin_title, asin_title_lower, crowd_counts, crowd_ids, element_counts, element_ids, festival_counts, festival_ids, sports_counts, sports_ids, style_counts, style_ids, " \
                  f"theme_counts, theme_ids, material_counts, material_ids, date_info_title from {self.db_save} limit 0"
            self.df_asin_title = self.spark.sql(sql).cache()

        self.handle_weight()
        self.handle_volume()
        self.df_save = self.df_asin_detail.select("asin", "date_info", "site_name")
        self.df_save = self.sort_by_latest(df=self.df_save)
        self.df_save = self.df_save.join(
            self.df_asin_img_url, on='asin', how='left'
        ).join(
            self.df_asin_title, on='asin', how='left'
        ).join(
            self.df_asin_weight, on='asin', how='left'
        ).join(
            self.df_asin_volume, on='asin', how='left'
        )
        # if self.site_name != 'us':
        #     # 由于其他站点没有这些主题数据
        #     self.df_save = self.df_save_std.unionByName(self.df_save, allowMissingColumns=True)
        # self.df_save = self.df_save.drop("created_time")

        print("self.df_save.columns:", self.df_save.columns)
        # self.df_save.show(10, truncate=False)

    def handle_img_url(self):
        self.df_asin_img_url = self.df_asin_detail.select("asin", "asin_img_url", "date_info").filter("asin_img_url is not null")
        self.df_asin_img_url = self.df_asin_img_url.filter(self.df_asin_img_url.asin_img_url.contains('amazon'))  # 保留包含amazon的字符串记录
        self.df_asin_img_url = self.sort_by_latest(df=self.df_asin_img_url)
        for i in range(1, 10, 1):
            self.df_asin_img_url = self.df_asin_img_url.withColumn(f"asin_trun_{i}", F.substring(self.df_asin_img_url.asin, 1, i))
        self.df_asin_img_url = self.df_asin_img_url.withColumn(
            "asin_img_path",
            F.concat(
                F.lit("/"), self.df_asin_img_url.asin_trun_1,
                F.lit("/"), self.df_asin_img_url.asin_trun_2,
                F.lit("/"), self.df_asin_img_url.asin_trun_3,
                F.lit("/"), self.df_asin_img_url.asin_trun_4,
                F.lit("/"), self.df_asin_img_url.asin_trun_5,
                F.lit("/"), self.df_asin_img_url.asin_trun_6,
                F.lit("/")
            )
        )
        self.df_asin_img_url = self.df_asin_img_url.withColumnRenamed("date_info", "date_info_img_url")
        print("self.df_asin_img_url.columns:", self.df_asin_img_url.columns)
        # self.df_asin_img_url.show(10, truncate=False)

    def handle_title(self):
        # 过滤null和none字符串
        self.df_asin_title = self.df_asin_detail.select("asin", "asin_title", "date_info").filter("asin_title is not null and asin_title not in ('none', 'null', 'nan')")
        # 小写
        self.df_asin_title = self.df_asin_title.withColumn("asin_title_lower", F.lower(self.df_asin_title["asin_title"]))  # 小写
        # self.df_asin_title.show(10, truncate=False)
        # 取最新的date_info对应的title
        self.df_asin_title = self.sort_by_latest(df=self.df_asin_title)
        # self.df_asin_title.show(10, truncate=False)
        # 匹配主题数据
        self.handle_title_theme()
        # 存储一份主题竖表数据
        self.reset_partitions(partitions_num=100)
        self.save_data_common(
            df_save=self.df_save_vertical,
            db_save=self.db_save_vertical,
            partitions_num=self.partitions_num,
            partitions_by=self.partitions_by
        )

    def handle_title_theme(self):
        pdf_theme = self.df_theme.toPandas()
        theme_list = list(set(pdf_theme.theme_en_lower))
        self.theme_list_str = str([f" {theme} " for theme in theme_list])
        print("self.theme_list_str:", self.theme_list_str[:100])
        # 匹配宽表时用到
        df_asin_title = self.df_asin_title.cache()  # 后面用作匹配asin_title

        self.df_asin_title = self.df_asin_title.withColumn("asin_title_lower", F.concat(F.lit(" "), "asin_title_lower", F.lit(" ")))  # 标题两头加空字符串用来匹配整个词
        self.df_asin_title = self.df_asin_title.withColumn("theme_en_lower", self.u_theme_pattern('asin_title_lower', F.lit(self.theme_list_str)))
        # 将列拆分为数组多列
        self.df_asin_title = self.df_asin_title.withColumn("theme_en_lower", F.split(self.df_asin_title["theme_en_lower"], ","))
        # 将数组合并到多行
        self.df_asin_title = self.df_asin_title.withColumn("theme_en_lower", F.explode(self.df_asin_title["theme_en_lower"]))
        self.df_asin_title = self.df_asin_title.join(
            self.df_theme, on=['theme_en_lower'], how='left'  # 改成inner, 这样避免正则匹配结果不准
        )
        # 1. 竖表
        self.df_save_vertical = self.df_asin_title.cache()
        self.df_save_vertical = self.df_save_vertical.withColumn("site_name", F.lit(self.site_name))
        print("self.df_save_vertical.columns:", self.df_save_vertical.columns)
        # print("self.df_save_vertical.count():", self.df_save_vertical.count())
        # self.df_save_vertical.show(30, truncate=False)
        # self.df_save_vertical.filter("theme_en_lower is not null").show(30, truncate=False)

        # 2. 宽表
        self.df_asin_title = self.df_asin_title.drop_duplicates(['asin', 'theme_type_en', 'theme_ch'])
        self.df_asin_title = self.df_asin_title.withColumn("theme_type_en_counts", F.concat("theme_type_en", F.lit("_counts")))
        self.df_asin_title = self.df_asin_title.withColumn("theme_type_en_ids", F.concat("theme_type_en", F.lit("_ids")))
        # self.df_asin_title.filter('theme_type_en_counts is null').show(20, truncate=False)   # 没有记录
        self.df_asin_title = self.df_asin_title.filter('theme_type_en_counts is not null')
        pivot_df1 = self.df_asin_title.groupBy("asin").pivot("theme_type_en_counts").agg(
            F.expr("IFNULL(count(*), 0) AS value"))
        pivot_df1 = pivot_df1.na.fill(0)
        pivot_df2 = self.df_asin_title.groupBy("asin").pivot("theme_type_en_ids").agg(
            F.concat_ws(",", F.collect_list("theme_id")))
        # pivot_df1.show(30, truncate=False)
        # pivot_df2.show(30, truncate=False)
        # self.df_save_wide = df_asin_title.join(
        self.df_asin_title = df_asin_title.join(
            pivot_df1, on='asin', how='left'
        ).join(
            pivot_df2, on='asin', how='left'
        )
        # self.df_save_wide.show(30, truncate=False)
        self.df_asin_title = self.df_asin_title.withColumnRenamed("date_info", "date_info_title")
        self.df_asin_title = self.df_asin_title.drop("site_name")
        print("self.df_asin_title.columns:", self.df_asin_title.columns)
        # self.df_asin_title.show(30, truncate=False)

    def handle_weight(self):
        self.df_asin_weight_new = self.df_asin_detail.filter("(date_info >= '2023-18' and date_type='week') or (date_type in ('month', 'month_week'))").select("asin", "weight", "weight_str", "date_info", "site_name").cache()
        self.df_asin_weight_old = self.df_asin_detail.filter("date_info < '2023-18' and date_type='week'").select("asin", "weight", "weight_str", "date_info", "site_name").cache()
        self.handle_weight_new()
        self.handle_weight_old()
        print("self.df_asin_weight.columns:", self.df_asin_weight.columns)
        print("self.df_asin_weight_old.columns:", self.df_asin_weight_old.columns)
        self.df_asin_weight = self.df_asin_weight_new.unionByName(self.df_asin_weight_old, allowMissingColumns=True)
        self.df_asin_weight = self.sort_by_latest(df=self.df_asin_weight)
        # 将weight列中的'none'转为null，并转为浮点数类型
        self.df_asin_weight = self.df_asin_weight.withColumn("weight", F.when(F.col("weight") == 'none', None).otherwise(
                                                                 F.col("weight").cast(FloatType())))
        # weight列中小于等于0.001的值设为0.001
        self.df_asin_weight = self.df_asin_weight.withColumn("weight", F.when(F.col("weight") <= 0.001, 0.001).otherwise(F.col("weight")))
        # 保留4位小数
        self.df_asin_weight = self.df_asin_weight.withColumn("weight", F.round(self.df_asin_weight["weight"], 4))
        # self.df_asin_weight.show(20, truncate=False)

        self.df_asin_weight = self.df_asin_weight.withColumnRenamed(
            "weight_str", "asin_weight_str"
        ).withColumnRenamed(
            "weight", "asin_weight"
        ).withColumnRenamed(
            "weight_type", "asin_weight_type"
        )
        self.df_asin_weight = self.df_asin_weight.withColumnRenamed("date_info", "date_info_weight")
        self.df_asin_weight = self.df_asin_weight.drop("site_name")
        print("self.df_asin_weight.columns:", self.df_asin_weight.columns)
        # self.df_asin_title.show(30, truncate=False)

    def handle_weight_new(self):
        print("开始处理重量数据: 2023-18周之后")
        # 将列类型转为字符串并转为小写
        self.df_asin_weight_new = self.df_asin_weight_new.withColumn("weight_str", F.lower(F.col("weight_str").cast(StringType())))
        # 提取体积字符串中的weight_info, weight_type
        self.df_asin_weight_new = self.df_asin_weight_new.withColumn('weight_detail', self.u_get_weight('weight_str', 'site_name'))
        self.df_asin_weight_new = self.df_asin_weight_new \
            .withColumn('weight', self.df_asin_weight_new.weight_detail.getField('weight')) \
            .withColumn('weight_type', self.df_asin_weight_new.weight_detail.getField('weight_type')) \
            .drop('weight_detail')

        # # 将weight列中的'none'转为null，并转为浮点数类型
        # self.df_asin_weight_new = self.df_asin_weight_new.withColumn("weight", F.when(F.col("weight") == 'none', None).otherwise(
        #                                                          F.col("weight").cast(FloatType())))
        #
        # # weight列中小于等于0.001的值设为0.001
        # self.df_asin_weight = self.df_asin_weight.withColumn("weight", F.when(F.col("weight") <= 0.001, 0.001).otherwise(F.col("weight")))

        # # 将weight_str列中的'none'转为null
        # self.df_asin_weight = self.df_asin_weight.withColumn("weight_str", F.when(F.col("weight_str") == 'none', None).otherwise(F.col("weight_str")))

    def handle_weight_old(self):
        print("开始处理重量数据: 2023-18周之前")
        self.df_asin_weight_old = self.df_asin_weight_old.withColumn("weight_type", F.lit(self.weight_type))
        window = Window.partitionBy(['asin']).orderBy(self.df_asin_weight_old.date_info.desc())
        self.df_asin_weight_old = self.df_asin_weight_old.withColumn(
            "row_number", F.row_number().over(window)
        )
        self.df_asin_weight_old = self.df_asin_weight_old.withColumn('row_number',
                                                                     F.row_number().over(window))  # 使用窗口函数为每个分区的行编号
        self.df_asin_weight_old = self.df_asin_weight_old.filter(self.df_asin_weight_old.row_number == 1).drop(
            'row_number')  # 只保留每个分区中 row_number 最大的行，并删除 row_number 列

    def handle_volume(self):
        self.df_asin_volume = self.df_asin_detail.select("asin", "asin_volume", "date_info")
        if self.site_name == 'us':
            self.handle_volume_us()
        else:
            self.handle_volume_others()
        self.df_asin_volume = self.df_asin_volume.withColumnRenamed("date_info", "date_info_volume")
        self.df_asin_volume = self.df_asin_volume.drop("site_name")
        self.handle_volume_sorted()
        print("self.df_asin_volume.columns:", self.df_asin_volume.columns)
        # self.df_asin_volume.show(30, truncate=False)

    def handle_volume_us(self):
        self.handle_filter_dirty_data()
        # self.handle_type_inches()
        # self.handle_type_cm()
        df_inches = self.handle_asin_volume_types_to_dimensions(asin_volume_type='inches')
        df_cm = self.handle_asin_volume_types_to_dimensions(asin_volume_type='cm')
        df_cm_inches = self.handle_asin_volume_types_to_dimensions(asin_volume_type='cm,inches')
        df_none = self.handle_asin_volume_types_to_dimensions(asin_volume_type='none')
        df_none_not_null = df_none.filter(~(df_none.length.isNull() & df_none.width.isNull() & df_none.height.isNull()))
        df_none_null = df_none.filter(df_none.length.isNull() & df_none.width.isNull() & df_none.height.isNull())
        df_none_not_null = df_none_not_null.withColumn("asin_volume_type", F.lit("inches"))
        print("df_none_not_null, df_none_null:", df_none_not_null.count(), df_none_null.count())
        # self.df_save = pd.concat([df_inches, df_cm, df_cm_inches, df_none])
        # 假设 df_inches、df_cm、df_cm_inches 和 df_none 都是 PySpark DataFrame
        self.df_asin_volume = df_inches.union(df_cm).union(df_cm_inches).union(df_none_not_null).union(df_none_null)
        self.df_asin_volume = self.df_asin_volume.drop("asin_volume_flag")
        self.df_asin_volume = self.df_asin_volume.withColumnRenamed("length", "asin_length"). \
            withColumnRenamed("width", "asin_width"). \
            withColumnRenamed("height", "asin_height")

    def handle_filter_dirty_data(self):
        # 将 asin_volume 列转换为小写
        self.df_asin_volume = self.df_asin_volume.withColumn("asin_volume", F.lower(self.df_asin_volume["asin_volume"]))
        # 使用自定义函数创建新列 asin_volume_flag
        self.df_asin_volume = self.df_asin_volume.withColumn("asin_volume_flag", self.u_contains_digit_udf(self.df_asin_volume["asin_volume"]))
        # 假设 df 是一个 PySpark DataFrame，asin_volume_flag 是 DataFrame 中的一列
        # self.df_asin_volume.groupBy('asin_volume_flag').agg(F.count('asin_volume_flag')).show()
        # self.df_asin_volume.show()
        self.df_asin_volume = self.df_asin_volume.filter('asin_volume_flag is True')
        # self.df_asin_volume.groupBy('asin_volume_flag').agg(F.count('asin_volume_flag')).show()
        # self.df_asin_volume.show()
        self.df_asin_volume = self.df_asin_volume.withColumn("asin_volume_type", self.u_asin_volume_type(self.df_asin_volume["asin_volume"]))
        self.df_asin_volume.groupBy('asin_volume_type').agg(F.count('asin_volume_type')).show()
        self.df_asin_volume.show()
        # 假设 df 是一个 PySpark DataFrame，date_info 是 DataFrame 中的一列
        window = Window.partitionBy('asin').orderBy(F.desc('date_info'))  # 按照 date_info 列进行分区，并按照 date 列进行排序
        self.df_asin_volume = self.df_asin_volume.withColumn('row_number', F.row_number().over(window))  # 使用窗口函数为每个分区的行编号
        self.df_asin_volume = self.df_asin_volume.filter(self.df_asin_volume.row_number == 1).drop('row_number')  # 只保留每个分区中 row_number 最大的行，并删除 row_number 列
        self.df_asin_volume.groupBy('asin_volume_type').agg(F.count('asin_volume_type')).show()
        self.df_asin_volume.show()

    def handle_asin_volume_types_to_dimensions(self, asin_volume_type='inches'):
        df = self.df_asin_volume.filter(f'asin_volume_type="{asin_volume_type}"').cache()
        # 使用 UDF 提取长宽高并添加新的列
        df = df.withColumn('dimensions', self.u_extract_dimensions('asin_volume', F.lit(asin_volume_type)))
        # 将新的列拆分成三个列并删除 dimensions 列
        df = df \
            .withColumn('length', df.dimensions.getField('length')) \
            .withColumn('width', df.dimensions.getField('width')) \
            .withColumn('height', df.dimensions.getField('height')) \
            .drop('dimensions')
        df.show(10, truncate=False)
        # # 假设 df_asin_none 是一个 PySpark DataFrame，length、width 和 height 是 DataFrame 中的列
        # df_null = df.filter(df.length.isNull() & df.width.isNull() & df.height.isNull())
        # print("asin_volume_type, df_null:", asin_volume_type, df_null.count())
        # df_null.show(50, truncate=False)
        return df

    def handle_volume_others(self):
        self.handle_filter_dirty_data()
        # 提取体积字符串中的length, width, height
        self.df_asin_volume = self.df_asin_volume.withColumn('dimensions', self.u_extract_dimensions_others('asin_volume', 'asin_volume_type'))
        self.df_asin_volume = self.df_asin_volume \
            .withColumn('asin_length', self.df_asin_volume.dimensions.getField('length')) \
            .withColumn('asin_width', self.df_asin_volume.dimensions.getField('width')) \
            .withColumn('asin_height', self.df_asin_volume.dimensions.getField('height')) \
            .drop('dimensions')
        self.df_asin_volume = self.df_asin_volume.drop("asin_volume_flag")

    def handle_volume_sorted(self):
        self.df_asin_volume = self.df_asin_volume.withColumn('dimensions', self.u_volume_sorted('asin_length', 'asin_width', 'asin_height'))
        # 将新的列拆分成三个列并删除 dimensions 列
        self.df_asin_volume = self.df_asin_volume \
            .withColumn('asin_length_sorted', self.df_asin_volume.dimensions.getField('asin_length')) \
            .withColumn('asin_width_sorted', self.df_asin_volume.dimensions.getField('asin_width')) \
            .withColumn('asin_height_sorted', self.df_asin_volume.dimensions.getField('asin_height')) \
            .drop('dimensions')
        # self.df_asin_volume.show(10, truncate=False)
        self.df_asin_volume = self.df_asin_volume.replace({'0': None})

        self.df_asin_volume = self.df_asin_volume.withColumn('asin_length_sorted', F.col('asin_length_sorted').cast('double')) \
            .withColumn('asin_width_sorted', F.col('asin_width_sorted').cast('double')) \
            .withColumn('asin_height_sorted', F.col('asin_height_sorted').cast('double'))
        self.df_asin_volume.show(10, truncate=False)


if __name__ == '__main__':
    site_name = sys.argv[1]  # 参数1：站点
    handle_obj = DimAsinStableInfo(site_name=site_name)
    handle_obj.run()