dim_asin_bs_info.py

import os
import sys
import re

sys.path.append(os.path.dirname(sys.path[0]))  # 上级目录
from utils.templates import Templates
# from ..utils.templates import Templates
from pyspark.sql import functions as F
from pyspark.sql.window import Window
from pyspark.sql.types import StructType, StructField, IntegerType, StringType
# 导入udf公共方法
from yswg_utils.common_udf import udf_parse_bs_category
# from ..yswg_utils.common_udf import udf_parse_bs_category


class DimBsAsinInfo(Templates):

    def __init__(self, site_name='us', date_type="month", date_info='2022-1'):
        super().__init__()
        self.site_name = site_name
        self.date_type = date_type
        self.date_info = date_info
        # 初始化self.spark对
        self.db_save = 'dim_asin_bs_info'
        self.spark = self.create_spark_object(
            app_name=f"{self.db_save}: {self.site_name}, {self.date_type}, {self.date_info}")
        self.df_save = self.spark.sql("select 1+1;")
        self.df_asin_node_id = self.spark.sql("select 1+1;")
        self.df_bs_asin_detail = self.spark.sql("select 1+1;")
        self.df_bs_category = self.spark.sql("select 1+1;")
        # 定义 UDF 的返回类型，即一个包含三个 DoubleType 字段的 StructType
        schema = StructType([
            StructField('asin_bs_cate_1_id', StringType(), True),
            StructField('asin_bs_cate_current_id', StringType(), True),
            StructField('asin_bs_cate_1_rank', IntegerType(), True),
            StructField('asin_bs_cate_current_rank', IntegerType(), True),
        ])
        # self.u_parse_bs_category = F.udf(self.udf_parse_bs_category, schema)
        self.u_parse_bs_category = F.udf(udf_parse_bs_category, schema)
        # self.pattern1_dict = {
        #     "us": "(\d+).*?See Top 100 in ".lower(),
        #     "uk": "(\d+).*?See Top 100 in ".lower(),
        #     "de": "(\d+).*?Siehe Top 100 in ".lower(),
        #     "es": "(\d+).*?Ver el Top 100 en ".lower(),
        #     "fr": "(\d+).*?Voir les 100 premiers en ".lower(),
        #     "it": "(\d+).*?Visualizza i Top 100 nella categoria ".lower(),
        # }
        self.pattern1_dict = {
            "us": "See Top 100 in ".lower(),
            "uk": "See Top 100 in ".lower(),
            "de": "Siehe Top 100 in ".lower(),
            "es": "Ver el Top 100 en ".lower(),
            "fr": "Voir les 100 premiers en ".lower(),
            "it": "Visualizza i Top 100 nella categoria ".lower(),
        }
        self.pattern_current_dict = {
            "us": "#(\d+) ",
            "uk": "(\d+) in ",
            "de": "(\d+) in ",
            "es": "(\d+) en ",
            "fr": "(\d+) en ",
            "it": "(\d+) in ",
        }
        self.partitions_by = ['site_name', 'date_type', 'date_info']
        self.reset_partitions(partitions_num=20)
        self.get_year_week_tuple()

    @staticmethod
    def udf_parse_bs_category(asin_bs_sellers_rank_lower, last_herf, all_best_sellers_href, cate_current_pattern, cate_1_pattern):

        # if (site_name == 'us' and date_type in ['month', 'month_week'] and date_info >= '2023-11') or (site_name != 'us' and date_type in ['week'] and date_info >= '2023-41'):
        #     href_list = all_best_sellers_href.split("&&&&")

        # 1. 判断用哪个字段来解析分类
        if str(all_best_sellers_href).lower() not in ['', 'none', 'null']:
            bs_href = all_best_sellers_href
        elif str(last_herf).lower() not in ['', 'none', 'null']:
            bs_href = last_herf
        else:
            bs_href = ''
        href_list = bs_href.replace("?tf=1", "").split("&&&&")

        # 2. 解析一级和当前 分类 + 排名
        # 2.1 提取分类
        if href_list:
            if len(href_list) == 1:
                cate_list = re.findall('bestsellers/(.*)/ref', href_list[0])
                if cate_list:
                    if "/" in cate_list[0]:
                        cate_1_id, cate_current_id = cate_list[0].split("/")[0], cate_list[0].split("/")[-1]
                    else:
                        cate_1_id, cate_current_id = cate_list[0].split("/")[0], None
                else:
                    cate_1_id, cate_current_id = None, None
            else:
                cate_1_id = re.findall('bestsellers/(.*)/ref', href_list[0])[0] if re.findall('bestsellers/(.*)/ref', href_list[0]) else None
                cate_current_id = re.findall('bestsellers/(.*)/ref', href_list[-1])[0] if re.findall('bestsellers/(.*)/ref', href_list[-1]) else None
                if "/" in cate_1_id:
                    cate_1_id = cate_1_id.split("/")[0]
                if "/" in cate_current_id:
                    cate_current_id = cate_current_id.split("/")[-1]

        else:
            cate_1_id, cate_current_id = None, None

        # 2.2 提取排名
        asin_bs_sellers_rank_lower2 = asin_bs_sellers_rank_lower.replace(",", "").replace(" 100 ", "")
        rank_list = re.findall(cate_current_pattern, asin_bs_sellers_rank_lower2)  # 匹配排名
        rank_list = [int(rank) for rank in rank_list]  # 转换成int类型
        if rank_list:
            if len(rank_list) == 1:
                if cate_1_pattern in asin_bs_sellers_rank_lower:
                    cate_1_rank, cate_current_rank = rank_list[0], None
                else:
                    cate_1_rank, cate_current_rank = None, rank_list[0]
            else:
                if cate_1_pattern in asin_bs_sellers_rank_lower:
                    cate_1_rank, cate_current_rank = rank_list[0], rank_list[-1]
                else:
                    cate_1_rank, cate_current_rank = None, rank_list[0]
        else:
            cate_1_rank, cate_current_rank = None, None

        return cate_1_id, cate_current_id, cate_1_rank, cate_current_rank

    def read_data(self):
        sql = f"select asin, category_id as asin_bs_cate_current_id_node, category_first_id as asin_bs_cate_1_id_node from dim_asin_detail where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info ='{self.date_info}';"  # and date_info>='2023-15'
        print(f"1. 读取dim_asin_detail表node_id数据: sql -- {sql}")
        self.df_asin_node_id = self.spark.sql(sqlQuery=sql).cache()
        self.df_asin_node_id.show(10, truncate=False)

        # 2. 读取ods_bs_category_asin_detail对应周期的详情表
        params = f" date_info <= '2022-42'" if max(self.year_week_tuple) <= '2022-42' and date_type == 'month' else f" date_info in {self.year_week_tuple}"
        sql = f"select asin, best_sellers_rank as asin_bs_sellers_rank, last_herf, all_best_sellers_href, date_type, date_info, created_at from ods_bs_category_asin_detail " \
              f"where site_name='{self.site_name}' and date_type='week' and {params};"

        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 asin, best_sellers_rank as asin_bs_sellers_rank, last_herf, all_best_sellers_href, date_type, date_info, created_at from ods_bs_category_asin_detail " \
                  f"where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info='{self.date_info}';"
        print(f"2. 读取ods_bs_category_asin_detail对应周期的详情表: sql -- {sql}")

        self.df_bs_asin_detail = self.spark.sql(sqlQuery=sql).cache()
        # self.df_bs_asin_detail = self.df_bs_asin_detail.drop_duplicates(['asin', 'date_info'])
        self.df_bs_asin_detail.show(10, truncate=False)

    def handle_df_asin_node_id(self):
        # 保留asin最新的node_id
        # self.df_asin_node_id = self.df_asin_node_id.filter("asin_bs_cate_current_id is not null")
        # window = Window.partitionBy(['asin']).orderBy(
        #     self.df_asin_node_id.date_info.desc()
        # )
        # self.df_asin_node_id = self.df_asin_node_id.withColumn(
        #     "row_number", F.row_number().over(window=window)
        # )
        # self.df_asin_node_id = self.df_asin_node_id.filter("row_number=1")
        # self.df_asin_node_id = self.df_asin_node_id.drop("row_number")
        pass  # dim_asin_detail已经去重过

    def handle_df_bs_asin_detail(self):
        # 保留asin最新的asin_bs_sellers_rank_lower
        window = Window.partitionBy(['asin']).orderBy(
            self.df_bs_asin_detail.created_at.desc()
        )
        self.df_bs_asin_detail = self.df_bs_asin_detail.withColumn(
            "row_number", F.row_number().over(window=window)
        )
        self.df_bs_asin_detail = self.df_bs_asin_detail.filter("row_number=1")
        self.df_bs_asin_detail = self.df_bs_asin_detail.drop("row_number", "date_info")

        # 小写
        self.df_bs_asin_detail = self.df_bs_asin_detail.withColumn("asin_bs_sellers_rank_lower",
                                                                   F.lower("asin_bs_sellers_rank"))
        # self.df_bs_asin_detail.show(10, truncate=False)
        # 提取分类字符串中的asin_bs_cate_1_rank, asin_bs_cate_current_rank
        # 生成当前分类匹配规则
        cate_current_pattern = self.pattern_current_dict[self.site_name]
        cate_1_pattern = self.pattern1_dict[self.site_name]
        self.df_bs_asin_detail = self.df_bs_asin_detail.withColumn(
            'asin_bs_cate_ranks',
            self.u_parse_bs_category('asin_bs_sellers_rank_lower', 'last_herf', 'all_best_sellers_href',
                                     F.lit(cate_current_pattern), F.lit(cate_1_pattern))
        )
        # self.df_bs_asin_detail.show(10, truncate=False)
        self.df_bs_asin_detail = self.df_bs_asin_detail \
            .withColumn('asin_bs_cate_1_id', self.df_bs_asin_detail.asin_bs_cate_ranks.getField('asin_bs_cate_1_id')) \
            .withColumn('asin_bs_cate_current_id',
                        self.df_bs_asin_detail.asin_bs_cate_ranks.getField('asin_bs_cate_current_id')) \
            .withColumn('asin_bs_cate_1_rank',
                        self.df_bs_asin_detail.asin_bs_cate_ranks.getField('asin_bs_cate_1_rank')) \
            .withColumn('asin_bs_cate_current_rank',
                        self.df_bs_asin_detail.asin_bs_cate_ranks.getField('asin_bs_cate_current_rank')) \
            .drop('asin_bs_cate_ranks')
        self.df_bs_asin_detail.show(10, truncate=False)
        # self.df_save = self.df_asin_node_id.join(
        #     self.df_bs_asin_detail, 'asin', how='left'
        # ).join(
        #     self.df_category_desc_id, 'asin_bs_cate_current_id', how='left'
        # )
        self.df_save = self.df_asin_node_id.join(
            self.df_bs_asin_detail, 'asin', how='left'
        )

        # 用node_id的分类去补充一级分类和当前分类
        self.df_save = self.df_save.withColumn(
            "asin_bs_cate_1_id",
            F.when(F.col("asin_bs_cate_1_id").isNull(), F.col("asin_bs_cate_1_id_node")).otherwise(F.col("asin_bs_cate_1_id"))
        ).withColumn(
            "asin_bs_cate_current_id",
            F.when(F.col("asin_bs_cate_current_id").isNull(), F.col("asin_bs_cate_current_id_node")).otherwise(F.col("asin_bs_cate_current_id"))
        )

        self.df_save = self.df_save.drop("asin_bs_sellers_rank_lower", "asin_bs_cate_1_id_node", "asin_bs_cate_current_id_node")
        self.df_save.show(20)

    def handle_data(self):
        self.handle_df_asin_node_id()
        self.handle_df_bs_asin_detail()
        self.df_save = self.df_save.withColumn("site_name", F.lit(self.site_name))
        self.df_save = self.df_save.withColumn("date_type", F.lit(self.date_type))
        self.df_save = self.df_save.withColumn("date_info", F.lit(self.date_info))


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