dim_asin_volume.py

"""
author: 汪瑞
description: 基于dim_asin_detail表，计算出asin维度下的产品体积相关信息
table_read_name: dim_asin_detail
table_save_name: dim_asin_volume
table_save_level: dwd
version: 3.0
created_date: 2022-05-12
updated_date: 2022-12-15
"""

import os
import re
import sys

from pyspark.storagelevel import StorageLevel

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, MapType, DoubleType


class DimAsinVolume(Templates):

    def __init__(self, site_name="us", date_type="week", date_info="2022-1"):
        super().__init__()
        self.site_name = site_name
        self.date_type = date_type
        self.date_info = date_info
        self.db_save = f"dim_asin_volume"
        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.df_save = self.spark.sql(f"select 1+1;")
        self.df_asin_detail = self.spark.sql(f"select 1+1;")
        self.week_counts = 1 if self.date_type == 'week' else len(self.year_week_tuple)
        self.partitions_by = ['site_name']
        self.reset_partitions(10)
        self.u_parse_asin_volume = self.spark.udf.register('u_parse_asin_volume', self.udf_parse_asin_volume,
                                                           MapType(StringType(), DoubleType(), True))

    @staticmethod
    def udf_parse_asin_volume(asin_volume):
        asin_volume = str(asin_volume)
        pattern_first = r"\d+\.?\d*[\'\"]?[LWHD]?"
        pattern_sec = r"\d+\.?\d*"
        asin_length = ''
        asin_width = ''
        asin_high = ''
        asin_element_unit = ''
        if asin_volume.lower() not in ['null', 'none'] and "A4" not in asin_volume and (
                asin_volume.count(" x ") > 0 or asin_volume.count("*") > 0) and not (
                asin_volume.count('cm') > 0 and asin_volume.count('inch') > 0):
            element_list = re.findall(pattern_first, asin_volume)
            if len(element_list) == 3:
                is_cm = asin_volume.lower().count('cm')
                if (("D" in asin_volume or "L" in asin_volume) and "W" in asin_volume and "H" in asin_volume):
                    for element in element_list:
                        if ("D" in element or "L" in element):
                            asin_length_list = re.findall(pattern_sec, element)
                            if len(asin_length_list) > 0:
                                asin_length = asin_length_list[0]
                        elif ("W" in element):
                            asin_width_list = re.findall(pattern_sec, element)
                            if len(asin_width_list) > 0:
                                asin_width = asin_width_list[0]
                        elif ("H" in element):
                            asin_high_list = re.findall(pattern_sec, element)
                            if len(asin_high_list) > 0:
                                asin_high = asin_high_list[0]
                    if (asin_length != '' and asin_width != '' and asin_high != ''):
                        asin_length = round(float(asin_length), 2)
                        asin_width = round(float(asin_width), 2)
                        asin_high = round(float(asin_high), 2)
                        if is_cm > 0:
                            asin_element_unit = 0
                        else:
                            asin_element_unit = 1
                    else:
                        asin_length = None
                        asin_width = None
                        asin_high = None
                else:
                    asin_element1 = element_list[0].strip()
                    asin_element1_list = re.findall(pattern_sec, asin_element1)
                    if len(asin_element1_list) > 0:
                        asin_element1 = asin_element1_list[0]
                    else:
                        asin_element1 = ''
                    asin_element2 = element_list[1].strip()
                    asin_element2_list = re.findall(pattern_sec, asin_element2)
                    if len(asin_element2_list) > 0:
                        asin_element2 = asin_element2_list[0]
                    else:
                        asin_element2 = ''
                    asin_element3 = element_list[-1].strip()
                    asin_element3_list = re.findall(pattern_sec, asin_element3)
                    if len(asin_element3_list) > 0:
                        asin_element3 = asin_element3_list[0]
                    else:
                        asin_element3 = ''
                    if (asin_element1 != '' and asin_element2 != '' and asin_element3 != ''):
                        asin_element1 = round(float(asin_element1), 2)
                        asin_element2 = round(float(asin_element2), 2)
                        asin_element3 = round(float(asin_element3), 2)
                        asin_element_list = [asin_element1, asin_element2, asin_element3]
                        asin_element_list.sort()
                        asin_high = asin_element_list[0]
                        asin_width = asin_element_list[1]
                        asin_length = asin_element_list[-1]
                        if is_cm > 0:
                            asin_element_unit = 0
                        else:
                            asin_element_unit = 1
                    else:
                        asin_length = None
                        asin_width = None
                        asin_high = None
                if asin_element_unit != '':
                    asin_element_unit = round(float(asin_element_unit), 2)
            return {"asin_length": asin_length, "asin_width": asin_width, "asin_high": asin_high,
                    "asin_element_unit": asin_element_unit}
        else:
            return {"asin_length": None, "asin_width": None, "asin_high": None,
                    "asin_element_unit": None}

    def read_data(self):
        print("1.读取dim_cal_asin_history_detail表")
        sql = f"select " \
              f"asin, " \
              f"asin_volume " \
              f"from dim_cal_asin_history_detail where site_name='{self.site_name}'"
        print("sql:" + sql)
        self.df_asin_detail = self.spark.sql(sqlQuery=sql).cache()
        self.df_asin_detail = self.df_asin_detail.na.fill({"asin_volume": "none"})
        self.df_asin_detail.show(10, truncate=False)

    def parse_asin_volume(self):
        # self.df_asin_detail = self.df_asin_detail.withColumn("asin_length", F.when(
        #     self.u_parse_asin_volume(self.df_asin_detail.asin_volume)["asin_length"].isNotNull(),
        #     self.u_parse_asin_volume(self.df_asin_detail.asin_volume)["asin_length"]).otherwise(F.lit(None)))
        # self.df_asin_detail = self.df_asin_detail.withColumn("asin_width", F.when(
        #     self.u_parse_asin_volume(self.df_asin_detail.asin_volume)["asin_width"].isNotNull(),
        #     self.u_parse_asin_volume(self.df_asin_detail.asin_volume)["asin_width"]).otherwise(F.lit(None)))
        # self.df_asin_detail = self.df_asin_detail.withColumn("asin_high", F.when(
        #     self.u_parse_asin_volume(self.df_asin_detail.asin_volume)["asin_high"].isNotNull(),
        #     self.u_parse_asin_volume(self.df_asin_detail.asin_volume)["asin_high"]).otherwise(F.lit(None)))
        # self.df_save = self.df_asin_detail.drop("asin_volume")
        try:
            asin_volume_map = self.u_parse_asin_volume(self.df_asin_detail.asin_volume)
            self.df_asin_detail = self.df_asin_detail.withColumn("asin_length",
                                                                 asin_volume_map["asin_length"])
            self.df_asin_detail = self.df_asin_detail.withColumn("asin_width",
                                                                 asin_volume_map["asin_width"])
            self.df_asin_detail = self.df_asin_detail.withColumn("asin_high",
                                                                 asin_volume_map["asin_high"])
            self.df_asin_detail = self.df_asin_detail.withColumn("asin_element_unit",
                                                                 F.when(asin_volume_map["asin_element_unit"] == 0,
                                                                        'cm').when(
                                                                     asin_volume_map["asin_element_unit"] == 1,
                                                                     'inches').otherwise(F.lit(None)))
            self.df_asin_detail = self.df_asin_detail.withColumn("asin_length", F.round(F.col("asin_length"), 2))
            self.df_asin_detail = self.df_asin_detail.withColumn("asin_width", F.round(F.col("asin_width"), 2))
            self.df_asin_detail = self.df_asin_detail.withColumn("asin_high", F.round(F.col("asin_high"), 2))
        except Exception as e:
            print(e)
        finally:
            self.df_save = self.df_asin_detail

    def handle_data(self):
        self.parse_asin_volume()
        self.df_save.show(10, truncate=False)
        self.df_save = self.df_save.withColumn("created_time",
                                               F.date_format(F.current_timestamp(), 'yyyy-MM-dd HH:mm:SS')). \
            withColumn("updated_time", F.date_format(F.current_timestamp(), 'yyyy-MM-dd HH:mm:SS'))
        self.df_save = self.df_save.withColumn("re_string_field1", F.lit("null"))
        self.df_save = self.df_save.withColumn("re_string_field2", F.lit("null"))
        self.df_save = self.df_save.withColumn("re_string_field3", F.lit("null"))
        self.df_save = self.df_save.withColumn("re_int_field1", F.lit(0))
        self.df_save = self.df_save.withColumn("re_int_field2", F.lit(0))
        self.df_save = self.df_save.withColumn("re_int_field3", F.lit(0))
        self.df_save = self.df_save.withColumn("site_name", F.lit(self.site_name))


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