dwt_st_info.py

"""
1. 热搜词，上升词，新出词，在售商品数等
2. 预估销量
3. bs销量， bs的category_id
4，st_ao_val
"""
"""
author: 方星钧(ffman)
description: 基于dwd层等表，计算出search_term和asin维度的基础信息表(包括预估销量)
table_read_name: dwd_st_counts系列, dwd_st_info系列, dwd_st_asin_info系列， dwd_asin_bs_info
table_save_name: dwt_st_info系列
table_save_level: dwt
version: 2.0
created_date: 2022-06-20
updated_date: 2022-12-25
"""

import os
import sys

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


class DwtStInfo(Templates):

    def __init__(self, site_name="us", date_type="week", date_info="2022-01"):
        super().__init__()
        self.site_name = site_name
        self.date_type = date_type
        self.date_info = date_info
        self.db_save = f"dwt_st_info"
        self.spark = self.create_spark_object(app_name=f"{self.db_save} {self.site_name}, {self.date_info}")
        self.df_date = self.get_year_week_tuple()
        self.get_date_info_tuple()
        self.df_save = 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_detail = self.spark.sql(f"select 1+1;")
        self.df_st_counts = self.spark.sql(f"select 1+1;")
        self.df_st_asin_info = self.spark.sql(f"select 1+1;")
        self.df_st_asin_zr = self.spark.sql(f"select 1+1;")
        self.df_st_key = self.spark.sql(f"select 1+1;")
        self.partitions_by = ['site_name', 'date_type', 'date_info']
        self.reset_partitions(1)
        self.u_is_title_appear = self.spark.udf.register("u_is_title_appear", self.udf_is_title_appear, IntegerType())
        print("self.date_info_tuple:", self.date_info_tuple)
        self.current_date = self.date_info_tuple[0]
        print("self.current_date:", self.current_date)

    @staticmethod
    def udf_is_title_appear(search_term, title):
        if str(search_term).lower() in str(title).lower():
            return 1
        else:
            return 0

    def read_data(self):
        print("1.1 读取asin维度： dim_cal_asin_history_detail表")
        sql = f"select asin, bsr_cate_1_id, bsr_cate_current_id as st_asin_bs_cate_current_id, " \
              f"asin_rank, asin_title, asin_launch_time, asin_price as asin1_price, asin_rating as asin1_rating, " \
              f"asin_total_comments as asin1_total_comments " \
              f"from dim_cal_asin_history_detail " \
              f"where site_name='{self.site_name}';"
        print("sql:", sql)
        self.df_asin_detail = self.spark.sql(sql).cache()
        self.df_asin_detail.show(10)
        print("1.2 读取bsr维度： ods_one_category_report表")
        sql = f"select cate_1_id as bsr_cate_1_id, rank as asin_rank, orders as asin_bs_orders from ods_one_category_report " \
              f"where site_name='{self.site_name}' and date_type='month' and date_info='{self.year}-{self.month}';"
        print("sql:", sql)
        self.df_bs_report = self.spark.sql(sqlQuery=sql).cache()
        self.df_bs_report.show(10)
        print("1.3 读取dim_st_detail系列表")
        sql = f"select * from dim_st_detail " \
              f"where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info = '{self.date_info}';"
        print("sql:", sql)
        self.df_st_detail = self.spark.sql(sql).cache()
        self.df_st_detail.show(10, truncate=False)
        print("1.4 读取dwd_st_counts系列表")
        sql = f"select search_term, st_ao_val, st_ao_val_rank, st_ao_val_rate, st_zr_counts, st_sp_counts, " \
              f"st_sb_counts, st_sb1_counts, st_sb2_counts, st_sb3_counts, st_adv_counts, " \
              f"st_ac_counts, st_bs_counts, st_er_counts, st_tr_counts " \
              f" from dwd_st_counts " \
              f"where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info = '{self.date_info}';"
        print("sql:", sql)
        self.df_st_counts = self.spark.sql(sql).cache()
        self.df_st_counts.show(10, truncate=False)
        print("1.5 读取dwd_st_asin_info系列表")
        sql = f"select search_term, asin, st_asin_zr_orders as st_asin_orders, st_asin_zr_orders_sum as st_asin_orders_sum from dwd_st_asin_info " \
              f"where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info = '{self.date_info}';"
        print("sql:", sql)
        self.df_st_asin_info = self.spark.sql(sql).cache()
        self.df_st_asin_info.show(10, truncate=False)
        sql = f"select search_term, asin from ods_search_term_zr " \
              f"where site_name ='{self.site_name}' and date_type='week' and date_info in {self.year_week_tuple} and page=1"
        self.df_st_asin_zr = self.spark.sql(sql).cache()
        self.df_st_asin_zr = self.df_st_asin_zr.drop_duplicates(['search_term', 'asin'])
        self.df_st_asin_zr.show(10)
        print("1.6 读取ods_st_key表")
        sql = f"select st_key, search_term from ods_st_key " \
              f"where site_name='{self.site_name}';"
        print("sql:", sql)
        self.df_st_key = self.spark.sql(sql).cache()
        self.df_st_key.show(10, truncate=False)

    def handle_data(self):
        self.handle_join()
        self.handle_data_asin_new()
        self.handle_data_st_orders()
        self.handle_data_asin_detail()
        self.handle_st_asin_zr_title()
        self.df_save = self.df_st_detail
        print(self.df_save.columns)
        self.df_save.show(10)
        self.df_save = self.df_save.drop("st_updated_time", "st_bsr_cate_1_id_new", "st_bsr_cate_current_id_new")
        # quit()

    def handle_join(self):
        # st维度
        self.df_st_detail = self.df_st_detail.join(
            self.df_st_counts, on="search_term", how="left"
        ).join(
            self.df_st_key, on="search_term", how="left"
        )
        # asin维度
        self.df_asin_detail = self.df_asin_detail.join(
            self.df_bs_report, on=['asin_rank', 'bsr_cate_1_id'], how='left'
        )
        # self.df_asin_detail = self.df_asin_detail.drop("bsr_cate_1_id")
        # st+asin维度
        self.df_st_asin_zr = self.df_st_asin_zr.join(
            self.df_asin_detail.select("asin", "asin_title"), on='asin', how='left'
        )

    def handle_data_asin_new(self):
        """
        1. 对self.df_asin_bs_info对象，选择asin最新一周的数据，并删掉不需要的字段
        2. 获取asin
        """
        # 获取新品的判定
        self.df_asin_detail = self.df_asin_detail.withColumn("current_date", F.lit(self.date_info_tuple[-1]))
        self.df_asin_detail = self.df_asin_detail.withColumn("days_diff", F.datediff("current_date", "asin_launch_time"))
        self.df_asin_detail = self.df_asin_detail.withColumn(
            "asin_new_flag",
            F.when(
                self.df_asin_detail.days_diff > 180, 0
            ).when(
                self.df_asin_detail.days_diff > 0, 1
            ).otherwise(2)
        )
        self.df_asin_detail.show(10, truncate=False)

    def handle_data_st_orders(self):
        """
        计算关键词维度的st_asin_bs_orders_sum和st_asin_orders_sum
        """
        self.df_st_asin_info = self.df_st_asin_info.join(
            self.df_asin_detail.select("asin", "asin_bs_orders", "asin_new_flag"), on="asin", how="left"
        )

        # df_st_search_sum = self.df_st_asin_info.groupby(['search_term']). \
        #     agg({"st_search_sum": "max"})
        # df_st_search_sum = df_st_search_sum.withColumnRenamed("max(st_search_sum)", "st_search_sum")
        self.df_st_asin_info = self.df_st_asin_info.withColumnRenamed("asin_bs_orders", "st_asin_bs_orders")
        df_st_asin_bs_orders_sum = self.df_st_asin_info.groupby(['search_term']). \
            agg({"st_asin_bs_orders": "sum"})
        df_st_asin_bs_orders_sum = df_st_asin_bs_orders_sum.withColumnRenamed("sum(st_asin_bs_orders)",
                                                                              "st_asin_bs_orders_sum")
        df_st_asin_orders_sum = self.df_st_asin_info.groupby(['search_term']). \
            agg({"st_asin_orders_sum": "max", "asin": "count"})
        # df_st_asin_orders_sum.show(10, truncate=False)
        df_st_asin_orders_sum = df_st_asin_orders_sum.withColumnRenamed("max(st_asin_orders_sum)", "st_asin_orders_sum")
        df_st_asin_orders_sum = df_st_asin_orders_sum.withColumnRenamed("count(asin)", "st_asin_counts")
        df_st_asin_new_orders_sum = self.df_st_asin_info.filter("asin_new_flag = 1").groupby(['search_term']). \
            agg({"st_asin_orders": "sum", "asin": "count"})
        df_st_asin_new_orders_sum = df_st_asin_new_orders_sum.withColumnRenamed("sum(st_asin_orders)", "st_asin_new_orders_sum")
        df_st_asin_new_orders_sum = df_st_asin_new_orders_sum.withColumnRenamed("count(asin)", "st_asin_new_counts")
        # df_st_asin_new_orders_sum.show(10, truncate=False)
        self.df_st_detail = self.df_st_detail.join(
            df_st_asin_bs_orders_sum, on="search_term", how="left"
        ).join(
            df_st_asin_orders_sum, on="search_term", how="left"
        ).join(
            df_st_asin_new_orders_sum, on="search_term", how="left"
        )
        self.df_st_detail = self.df_st_detail.withColumn("st_asin_new_orders_rate", self.df_st_detail.st_asin_new_orders_sum/self.df_st_detail.st_asin_orders_sum)
        self.df_st_detail = self.df_st_detail.withColumn("st_asin_new_counts_rate", self.df_st_detail.st_asin_new_counts/self.df_st_detail.st_asin_counts)

    def handle_data_asin_detail(self):
        self.df_st_detail = self.df_st_detail.join(
            self.df_asin_detail.select("asin", "asin_bs_orders", "asin1_price", "asin1_rating", "asin1_total_comments").withColumnRenamed("asin", "st_top_asin1"), on="st_top_asin1", how="left"
        ).withColumnRenamed("asin_bs_orders", "st_asin1_bs_orders")

    def handle_st_asin_zr_title(self):
        self.df_st_asin_zr = self.df_st_asin_zr.withColumn(
            "st_asin_in_title_flag",
            self.u_is_title_appear(self.df_st_asin_zr.search_term, self.df_st_asin_zr.asin_title)
        )
        df_st_zr_page1_counts = self.df_st_asin_zr.groupby("search_term").count()
        df_st_zr_page1_counts = df_st_zr_page1_counts.withColumnRenamed("count", "st_zr_page1_counts")

        df_st_zr_page1_in_title_counts = self.df_st_asin_zr.filter("st_asin_in_title_flag=1").groupby(
            "search_term").count()
        df_st_zr_page1_in_title_counts = df_st_zr_page1_in_title_counts.withColumnRenamed("count", "st_zr_page1_in_title_counts")
        df_st_zr_page1_counts = df_st_zr_page1_counts.join(
            df_st_zr_page1_in_title_counts, on='search_term', how='left'
        )
        df_st_zr_page1_counts = df_st_zr_page1_counts.fillna(0)
        df_st_zr_page1_counts = df_st_zr_page1_counts.withColumn(
            "st_zr_page1_in_title_rate", df_st_zr_page1_counts.st_zr_page1_in_title_counts / df_st_zr_page1_counts.st_zr_page1_counts
        )
        df_st_zr_page1_counts.show(10, truncate=False)
        self.df_st_detail = self.df_st_detail.join(
            df_st_zr_page1_counts, on='search_term', how='left'
        )


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 = DwtStInfo(site_name=site_name, date_type=date_type, date_info=date_info)
    handle_obj.run()