dwd_st_asin_measure.py

import os
import re
import sys

sys.path.append(os.path.dirname(sys.path[0]))  # 上级目录
from pyspark.storagelevel import StorageLevel
from utils.templates import Templates
# from ..utils.templates import Templates
# from AmazonSpider.pyspark_job.utils.templates_test import Templates
from pyspark.sql.types import StringType, IntegerType
# 分组排序的udf窗口函数
from pyspark.sql.window import Window
from pyspark.sql import functions as F
from textblob import Word


class DwdStMeasure(Templates):

    def __init__(self, site_name='us', date_type="month", date_info='2022-01'):
        super().__init__()
        self.site_name = site_name
        self.date_type = date_type
        self.date_info = date_info
        self.db_save_st_asin = f'dwd_st_asin_measure'
        self.db_save_st = f'dwd_st_measure'
        self.db_save_asin = f'dwd_asin_measure'
        self.spark = self.create_spark_object(
            app_name=f"{self.db_save_st_asin}, {self.db_save_st}, {self.db_save_asin}: {self.site_name}, {self.date_type}, {self.date_info}")
        # self.df_date = self.get_year_week_tuple()  # pandas的df对象
        self.get_date_info_tuple()
        self.get_year_week_tuple()
        self.get_year_month_days_dict(year=int(self.year))
        self.orders_transform_rate = self.get_orders_transform_rate()  # 获取月销-->日销,周销
        self.df_st_asin = self.spark.sql(f"select 1+1;")
        self.df_st_asin_flow = self.spark.sql(f"select 1+1;")
        self.df_st = self.spark.sql(f"select 1+1;")
        self.df_brand_analytics = self.spark.sql(f"select 1+1;")
        self.df_st_rate = self.spark.sql(f"select 1+1;")
        self.df_st_quantity = self.spark.sql(f"select 1+1;")
        self.df_asin_bs = self.spark.sql(f"select 1+1;")
        self.df_asin_detail = self.spark.sql(f"select 1+1;")
        self.df_bs_report = self.spark.sql(f"select 1+1;")
        self.df_st_asin_duplicated = self.spark.sql(f"select 1+1;")
        self.df_save_st_asin = self.spark.sql(f"select 1+1;")
        self.df_save_asin = self.spark.sql(f"select 1+1;")
        self.df_save_st = self.spark.sql(f"select 1+1;")
        self.df_asin_stable = self.spark.sql(f"select 1+1;")
        self.df_asin_price_weight = self.spark.sql(f"select 1+1;")
        self.df_asin_amazon_orders = self.spark.sql(f"select 1+1;")
        self.df_asin_self = self.spark.sql(f"select 1+1;")
        self.df_st_templates = self.spark.sql("select st_zr_counts, st_sp_counts, st_sb1_counts,st_sb2_counts,st_sb3_counts,st_ac_counts,st_bs_counts,st_er_counts,st_tr_counts from dwd_st_measure limit 0")
        self.df_asin_templates = self.spark.sql("select asin_zr_counts, asin_sp_counts, asin_sb1_counts,asin_sb2_counts,asin_sb3_counts,asin_ac_counts,asin_bs_counts,asin_er_counts,asin_tr_counts from dwd_asin_measure limit 0")
        self.partitions_by = ['site_name', 'date_type', 'date_info']
        self.u_is_title_appear = self.spark.udf.register("u_is_title_appear", self.udf_is_title_appear, IntegerType())

    def get_orders_transform_rate(self):
        month_days = self.year_month_days_dict[int(self.month)]
        if self.date_type in ['day', 'week']:
            if self.date_type == 'day':
                return 1 / month_days
            if self.date_type == 'week':
                return 7 / month_days
        else:
            return 1

    @staticmethod
    def udf_is_title_appear(search_term, title):
        english_prepositions = ["aboard", "about", "above", "across", "after", "against", "along", "amid", "among",
                                "around", "as", "at", "before", "behind", "below", "beneath", "beside", "between",
                                "beyond", "but", "by", "concerning", "considering", "despite", "down", "during",
                                "except", "for", "from", "in", "inside", "into", "like", "near", "of", "off", "on",
                                "onto", "out", "outside", "over", "past", "regarding", "round", "since", "through",
                                "to", "toward", "under", "underneath", "until", "unto", "up", "upon", "with", "within",
                                "without"]

        symbol_list = ['，', '。', '？', '！', '：', '?', '!', '-', '%', '|', '；', '·', '…', '~', '&', '@', '#', '、', '…', '～', '＆', '＠', '＃', '“', '”', '‘', '’', '〝', '〞', '"', "'", '＂', '＇', '´', '＇', '(', ')', '【', '】', '《', '》', '＜', '＞', '﹝', '﹞', '<', '>', '«', '»', '‹', '›', '〔', '〕', '〈', '〉', '{', '}', '［', '］', '「', '」', '｛', '｝', '〖', '〗', '『', '』', '︵', '︷', '︹', '︿', '︽', '﹁', '﹃', '︻', '︗', '/', '\\', '︶', '︸', '︺', '﹀', '︾', '﹂', '﹄', '︼', '︘', '／', '｜', '＼', '_', '＿', '﹏', '﹍', '﹎', '``', '¦', '¡', '^', '\xad', '¨', 'ˊ', '¯', '￣', '﹋', '﹉', '﹊', 'ˋ', '︴', '¿', 'ˇ']

        # 小写
        search_term = str(search_term).lower().replace(",", " ").replace(":", " ").replace(";", " ")  # 新增逗号匹配
        title = f" {str(title).lower().replace(',', ' ').replace(';', ' ').replace(':', ' ')} "
        # 1. 去掉特殊符号
        # search_term = re.sub(r'[,:()]', '', search_term)  # 去掉特殊符号
        # title = re.sub(r'[,:()]', '', title)  # 去掉特殊符号
        # for symbol in symbol_list:
        #     if symbol in title:
        #         title = title.replace(symbol, "")
        #     if symbol in search_term:
        #         search_term = search_term.replace(symbol, "")

        # 改成正则去掉特殊符号
        symbols = "".join(symbol_list)  # 将列表中的所有字符连接成一个字符串
        search_term = re.sub('[' + symbols + ']', '', search_term)  # 去掉特殊符号
        title = re.sub('[' + symbols + ']', '', title)  # 去掉特殊符号

        # 2. 去掉介词(关键词去掉就行)
        st_list = [f" {st} " for st in search_term.split(" ") if st not in english_prepositions]  # 去掉介词

        # 3. 复数一起匹配
        for st in st_list:
            # if st in symbol_list:
            #     st = st.replace(symbol, "")
            if st not in title:
                if Word(st) not in title:
                    return 0
        return 1

        # 旧版
        # if str(search_term).lower() in str(title).lower():
        #     return 1
        # else:
        #     return 0

    def read_data(self):
        print("self.year, self.month:", self.year, self.month)

        print("1 读取st+asin两个维度： dim_st_asin_info表和ods_rank_flow表")
        print("1.1 读取dim_st_asin_info表")
        if self.date_type == 'month_old':
            # self.get_year_week_tuple()
            if int(self.month) <= 9 and int(self.year) <= 2022:
                sql = f"select * from dim_st_asin_info where site_name='{self.site_name}' and date_type='month' and date_info ='{self.date_info}'"
            else:
                sql = f"select * from dim_st_asin_info where site_name='{self.site_name}' and date_type='week' and date_info in {self.year_week_tuple}"
        else:
            sql = f"select * from dim_st_asin_info where site_name='{self.site_name}' and date_type='week' and date_info in {self.year_week_tuple}"

        if date_type in ['month', 'month_week'] and ((self.site_name == 'us' and date_info >= '2023-10') or (self.site_name in ['uk', 'de'] and self.date_info >= '2024-05')):
            sql = f"select * from dim_st_asin_info where site_name='{self.site_name}' and date_type='month' and date_info ='{self.date_info}'"

        # else:
        #     if (int(self.year) == 2022 and int(self.month) < 10) or int(self.year) <= 2021:
        #         sql = f"select * from dim_st_asin_info where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info='{self.date_info}'"
        #     else:
        #         sql = f"select * from dim_st_asin_info where site_name='{self.site_name}' and date_type='day' and date_info in {self.date_info_tuple}"
        print("sql:", sql)
        self.df_st_asin = self.spark.sql(sqlQuery=sql).cache()
        self.df_st_asin.show(10, truncate=False)
        # self.df_st_asin.filter("search_term='abiie high chair'").show(100, truncate=False)
        # quit()
        # self.df_st_asin = self.df_st_asin.drop_duplicates(["search_term", "asin", "data_type", "date_info"]).cache()
        # self.df_st_asin_duplicated = self.df_st_asin.drop_duplicates(['search_term', 'asin']).cache()
        # print("self.df_st_asin:", self.df_st_asin.count())
        # print("self.df_st_asin_duplicated:", self.df_st_asin_duplicated.count())
        # self.df_st_asin.show(10, truncate=False)
        # self.df_asin = self.df_st_asin.select("asin").drop_duplicates(["asin"])
        # self.df_st = self.df_st_asin.select("search_term").drop_duplicates(["search_term"])
        print("1.2 读取ods_rank_flow表")
        sql = f"select rank as page_rank, flow from ods_rank_flow " \
              f"where site_name='{self.site_name}'"
        self.df_st_asin_flow = self.spark.sql(sql).cache()
        # self.df_st_asin_flow.persist(StorageLevel.MEMORY_ONLY)
        self.df_st_asin_flow.show(10, truncate=False)
        print("2 读取st维度： dim_st_detail表和ods_brand_analytics表")
        print("self.year, self.month:", self.year, self.month)
        print("2.1 读取dim_st_detail和ods_brand_analytics表")
        sql = f"select search_term, st_rank, st_search_sum from dim_st_detail where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info ='{self.date_info}';"
        print("sql:", sql)
        self.df_st = self.spark.sql(sqlQuery=sql)
        self.df_st.persist(StorageLevel.MEMORY_ONLY)
        self.df_st.show(10, truncate=False)
        # 统计词频
        print("2.2 读取ods_brand_analytics表")
        # sql = f"select search_term, date_info from ods_brand_analytics where site_name='{self.site_name}' and date_type='day' and date_info in {self.date_info_tuple}"
        sql = f"select search_term, date_info from ods_brand_analytics where site_name='{self.site_name}' and date_type='week' and date_info in {self.year_week_tuple}"
        print("sql:", sql)
        self.df_brand_analytics = self.spark.sql(sqlQuery=sql)
        self.df_brand_analytics.persist(StorageLevel.MEMORY_ONLY)
        self.df_brand_analytics.show(10, truncate=False)
        print("3 读取asin维度： dim_asin_bs_info+dim_asin_detail表")
        print("3.1 读取dim_asin_bs_info表")

        sql = f"select asin, asin_bs_cate_1_rank, asin_bs_cate_1_id " \
              f"from dim_asin_bs_info where site_name='{self.site_name}' and date_type='{self.date_type.replace('_old', '')}' and date_info='{self.date_info}';"
        print("sql:", sql)
        self.df_asin_bs = self.spark.sql(sql).cache()
        self.df_asin_bs.show(10)

        sql = f"select asin, asin_title, asin_price, parent_asin " \
              f"from dim_asin_detail where site_name='{self.site_name}' and date_type='{self.date_type.replace('_old', '')}' and date_info='{self.date_info}';"
        print("sql:", sql)
        self.df_asin_detail = self.spark.sql(sql).cache()

        self.df_asin_detail.show(10)
        print("4 读取bsr维度： ods_one_category_report表")
        print("4.1 读取ods_one_category_report表")

        if int(self.year) == 2022 and int(self.month) < 3:
            sql = f"select category_id as asin_bs_cate_1_id, rank as asin_bs_cate_1_rank, ceil(orders*{self.orders_transform_rate}) as asin_bsr_orders from ods_one_category_report " \
                  f"where site_name='{self.site_name}' and date_type='month' and date_info='2022-12';"
        else:
            month = f"0{str(self.month)}" if len(str(self.month)) == 1 else str(self.month)
            sql = f"select category_id as asin_bs_cate_1_id, rank as asin_bs_cate_1_rank, ceil(orders*{self.orders_transform_rate}) as asin_bsr_orders from ods_one_category_report " \
                  f"where site_name='{self.site_name}' and date_type='month' and date_info='{self.year}-{month}';"
        print("sql:", sql)
        self.df_bs_report = self.spark.sql(sqlQuery=sql)
        self.df_bs_report.persist(StorageLevel.MEMORY_ONLY)
        self.df_bs_report.show(10, truncate=False)
        print("5 读取asin维度-体积信息： dim_asin_stable_info表")
        sql = f"select asin, asin_length * asin_width * asin_height as asin_volume, asin_weight from dim_asin_stable_info where site_name='{self.site_name}'"
        print("sql:", sql)
        self.df_asin_stable = self.spark.sql(sqlQuery=sql).cache()
        self.df_asin_stable.show(10, truncate=False)
        print("6 读取asin维度-月销数据： dim_asin_amorders_info表")
        sql = f"select asin, asin_amazon_orders from dim_asin_amorders_info where site_name='{self.site_name}' and date_type='{self.date_type.replace('_old', '')}' and date_info='{self.date_info}'"
        print("sql:", sql)
        self.df_asin_amazon_orders = self.spark.sql(sqlQuery=sql).cache()
        self.df_asin_amazon_orders.show(10, truncate=False)
        print("7 读取asin维度-内部asin： ods_self_asin")
        sql = f"select asin, 1 as is_self_asin from ods_self_asin where site_name='{self.site_name}' group by asin"
        print("sql:", sql)
        self.df_asin_self = self.spark.sql(sqlQuery=sql)
        self.df_asin_self = F.broadcast(self.df_asin_self)
        self.df_asin_self.show(10, truncate=False)

    def save_data(self):
        self.reset_partitions(partitions_num=50)
        self.save_data_common(
            df_save=self.df_save_st_asin,
            db_save=self.db_save_st_asin,
            partitions_num=self.partitions_num,
            partitions_by=self.partitions_by
        )
        self.reset_partitions(partitions_num=5)
        self.save_data_common(
            df_save=self.df_save_st,
            db_save=self.db_save_st,
            partitions_num=self.partitions_num,
            partitions_by=self.partitions_by
        )
        self.reset_partitions(partitions_num=10)
        self.save_data_common(
            df_save=self.df_save_asin,
            db_save=self.db_save_asin,
            partitions_num=self.partitions_num,
            partitions_by=self.partitions_by
        )

    def handle_data(self):
        self.handle_join()
        self.df_save_asin = self.handle_st_asin_counts(cal_type="asin", df_templates=self.df_asin_templates, page=3)
        self.df_save_st = self.handle_st_asin_counts(cal_type="st", df_templates=self.df_st_templates)
        # self.handle_st_zr_page1_title_rate()
        self.handle_st_zr_sp_page123_title_rate(data_type='zr', page_type='page1')
        self.handle_st_zr_sp_page123_title_rate(data_type='zr', page_type='page123')
        self.handle_st_zr_sp_page123_title_rate(data_type='sp', page_type='page123')
        self.handle_st_asin_orders()  # 预估销量和bsr销量
        self.handle_asin_ao_and_zr_flow_proportion()
        self.handle_st_ao_and_zr_flow_proportion()
        self.handle_st_num()
        self.handle_st_weight_price_volume()
        # self.df_save_st.filter("search_term='abiie high chair'").show(10, truncate=False)
        # self.df_save_st_asin.filter("search_term='abiie high chair'").show(100, truncate=False)
        # self.df_save_st_asin.show(10, truncate=False)
        # self.df_save_st.show(10, truncate=False)
        del self.df_st_asin_duplicated
        del self.df_st_asin
        # self.df_save_asin.show(10, truncate=False)
        # quit()

    def handle_st_attributes_avg(self, df_st_asin, attributes_type, st_type):
        # 根据基准值，计算平均值
        df_st_asin = df_st_asin.select("search_term", f"{attributes_type}").filter(f'{attributes_type} > 0')
        # 过滤大于基准值的几率
        df_st_asin = df_st_asin.filter(F.col(f"{attributes_type}") <= F.col(st_type))

        df_st_avg = df_st_asin.groupby(['search_term']).agg(
            F.round(F.avg(f"{attributes_type}"), 4).alias(f'{st_type.replace("_std", "_avg")}')
        )
        return df_st_avg

    def handle_st_attributes_std(self, df, attributes_type='asin_volume'):
        # 计算基准值
        # 定义窗口函数
        window = Window.partitionBy(['search_term']).orderBy(F.desc(f"{attributes_type}"))

        # 计算百分比排名并筛选 <= 0.25 的记录
        df = df.select("search_term", f"{attributes_type}").filter(f'{attributes_type} > 0') \
            .withColumn(f"{attributes_type}_percent_rank", F.percent_rank().over(window)) \
            .filter(f'{attributes_type}_percent_rank <= 0.25') \

        # 使用 row_number() 方法获取每个 search_term 的最大百分比排名记录
        window = Window.partitionBy(['search_term']).orderBy(F.desc(f"{attributes_type}_percent_rank"))
        df = df.withColumn(f"{attributes_type}_row_number", F.row_number().over(window)) \
            .filter(f'{attributes_type}_row_number = 1')

        # 显示结果
        df = df.drop(f"{attributes_type}_percent_rank", f"{attributes_type}_row_number")
        df = df.withColumnRenamed(f"{attributes_type}", f"{attributes_type.replace('asin', 'st')}_25_percent")
        df.show(10, truncate=False)
        return df

    def handle_st_weight_price_volume(self):
        # self.df_st_asin_duplicated = self.df_st_asin_duplicated.drop_duplicates(['search_term', 'asin']).cache()
        df_st_asin = self.df_st_asin_duplicated.select('search_term', 'asin').drop_duplicates(['search_term', 'asin']).cache()
        df_asin_label = self.df_asin_detail.select("asin", "asin_price", "asin_weight", "asin_volume").cache()

        df_st_asin = df_st_asin.join(
            df_asin_label, on='asin', how='inner'
        )
        # df_st_asin.filter("search_term='airpods'").show(100, truncate=False)
        #
        # quit()
        # 取四分位值
        df_st_volume = self.handle_st_attributes_std(df=df_st_asin, attributes_type='asin_volume')
        df_st_price = self.handle_st_attributes_std(df=df_st_asin, attributes_type='asin_price')
        df_st_weight = self.handle_st_attributes_std(df=df_st_asin, attributes_type='asin_weight')
        # 取最小值
        df_st_min = df_st_asin.groupby(['search_term']).agg(
            F.round(F.min("asin_volume"), 4).alias('st_volume_min'),
            F.round(F.min("asin_price"), 4).alias('st_price_min'),
            F.round(F.min("asin_weight"), 4).alias('st_weight_min')
        )
        df_st_min = df_st_min.join(
            df_st_volume, on='search_term', how='left'
        ).join(
            df_st_price, on='search_term', how='left'
        ).join(
            df_st_weight, on='search_term', how='left'
        )
        # 基准值
        df_st_std = df_st_min.withColumn(
            "st_volume_std",
            F.round(1.5 * (df_st_min.st_volume_25_percent - df_st_min.st_volume_min) + df_st_min.st_volume_min, 4)
        ).withColumn(
            "st_price_std",
            F.round(1.5 * (df_st_min.st_price_25_percent - df_st_min.st_price_min) + df_st_min.st_price_min, 4)
        ).withColumn(
            "st_weight_std",
            F.round(1.5 * (df_st_min.st_weight_25_percent - df_st_min.st_weight_min) + df_st_min.st_weight_min, 4)
        )
        # df_st_min.show(10, truncate=False)
        # 四分位平均值
        df_st_asin = df_st_asin.join(
            df_st_std, on="search_term", how="left"
        )
        df_st_volume_avg = self.handle_st_attributes_avg(df_st_asin=df_st_asin, attributes_type='asin_volume', st_type="st_volume_std")
        df_st_price_avg = self.handle_st_attributes_avg(df_st_asin=df_st_asin, attributes_type='asin_price', st_type="st_price_std")
        df_st_weight_avg = self.handle_st_attributes_avg(df_st_asin=df_st_asin, attributes_type='asin_weight', st_type="st_weight_std")

        df_st_avg = df_st_std.join(
            df_st_volume_avg, on='search_term', how='left'
        ).join(
            df_st_price_avg, on='search_term', how='left'
        ).join(
            df_st_weight_avg, on='search_term', how='left'
        )
        # df_st_avg.show(10, truncate=False)
        # df_st_avg.filter("search_term='airpods'").show(10, truncate=False)

        self.df_save_st = self.df_save_st.join(
            df_st_avg, on='search_term', how='left'
        )

    def handle_join(self):
        # st+asin
        self.df_st_asin = self.df_st_asin.join(
            self.df_st_asin_flow, on=['page_rank'], how='left'
        )
        # st -- dim_st_detail已经有
        # asin
        self.df_asin_bs = self.df_asin_bs.join(
            self.df_bs_report, on=['asin_bs_cate_1_rank', 'asin_bs_cate_1_id'], how='left'
        )
        self.df_asin_detail = self.df_asin_detail.join(
            self.df_asin_bs, on='asin', how='left'
        ).join(
            self.df_asin_stable, on='asin', how='left'
        )
        # 合并
        self.df_st_asin = self.df_st_asin.join(
            self.df_st, on=['search_term'], how='left'
        ).join(
            self.df_asin_detail, on=['asin'], how='left'
        )

        window = Window.partitionBy(['search_term', 'asin', 'data_type']).orderBy('page')
        self.df_st_asin = self.df_st_asin.withColumn("rk", F.row_number().over(window=window))
        self.df_st_asin_duplicated = self.df_st_asin.filter("rk=1").drop("rk").cache()

        # self.df_st_asin.show(10, truncate=False)
        # self.df_st_asin = self.df_st_asin.drop_duplicates(['search_term', 'asin', 'data_type'])
        # self.df_st_asin = self.df_st_asin.cache()
        # self.df_st_asin_duplicated = self.df_st_asin.drop_duplicates(['search_term', 'asin', 'data_type']).cache()
        # self.df_st_asin.persist(StorageLevel.MEMORY_ONLY)

    def handle_st_asin_counts(self, cal_type="asin", df_templates=None, page=3):
        print(f"计算{cal_type}_counts")
        cal_type_complete = "search_term" if cal_type == "st" else cal_type
        self.df_st_asin_duplicated = self.df_st_asin_duplicated.filter(f"page <= {page}").withColumn(
            f"{cal_type}_data_type",
            F.concat(F.lit(f"{cal_type}_"), self.df_st_asin_duplicated.data_type, F.lit(f"_counts"))
        )
        df = self.df_st_asin_duplicated.groupby([f'{cal_type_complete}']). \
            pivot(f"{cal_type}_data_type").count()

        df = df_templates.unionByName(df, allowMissingColumns=True)  # 防止爬虫数据没有导致程序运行出错
        df = df.fillna(0)
        # df.show(10, truncate=False)
        df = df.withColumn(
            f"{cal_type}_sb_counts",
            df[f"{cal_type}_sb1_counts"] + df[f"{cal_type}_sb2_counts"] + df[f"{cal_type}_sb3_counts"]
        )
        df = df.withColumn(
            f"{cal_type}_adv_counts",
            df[f"{cal_type}_sb_counts"] + df[f"{cal_type}_sp_counts"]
        )
        if cal_type == "asin":
            df_st_asin_duplicated = self.df_st_asin_duplicated.drop_duplicates(['search_term', 'asin'])
            df_st_asin_agg = df_st_asin_duplicated.groupby(['asin']).agg(
                F.count('search_term').alias("asin_st_counts")
            )
            df = df.join(
                df_st_asin_agg, on=['asin'], how='left'
            )
        elif cal_type == "st":
            df_st_asin_agg = self.df_st_asin_duplicated.select("search_term", "asin").join(
                self.df_asin_self, on='asin', how='left'
            ).withColumn(
                "is_self_asin",
                F.when(F.col("is_self_asin").isNotNull(), F.col("is_self_asin")).otherwise(F.lit(0))
            ).groupby(['search_term']).agg(
                F.sum('is_self_asin').alias("st_self_asin_counts"),
                F.count('asin').alias("st_total_asin_counts")
            ).withColumn(
                'st_self_asin_proportion',
                F.round(F.col('st_self_asin_counts') / F.col('st_total_asin_counts'), 4)
            )
            df = df.join(
                df_st_asin_agg, on=['search_term'], how='left'
            )
        df = df.withColumn(f"site_name", F.lit(self.site_name))
        df = df.withColumn(f"date_type", F.lit(self.date_type))
        df = df.withColumn(f"date_info", F.lit(self.date_info))
        # df.show(10, truncate=False)
        return df

    def handle_st_zr_sp_page123_title_rate(self, data_type='zr', page_type="page1"):
        params = ' and page=1' if page_type == 'page1' else ""
        df = self.df_st_asin.filter(f"data_type='{data_type}' {params}")
        df = df.select("search_term", "asin", "asin_title").drop_duplicates(["search_term", "asin"])
        df = df.withColumn(
            "st_asin_in_title_flag",
            self.u_is_title_appear(df.search_term, df.asin_title)
        )
        # 测试--保留需要的数据
