"""
   @Author      : HuangJian
   @SourceTable :
                  ①ods_seller_account_syn
                  ②ods_seller_asin_account
                  ③ods_seller_account_feedback
   @SinkTable   :
                  ①dim_fd_asin_info
   @CreateTime  : 2022/12/19 9:56
   @UpdateTime  : 2022/12/19 9:56
"""

import os
import sys

sys.path.append(os.path.dirname(sys.path[0]))  # 上级目录

from pyspark.sql.window import Window
from pyspark.sql import functions as F
from utils.spark_util import SparkUtil
from utils.hdfs_utils import HdfsUtils
from utils.common_util import CommonUtil

class DwtFdAsinInfo(object):
    def __init__(self, site_name='us', date_type='month', date_info='2026-06'):
        super().__init__()
        self.hive_tb = "dim_fd_asin_info"
        self.site_name = site_name
        self.date_type = date_type
        self.date_info = date_info
        self.partition_dict = {
            "site_name": site_name,
        }
        # 落表路径校验
        self.hdfs_path = CommonUtil.build_hdfs_path(self.hive_tb, partition_dict=self.partition_dict)
        app_name = f"{self.hive_tb}:{self.site_name}_{self.date_type}_{self.date_info}"
        self.spark = SparkUtil.get_spark_session(app_name)
        self.partitions_num = CommonUtil.reset_partitions(self.site_name, 80)

        # 初始化全局变量df--ods获取数据的原始df
        self.df_seller_account_syn = self.spark.sql("select 1+1;")
        self.df_seller_account_feedback = self.spark.sql("select 1+1;")
        self.df_fd_asin = self.spark.sql("select 1+1;")

        # 初始化全局变量df--dwd层转换输出的df
        self.df_save = self.spark.sql(f"select 1+1;")

    def read_data(self):
        # 获取爬虫店铺记录表
        print("获取 ods_seller_account_syn")
        sql = f"""
            select id as fd_account_id, seller_id as unique_id, account_name as fd_account_name, 
            lower(account_name) as fd_account_name_lower, url as fd_url
            from ods_seller_account_syn where site_name = '{self.site_name}'
        """
        # seller_id 本身在 ods_seller_account_syn 里是唯一的，不需要再去重
        self.df_seller_account_syn = self.spark.sql(sqlQuery=sql)
        print(sql)

        # 获取店铺详情表：只读传参指定的这一个分区（date_type+date_info 即调度传入的最新分区）
        print("获取 ods_seller_account_feedback")
        sql = f"""
            select seller_id as unique_id, country_name as fd_country_name, created_at
            from ods_seller_account_feedback
            where site_name = '{self.site_name}' and date_type = '{self.date_type}' and date_info = '{self.date_info}'
        """
        self.df_seller_account_feedback = self.spark.sql(sqlQuery=sql)
        print(sql)
        window = Window.partitionBy('unique_id').orderBy(F.col('created_at').desc())
        self.df_seller_account_feedback = self.df_seller_account_feedback.withColumn(
            'rank', F.row_number().over(window)
        ).filter('rank = 1').withColumn(
            'fb_crawl_date', F.date_format(F.col('created_at'), 'yyyy-MM-dd HH:mm:ss')
        ).drop('rank', 'created_at')

        # 获取店铺与asin的对应关系库(店铺与asin所有历史对应关系表)
        print("获取 ods_seller_asin_account")
        sql = f"""
            select seller_id as unique_id, asin from ods_seller_asin_account where site_name='{self.site_name}'
        """
        self.df_fd_asin = self.spark.sql(sqlQuery=sql)
        self.df_fd_asin = self.df_fd_asin.drop_duplicates(['unique_id', 'asin'])
        print(sql)

    def save_data(self):
        df_save = self.df_seller_account_syn.join(
            self.df_seller_account_feedback, on='unique_id', how='left'
        ).join(
            self.df_fd_asin, on='unique_id', how='left'
        )

        df_save = df_save.select(
            F.col('fd_account_id'),
            F.col('unique_id').alias('fd_unique'),
            F.col('fd_account_name'),
            F.col('fd_account_name_lower'),
            F.col('fd_country_name'),
            F.col('fd_url'),
            F.col('asin'),
            F.date_format(F.current_timestamp(), 'yyyy-MM-dd HH:mm:ss').alias('created_at'),
            F.date_format(F.current_timestamp(), 'yyyy-MM-dd HH:mm:ss').alias('updated_at'),
            F.col('fb_crawl_date'),
            F.lit(self.site_name).alias('site_name'),
        )

        print(f"清除hdfs目录中:{self.hdfs_path}")
        HdfsUtils.delete_file_in_folder(self.hdfs_path)
        df_save = df_save.repartition(self.partitions_num)
        partition_by = ["site_name"]
        print(f"当前存储的表名为：{self.hive_tb},分区为{partition_by}", )
        df_save.write.saveAsTable(name=self.hive_tb, format='hive', mode='append', partitionBy=partition_by)
        print("success")

    def run(self):
        self.read_data()
        self.save_data()

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