import os
import re
import sys
import time
import traceback

sys.path.append("/opt/module/spark-3.2.0-bin-hadoop3.2/demo/py_demo/")

sys.path.append(os.path.dirname(sys.path[0]))  # 上级目录
from utils.templates import Templates
from pyspark.sql import functions as F
from pyspark.sql.types import *
from utils.db_util import DBUtil
from utils.common_util import CommonUtil
from utils.spark_util import SparkUtil
from functools import reduce
from pyspark.sql import Window
from pyspark.storagelevel import StorageLevel
from utils.DorisHelper import DorisHelper
from yswg_utils.common_df import get_node_first_id_df, get_first_id_from_category_desc_df
from yswg_utils.common_udf import udf_parse_bs_category, parse_weight_str, udf_extract_volume_dimensions, udf_get_package_quantity_with_flag as udf_get_package_quantity, udf_parse_seller_json, udf_parse_amazon_orders, resolve_asin_jump

# asin_is_need 判断中需要过滤的类目（对齐 dwt.handle_asin_is_hide）
NEED_FILTER_CATEGORIES = (
    'mobile-apps', 'audible', 'books', 'music', 'dmusic', 'digital-text',
    'magazines', 'movies-tv', 'software', 'videogames',
    'amazon-devices', 'boost', 'us-live-explorations', 'amazon-renewed'
)