#         s_str = """transformers
# barbie
# ring doorbell
# iphone 14 pro max case
# nintendo switch
# prime day deals today 2023
# the boys
# lube
# marvelous mrs maisel
# hydrogen peroxide"""
#         s_tuple = tuple(s_str.split("\n"))
#         df_csv = df.filter(f"search_term in {s_tuple}").toPandas()
#         df_csv.to_csv(rf"/root/{data_type}_{page_type}.csv", index=False)
        df = df.groupby(['search_term']).agg(
            {
                "search_term": "count",
                "st_asin_in_title_flag": "sum",
            }
        )
        df = df.withColumnRenamed(
            "sum(st_asin_in_title_flag)", f"st_{data_type}_{page_type}_title_appear_counts"
        ).withColumnRenamed(
            "count(search_term)", f"st_{data_type}_{page_type}_title_counts"
        )
        df = df.withColumn(
            f"st_{data_type}_{page_type}_title_appear_rate",
            # df.st_zr_page1_title_appear_counts / df.st_zr_page1_title_counts
            F.round(F.col(f"st_{data_type}_{page_type}_title_appear_counts") / F.col(f"st_{data_type}_{page_type}_title_counts"), 4)
        )
        self.df_save_st = self.df_save_st.join(
            df, on=['search_term'], how='left'
        )

    def handle_st_zr_page1_title_rate(self):
        print("计算关键词的zr类型page=1的去重asin的标题密度")
        df_zr_page1 = self.df_st_asin.filter(
            "data_type='zr' and page=1"
        )
        df_zr_page1 = df_zr_page1.select("search_term", "asin", "asin_title").drop_duplicates(["search_term", "asin"])
        df_zr_page1 = df_zr_page1.withColumn(
            "st_asin_in_title_flag",
            self.u_is_title_appear(df_zr_page1.search_term, df_zr_page1.asin_title)
        )
        # df_zr_page1.show(10, truncate=False)
        df_zr_page1 = df_zr_page1.groupby(['search_term']).agg(
            {
                "search_term": "count",
                "st_asin_in_title_flag": "sum",
            }
        )
        df_zr_page1 = df_zr_page1.withColumnRenamed(
            "sum(st_asin_in_title_flag)", "st_zr_page1_title_appear_counts"
        ).withColumnRenamed(
            "count(search_term)", "st_zr_page1_title_counts"
        )
        df_zr_page1 = df_zr_page1.withColumn(
            "st_zr_page1_title_appear_rate", df_zr_page1.st_zr_page1_title_appear_counts / df_zr_page1.st_zr_page1_title_counts
        )
        self.df_save_st = self.df_save_st.join(
            df_zr_page1, on=['search_term'], how='left'
        )
        # df_zr_page1.show(10, truncate=False)
        # quit()
        del df_zr_page1

    def handle_st_asin_orders(self):
        # 预估销量+bsr销量
        print("1. 预估销量:zr, sp的销量")
        # 1.1 st+asin
        self.df_st_asin = self.df_st_asin.withColumn(
            "st_asin_orders",
            F.ceil(self.df_st_asin.flow * self.df_st_asin.st_search_sum * self.orders_transform_rate)
        )
        self.df_save_st_asin = self.df_st_asin.withColumn(
            "st_asin_orders_data_type", F.concat(F.lit("st_asin_"), self.df_st_asin.data_type, F.lit("_orders"))
        )
        self.df_save_st_asin = self.df_save_st_asin.groupby(["search_term", "asin"]). \
            pivot("st_asin_orders_data_type").agg(F.mean(f"st_asin_orders"))
        self.df_save_st_asin = self.df_save_st_asin.select(
            "search_term", "asin", "st_asin_zr_orders", "st_asin_sp_orders"
        )
        # self.df_save_st_asin = self.df_save_st_asin.cache()
        self.df_save_st_asin.persist(StorageLevel.MEMORY_ONLY)
        self.df_save_st_asin = self.df_save_st_asin.withColumn(f"site_name", F.lit(self.site_name))
        self.df_save_st_asin = self.df_save_st_asin.withColumn(f"date_type", F.lit(self.date_type))
        self.df_save_st_asin = self.df_save_st_asin.withColumn(f"date_info", F.lit(self.date_info))
        # self.df_save_st_asin.show(10, truncate=False)
        # 1.2 st维度的zr和sp预估销量
        df_st_orders = self.df_save_st_asin.groupby(['search_term']).agg(
            F.sum('st_asin_zr_orders').alias("st_zr_orders"),
            F.sum('st_asin_sp_orders').alias("st_sp_orders"),
            # F.sum('st_asin_zr_orders').alias("st_zr_orders"),
            # F.sum('st_asin_sp_orders').alias("st_zr_orders"),
        )
        # df_st_orders = df_st_orders.withColumnRenamed(
        #     "sum(st_asin_zr_orders)", "st_zr_orders"
        # ).withColumnRenamed(
        #     "sum(st_asin_sp_orders)", "st_sp_orders"
        # )
        self.df_save_st = self.df_save_st.join(
            df_st_orders, on=['search_term'], how='left'
        )
        # 1.3 asin维度的zr和sp预估销量
        df_asin_orders = self.df_save_st_asin.groupby(['asin']).agg(
            F.mean('st_asin_zr_orders').alias("asin_zr_orders"),
            F.mean('st_asin_sp_orders').alias("asin_sp_orders"),
            F.sum('st_asin_zr_orders').alias("asin_zr_orders_sum"),
            F.sum('st_asin_sp_orders').alias("asin_sp_orders_sum"),
        )

        # df_asin_orders = df_asin_orders.withColumnRenamed(
        #     "avg(st_asin_zr_orders)", "asin_zr_orders"
        # ).withColumnRenamed(
        #     "avg(st_asin_sp_orders)", "asin_sp_orders"
        # ).withColumnRenamed(
        #     "sum(st_asin_zr_orders)", "asin_zr_orders_sum"
        # ).withColumnRenamed(
        #     "sum(st_asin_sp_orders)", "asin_sp_orders_sum"
        # )
        self.df_save_asin = self.df_save_asin.join(
            df_asin_orders, on=['asin'], how='left'
        )
        # 向上取整
        self.df_save_asin = self.df_save_asin.withColumn(
            "asin_zr_orders", F.ceil(self.df_save_asin.asin_zr_orders)
        ).withColumn(
            "asin_sp_orders", F.ceil(self.df_save_asin.asin_sp_orders)
        ).withColumn(
            "asin_zr_orders_sum", F.ceil(self.df_save_asin.asin_zr_orders_sum)
        ).withColumn(
            "asin_sp_orders_sum", F.ceil(self.df_save_asin.asin_sp_orders_sum)
        )

        print("2. bsr销量")
        # 2.1 st_bsr_orders
        df_st_bsr_orders = self.df_st_asin.select("search_term", "asin", "asin_bsr_orders").drop_duplicates(["search_term", "asin"])
        df_st_bsr_orders = df_st_bsr_orders.groupby(['search_term']).agg({"asin_bsr_orders": "sum"})
        df_st_bsr_orders = df_st_bsr_orders.withColumnRenamed(
            "sum(asin_bsr_orders)", "st_bsr_orders"
        )
        # 2.2 asin_bsr_orders
        df_asin_bsr_orders = self.df_st_asin.select("asin", "asin_bsr_orders").drop_duplicates(['asin'])
        # df_st_bsr_orders.show(10, truncate=False)
        # df_asin_bsr_orders.show(10, truncate=False)
        self.df_save_st = self.df_save_st.join(
            df_st_bsr_orders, on='search_term', how='left'
        )
        self.df_save_asin = self.df_save_asin.join(
            df_asin_bsr_orders, on='asin', how='left'
        ).join(
            self.df_asin_amazon_orders, on='asin', how='left'
        )

    def handle_asin_ao_and_zr_flow_proportion(self):
        print("计算asin维度的ao+zr流量占比")
        # 1.计算asin的ao值
        self.df_save_asin = self.df_save_asin.withColumn(
            "asin_ao_val", F.round(self.df_save_asin.asin_adv_counts / self.df_save_asin.asin_zr_counts, 3)
        )
        # 2.计算asin的ao竞争比例
        window = Window.orderBy(self.df_save_asin.asin_ao_val.asc_nulls_last())
        self.df_save_asin = self.df_save_asin.withColumn(
            "asin_ao_val_rate",
            F.round(F.percent_rank().over(window=window), 4)
        )
        # 3.计算asin的自然流量占比
        self.df_save_asin = self.df_save_asin.withColumn(
            "asin_zr_flow_proportion",
            F.when(F.col("asin_st_counts").isNotNull(), F.round(F.col("asin_zr_counts") / F.col("asin_st_counts"), 3))
        )
        self.df_save_asin = self.df_save_asin.join(
            self.df_asin_detail.select("asin", "parent_asin"), on='asin', how='left'
        )
        # 4.计算asin的母体ao值和母体zr流量占比
        df_asin_variation = self.df_save_asin.filter("parent_asin is not null").select("parent_asin",
                                                                                         "asin_zr_counts",
                                                                                         "asin_st_counts",
                                                                                         "asin_adv_counts")
        df_asin_variation_agg = df_asin_variation.groupby(['parent_asin']).agg(
            F.sum("asin_zr_counts").alias("sum_asin_zr_counts"),
            F.sum("asin_st_counts").alias("sum_asin_st_counts"),
            F.sum("asin_adv_counts").alias("sum_asin_adv_counts")
        ).withColumn(
            "asin_flow_proportion_matrix",
            F.when(
                F.col("sum_asin_st_counts").isNotNull(),
                F.round(F.col("sum_asin_zr_counts") / F.col("sum_asin_st_counts"), 3)
            )
        ).withColumn(
            "asin_ao_val_matrix",
            F.when(
                F.col("sum_asin_zr_counts").isNotNull(),
                F.round(F.col("sum_asin_adv_counts") / F.col("sum_asin_zr_counts"), 3)
            )
        ).drop("sum_asin_zr_counts", "sum_asin_st_counts", "sum_asin_adv_counts")
        self.df_save_asin = self.df_save_asin.join(
            df_asin_variation_agg, on=['parent_asin'], how='left'
        )
        # 5.若母体自然流量占比为null，则用asin的自然流量占比替代，ao同理
        self.df_save_asin = self.df_save_asin.withColumn(
            "asin_flow_proportion_matrix",
            F.coalesce(F.col("asin_flow_proportion_matrix"), F.col("asin_zr_flow_proportion"))
        ).withColumn(
            "asin_ao_val_matrix",
            F.coalesce(F.col("asin_ao_val_matrix"), F.col("asin_ao_val"))
        )
        self.df_save_asin.show(10, truncate=False)
        self.df_save_asin = self.df_save_asin.drop("parent_asin")

    def handle_st_ao_and_zr_flow_proportion(self):
        print("计算st维度的ao+zr流量占比")
        # 1.得到asin的ao值和zr流量占比
        df_asin_ao_and_zr_flow_proportion = self.df_save_asin.select("asin", "asin_ao_val", "asin_zr_flow_proportion", "asin_ao_val_matrix", "asin_flow_proportion_matrix")
        df_st_ao_and_zr_flow_proportion = self.df_st_asin_duplicated.filter("data_type='zr'").select("search_term", "asin", "page").join(
            df_asin_ao_and_zr_flow_proportion, on=['asin'], how='left'
        )
        # 2.新增asin的ao值升序排序，计算排名4到20的均值
        window = Window.partitionBy(['search_term']).orderBy(df_st_ao_and_zr_flow_proportion.asin_ao_val.asc_nulls_last())
        df_st_ao_4_20 = df_st_ao_and_zr_flow_proportion.withColumn(
            "asin_ao_val_rank",
            F.row_number().over(window=window)
        ).filter(
            "asin_ao_val_rank between 4 and 20"
        ).select(
            "search_term", "asin_ao_val"
        )
        df_st_ao_4_20 = df_st_ao_4_20\
            .groupby(["search_term"])\
            .agg(F.round(F.mean(df_st_ao_4_20.asin_ao_val), 3).alias("st_4_20_ao_avg"))
        # 3.计算st的ao值和zr流量占比--首页zr位asin的平均值
        df_st_ao = df_st_ao_and_zr_flow_proportion\
            .filter("page=1 and asin_ao_val is not null")\
            .groupby(["search_term"])\
            .agg(F.round(F.mean("asin_ao_val"), 3).alias("st_ao_val"))
        df_st_zr_flow_proportion = df_st_ao_and_zr_flow_proportion\
            .filter("page=1 and asin_zr_flow_proportion is not null")\
            .groupby(["search_term"])\
            .agg(F.round(F.mean("asin_zr_flow_proportion"), 3).alias("st_zr_flow_proportion"))
        self.df_save_st = self.df_save_st.join(
            df_st_ao, on=['search_term'], how='left'
        ).join(
            df_st_zr_flow_proportion, on=['search_term'], how='left'
        ).join(
            df_st_ao_4_20, on=['search_term'], how='left'
        )
        # 4.计算st的ao竞争比例
        window = Window.orderBy(self.df_save_st.st_ao_val.asc())
        self.df_save_st = self.df_save_st.withColumn(
            "st_ao_val_rate",
            F.round(F.percent_rank().over(window=window), 4)
        )
        # 5.计算st的母体ao值和母体zr流量占比--首页zr位asin的母体平均值
        df_st_ao_val_matrix = df_st_ao_and_zr_flow_proportion\
            .filter("page=1 and asin_ao_val_matrix is not null")\
            .groupby(["search_term"])\
            .agg(F.round(F.mean("asin_ao_val_matrix"), 3).alias("st_ao_val_matrix"))
        df_st_flow_proportion_matrix = df_st_ao_and_zr_flow_proportion\
            .filter("page=1 and asin_flow_proportion_matrix is not null")\
            .groupby(["search_term"])\
            .agg(F.round(F.mean("asin_flow_proportion_matrix"), 3).alias("st_flow_proportion_matrix"))
        self.df_save_st = self.df_save_st.join(
            df_st_ao_val_matrix, on=['search_term'], how='left'
        ).join(
            df_st_flow_proportion_matrix, on=['search_term'], how='left'
        )
        self.df_save_st.show(10, truncate=False)

    def handle_st_num(self):
        df_num = self.df_brand_analytics.drop_duplicates(['search_term', 'date_info'])
        df_num = df_num.groupby(['search_term']).count()
        df_num = df_num.withColumnRenamed("count", "st_num")
        # self.df_save_st = self.df_save_st.withColumn("st_num", F.lit(1))
        self.df_save_st = self.df_save_st.join(
            df_num, on=['search_term'], how='left'
        )


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