class KafkaFlowAsinDetail(Templates):
    def __init__(self, site_name='us', date_type="month", date_info='2026-03', consumer_type='history', test_flag='test', batch_size=1200000):
        super().__init__()
        self.site_name = site_name
        self.date_type = date_type
        self.date_info = date_info
        self.consumer_type = consumer_type
        self.test_flag = test_flag
        # 用于父 ASIN 详情表 / ASIN 最新详情表的 date_info 字段写入
        self.date_info_month = self.date_info[:7]
        # actual_last_month：环比实际基线月，兼容重跑历史数据——
        # day模式：从workflow_everyday取最新已产出月份（实时口径，始终对比最新完成的月）；
        # month模式：直接取上一个月（按date_info推算），避免重跑历史月份时基线漂移到"当前最新月"
        if self.date_type == 'day':
            _engine_mysql = DBUtil.get_db_engine('mysql', 'us')
            _sql_latest = (
                f"SELECT MAX(report_date) "
                f"FROM workflow_everyday "
                f"WHERE site_name='{self.site_name}' AND date_type='month' "
                f"AND page='流量选品' AND status_val=14 AND is_end='是'"
            )
            self.actual_last_month = list(DBUtil.engine_exec_sql(_engine_mysql, _sql_latest))[0][0]
        elif self.date_type == 'month':
            self.actual_last_month = CommonUtil.get_month_offset(self.date_info, -1)
        else:
            print(f"不支持的date_type: {self.date_type}，程序退出")
            sys.exit(1)
        self.actual_last_year = CommonUtil.get_month_offset(self.actual_last_month, -11)
        print(f"环比基线月：{self.actual_last_month}，同比基线月：{self.actual_last_year}")
        # spark相关参数
        self.app_name = self.get_app_name()
        self.spark = SparkUtil.get_stream_spark(app_name=self.app_name)
        self.processing_time = 900 if self.site_name == 'us' else 600
        self.repartition_num = 80
        # kafka相关参数（topic 按 date_type 动态：day → {site}_asin_detail_day_{yyyy_MM_dd}，month → {site}_asin_detail_month_{yyyy_MM}）
        self.topic_name = f"{self.site_name}_asin_detail_{self.date_type}_{str(self.date_info).replace('-', '_')}"
        self.batch_size = batch_size
        self.batch_size_history = 100000
        self.check_path = f"/home/big_data_selection/tmp/kafka_checkpoint/{self.topic_name}_{self.consumer_type}_test" if self.test_flag == 'test' else f"/home/big_data_selection/tmp/kafka_checkpoint/{self.topic_name}_{self.consumer_type}"
        self.schema = self.init_schema()
        # doris相关参数：主表落 dwt 库，最新详情/父 ASIN 详情表落 selection 库
        self.doris_db_dwt        = "test" if self.test_flag == "test" else "dwt"
        self.doris_db_selection  = "test" if self.test_flag == "test" else "selection"
        self.doris_30day_table   = f"{self.site_name}_flow_asin_30day_test" if self.test_flag == "test" else f"{self.site_name}_flow_asin_30day"
        self.parent_asin_latest_detail_table = f"{self.site_name}_parent_asin_latest_detail_test" if self.test_flag == "test" else f"{self.site_name}_parent_asin_latest_detail"
        self.asin_latest_detail_table        = f"{self.site_name}_asin_latest_detail_test" if self.test_flag == "test" else f"{self.site_name}_asin_latest_detail"
        # 分类解析模板
        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.pattern_str = {
            "us": r"(\d+ in [\w&' -]+)",
            "uk": r"(\d+ in [\w&' -]+)",
            # de: 分类名字符集 [\w&' -] 含字母和空格，会贪婪吃后续 "nr 数字" 段；
            # 用非贪婪 +? 配合 lookahead 限定结尾：下一段 nr 数字 / 括号注释 / 字符串结束
            "de": r"nr (\d+ in [\w&' -]+?)(?=\s*\(|\s+nr\s+\d|$)",
            "es": r"nº(\d+ en [\w&' -]+)",
            "fr": r"(\d+ en [\w&' -]+)",
            "it": r"n (\d+ in [\w&' -]+)",    # n. → n
        }  # 匹配排名和分类名称（与 best_sellers_rank lower + replace(".", "") 后匹配）
        # DataFrame初始化
        self.df_previous_flow_asin = self.spark.sql("select 1+1;")
        self.df_previous_flow_asin_lastyear = self.spark.sql("select 1+1;")
        self.df_seller_info = self.spark.sql("select 1+1;")
        self.df_seller_country = self.spark.sql("select 1+1;")
        self.df_asin_seller = self.spark.sql("select 1+1;")
        self.df_asin_label_info = self.spark.sql("select 1+1;")
        self.df_asin_measure = self.spark.sql("select 1+1;")
        self.df_bs_report = self.spark.sql("select 1+1;")
        self.df_asin_new_cate = self.spark.sql("select 1+ 1;")
        self.df_asin_category = self.spark.sql("select 1+1;")
        self.df_self_asin = self.spark.sql("select 1+1;")
        self.df_self_brand = self.spark.sql("select 1+1;")
        self.df_hide_category = self.spark.sql("select 1+1;")
        self.df_ai_hide_category = self.spark.sql("select 1+1;")
        self.color_set = set()
        self.save_parent_asin_latest_detail_cache = None
        self.save_asin_latest_detail_cache = None
        # udf函数注册
        package_schema = StructType([
            StructField("parse_package_quantity", IntegerType(), True),
            StructField("is_package_quantity_abnormal", IntegerType(), True),
        ])
        self.u_parse_package_quantity = self.spark.udf.register('u_parse_package_quantity', udf_get_package_quantity, package_schema)
        bs_category_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 = self.spark.udf.register('u_parse_bs_category', udf_parse_bs_category, bs_category_schema)
        weight_schema = StructType([
            StructField('weight', FloatType(), True),
            StructField('weight_type', StringType(), True)
        ])
        self.u_parse_weight = self.spark.udf.register('u_parse_weight', parse_weight_str, weight_schema)
        volume_schema = StructType([
            StructField("length", FloatType(), True),
            StructField("width", FloatType(), True),
            StructField("height", FloatType(), True),
            StructField("asin_volume_type", StringType(), True)
        ])
        self.u_parse_volume = self.spark.udf.register('u_parse_volume', udf_extract_volume_dimensions, volume_schema)
        seller_schema = StructType([
            StructField("buy_box_seller_type", IntegerType(), True),
            StructField("account_name", StringType(), True),
            StructField("account_id", StringType(), True)
        ])
        self.u_parse_seller_info = self.spark.udf.register('u_parse_seller_info', udf_parse_seller_json, seller_schema)
        self.u_rank_and_category = self.spark.udf.register('u_rank_and_category', self.udf_rank_and_category, StringType())
        self.u_parse_amazon_orders = self.spark.udf.register('u_parse_amazon_orders', udf_parse_amazon_orders, IntegerType())

    @staticmethod
    def init_schema():
        schema = StructType([
            StructField("asin", StringType(), True),
            StructField("current_asin", StringType(), True),
            StructField("title", StringType(), True),
            StructField("img_url", StringType(), True),
            StructField("rating", DoubleType(), True),
            StructField("total_comments", IntegerType(), True),
            StructField("price", FloatType(), True),
            StructField("category", StringType(), True),
            StructField("launch_time", StringType(), True),
            StructField("volume", StringType(), True),
            StructField("page_inventory", IntegerType(), True),
            StructField("asin_vartion_list", ArrayType(ArrayType(StringType()), True), True),
            StructField("title_len", IntegerType(), True),
            StructField("img_num", IntegerType(), True),
            StructField("img_type", StringType(), True),
            StructField("activity_type", StringType(), True),
            StructField("one_two_val", StringType(), True),
            StructField("three_four_val", StringType(), True),
            StructField("five_six_val", StringType(), True),
            StructField("eight_val", StringType(), True),
            StructField("node_id", StringType(), True),
            StructField("five_star", IntegerType(), True),
            StructField("four_star", IntegerType(), True),
            StructField("three_star", IntegerType(), True),
            StructField("two_star", IntegerType(), True),
            StructField("one_star", IntegerType(), True),
            StructField("low_star", IntegerType(), True),
            StructField("together_asin", StringType(), True),
            StructField("brand", StringType(), True),
            StructField("ac_name", StringType(), True),
            StructField("material", StringType(), True),
            StructField("data_type", IntegerType(), True),
            StructField("weight_str", StringType(), True),
            StructField("seller_id", StringType(), True),
            StructField("variat_num", IntegerType(), True),
            StructField("best_sellers_rank", StringType(), True),
            StructField("best_sellers_herf", StringType(), True),
            StructField("account_name", StringType(), True),
            StructField("parentAsin", StringType(), True),
            StructField("asinUpdateTime", StringType(), True),
            StructField("all_best_sellers_herf", StringType(), True),
            StructField("image_view", IntegerType(), True),
            StructField("product_description", StringType(), True),
            StructField("describe", StringType(), True),
            StructField("buy_sales", StringType(), True),
            StructField("lob_asin_json", StringType(), True),
            StructField("seller_json", StringType(), True),
            StructField("customer_reviews_json", StringType(), True),
            StructField("img_list", StringType(), True),
            StructField("follow_sellers", IntegerType(), True),
            StructField("fbm_delivery_price", FloatType(), True),
            StructField("product_json", StringType(), True),
            StructField("product_detail_json", StringType(), True),
            StructField("review_json_list", StringType(), True),
            StructField("amazon_label", StringType(), True),
            StructField("sub_title", StringType(), True)
        ])
        return schema

    # 覆写模板去重方法：第一步处理，在去重前统一完成——
    # 1) 处理跳转asin（判断规则见 resolve_asin_jump），确保跳转后同 asin 多条记录在去重时被合并
    # 2) 标题拼接+转小写：title+sub_title用' | '拼接后统一转小写
    def deduplication_kafka_data(self, kafka_df, deduplicaiton_key_field, deduplication_time_field):
        if deduplicaiton_key_field == "asin" and "current_asin" in kafka_df.columns:
            kafka_df = kafka_df.withColumn("asin", resolve_asin_jump())
        kafka_df = kafka_df.withColumn(
            "title", F.lower(F.concat_ws(
                " | ", F.col("title"),
                F.when(F.trim(F.col("sub_title")) == '', F.lit(None)).otherwise(F.trim(F.col("sub_title")))
            ))
        ).withColumn("title_len", F.length(F.col("title"))).drop("sub_title")
        return super().deduplication_kafka_data(kafka_df, deduplicaiton_key_field, deduplication_time_field)

    # 1. 处理asin分类及排名以及排名类型字段
    def handle_asin_bs_category_info(self, df):
        df = df.withColumnRenamed("parentAsin", "parent_asin")
        cate_current_pattern = self.pattern_current_dict[self.site_name]
        cate_1_pattern = self.pattern1_dict[self.site_name]
        df = df.withColumn("asin_bs_sellers_rank_lower", F.lower("best_sellers_rank"))
        df = df.withColumn("asin_bs", self.u_parse_bs_category(
            "asin_bs_sellers_rank_lower", "best_sellers_herf", "all_best_sellers_herf", F.lit(cate_current_pattern), F.lit(cate_1_pattern), "node_id"))
        # 新增：解析 best_sellers_rank 为 rank_and_category（&&&& 拼接），作为最终 Doris 字段值
        df = df.withColumn("rank_cate", self.u_rank_and_category(
            F.col("best_sellers_rank"),
            F.lit(self.pattern_str[self.site_name]),
            F.lit(self.pattern1_dict[self.site_name])
        ))
        df = df.withColumn("asin_bs_cate_1_id", df.asin_bs.getField("asin_bs_cate_1_id")) \
            .withColumn("asin_bs_cate_current_id", df.asin_bs.getField("asin_bs_cate_current_id")) \
            .withColumn("asin_bs_cate_1_rank", df.asin_bs.getField("asin_bs_cate_1_rank")) \
            .withColumn("asin_bs_cate_current_rank", df.asin_bs.getField("asin_bs_cate_current_rank")) \
            .drop("asin_bs", "asin_bs_sellers_rank_lower", "all_best_sellers_herf")
        # 用解析后的 rank_and_category 覆盖原 best_sellers_rank 原文
        df = df.drop("best_sellers_rank") \
            .withColumnRenamed("rank_cate", "best_sellers_rank")
        df = df.withColumn("rank_type", F.expr(""" 
            CASE WHEN asin_bs_cate_1_rank IS NOT NULL AND asin_bs_cate_1_rank BETWEEN 0 AND 1000 THEN 1 
            WHEN asin_bs_cate_1_rank BETWEEN 1000 AND 5000 THEN 2 WHEN asin_bs_cate_1_rank BETWEEN 5000 AND 10000 THEN 3 
            WHEN asin_bs_cate_1_rank BETWEEN 10000 AND 20000 THEN 4 WHEN asin_bs_cate_1_rank BETWEEN 20000 AND 30000 THEN 5 
            WHEN asin_bs_cate_1_rank BETWEEN 30000 AND 50000 THEN 6 WHEN asin_bs_cate_1_rank BETWEEN 50000 AND 70000 THEN 7 
            WHEN asin_bs_cate_1_rank >= 70000 THEN 8 ELSE 0 END"""))
        return df

    # 2. 利用node_id进行分类补充（此时无排名信息）及解析全路径分类
    def handle_asin_category_supplement(self, df):
        df = df.join(self.df_asin_new_cate, on=['node_id'], how='left')
        df = df.withColumn("asin_bs_cate_current_id", F.coalesce(F.col("asin_bs_cate_current_id"), F.col("node_id"))). \
            withColumn("asin_bs_cate_1_id", F.coalesce(F.col("asin_bs_cate_1_id"), F.col("category_first_id"))). \
            drop("category_first_id", "node_id")
        df = df.withColumn("desc_category_first_name", F.lower(F.trim(F.split(F.col("category"), "›").getItem(0))))
        df = df.join(self.df_asin_category, on=['desc_category_first_name'], how='left')
        df = df.withColumn("asin_bs_cate_1_id", F.coalesce(F.col("asin_bs_cate_1_id"), F.col("desc_category_first_id"))). \
            drop("desc_category_first_name")
        return df

    # 3. 处理bsr销量、价格类型字段以及BSR销售额信息
    def handle_asin_bsr_orders(self, df):
        df = df.withColumn("price", F.when(F.col("price") < 0, F.lit(None)).otherwise(F.col("price")))
        df = df.join(self.df_bs_report, on=['asin_bs_cate_1_id', 'asin_bs_cate_1_rank'], how='left')
        df = df.withColumn("price_type", F.expr("""
        CASE WHEN price IS NOT NULL AND price > 0 AND price < 10 THEN 1 WHEN price >= 10 AND price < 15 THEN 2 
        WHEN price >= 15 AND price < 20 THEN 3 WHEN price >= 20 AND price < 30 THEN 4 
        WHEN price >= 30 AND price < 50 THEN 5 WHEN price >= 50 THEN 6 ELSE 0 END""")).\
            withColumn("bsr_orders_sale", F.round(F.col("bsr_orders") * F.col("price"), 2))
        return df

    # 4.解析Make-It-A-Bundle信息
    def handle_asin_lob_info(self, df):
        df = df.withColumn("is_contains_lob_info",
                           F.when(F.col("lob_asin_json").isNotNull(), F.lit(1)).otherwise(F.lit(0)))
        df_parsed = df.withColumn("parse_asin_lob",
                                  F.when(F.col("is_contains_lob_info") == 1, F.from_json("lob_asin_json", "array<struct<lob_asin:string>>")))
        df_result = df_parsed.withColumn("asin_lob_info", F.expr("transform(parse_asin_lob, x -> x.lob_asin)"))
        df = df_result.withColumn(
            "asin_lob_info", F.regexp_replace(F.concat_ws(",", "asin_lob_info"), "[{}]", "")).drop(
            "parse_asin_lob", "lob_asin_json")
        return df

    # 5. 处理配送方式、卖家所在地以及卖家所在地类型
    def handle_asin_buy_box_seller_type(self, df):
        df = df.withColumn("seller_json_parsed", self.u_parse_seller_info(df.seller_json))
        df = df.withColumn("buy_box_seller_type", df.seller_json_parsed.buy_box_seller_type).withColumn(
            "account_name", df.seller_json_parsed.account_name).drop("seller_json_parsed")

        # 1. 关联全局缓存的df_asin_seller，用于填充seller_id为空的情况
        df = df.join(self.df_asin_seller, on=['asin'], how='left')

        # 2. 优先使用kafka消息中的seller_id，为空则使用df_asin_seller的数据
        df = df.withColumn(
            "seller_id", F.coalesce(F.col("seller_id"), F.col("fd_seller_id"))
        ).withColumn(
            "account_name", F.coalesce(F.col("account_name"), F.col("fd_account_name"))
        ).drop("fd_seller_id", "fd_account_name")

        # 3. 关联全局缓存的df_seller_country获取seller_country_name
        df = df.join(self.df_seller_country, on=['seller_id'], how='left')

        df = df.withColumn("site_name_type", F.expr("""
        CASE WHEN buy_box_seller_type = 1 THEN 4
        WHEN buy_box_seller_type != 1 AND seller_country_name is not null AND seller_country_name like '%US%' THEN 1
        WHEN buy_box_seller_type != 1 AND seller_country_name is not null AND seller_country_name like '%CN%' THEN 2
        WHEN buy_box_seller_type != 1 AND seller_country_name is not null AND seller_country_name like '%HK%' THEN 5
        WHEN buy_box_seller_type != 1 AND seller_country_name is not null AND seller_country_name like '%TW%' THEN 6
        ELSE 3 END"""))
        return df

    # 6. 处理asin基础属性信息(长宽高重量等)
    def handle_asin_basic_attribute_info(self, df):
        # 1.解析ASIN重量相关信息
        df = df.withColumn("weight_str", F.lower(F.col("weight_str"))).withColumn("asin_weight", self.u_parse_weight("weight_str", F.lit(self.site_name)))
        df = df.withColumnRenamed("weight_str", "asin_weight_str")
        df = df.withColumn(
            "weight", F.when(df.asin_weight.getField("weight_type") == 'pounds', df.asin_weight.getField("weight")).otherwise(F.lit(0))).drop("asin_weight")
        # 2.处理重量类型
        df = df.withColumn("weight_type", F.expr("""
        CASE WHEN weight BETWEEN 0 AND 0.2 THEN 1 WHEN weight BETWEEN 0.2 AND 0.4 THEN 2 
        WHEN weight BETWEEN 0.4 AND 0.6 THEN 3 WHEN weight BETWEEN 0.6 AND 1 THEN 4 
        WHEN weight BETWEEN 1 AND 2 THEN 5 WHEN weight >= 2 THEN 6 ELSE 0 END"""))
        # 3.解析ASIN体积相关信息
        df = df.withColumn("asin_volume", self.u_parse_volume("volume"))
        df = df.withColumn("asin_volume_type", df.asin_volume.getField("asin_volume_type")) \
            .withColumn("asin_length", F.when(F.col("asin_volume_type") == 'inches', df.asin_volume.getField("length"))) \
            .withColumn("asin_width", F.when(F.col("asin_volume_type") == 'inches', df.asin_volume.getField("width"))) \
            .withColumn("asin_height", F.when(F.col("asin_volume_type") == 'inches', df.asin_volume.getField("height"))) \
            .drop("asin_volume", "asin_volume_type")
        # 4.获取体积重/毛重相关信息
        df = df.withColumn(
            "asin_weight_ratio", F.when(
                F.col("asin_length").isNotNull() & (F.col("asin_width").isNotNull()) &
                (F.col("asin_height").isNotNull()) &
                (F.col("weight") > 0), F.round(F.col("asin_length") * F.col("asin_width") * F.col("asin_height") * 3.2774128 / (F.col("weight") * 453.59), 4))
            .otherwise(F.lit(-1)))
        # 5.处理尺寸类型
        if self.site_name == 'us':
            expr_str = f"""
            CASE WHEN weight > 0 AND weight * 16 <= 16 AND asin_length > 0 AND asin_length <= 15 AND asin_width > 0 AND asin_width <= 12 AND asin_height > 0 AND asin_height <= 0.75 THEN 1 
            WHEN weight > 0 AND weight <= 20 AND asin_length > 0 AND asin_length <= 18 AND  asin_width > 0 AND asin_width <= 14 AND asin_height > 0 AND asin_height <= 8 THEN 2 
            WHEN weight > 0 AND weight <= 70 AND asin_length > 0 AND asin_length <= 60 AND asin_width > 0 AND asin_width <= 30 AND asin_length + asin_length + (asin_width + asin_height) * 2 <= 130  THEN 3 
            WHEN weight > 0 AND weight <= 150 AND asin_length > 0 AND asin_length <= 108 AND asin_length + asin_length + (asin_width + asin_height) * 2 <= 130  THEN 4 
            WHEN weight > 0 AND weight <= 150 AND asin_length > 0 AND asin_length <= 108 AND asin_length + asin_length + (asin_width + asin_height) * 2 <= 165  THEN 5 
            WHEN weight > 150 AND asin_length > 108 AND asin_length + asin_length + (asin_width + asin_height) * 2 > 165  THEN 6 ELSE 0 END"""
        else:
            expr_str = f"""
            CASE WHEN weight > 0 AND weight <= 100 AND asin_length > 0 AND asin_length <= 20 AND asin_width > 0 AND asin_width <= 15 AND asin_height > 0 AND asin_height <= 1 THEN 1 
            WHEN weight > 0 AND weight <= 500 AND asin_length > 0 AND asin_length <= 33 AND asin_width > 0 AND asin_width <= 23 AND asin_height > 0 AND asin_height <= 2.5 THEN 2 
            WHEN weight > 0 AND weight <= 1000 AND asin_length > 0 AND asin_length <= 33 AND asin_width > 0 AND asin_width <= 23 AND asin_height > 0 AND asin_height <= 5 THEN 3 
            WHEN weight > 0 AND weight <= 12000 AND asin_length > 0 AND asin_length <= 45 AND asin_width > 0 AND asin_width <= 34 AND asin_height > 0 AND asin_height <= 26 THEN 4 
            WHEN weight > 0 AND weight <= 2000 AND asin_length > 0 AND asin_length <= 61 AND asin_width > 0 AND asin_width <= 46 AND asin_height > 0 AND asin_height <= 46 THEN 5 
            WHEN asin_length > 0 AND asin_length <= 150 AND asin_length + asin_length + (asin_width + asin_height) <= 300 THEN 6 
            WHEN asin_length > 150 AND asin_length + asin_length + (asin_width + asin_height) > 300 THEN 7 ELSE 0 END"""
        df = df.withColumn("size_type", F.expr(expr_str)).drop("asin_length", "asin_width", "asin_height")
        # 6.处理五点描述长度
        df = df.withColumn("describe_len", F.length(F.regexp_replace(F.col("describe"), "\\|-\\|", "")))
        return df

    # 7. 处理asin图片信息
    def handle_asin_img_info(self, df):
        img_schema = ArrayType(ArrayType(StringType()))
        df = df.withColumn("img_list", F.from_json(F.col("img_list"), img_schema))
        df_with_img = df.filter(F.size("img_list") > 0).select("asin", "img_list")
        df_with_img_attribute = df_with_img.select(
            "asin", F.explode("img_list").alias("img_attributes")
        ).select(
            "asin", F.col("img_attributes")[1].alias("img_url"), F.col("img_attributes")[2].alias("img_order_by"),
            F.col("img_attributes")[3].alias("data_type")
        )
        df_with_img_attribute_agg = df_with_img_attribute.groupby("asin").agg(
            F.to_json(F.collect_list(F.struct(F.col("img_url"), F.col("img_order_by"), F.col("data_type")))).alias(
                "img_info")
        )
        df = df.drop("img_list")
        df = df.join(df_with_img_attribute_agg, on=['asin'], how='left')
        df = df.withColumn("img_type", F.split(F.col("img_type"), ",")) \
               .withColumn("img_type", F.expr("transform(img_type, x -> cast(x as int))"))
        return df

    # 8. 处理变体相关（ao及母体相关，自然占比及母体自然占比，各类型数量，月销信息等）
    def handle_asin_measure(self, df):
        df = CommonUtil.get_asin_variant_attribute(df_asin_detail=df, df_asin_measure=self.df_asin_measure,
                                                   partition_num=self.repartition_num, use_type=1)
        # 是否数量变体类型和ao的类型
        df = df.withColumn("quantity_variation_type", F.expr("""
        CASE WHEN size is not null and size != '' and lower(size) like '%quantity%' THEN 1 ELSE 0 END""")).withColumn(
            "ao_val_type", F.expr("""
            CASE WHEN asin_ao_val BETWEEN 0 AND 0.1 THEN 1 WHEN asin_ao_val BETWEEN 0.1 AND 0.2 THEN 2 
            WHEN asin_ao_val BETWEEN 0.2 AND 0.4 THEN 3 WHEN asin_ao_val BETWEEN 0.4 AND 0.8 THEN 4 
            WHEN asin_ao_val BETWEEN 0.8 AND 1.2 THEN 5 WHEN asin_ao_val BETWEEN 1.2 AND 2 THEN 6 
            WHEN asin_ao_val >= 2 THEN 7 ELSE 0 END"""))
        df = df.withColumnRenamed("asin_zr_counts", "zr_counts").withColumnRenamed("asin_ao_val", "ao_val") \
            .withColumnRenamed("asin_zr_flow_proportion", "zr_flow_proportion") \
            .withColumnRenamed("asin_amazon_orders", "asin_bought_month") \
            .drop("asin_st_counts", "asin_adv_counts")
        # 获取parent_asin下最新ASIN信息，导出到 doris 父ASIN最新详情表（仅 latest+normal 模式）
        if self.test_flag == 'normal':
            df_parent_asin_info = df.filter("parent_asin is not null").select("parent_asin", "asin_vartion_list", "asinUpdateTime", "matrix_flow_proportion", "matrix_ao_val")
            parent_asin_window = Window.partitionBy(['parent_asin']).orderBy(F.desc_nulls_last("asinUpdateTime"))
            df_parent_asin_info = df_parent_asin_info.withColumn("u_rank", F.row_number().over(window=parent_asin_window))
            df_parent_asin_info = df_parent_asin_info.repartition(self.repartition_num)
            df_parent_asin_info = df_parent_asin_info.filter("u_rank = 1").drop("u_rank")
            df_asin_variat = df_parent_asin_info.filter(F.size("asin_vartion_list") > 0).\
                select("parent_asin", "asinUpdateTime", F.explode("asin_vartion_list").alias("variant_attribute")).\
                select("parent_asin", F.col("asinUpdateTime").alias("asin_crawl_date"),
                       F.col("variant_attribute")[0].alias("asin"), F.col("variant_attribute")[1].alias("color"),
                       F.col("variant_attribute")[3].alias("size"), F.col("variant_attribute")[5].alias("style"))
            df_asin_variat_agg = df_asin_variat.groupby(['parent_asin']).agg(
                F.first("asin_crawl_date").alias("asin_crawl_date"),
                F.concat_ws(',', F.collect_list("asin")).alias("variation_info"),
                F.to_json(F.collect_list(F.struct(F.col("color"), F.col("size"), F.col("style")))).alias("attr_info")
            )
            # 关联母体 AO 值和自然流量占比（取 parent_asin 维度最新记录的值）
            df_parent_matrix = df_parent_asin_info.select("parent_asin", "matrix_flow_proportion", "matrix_ao_val")
            df_asin_variat_agg = df_asin_variat_agg.join(df_parent_matrix, on="parent_asin", how="left")
            print("导出父ASIN最新变体信息到doris：")
            df_doris = df_asin_variat_agg.select(
                "parent_asin",
                F.lit(self.date_info_month).alias("date_info"),
                # Doris 新表 asin_crawl_date 是 DATETIME，需 string → timestamp
                F.to_timestamp(F.col("asin_crawl_date")).alias("asin_crawl_date"),
                "variation_info", "attr_info",
                F.current_timestamp().alias("updated_at"),
                F.round(F.col("matrix_flow_proportion"), 4).alias("matrix_flow_proportion"),
                F.round(F.col("matrix_ao_val"), 4).alias("matrix_ao_val"))
            table_columns = "parent_asin, date_info, asin_crawl_date, variation_info, attr_info, updated_at, matrix_flow_proportion, matrix_ao_val"
            DorisHelper.spark_export_with_columns(df_save=df_doris, db_name=self.doris_db_selection, table_name=self.parent_asin_latest_detail_table, table_columns=table_columns)
        return df

    # 9. 提取打包数量字段
    def handle_asin_package_quantity(self, df):
        df = df.withColumn("material",
            F.when(F.lower(F.col("material")) == 'nan', F.lit(None)).otherwise(F.col("material"))
        )
        df = df.withColumn(
            "variat_attribute", F.concat_ws("&&&%", F.col("color"), F.col("style"), F.col("size"), F.col("material")))
        df = df.withColumn("title_parse", self.u_parse_package_quantity(df.title)).withColumn(
            "variat_parse", self.u_parse_package_quantity(df.variat_attribute))
        df = df.withColumn("title_package_quantity", df.title_parse.getField("parse_package_quantity")). \
            withColumn("variat_package_quantity", df.variat_parse.getField("parse_package_quantity")). \
            withColumn("title_package_quantity_is_abnormal", df.title_parse.getField("is_package_quantity_abnormal")). \
            withColumn("variat_package_quantity_is_abnormal", df.variat_parse.getField("is_package_quantity_abnormal")). \
            drop("title_parse", "variat_parse", "variat_attribute")
        # 从 product_json 额外提取 Color 字段作为颜色来源备用
        df = df.withColumn(
            "_product_json_color",
            F.lower(F.get_json_object(F.col("product_json"), "$.Color"))
        )
        # Number of Items：从 product_json 提取，cast 失败（脏数据）自动为 null，提取后立即 drop
        df = df.withColumn(
            "number_of_items",
            F.get_json_object(F.col("product_json"), "$.Number of Items").cast("int")
        )  # product_json 信息库PG导出需要用到，不再drop
        # 优先级：Number of Items > 属性字段 > 标题解析 > 默认1
        df = df.withColumn(
            "package_quantity", F.expr("""
            CASE WHEN number_of_items IS NOT NULL AND number_of_items > 0 THEN number_of_items
                 WHEN variat_package_quantity IS NOT NULL THEN variat_package_quantity
                 WHEN title_package_quantity IS NOT NULL THEN title_package_quantity
                 ELSE 1 END""")
        ).withColumn(
            "is_package_quantity_abnormal", F.expr("""
            CASE WHEN number_of_items IS NOT NULL AND number_of_items > 0 THEN 0
                 WHEN variat_package_quantity IS NOT NULL THEN variat_package_quantity_is_abnormal
                 WHEN title_package_quantity IS NOT NULL THEN title_package_quantity_is_abnormal
                 ELSE 2 END""")
        ).drop("number_of_items", "title_package_quantity", "variat_package_quantity",
               "title_package_quantity_is_abnormal", "variat_package_quantity_is_abnormal")
        # color 优先使用变体属性颜色，fallback 到 product_json 中的 Color 字段
        df = df.withColumn("color", F.coalesce(F.col("color"), F.col("_product_json_color"))).drop("_product_json_color")
        return df

    # 10. 处理品牌标签、是否告警品牌、处理asin_lqs_rating信息
    def handle_asin_lqs_and_brand(self, df):
        # 1.品牌标签以及是否告警品牌
        df = df.withColumn("is_brand_label", F.expr("""CASE WHEN brand is not null THEN 1 ELSE 0 END"""))
        df = df.withColumn("brand", F.lower("brand"))
        # 2. lqs评分
        df = df.withColumn("category_node_rating",
                           F.expr(f"""CASE WHEN asin_bs_cate_current_id is not null THEN 1 ELSE 0 END""")) \
            .withColumn("zr_rating", F.expr(f"""CASE WHEN zr_counts > 0 THEN 0.5 ELSE 0 END""")) \
            .withColumn("sp_rating", F.expr(f"""CASE WHEN sp_counts > 0 THEN 1 ELSE 0 END""")) \
            .withColumn("a_add_rating", F.expr(f"""CASE WHEN array_contains(img_type, 3) THEN 1 ELSE 0 END""")) \
            .withColumn("video_rating", F.expr(f"""CASE WHEN array_contains(img_type, 2) THEN 0.5 ELSE 0 END""")) \
            .withColumn("brand_rating", F.expr(f"""CASE WHEN is_brand_label = 1 THEN 0.2 ELSE 0 END""")) \
            .withColumn("product_describe_rating",
                        F.expr(f"""CASE WHEN product_description is not null THEN 0.2 ELSE 0 END""")) \
            .withColumn("highlight_rating", F.expr(f"""
            CASE WHEN describe is not null AND size(split(describe, '\\|-\\|')) <= 4 THEN size(split(describe, '\\|-\\|')) * 0.4 
            WHEN describe is not null AND size(split(describe, '\\|-\\|')) > 4 THEN 1.6 ELSE 0 END""")) \
            .withColumn("title_len_rating", F.expr(f"""CASE WHEN title_len >= 50 AND title_len <=200 THEN 0.5 ELSE 0 END""")) \
            .withColumn("title_brand_rating", F.expr(f"""
            CASE WHEN brand is not null AND lower(regexp_replace(title, '[^a-zA-Z0-9\\s]', '')) LIKE lower(regexp_replace(brand, '[^a-zA-Z0-9\\s]', '')) || '%' THEN 0.5 
            ELSE 0 END""")) \
            .withColumn("img_num_rating", F.expr(f"""
            CASE WHEN img_num <= 4 THEN img_num * 0.5 WHEN img_num >4 THEN 2 ELSE 0 END""")) \
            .withColumn("img_enlarge_rating", F.expr(f"""CASE WHEN image_view = 1 THEN 0.5 ELSE 0 END"""))
        df = df.withColumn(
            "asin_lqs_rating",
            (F.col("category_node_rating") + F.col("zr_rating") + F.col("sp_rating") + F.col("a_add_rating") +
             F.col("video_rating") + F.col("brand_rating") + F.col("product_describe_rating") +
             F.col("highlight_rating") + F.col("title_len_rating") + F.col("title_brand_rating") +
             F.col("img_num_rating") + F.col("img_enlarge_rating")).cast("double")).withColumn(
            "asin_lqs_rating_detail", F.to_json(
                F.struct(F.col("category_node_rating"), F.col("zr_rating"), F.col("sp_rating"), F.col("a_add_rating"),
                         F.col("video_rating"), F.col("brand_rating"), F.col("product_describe_rating"),
                         F.col("highlight_rating"), F.col("title_len_rating"), F.col("title_brand_rating"),
                         F.col("img_num_rating"), F.col("img_enlarge_rating")))
        )
        df = df.drop("product_description", "image_view", "category_node_rating", "zr_rating", "sp_rating",
                     "a_add_rating", "video_rating", "brand_rating", "product_describe_rating", "highlight_rating",
                     "title_len_rating", "title_brand_rating", "img_num_rating", "img_enlarge_rating")
        return df

    def handle_multi_color_flag(self, df):
        """判断 ASIN 是否为颜色组合产品
        multi_color_flag:
          0 = 非多色
          1 = 从颜色变体属性字段（color）解析为多色
          2 = 从标题或五点描述解析为多色（降级 fallback）
        颜色变体属性三层判断：
          1. 命中颜色组合关键词
          2. 含分隔符 (/ + & ; , and) 或 "数字 color"
          3. 含 2 个及以上颜色表中的单色词
        """
        # ── 第1层：关键词正则（支持 color/colors/colour/colours）──
        KEYWORD_PATTERN = (
            r"(?i)("
            r"\bmulticolou?rs?\b|\bmulti[\s\-]colou?rs?\b|\bmulti[\s\-]colored\b|\bmulticolored\b|"
            r"\bassorted\b|\bmorandi\b|\bpastel\b|\bvibrant\b|\bvintage\b|\bboho\b|\bgradient\b|"
            r"\bcandy\b|\bdusty\b|\bfluorescent\b|\bgentle\b|\bneutral\b|\bsoft\b|\bmuted\b|"
            r"\brainbow\b|\bmaillard\b|\bcolorful\b|\bcolourful\b|\bmulti\b|"
            r"\baesthetic colou?rs?\b|\bdreamy colou?rs?\b|\bearthy colou?rs?\b|\bshades of\b|"
            r"\bvarious colou?rs?\b|\bsolid colou?rs?\b|\bmix colou?rs?\b|\bmixed colou?rs?\b|"
            r"\bbasic colou?rs?\b|\blightcolor\b|\bdarkcolor\b|\battractive colou?rs?\b|"
            r"\bmultiple colou?rs?\b|\bbright colorful\b|\bdifferent colou?rs?\b|\bclassic colou?rs?\b|"
            r"\bfriendly colou?rs?\b|\bwarm colou?rs?\b|\bfun colou?rs?\b|\bmetallic colou?rs?\b|"
            r"\bbright colou?rs?\b|\bdark colou?rs?\b|\blight colou?rs?\b|"
            r"\bautumn colou?rs?\b|\bsummer colou?rs?\b|\bwinter colou?rs?\b|\bspring colou?rs?\b"
            r")"
        )
        # ── 第2层：分隔符正则（/ + & ; 逗号 and 数字+color）──
        SEPARATOR_PATTERN = r"(?i)[/+&；;，,]|\band\b|\d+\s*colou?rs?"

        # ── 第3层：颜色表多色词检测 UDF ──
        single_colors = sorted(
            [c for c in self.color_set
             if c.strip() and not re.search(r'[/+&;；,，]|\band\b', c)],
            key=len, reverse=True
        )
        if single_colors:
            color_regex = re.compile(
                r'(?i)\b(' + '|'.join(re.escape(c) for c in single_colors) + r')\b'
            )
        else:
            color_regex = None

        def _get_matched_colors(color_str):
            if not color_str or color_regex is None:
                return None
            matched = sorted({m.group(1).lower() for m in color_regex.finditer(color_str)})
            return "/".join(matched) if len(matched) >= 2 else None

        udf_matched_colors = F.udf(_get_matched_colors, StringType())

        # ── 降级：标题 + 五点描述关键词正则 ──
        FALLBACK_PATTERN = (
            r"(?i)("
            r"\bmulticolou?rs?\b|\bmulti[\s\-]colou?rs?\b|\bmulti[\s\-]colored\b|\bmulticolored\b|"
            r"\bassorted colou?rs?\b|\bfluorescent colou?rs?\b|\bdifferent colou?rs?\b|"
            r"\bbright colou?rs?\b|\bcolorful\b|\bcolourful\b|\battractive colou?rs?\b|"
            r"\bvibrant colou?rs?\b|\d+\s*colou?rs?"
            r")"
        )

        # 提前计算 UDF 结果，避免 multi_color_flag / multi_color_str 各调用一次（性能优化）
        df = df.withColumn("_matched_colors_str", udf_matched_colors(F.col("color")))
        df = df.withColumn(
            "multi_color_flag",
            F.when(
                F.col("color").isNotNull() & F.col("color").rlike(KEYWORD_PATTERN), 1
            ).when(
                F.col("color").isNotNull() & F.col("color").rlike(SEPARATOR_PATTERN), 1
            ).when(
                F.col("color").isNotNull() & F.col("_matched_colors_str").isNotNull(), 1
            ).when(
                F.lower(F.concat_ws(" ", F.col("title"), F.col("describe"))).rlike(FALLBACK_PATTERN), 2
            ).otherwise(0)
        ).withColumn(
            "multi_color_str",
            F.when(
                F.col("color").isNotNull() & F.col("color").rlike(KEYWORD_PATTERN),
                F.regexp_extract(F.col("color"), KEYWORD_PATTERN, 1)
            ).when(
                F.col("color").isNotNull() & F.col("color").rlike(SEPARATOR_PATTERN),
                F.col("color")
            ).when(
                F.col("color").isNotNull() & F.col("_matched_colors_str").isNotNull(),
                F.col("_matched_colors_str")
            ).when(
                F.lower(F.concat_ws(" ", F.col("title"), F.col("describe"))).rlike(FALLBACK_PATTERN),
                F.regexp_extract(
                    F.lower(F.concat_ws(" ", F.col("title"), F.col("describe"))),
                    FALLBACK_PATTERN, 1
                )
            ).otherwise(F.lit(None))
        ).drop("_matched_colors_str")
        return df

    # 11. 通过ASIN页面信息处理(评分类型、电影标签)
    def handle_asin_detail_all_type(self, df):
        # 1. 评分类型
        df = df.withColumn("rating_type", F.expr("""
        CASE WHEN rating >= 4.5 THEN 1 WHEN rating >= 4 AND rating < 4.5 THEN 2 WHEN rating >= 3.5 AND rating < 4 THEN 3
        WHEN rating >= 3 AND rating < 3.5 THEN 4 WHEN rating < 3 AND rating >= 0 THEN 5 ELSE 0 END"""))
        # 2. 电影标签
        movie_label_list = ['prime video', 'dvd', 'blu-ray', 'kindle', 'app', 'paperback', 'audible audiobook',
                            'kindle edition', 'kindle & comixology', 'hardcover', 'comic', 'multi-format', '4k',
                            'library binding', 'vinyl', 'audio cd', 'mp3 music', 'single issue magazine',
                            'print magazine', 'unknown binding']
        df = df.join(self.df_asin_label_info, on=['asin'], how='left')
        condition = reduce(
            lambda acc, keyword: acc | F.expr(f"exists(asin_label_list, x -> x like '%{keyword}%')"),
            movie_label_list,
            F.lit(False)
        )
        df = df.withColumn("is_movie_label", condition.cast("int"))
        # 3. amazon 新品标签: badge_list 任意元素匹配 %new on amazon% → 1, 否则 0
        # otherwise(0) 已覆盖 badge_list 为 NULL 的情况(exists(NULL,...)=NULL → 走 otherwise)
        df = df.withColumn(
            "is_amazon_new", F.when(F.expr("exists(badge_list, x -> x like '%new on amazon%')"), F.lit(1)).otherwise(F.lit(0))
        )
        df = df.drop("asin_label_list", "badge_list")
        # 4. asin_type（对齐 dwt.handle_asin_is_hide：内部asin=1 / 数字虚拟类目=2 / 隐藏分类=3 / 默认0）
        # 内部asin判断两步：asin级(ods_self_asin) 或 品牌级(amazon_brand,brand_type=1) 命中任一即可
        df = df.join(self.df_self_asin, on=['asin'], how='left')
        df = df.join(self.df_self_brand, on=['brand'], how='left')
        df = df.join(self.df_hide_category, on=['asin_bs_cate_current_id'], how='left')
        df = df.withColumn(
            "asin_is_need",
            F.when(
                F.col("asin_bs_cate_1_id").isin(*NEED_FILTER_CATEGORIES) &
                F.col("desc_category_first_id").isin(*NEED_FILTER_CATEGORIES),
                F.lit(1)
            ).when(
                ~F.col("asin").like("B0%"), F.lit(1)
            ).otherwise(F.lit(0))
        )
        df = df.withColumn(
            "asin_type",
            F.when((F.col("asin_is_self") == 1) | (F.col("asin_is_self_brand") == 1), F.lit(1))
             .when(F.col("asin_is_need") == 1, F.lit(2))
             .when(F.col("hide_flag") == 1, F.lit(3))
             .otherwise(F.lit(0)).cast("tinyint")
        ).drop("asin_is_self", "asin_is_self_brand", "asin_is_need", "hide_flag")
        return df

    # 12. 处理变化率相关字段（环比_mom / 同比_yoy 统一处理）
    def handle_asin_attribute_change(self, df):
        df = df.join(self.df_previous_flow_asin, on=['asin'], how='left')
        df = df.join(self.df_previous_flow_asin_lastyear, on=['asin'], how='left')
        # (current_col, prev_col, lastyear_col, suffix, rise_round)
        # rise_round: None=不计算_rise; 整数=round精度; 'int'=IntegerType
        # 哨兵语义（全字段统一）：
        #   both null  → null    （新ASIN，双边均无历史，无法比较）
        #   current null → -1000 （本期无数据，上期有数据，ASIN消失/下架）
        #   previous null/0 → 1000 （上期无数据或为0，本期新出现）
        columns_to_change = [
            ("ao_val",            "previous_asin_ao_val",         "lastyear_asin_ao_val",         "ao",          4),
            ("price",             "previous_asin_price",          "lastyear_asin_price",          "price",       2),
            ("asin_bs_cate_1_rank","previous_first_category_rank","lastyear_first_category_rank", "rank",        "int"),
            ("bsr_orders",        "previous_asin_bsr_orders",     "lastyear_asin_bsr_orders",     "bsr_orders",  "int"),
            ("rating",            "previous_asin_rating",         "lastyear_asin_rating",         "rating",      1),
            ("total_comments",    "previous_asin_total_comments", "lastyear_asin_total_comments", "comments",    "int"),
            ("variat_num",        "previous_asin_variation_num",  "lastyear_asin_variation_num",  "variation",   "int"),
            ("bsr_orders_sale",   "previous_sales",               "lastyear_sales",               "sales",       2),
            ("asin_bought_month", "previous_asin_bought_month",   "lastyear_asin_bought_month",   "bought_month", None),
        ]

        def calculate_change(current_col, previous_col):
            curr = F.when(F.col(current_col) < 0, F.lit(None)).otherwise(F.col(current_col))
            prev = F.when(F.col(previous_col) < 0, F.lit(None)).otherwise(F.col(previous_col))
            rise_col = curr - prev
            change_col = F.when(
                curr.isNull() & prev.isNull(), F.lit(None)
            ).when(
                curr.isNull() & (prev == 0), F.lit(0.0)
            ).when(
                curr.isNull(), F.lit(-1000.0)
            ).when(
                (prev.isNull() | (prev == 0)) & (curr == 0), F.lit(0.0)
            ).when(
                prev.isNull() | (prev == 0), F.lit(1000.0)
            ).otherwise(
                F.round((curr - prev) / prev, 4)
            )
            return rise_col, change_col

        for current_col, prev_col, lastyear_col, suffix, rise_round in columns_to_change:
            rise_col, mom_col = calculate_change(current_col, prev_col)
            _, yoy_col = calculate_change(current_col, lastyear_col)
            if rise_round is not None:
                if rise_round == "int":
                    df = df.withColumn(f"{suffix}_rise", rise_col.cast(IntegerType()))
                else:
                    df = df.withColumn(f"{suffix}_rise", F.round(rise_col, rise_round))
            df = df.withColumn(f"{suffix}_mom", F.round(mom_col, 4)) \
                   .withColumn(f"{suffix}_yoy", F.round(yoy_col, 4))
        df = df.drop('previous_asin_bought_month', 'lastyear_asin_bought_month')
        return df

    # 13. 字段标准化
    def handle_column_name(self, df):
        df = df.withColumnRenamed("asin_bs_cate_1_id", "category_first_id")\
            .withColumnRenamed("asin_bs_cate_current_id", "category_id") \
            .withColumnRenamed("asin_bs_cate_1_rank", "first_category_rank")\
            .withColumnRenamed("asin_bs_cate_current_rank", "current_category_rank") \
            .withColumnRenamed("variat_num", "variation_num")\
            .withColumnRenamed("seller_id", "account_id").withColumnRenamed("seller_country_name", "site_name") \
            .withColumnRenamed("asinUpdateTime", "asin_crawl_date")\
            .withColumnRenamed("customer_reviews_json", "product_features")\
            .withColumn("asin_crawl_date", F.coalesce(
                F.to_timestamp(F.col("asin_crawl_date")),
                F.current_timestamp()
            )) \
            .withColumn("launch_time", F.to_timestamp(F.col("launch_time"))) \
            .withColumn("rating", F.round(F.col("rating"), 1)) \
            .withColumn("collapse_asin", F.coalesce(F.col("parent_asin"), F.col("asin")))\
            .withColumn("amazon_label", F.when(
                F.coalesce(F.get_json_object(F.col("amazon_label"), "$[0].badge_type"),
                           F.get_json_object(F.col("amazon_label"), "$.badge_type")) != "unknown",
                F.coalesce(F.get_json_object(F.col("amazon_label"), "$[0].badge_type"),
                           F.get_json_object(F.col("amazon_label"), "$.badge_type"))
            ))
        df_save = df.select(
            # ── 主键/标识 ──────────────────────────────────────────────────
            "asin",             # Kafka，ASIN 唯一标识
            "parent_asin",      # Kafka，父 ASIN
            "collapse_asin",    # 派生：coalesce(parent_asin, asin)，折叠用 ASIN
            "asin_crawl_date",  # Kafka asinUpdateTime，本次爬取时间
            # ── 标题/描述 ──────────────────────────────────────────────────
            "title",            # Kafka，商品标题（已转小写）
            "title_len",        # Kafka，标题字符数
            "brand",            # Kafka，品牌（已转小写）
            F.col("describe").alias("asin_describe"),  # Kafka describe，五点描述
            "describe_len",     # 派生：五点描述去分隔符后长度
            "product_features", # Kafka customer_reviews_json，买家评测特性
            "together_asin",    # Kafka，组合购买推荐 ASIN
            # ── 价格 ───────────────────────────────────────────────────────
            F.round(F.col("price").cast("double"), 2).alias("price"),  # Kafka，商品售价
            F.round(F.col("fbm_delivery_price").cast("double"), 2).alias("fbm_price"),  # Kafka，FBM 配送价
            # ── 评分/评论 ──────────────────────────────────────────────────
            "rating",           # Kafka，综合评分
            "total_comments",   # Kafka，总评论数
            "one_star", "two_star", "three_star", "four_star", "five_star", "low_star",  # Kafka，各星级分布
            # ── 销量/BSR ───────────────────────────────────────────────────
            "bsr_orders",       # df_bs_report，BSR 排名对应月销量估算
            F.round(F.col("bsr_orders_sale").cast("double"), 2).alias("bsr_orders_sale"),  # 派生：bsr_orders × price，BSR 销售额估算
            "asin_bought_month",  # df_asin_measure，页面月购买量
            # ── 流量/占比 ──────────────────────────────────────────────────
            "ao_val",           # df_asin_measure，ASIN 自然流量占比值
            "zr_counts",        # df_asin_measure，自然排名位置数
            "sp_counts",        # df_asin_measure，SP 广告位置数
            "sb_counts",        # df_asin_measure，SB 广告位置数
            "vi_counts",        # df_asin_measure，视频广告位置数
            "bs_counts",        # df_asin_measure，BS 版位数
            "ac_counts",        # df_asin_measure，AC 标数
            "tr_counts",        # df_asin_measure，趋势标数
            "er_counts",        # df_asin_measure，编辑推荐位数
            "zr_flow_proportion",      # df_asin_measure，自然搜索流量占比
            F.round(F.col("matrix_flow_proportion"), 4).alias("matrix_flow_proportion"),  # get_asin_variant_attribute，母体自然流量占比
            F.round(F.col("matrix_ao_val"), 4).alias("matrix_ao_val"),                 # get_asin_variant_attribute，母体 ao_val
            F.round(F.col("one_two_val").cast("double"),    4).cast("decimal(20,4)").alias("one_two_val"),    # Kafka，1-2 位流量占比
            F.round(F.col("three_four_val").cast("double"), 4).cast("decimal(20,4)").alias("three_four_val"), # Kafka，3-4 位流量占比
            F.round(F.col("five_six_val").cast("double"),   4).cast("decimal(20,4)").alias("five_six_val"),   # Kafka，5-6 位流量占比
            F.round(F.col("eight_val").cast("double"),      4).cast("decimal(20,4)").alias("eight_val"),      # Kafka，8 位以上流量占比
            # ── 分类 ───────────────────────────────────────────────────────
            "category_first_id",      # BSR 解析 + node_id 补充，一级分类 ID
            "category_id",            # BSR 解析 + node_id 补充，当前分类 ID
            "desc_category_first_id", # category 文本截取后匹配得到的一级分类 ID
            "first_category_rank",    # BSR 解析，一级分类排名
            "current_category_rank",  # BSR 解析，当前分类排名
            # ── 商品属性 ───────────────────────────────────────────────────
            F.round(F.col("weight").cast("double"), 4).alias("weight"),  # 派生：weight_str 解析（磅）
            "volume",            # Kafka，体积原始字符串
            F.round(F.col("asin_weight_ratio").cast("double"), 4).cast("decimal(20,4)").alias("asin_weight_ratio"), # 派生：体积重 / 实重比率，cast decimal与月流程一致，超限/NaN自动置空
            "color",             # 变体属性 + product_json Color 补充
            "size",              # 变体属性
            "style",             # 变体属性
            "material",          # Kafka，材质
            "package_quantity",             # 派生：包装数量（Number of Items / 属性 / 标题解析）
            "is_package_quantity_abnormal", # 派生：包装数量异常（0正常/1异常/2默认）
            "variation_num",     # Kafka variat_num，变体总数
            "page_inventory",    # Kafka，页面展示库存量
            "activity_type",     # Kafka，促销活动类型
            "launch_time",       # Kafka，上架时间
            # ── 图片 ───────────────────────────────────────────────────────
            "img_url",   # Kafka，主图 URL
            "img_num",   # Kafka，图片数量
            "img_type",  # Kafka → 处理为 array<int>，图片类型列表
            "img_info",  # 派生：图片详情 JSON（url/order/type）
            # ── 卖家/店铺 ──────────────────────────────────────────────────
            "account_id",           # Kafka seller_id + df_asin_seller 补充，卖家 ID
            "account_name",         # UDF + df_asin_seller 补充，卖家名称
            "buy_box_seller_type",  # UDF，配送类型（1.自营/2.FBA/3.FBM/4.其他/0.未知）
            "site_name",            # df_seller_country，卖家所在地
            "asin_type",            # 派生：handle_asin_detail_all_type 算好(0默认/1内部/2书籍影视/3隐藏)
            F.col("follow_sellers").alias("follow_sellers_count"),  # Kafka，关注该 ASIN 的卖家数
            # ── LOB（Make-It-A-Bundle）────────────────────────────────────
            "asin_lob_info",        # 派生：捆绑销售关联 ASIN 列表
            "is_contains_lob_info", # 派生：是否含捆绑销售信息（0/1）
            # ── LQS 评分 ───────────────────────────────────────────────────
            F.round(F.col("asin_lqs_rating"), 1).alias("asin_lqs_rating"),  # 派生：LQS 综合评分
            "asin_lqs_rating_detail", # 派生：LQS 各维度评分 JSON
            # ── 标识/标签 ──────────────────────────────────────────────────
            "amazon_label",      # Kafka → 解析 badge_type（兼容新旧格式）
            "is_amazon_new",     # 派生：badge_text 含 "new on amazon" 则 1
            "is_movie_label",    # 派生：是否电影/媒体类（0/1）
            "is_brand_label",    # 派生：是否有品牌（0/1）
            "multi_color_flag",  # 派生：多色标识（0.非多色/1.属性多色/2.标题降级）
            "multi_color_str",   # 派生：多色描述字符串
            # ── 区间类型 ───────────────────────────────────────────────────
            "rank_type",               # 一级排名区间（0-8）
            "ao_val_type",             # ao_val 区间（0-7）
            "price_type",              # 价格区间（0-6）
            "rating_type",             # 评分区间（0-5）
            "size_type",               # 尺寸类型（0-6/7）
            "weight_type",             # 重量类型（0-6）
            "site_name_type",          # 卖家所在地类型（1.US/2.CN/3.其他/4.自营/5.HK/6.TW）
            "quantity_variation_type", # 是否数量变体（0/1）
            # ── 变化率（绝对值/环比/同比）────────────────────────────────
            "rank_rise", "rank_mom", "rank_yoy",
            "ao_rise", "ao_mom", "ao_yoy",
            "price_rise", "price_mom", "price_yoy",
            "rating_rise", "rating_mom", "rating_yoy",
            "comments_rise", "comments_mom", "comments_yoy",
            "bsr_orders_rise", "bsr_orders_mom", "bsr_orders_yoy",
            "sales_rise", "sales_mom", "sales_yoy",
            "variation_rise", "variation_mom", "variation_yoy",
            "bought_month_mom", "bought_month_yoy",  # 月销无绝对值，仅环比/同比
        )
        df_save = df_save.na.fill(
            {"zr_counts": 0, "sp_counts": 0, "sb_counts": 0, "vi_counts": 0, "bs_counts": 0, "ac_counts": 0,
             "tr_counts": 0, "er_counts": 0, "title_len": 0, "total_comments": 0, "variation_num": 0, "img_num": 0,
             "one_two_val": 0.0, "three_four_val": 0.0, "five_six_val": 0.0, "eight_val": 0.0,
             "one_star": 0, "two_star": 0, "three_star": 0, "four_star": 0, "five_star": 0, "low_star": 0,
             "size_type": 0, "rating_type": 0, "site_name_type": 0, "weight_type": 0,
             "ao_val_type": 0, "rank_type": 0, "price_type": 0, "quantity_variation_type": 0, "package_quantity": 1,
             "is_movie_label": 0, "is_brand_label": 0, "asin_lqs_rating": 0.0, "follow_sellers_count": -1,
             "describe_len": 0, "multi_color_flag": 0}
        )
        print("asin的标准信息：")
        df_save.show(10, truncate=False)
        return df_save

    def read_data(self):
        # 1a/1b 共用读取逻辑：单表读取 + PySpark DataFrame join，避免 Hive 服务端 3 表 JOIN 性能差
        def _load_baseline(date_info_target, alias_prefix):
            sql_dwt = f"""
                select asin, asin_ao_val, asin_price, variation_num, asin_rating, asin_total_comments,
                       first_category_rank, bsr_orders, asin_bought_month
                from dwt_flow_asin
                where site_name = '{self.site_name}' and date_type = 'month' and date_info = '{date_info_target}'
            """
            print(f"sql_dwt({alias_prefix})=", sql_dwt)
            df = self.spark.sql(sqlQuery=sql_dwt).repartition(self.repartition_num, 'asin')
            return df.select(
                F.col('asin'),
                F.round(F.col('asin_ao_val'), 3).alias(f'{alias_prefix}_asin_ao_val'),
                F.col('asin_price').alias(f'{alias_prefix}_asin_price'),
                F.col('variation_num').alias(f'{alias_prefix}_asin_variation_num'),
                F.col('asin_rating').alias(f'{alias_prefix}_asin_rating'),
                F.col('asin_total_comments').alias(f'{alias_prefix}_asin_total_comments'),
                F.col('first_category_rank').alias(f'{alias_prefix}_first_category_rank'),
                F.col('bsr_orders').alias(f'{alias_prefix}_asin_bsr_orders'),
                F.round(F.col('bsr_orders') * F.col('asin_price'), 2).alias(f'{alias_prefix}_sales'),
                F.col('asin_bought_month').alias(f'{alias_prefix}_asin_bought_month'),
            ).persist(StorageLevel.DISK_ONLY)

        print(f"1a. 读取上个月维度的flow_asin（date_type=month, date_info={self.actual_last_month}）")
        self.df_previous_flow_asin = _load_baseline(self.actual_last_month, 'previous')
        self.df_previous_flow_asin.show(10, truncate=False)
        print(f"1b. 读取同比去年的flow_asin（date_type=month, date_info={self.actual_last_year}）")
        self.df_previous_flow_asin_lastyear = _load_baseline(self.actual_last_year, 'lastyear')
        self.df_previous_flow_asin_lastyear.show(10, truncate=False)
        print("2. 获取店铺相关信息")
        sql = f"""
            select seller_id, account_name, upper(fb_country_name) as seller_country_name, asin, fb_crawl_date
            from dim_fb_asin_info where site_name='{self.site_name}' and date_type = '30day' and seller_id is not null"""
        print("sql=", sql)
        self.df_seller_info = self.spark.sql(sqlQuery=sql)
        self.df_seller_info = self.df_seller_info.repartition(self.repartition_num).persist(StorageLevel.DISK_ONLY)
        self.df_seller_info.show(10, truncate=False)
        # df_seller_country: 按 seller_id 去重，保留 seller_country_name 非空的最新记录
        window_seller = Window.partitionBy('seller_id').orderBy(F.col('fb_crawl_date').desc())
        self.df_seller_country = self.df_seller_info \
            .filter(F.col('seller_country_name').isNotNull()) \
            .withColumn('rank', F.row_number().over(window_seller)) \
            .filter(F.col('rank') == 1) \
            .select('seller_id', 'seller_country_name') \
            .persist(StorageLevel.DISK_ONLY)
        # df_asin_seller: 按 asin 去重，保留最新记录
        window_asin = Window.partitionBy('asin').orderBy(F.col('fb_crawl_date').desc())
        self.df_asin_seller = self.df_seller_info \
            .withColumn('rank', F.row_number().over(window_asin)) \
            .filter(F.col('rank') == 1) \
            .select('asin', F.col('seller_id').alias('fd_seller_id'), F.col('account_name').alias('fd_account_name')) \
            .persist(StorageLevel.DISK_ONLY)
        # df_seller_info 已完成派生，及时释放缓存
        self.df_seller_info.unpersist()
        print("3. 读取asin_label信息(含 badge_text)")
        sql = f"""
            select asin, lower(label) as label, lower(badge_text) as badge_text
            from ods_other_search_term_data
            where site_name = '{self.site_name}' and date_type = '30day'
        """
        print("sql=", sql)
        df_label_raw = self.spark.sql(sqlQuery=sql) \
            .withColumn("label",      F.when(F.trim(F.col("label"))      .isin('null', ''), F.lit(None)).otherwise(F.col("label"))) \
            .withColumn("badge_text", F.when(F.trim(F.col("badge_text")) .isin('null', ''), F.lit(None)).otherwise(F.col("badge_text")))
        # 按 asin 聚合: collect_set 自动去重 + 过滤 NULL
        self.df_asin_label_info = df_label_raw.groupby(['asin']).agg(
            F.collect_set("label").alias("asin_label_list"),
            F.collect_set("badge_text").alias("badge_list"),
        )
        self.df_asin_label_info = self.df_asin_label_info.repartition(self.repartition_num).persist(StorageLevel.DISK_ONLY)
        self.df_asin_label_info.show(10, truncate=False)
        print("4. 读取dwd_asin_measure拿取ao及各类型数量")
        sql = f"""
        select asin, asin_sp_counts as sp_counts, (asin_sb1_counts + asin_sb2_counts) as sb_counts, asin_sb3_counts as vi_counts,
        asin_bs_counts as bs_counts, asin_ac_counts as ac_counts, asin_tr_counts as tr_counts, asin_er_counts as er_counts, 
        asin_st_counts, asin_zr_counts, asin_adv_counts, round(asin_zr_flow_proportion, 4) as asin_zr_flow_proportion,
        round(asin_ao_val, 4) as asin_ao_val, asin_amazon_orders
        from dwd_asin_measure where site_name='{self.site_name}' and date_type='30day'
        """
        print("sql=", sql)
        self.df_asin_measure = self.spark.sql(sqlQuery=sql)
        self.df_asin_measure = self.df_asin_measure.repartition(self.repartition_num).persist(StorageLevel.DISK_ONLY)
        self.df_asin_measure.show(10, truncate=False)
        print("5. 读取one_category_report表")
        sql = f"select category_id as asin_bs_cate_1_id, rank as asin_bs_cate_1_rank, orders as bsr_orders from ods_one_category_report " \
              f"where site_name='{self.site_name}' and date_type='30day'"
        print("sql=", sql)
        self.df_bs_report = self.spark.sql(sqlQuery=sql)
        self.df_bs_report = self.df_bs_report.repartition(self.repartition_num).persist(StorageLevel.DISK_ONLY)
        self.df_bs_report.show(10, truncate=False)
        print("6. 通过node_id获取一级分类进行分类补充")
        df_asin_new_cate = get_node_first_id_df(self.site_name, self.spark)
        self.df_asin_new_cate = F.broadcast(df_asin_new_cate)
        self.df_asin_new_cate.show(10, truncate=False)
        print("7. 获取分类ID与分类名称的对应关系")
        df_asin_category = get_first_id_from_category_desc_df(self.site_name, self.spark)
        self.df_asin_category = F.broadcast(df_asin_category.select(
            F.lower(F.col('category_first_name')).alias('desc_category_first_name'),
            F.col('category_first_id').alias('desc_category_first_id')
        ))
        self.df_asin_category.show(10, truncate=False)
        print("8. 读取颜色词表 dim_asin_color_info")
        color_rows = self.spark.sql(
            f"SELECT lower(en_name) as en_name FROM dim_asin_color_info WHERE site_name='{self.site_name}'"
        ).collect()
        self.color_set = {row.en_name for row in color_rows}
        print(f"颜色词表共 {len(self.color_set)} 条")
        print("9. 读取内部asin信息(ods_self_asin)，用于asin_type计算")
        sql = f"select asin, 1 as asin_is_self from ods_self_asin where site_name='{self.site_name}'"
        print("sql=", sql)
        self.df_self_asin = self.spark.sql(sqlQuery=sql).dropDuplicates(['asin']).persist(StorageLevel.DISK_ONLY)
        self.df_self_asin.show(10, truncate=False)
        print("9b. 读取内部品牌信息(amazon_brand)，用于asin_type计算")
        mysql_con = DBUtil.get_connection_info("mysql", "us")
        mysql_brand_sql = """
            SELECT DISTINCT LOWER(TRIM(brand_name)) AS brand, 1 as asin_is_self_brand FROM amazon_brand WHERE brand_type = '1'
        """
        self.df_self_brand = F.broadcast(SparkUtil.read_jdbc_query(
            session=self.spark, url=mysql_con['url'], pwd=mysql_con['pwd'],
            username=mysql_con['username'], query=mysql_brand_sql
        ).persist(StorageLevel.MEMORY_ONLY))
        self.df_self_brand.show(10, truncate=False)
        print("10. 读取隐藏分类(流量选品模块，id_path前缀匹配)，用于asin_type计算")
        # category_full_name + category_disable_config 按 id_path 前缀匹配（对齐 dwt.handle_asin_is_hide）
        sql = f"""
            SELECT DISTINCT category_id AS asin_bs_cate_current_id, 1 as hide_flag FROM category_full_name a
            WHERE EXISTS (
                SELECT 1 FROM category_disable_config b
                WHERE b.site = a.site AND b.module = '流量选品:名称前缀筛选'
                  AND a.id_path LIKE CONCAT(b.id_path, '%')
            ) AND a.site = '{self.site_name}'
        """
        self.df_hide_category = F.broadcast(SparkUtil.read_jdbc_query(
            session=self.spark, url=mysql_con['url'], pwd=mysql_con['pwd'],
            username=mysql_con['username'], query=sql
        ).persist(StorageLevel.MEMORY_ONLY))
        self.df_hide_category.show(10, truncate=False)
        print("11. 读取禁用分类(信息库模块，id_path前缀匹配)，用于信息库PG导出过滤")
        # 跟df_hide_category同样的匹配方式，只是module换成信息库自己的配置，避免误用流量选品的禁用分类
        sql = f"""
            SELECT DISTINCT category_id FROM category_full_name a
            WHERE EXISTS (
                SELECT 1 FROM category_disable_config b
                WHERE b.site = a.site AND b.module = '信息库:名称前缀筛选'
                  AND a.id_path LIKE CONCAT(b.id_path, '%')
            ) AND a.site = '{self.site_name}'
        """
        self.df_ai_hide_category = F.broadcast(SparkUtil.read_jdbc_query(
            session=self.spark, url=mysql_con['url'], pwd=mysql_con['pwd'],
            username=mysql_con['username'], query=sql
        ).persist(StorageLevel.MEMORY_ONLY))
        self.df_ai_hide_category.show(10, truncate=False)

    # 字段处理逻辑综合
    def handle_all_field(self, df):
        # 1. 处理asin分类及排名以及排名类型字段
        df = self.handle_asin_bs_category_info(df)
        # 2. 利用node_id进行分类补充
        df = self.handle_asin_category_supplement(df)
        # 3. 处理bsr销量及销售额信息以及价格类型字段
        df = self.handle_asin_bsr_orders(df)
        # 4. 解析Make-It-A-Bundle信息
        df = self.handle_asin_lob_info(df)
        # 5. 处理配送方式、卖家所在地以及卖家所在地类型
        df = self.handle_asin_buy_box_seller_type(df)
        # 6. 处理asin基础属性信息(长宽高重量等)
        df = self.handle_asin_basic_attribute_info(df)
        # 7. 处理asin图片信息
        df = self.handle_asin_img_info(df)
        # persist：第8步内部(test_flag=normal时会写一次parent_asin_latest_detail)会触发一次action，
        # 不persist的话后面第8~14步还会把第1~7步重新算一遍。批次末尾统一unpersist，见_unpersist_batch_cache
        df = df.persist(StorageLevel.DISK_ONLY)
        self.save_parent_asin_latest_detail_cache = df
        # 8. 处理变体相关（ao及母体相关，自然占比及母体自然占比，各类型数量，月销信息等）
        df = self.handle_asin_measure(df)
        # 9. 提取打包数量字段
        df = self.handle_asin_package_quantity(df)
        # 9.5. 多色判断（依赖 color 字段已从变体属性+product_json 补全）
        df = self.handle_multi_color_flag(df)
        # 10. 处理品牌标签、是否告警品牌、处理asin_lqs_rating信息
        df = self.handle_asin_lqs_and_brand(df)
        # 11.通过ASIN页面信息处理(评分类型、上架时间类型、电影标签、ASIN类型、有效类型)
        df = self.handle_asin_detail_all_type(df)
        # 12. 处理变化率相关字段
        df = self.handle_asin_attribute_change(df)
        # persist：第13步(asin_latest_detail写入)、信息库PG导出、字段标准化后的doris主表写入，
        # 三个action共用这份结果，避免第8~12步被重复算3次
        df = df.persist(StorageLevel.DISK_ONLY)
        self.save_asin_latest_detail_cache = df
        # 13. 写入 ASIN 最新详情表（在字段标准化 select 过滤前，趁 category/seller_json 等原始字段还在）
        self.save_asin_latest_detail(df)
        # 13.5 导出信息库ASIN详情到PG（倒数第二步，同样要在字段标准化前，用的还是原始字段名）
        self.save_ai_asin_detail(df)
        # 14. 字段标准化
        df_save = self.handle_column_name(df)
        return df_save

    def _unpersist_batch_cache(self):
        """及时释放handle_all_field里persist的中间结果，避免多批次/多次重试导致缓存堆积"""
        if self.save_parent_asin_latest_detail_cache is not None:
            self.save_parent_asin_latest_detail_cache.unpersist()
            self.save_parent_asin_latest_detail_cache = None
        if self.save_asin_latest_detail_cache is not None:
            self.save_asin_latest_detail_cache.unpersist()
            self.save_asin_latest_detail_cache = None

    @staticmethod
    def udf_rank_and_category(best_sellers_rank, pattern_str, top100_prefix):
        import re
        # 原文（去掉 . 和 ,）用于截取分类名，保留大小写
        original = str(best_sellers_rank).replace(".", "").replace(",", "")
        # de 数据形如 "Nr X in 分类 (Siehe Top 100 in 分类)" — 先把括号注释整段删除，
        # 避免分类名尾部贪婪吃到 "(" 前空格，保证与 us_asin_latest_detail.py 输出一致
        original = re.sub(r"\s*\([^)]*\)", "", original)
        # lower 版本仅用于正则匹配定位（pattern_str 与 top100_prefix 都是小写）
        lowered = original.lower().replace(top100_prefix, "")
        # 同步从原文剔除 Top100 前缀（大小写不敏感）
        cleaned_original = re.sub(re.escape(top100_prefix), "", original, flags=re.IGNORECASE)
        # 用小写匹配定位每个 "排名 + 分类" 在原文中的 span，再切原文保留大小写
        matches = list(re.finditer(pattern_str, lowered))
        if not matches:
            return None
        return "&&&&".join([cleaned_original[m.start():m.end()] for m in matches])

    # 写入 ASIN 最新详情表 Doris（仅 normal 模式）
    def save_asin_latest_detail(self, df):
        if self.test_flag != 'normal':
            return
        print(f"导出ASIN最新详情信息到doris：")
        start_time = time.time()
        df_asin_latest_detail = df.select(
            "asin",
            F.col("ao_val").alias("asin_ao_val"),
            F.col("title").alias("asin_title"),
            F.col("title_len").alias("asin_title_len"),
            F.col("category").alias("asin_category_desc"),
            F.col("volume").alias("asin_volume"),
            F.col("weight").alias("asin_weight"),
            # Doris 端 DATETIME 严格类型检查，string → timestamp 需显式转换
            F.to_timestamp(F.col("launch_time")).alias("asin_launch_time"),
            F.col("brand").alias("asin_brand_name"),
            "one_star", "two_star", "three_star", "four_star", "five_star", "low_star",
            "account_name",
            F.col("seller_id").alias("account_id"),  # handle_column_name 之前还叫 seller_id
            "seller_country_name",                     # handle_column_name 之前还叫 seller_country_name（不是 site_name）
            F.col("asin_bs_cate_1_id").alias("category_first_id"),
            "parent_asin",
            F.col("variat_num").alias("variation_num"),
            "img_info",
            F.to_timestamp(F.col("asinUpdateTime")).alias("asin_crawl_date"),
            F.col("price").alias("asin_price"),
            F.col("rating").alias("asin_rating"),
            F.col("total_comments").alias("asin_total_comments"),
            "matrix_ao_val", "zr_flow_proportion", "matrix_flow_proportion",
            F.lit(self.date_info_month).alias("date_info"),
            "img_url",
            F.col("asin_bs_cate_current_id").alias("category_current_id"),
            F.col("asin_bs_cate_1_rank").alias("category_first_rank"),
            F.col("asin_bs_cate_current_rank").alias("category_current_rank"),
            "bsr_orders", "bsr_orders_sale", "page_inventory", "asin_bought_month",
            "seller_json", "buy_box_seller_type",
            F.col("describe").alias("asin_describe"),
            F.col("fbm_delivery_price").alias("asin_fbm_price"),
            F.col("describe_len").alias("asin_describe_len"),
            "asin_weight_str",
            "best_sellers_rank",
            "best_sellers_herf",
            "asin_type",
            "is_amazon_new",
        )
        table_columns = """asin, asin_ao_val, asin_title, asin_title_len, asin_category_desc, asin_volume,
        asin_weight, asin_launch_time, asin_brand_name, one_star, two_star, three_star, four_star, five_star, low_star,
        account_name, account_id, seller_country_name, category_first_id, parent_asin, variation_num, img_info,
        asin_crawl_date, asin_price, asin_rating, asin_total_comments, matrix_ao_val, zr_flow_proportion, matrix_flow_proportion,
        date_info, img_url, category_current_id, category_first_rank, category_current_rank, bsr_orders, bsr_orders_sale,
        page_inventory, asin_bought_month, seller_json, buy_box_seller_type, asin_describe, asin_fbm_price, asin_describe_len,
        asin_weight_str, best_sellers_rank, best_sellers_herf, asin_type, is_amazon_new"""
        DorisHelper.spark_export_with_columns(
            df_save=df_asin_latest_detail, db_name=self.doris_db_selection,
            table_name=self.asin_latest_detail_table, table_columns=table_columns
        )
        end_time = time.time()
        print(f"Doris {self.asin_latest_detail_table} 写入完毕，耗时：{end_time - start_time:.1f}s")

    # 写入doris逻辑
    def save_to_doris(self, df, batch_num):
        print(f"写入Doris，数据量：{batch_num}")
        start_time = time.time()

        # ---- 30day 主表写入 ----
        # array 序列化为 JSON 字符串（Doris StreamLoad ARRAY<INT> 格式）
        df = df.withColumn("img_type", F.to_json(F.col("img_type")))

        doris_30day_columns = """asin, ao_val, zr_counts, sp_counts, sb_counts, vi_counts, bs_counts, ac_counts,
            tr_counts, er_counts, bsr_orders, bsr_orders_sale, title, title_len, price, rating, total_comments,
            buy_box_seller_type, page_inventory, volume, weight, color, size, style, material, launch_time,
            img_num, parent_asin, img_type, img_url, activity_type, one_two_val, three_four_val, five_six_val,
            eight_val, brand, variation_num, one_star, two_star, three_star, four_star, five_star, low_star,
            together_asin, account_name, account_id,
            rank_rise, rank_mom, rank_yoy, ao_rise, ao_mom, ao_yoy,
            price_rise, price_mom, price_yoy, rating_rise, rating_mom, rating_yoy,
            comments_rise, comments_mom, comments_yoy, bsr_orders_rise, bsr_orders_mom, bsr_orders_yoy,
            sales_rise, sales_mom, sales_yoy, variation_rise, variation_mom, variation_yoy,
            bought_month_mom, bought_month_yoy,
            size_type, rating_type, site_name_type, weight_type, ao_val_type, rank_type,
            price_type, quantity_variation_type, package_quantity,
            is_movie_label, is_brand_label, asin_crawl_date,
            category_first_id, category_id, first_category_rank, current_category_rank, desc_category_first_id, asin_weight_ratio,
            site_name, asin_type, asin_bought_month, asin_lqs_rating, asin_lqs_rating_detail, asin_lob_info,
            is_contains_lob_info, is_package_quantity_abnormal, zr_flow_proportion, matrix_flow_proportion,
            matrix_ao_val, product_features, img_info, collapse_asin, follow_sellers_count, asin_describe,
            fbm_price, describe_len, multi_color_flag, multi_color_str, amazon_label, is_amazon_new"""
        print(f"写入Doris {self.doris_30day_table}")
        DorisHelper.spark_export_with_columns(
            df_save=df, db_name=self.doris_db_dwt,
            table_name=self.doris_30day_table, table_columns=doris_30day_columns
        )
        end_time = time.time()
        print(f"Doris {self.doris_30day_table} 写入完毕，耗时：{end_time - start_time:.1f}s")

    # 导出信息库ASIN详情到PG（仅normal+latest模式；允许重复写入，同事AI分析工具按asin去重读取，不做唯一性判重）
    # 注意：在handle_column_name(字段标准化)之前调用，字段名都是renamed前的原始名
    def save_ai_asin_detail(self, df):
        if self.test_flag != 'normal' or self.consumer_type != 'latest':
            return
        print("导出信息库ASIN详情到PG：")
        start_time = time.time()
        df_info_base = df.filter(
            "asin_type in (0, 1) and asin_bought_month >= 50"
        ).join(
            self.df_ai_hide_category,
            df["asin_bs_cate_current_id"] == self.df_ai_hide_category["category_id"],
            'left_anti'
        )
        df_info_base = df_info_base.select(
            F.lit(self.site_name).alias('site_name'),
            "asin",
            F.round(F.col('weight').cast('double'), 2).alias('weight'),
            F.col('asin_bought_month').alias('bought_month'),
            "category",
            F.col('img_url').alias('img'),
            "title",
            "brand",
            "account_name",
            "buy_box_seller_type",
            F.date_format(F.to_timestamp(F.col('launch_time')), 'yyyy-MM-dd').alias('launch_time'),
            "img_num",
            F.when(F.col('variat_num') > 0, F.lit(1)).otherwise(F.lit(0)).alias('variation_flag'),
            F.col('variat_num').alias('variation_num'),
            F.round(F.col('ao_val').cast('double'), 2).alias('ao_val'),
            F.col('asin_bs_cate_1_id').alias('category_first_id'),
            F.col('asin_bs_cate_current_id').alias('category_current_id'),
            "parent_asin",
            F.col('asin_bs_cate_1_rank').alias('bsr_rank'),
            F.round(F.col('price').cast('double'), 2).alias('price'),
            F.round(F.col('rating').cast('double'), 2).alias('rating'),
            "total_comments",
            "seller_id",  # 已在第5步 coalesce 过（优先kafka自带，缺失时补df_asin_seller），跟新表seller_id同名
            F.col('seller_country_name').alias('fb_country_name'),
            "review_json_list",
            "describe",
            "product_json",
            "product_detail_json",
            "bought_month_mom",
            "bought_month_yoy",
            F.lit(0).cast('short').alias('chat_flag'),
        ).na.fill({'total_comments': 0, 'img_num': 0}).persist(StorageLevel.DISK_ONLY)
        export_count = df_info_base.count()
        print(f"待导出信息库PG的ASIN数量：{export_count}")
        if export_count > 0:
            pg_con = DBUtil.get_connection_info("postgresql", "us")
            df_info_base.write.format("jdbc") \
                .option("url", pg_con["url"]) \
                .option("dbtable", f"{self.site_name}_ai_asin_detail") \
                .option("user", pg_con["username"]) \
                .option("password", pg_con["pwd"]) \
                .mode("append") \
                .save()
        df_info_base.unpersist()
        end_time = time.time()
        print(f"信息库PG导出完毕，耗时：{end_time - start_time:.1f}s")

    # 实时消费中批次数据的处理逻辑（latest 模式）：失败重试2次，仍失败则抛出异常（避免checkpoint误提交导致丢数据）
    def handle_kafka_stream(self, df, batch_id):
        max_retries = 3
        for retry in range(max_retries):
            try:
                batch_num = df.count()
                if batch_num == 0:
                    print("当前批次没有数据")
                    return
                start_time = time.time()
                print(f"当前批次:{batch_id}; 该批次数据量为：{batch_num}")
                df_repartitioned = df.repartition(self.repartition_num)
                df_save = self.handle_all_field(df_repartitioned)
                self.save_to_doris(df_save, batch_num)
                end_time = time.time()
                print(f"当前批次:{batch_id} 执行完毕, 执行时长为：{end_time - start_time:.1f}s")
                return
            except Exception as e:
                print(f"当前批次:{batch_id} 处理失败(第{retry + 1}次)，{e}", traceback.format_exc())
                if retry == max_retries - 1:
                    CommonUtil.send_wx_msg(
                        ['chenyuanjie'], "⚠实时消费批次处理失败⚠", f"topic: {self.topic_name}, batch_id: {batch_id}, 重试{max_retries}次后仍失败: {e}"
                    )
                    raise
                time.sleep(10)
            finally:
                # 及时释放本批次persist的缓存（成功/失败/每次重试都要释放），避免资源堆积
                self._unpersist_batch_cache()

    # 消费主题下的所有历史数据
    def handle_kafka_history(self, kafka_df):
        print("处理kafka历史数据")
        batch_num = kafka_df.count()
        if batch_num > 0:
            try:
                start_time = time.time()
                kafka_df = kafka_df.repartition(self.repartition_num)
                kafka_df = self.handle_all_field(kafka_df)
                self.save_to_doris(kafka_df, batch_num)
                end_time = time.time()
                print("该批次数据处理完毕, 执行时长为：" + str(end_time - start_time))
            finally:
                self._unpersist_batch_cache()
        else:
            raise ValueError("当前主题中没有数据,请注意检查！")


if __name__ == '__main__':
    arguments = sys.argv[1:]
    site_name = sys.argv[1]      # 参数1：站点，如 us
    date_type = sys.argv[2]      # 参数2：类型，如 month
    date_info = sys.argv[3]      # 参数3：年-月，如 2026-03
    consumer_type = sys.argv[4]  # 参数4：history / latest
    if len(arguments) == 5:
        test_flag = sys.argv[5]
    else:
        test_flag = 'normal'
    handle_obj = KafkaFlowAsinDetail(site_name=site_name, date_type=date_type, date_info=date_info, consumer_type=consumer_type, test_flag=test_flag, batch_size=1200000)
    handle_obj.run_kafka()
