"""
   @Author      : CT
   @Description : 月流量选品 Doris 落地脚本，同时关联产出年度流量选品数据
                  流程：
                    [Step 1] Doris 建表 selection.{site}_flow_asin_month_{yyyy_mm}[_test]
                    [Step 2] 读 Hive dwt_flow_asin 月数据 + 字段规范化（与 dwt.us_flow_asin_30day DDL 对齐）
                    [Step 3] 写入 Doris dwt 月表 dwt.{site}_flow_asin_month
                    [Step 4] 写入 Doris dwt 年表 dwt.{site}_flow_asin_365day，并删除近12月窗口之外的过期数据
                    [Step 5] 同步年度聚合指标 dwt.{site}_flow_asin_365day_extra：从 Hive dwt_flow_asin_year 同步过来
                    [Step 6] Doris INSERT OVERWRITE 到 selection 月物化表 selection.{site}_flow_asin_month_{yyyy_mm}[_test]
                    [Step 7] Doris INSERT 到 selection 年表 selection.{site}_flow_asin_365day[_test]
                    [Step 8] 更新 MySQL workflow_everyday 流程记录表（月/年各写一条，仅 formal 模式）
                  支持 us / uk / de 三站点
                  支持 formal / test 模式：
                    - formal：selection 表名无后缀，更新流程记录表
                    - test  ：selection 表名加 _test 后缀，不更新流程记录表（dwt 主表不变）
执行示例:
    spark-submit dwt_flow_asin_month.py us 2026-05            # 默认 formal
    spark-submit dwt_flow_asin_month.py us 2026-05 formal
    spark-submit dwt_flow_asin_month.py us 2026-05 test       # test 模式
"""
import os
import sys

sys.path.append(os.path.dirname(sys.path[0]))

import pandas as pd
from pyspark.sql import functions as F

from utils.spark_util import SparkUtil
from utils.DorisHelper import DorisHelper
from utils.common_util import CommonUtil
from utils.db_util import DBUtil, DbTypes


DORIS_DB = "dwt"
SUPPORTED_SITES = ("us", "uk", "de")


def _doris_connect(use_type='selection'):
    """统一 pymysql 连接，database='selection' 让 UDF（如 stem_en）与 selection 库可见"""
    import pymysql
    conn_info = DorisHelper.get_connection_info(use_type)
    return pymysql.connect(
        host=conn_info['ip'],
        port=conn_info['jdbc_port'],
        user=conn_info['user'],
        password=conn_info['pwd'],
        database='selection',
        charset='utf8mb4',
        autocommit=True,
    )


def _exec_doris_sql(sql_list, use_type='selection'):
    """通过 pymysql 走 Doris jdbc_port 执行 DDL / INSERT 等"""
    conn = _doris_connect(use_type)
    try:
        cur = conn.cursor()
        for sql in sql_list:
            print(f"[Doris SQL] {sql[:200]}{'...' if len(sql) > 200 else ''}")
            cur.execute(sql)
        cur.close()
    finally:
        conn.close()


def _query_doris(sql, use_type='selection'):
    """执行查询并返回所有行"""
    conn = _doris_connect(use_type)
    try:
        cur = conn.cursor()
        cur.execute(sql)
        rows = cur.fetchall()
        cur.close()
        return rows
    finally:
        conn.close()


# ============================================================
# [Step 1] selection 月物化表建表
# ============================================================

def build_month_create_table_sql(table_name, date_info):
    """构建 selection.{table_name} 建表语句（与 DDL 一致）；
    table_name 由外层拼接：{site}_flow_asin_month_{yyyy_mm}[_test]"""
    return f"""
CREATE TABLE IF NOT EXISTS `selection`.`{table_name}`
(
    `asin`                        VARCHAR(20)      NOT NULL COMMENT 'ASIN',
    `parent_asin`                 STRING           NULL     COMMENT '父ASIN',
    `collapse_asin`               STRING           NULL     COMMENT '聚合ASIN',
    `asin_crawl_date`             DATETIME         NULL     COMMENT 'Kafka消息抓取时间',
    `title`                       STRING           NULL     COMMENT '标题(lowercase)',
    `title_stem`                  STRING           NULL     COMMENT '标题词干(stem_en派生)',
    `title_len`                   INT              NULL     COMMENT '标题字符长度',
    `brand`                       STRING           NULL     COMMENT '品牌(lowercase)',
    `asin_describe`               STRING           NULL     COMMENT '五点描述',
    `describe_len`                INT              NULL     COMMENT '五点描述字符长度',
    `product_features`            STRING           NULL     COMMENT '产品评论/特色JSON',
    `together_asin`               STRING           NULL     COMMENT '组合购买推荐ASIN',
    `price`                       DECIMAL(20,2)    NULL,
    `fbm_price`                   DECIMAL(20,2)    NULL,
    `rating`                      DECIMAL(10,1)    NULL,
    `total_comments`              INT              NULL,
    `one_star`                    INT              NULL,
    `two_star`                    INT              NULL,
    `three_star`                  INT              NULL,
    `four_star`                   INT              NULL,
    `five_star`                   INT              NULL,
    `low_star`                    INT              NULL,
    `bsr_orders`                  INT              NULL,
    `bsr_orders_sale`             DECIMAL(20,2)    NULL,
    `asin_bought_month`           INT              NULL,
    `ao_val`                      DECIMAL(20,4)    NULL,
    `zr_counts`                   INT              NULL,
    `sp_counts`                   INT              NULL,
    `sb_counts`                   INT              NULL,
    `vi_counts`                   INT              NULL,
    `bs_counts`                   INT              NULL,
    `ac_counts`                   INT              NULL,
    `tr_counts`                   INT              NULL,
    `er_counts`                   INT              NULL,
    `zr_flow_proportion`          DECIMAL(20,4)    NULL,
    `matrix_flow_proportion`      DECIMAL(20,4)    NULL,
    `matrix_ao_val`               DECIMAL(20,4)    NULL,
    `one_two_val`                 DECIMAL(20,4)    NULL,
    `three_four_val`              DECIMAL(20,4)    NULL,
    `five_six_val`                DECIMAL(20,4)    NULL,
    `eight_val`                   DECIMAL(20,4)    NULL,
    `category_first_id`           STRING           NULL,
    `category_id`                 STRING           NULL,
    `first_category_rank`         INT              NULL,
    `current_category_rank`       INT              NULL,
    `weight`                      DECIMAL(20,4)    NULL,
    `volume`                      STRING           NULL,
    `asin_weight_ratio`           DECIMAL(20,4)    NULL,
    `color`                       STRING           NULL,
    `size`                        STRING           NULL,
    `style`                       STRING           NULL,
    `material`                    STRING           NULL,
    `package_quantity`            INT              NULL,
    `is_package_quantity_abnormal` TINYINT         NULL,
    `variation_num`               INT              NULL,
    `page_inventory`              INT              NULL,
    `activity_type`               STRING           NULL,
    `launch_time`                 DATETIME         NULL,
    `launch_time_type`            INT              NULL,
    `img_url`                     STRING           NULL,
    `img_num`                     INT              NULL,
    `img_type_arr`                ARRAY<INT>       NULL,
    `img_info`                    STRING           NULL,
    `account_id`                  STRING           NULL,
    `account_name`                STRING           NULL,
    `buy_box_seller_type`         TINYINT          NULL,
    `site_name`                   STRING           NULL,
    `follow_sellers_count`        INT              NULL,
    `asin_lob_info`               STRING           NULL,
    `is_contains_lob_info`        TINYINT          NULL,
    `asin_lqs_rating`             DECIMAL(20,1)    NULL,
    `asin_lqs_rating_detail`      STRING           NULL,
    `amazon_label`                STRING           NULL,
    `is_movie_label`              TINYINT          NULL,
    `is_brand_label`              TINYINT          NULL,
    `multi_color_flag`            TINYINT          NULL,
    `multi_color_str`             STRING           NULL,
    `rank_type`                   TINYINT          NULL,
    `ao_val_type`                 TINYINT          NULL,
    `price_type`                  TINYINT          NULL,
    `rating_type`                 TINYINT          NULL,
    `size_type`                   TINYINT          NULL,
    `weight_type`                 TINYINT          NULL,
    `site_name_type`              TINYINT          NULL,
    `quantity_variation_type`     TINYINT          NULL,
    `rank_rise`                   INT              NULL,
    `rank_change`                 DECIMAL(20,4)    NULL,
    `rank_yoy`                    DECIMAL(20,4)    NULL,
    `ao_rise`                     DECIMAL(20,4)    NULL,
    `ao_change`                   DECIMAL(20,4)    NULL,
    `ao_yoy`                      DECIMAL(20,4)    NULL,
    `price_rise`                  DECIMAL(20,2)    NULL,
    `price_change`                DECIMAL(20,4)    NULL,
    `price_yoy`                   DECIMAL(20,4)    NULL,
    `rating_rise`                 DECIMAL(20,1)    NULL,
    `rating_change`               DECIMAL(20,4)    NULL,
    `rating_yoy`                  DECIMAL(20,4)    NULL,
    `comments_rise`               INT              NULL,
    `comments_change`             DECIMAL(20,4)    NULL,
    `comments_yoy`                DECIMAL(20,4)    NULL,
    `bsr_orders_rise`             INT              NULL,
    `bsr_orders_change`           DECIMAL(20,4)    NULL,
    `bsr_orders_yoy`              DECIMAL(20,4)    NULL,
    `sales_rise`                  DECIMAL(20,2)    NULL,
    `sales_change`                DECIMAL(20,4)    NULL,
    `sales_yoy`                   DECIMAL(20,4)    NULL,
    `variation_rise`              INT              NULL,
    `variation_change`            DECIMAL(20,4)    NULL,
    `variation_yoy`               DECIMAL(20,4)    NULL,
    `bought_month_mom`            DECIMAL(20,4)    NULL,
    `bought_month_yoy`            DECIMAL(20,4)    NULL,
    `ocean_profit`                DECIMAL(20,4)    NULL,
    `air_profit`                  DECIMAL(20,4)    NULL,
    `tracking_since`              STRING           NULL,
    `tracking_since_type`         INT              NULL,
    `package_length`              INT              NULL,
    `package_width`               INT              NULL,
    `package_height`              INT              NULL,
    `item_weight`                 INT              NULL,
    `is_alarm_brand`              INT              NULL,
    `bsr_best_orders_type`        SMALLINT         NULL,
    `asin_source_flag`            ARRAY<INT>       NULL,
    `bsr_last_seen_at`            DATE             NULL,
    `bsr_seen_count_30d`          INT              NULL,
    `nsr_last_seen_at`            DATE             NULL,
    `nsr_seen_count_30d`          INT              NULL,
    `asin_type`                   SMALLINT         NULL,
    `usr_mask_progress`           STRING           NULL,
    `usr_mask_type`               STRING           NULL,
    `auctions_num`                INT              NULL,
    `auctions_num_all`            INT              NULL,
    `skus_num_creat`              INT              NULL,
    `skus_num_creat_all`          INT              NULL,
    `title_matching_degree`       DECIMAL(20,4)    NULL,
    `brand_badge_reason`          STRING           NULL,
    `title_stem_15`               STRING           NULL     COMMENT '标题词干前15个词(按空格分词截取)',
    `seller_province`             STRING           NULL     COMMENT '卖家所在省份(dim_seller_address关联,仅国内卖家有值)',
    `seller_city`                 STRING           NULL     COMMENT '卖家所在城市(dim_seller_address关联,仅国内卖家有值)',
    `variation_bought_month`      INT              NULL     COMMENT '变体销量和(12个月变体销量之和口径见年度指标,此处为当月值)',
    `variation_bought_month_mom`  DECIMAL(20,4)    NULL     COMMENT '变体销量和环比',
    `variation_bought_month_yoy`  DECIMAL(20,4)    NULL     COMMENT '变体销量和同比',
    INDEX idx_title      (`title`)      USING INVERTED PROPERTIES("parser" = "english") COMMENT '标题倒排索引',
    INDEX idx_title_stem (`title_stem`) USING INVERTED PROPERTIES("parser" = "english") COMMENT '标题词干倒排索引',
    INDEX idx_title_stem_15 (`title_stem_15`) USING INVERTED PROPERTIES("parser" = "english") COMMENT '标题词干前15个词倒排索引',
    INDEX idx_color      (`color`)      USING INVERTED PROPERTIES("parser" = "english") COMMENT '颜色倒排索引',
    INDEX idx_brand      (`brand`)      USING INVERTED PROPERTIES("parser" = "english") COMMENT '品牌倒排索引',
    INDEX idx_account_name (`account_name`) USING INVERTED PROPERTIES("parser" = "english") COMMENT '卖家名称倒排索引'
) ENGINE=OLAP
UNIQUE KEY(`asin`)
COMMENT '流量选品月维度物化表 {date_info}'
DISTRIBUTED BY HASH(`asin`) BUCKETS 32
PROPERTIES (
    "replication_num" = "3",
    "enable_unique_key_merge_on_write" = "true"
)
"""


# ============================================================
# [Step 2] 读 Hive dwt_flow_asin 月数据 + 字段规范化
# ============================================================

def read_and_normalize_month_data(spark, site_name, date_info):
    """读 Hive dwt_flow_asin 月数据，规范化字段（与 dwt.us_flow_asin_30day DDL 对齐）
    返回 cache 好的 df_save，调用方用完需自行 unpersist（见 write_dwt_month_table）"""
    sql = f"""
        SELECT
            asin,
            asin_ao_val,
            asin_zr_counts, asin_sp_counts, asin_sb_counts, asin_vi_counts,
            asin_bs_counts, asin_ac_counts, asin_tr_counts, asin_er_counts,
            bsr_orders,
            sales AS bsr_orders_sale,
            asin_title, asin_title_len,
            asin_price,
            asin_rating, asin_total_comments,
            asin_buy_box_seller_type,
            asin_page_inventory,
            asin_volume, asin_weight,
            asin_color, asin_size, asin_style, asin_material,
            asin_launch_time,
            asin_img_num,
            parent_asin,
            asin_img_type,
            asin_img_url,
            asin_activity_type,
            act_one_two_val, act_three_four_val, act_five_six_val, act_eight_val,
            asin_brand_name,
            variation_num,
            one_star, two_star, three_star, four_star, five_star, low_star,
            together_asin,
            account_name, account_id,
            asin_rank_rise, asin_rank_mom, asin_rank_yoy,
            asin_ao_rise, asin_ao_mom, asin_ao_yoy,
            asin_price_rise, asin_price_mom, asin_price_yoy,
            asin_rating_rise, asin_rating_mom, asin_rating_yoy,
            asin_comments_rise, asin_comments_mom, asin_comments_yoy,
            asin_bsr_orders_rise, asin_bsr_orders_mom, asin_bsr_orders_yoy,
            asin_sales_rise, asin_sales_mom, asin_sales_yoy,
            asin_variation_rise, asin_variation_mom, asin_variation_yoy,
            asin_bought_mom, asin_bought_yoy,
            asin_size_type, asin_rating_type, asin_site_name_type, asin_weight_type,
            asin_ao_val_type, asin_rank_type, asin_price_type,
            asin_quantity_variation_type, package_quantity,
            is_movie_label, is_brand_label, is_alarm_brand,
            asin_crawl_date,
            category_first_id, category_id, desc_category_first_id,
            first_category_rank, current_category_rank,
            asin_weight_ratio,
            seller_country_name,
            asin_type,
            asin_bought_month,
            variation_bought_month, variation_bought_month_mom, variation_bought_month_yoy,
            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,
            customer_reviews_json,
            img_info,
            coalesce(parent_asin, asin) AS collapse_asin,
            follow_sellers_count,
            asin_describe, asin_fbm_price, describe_len,
            title_matching_degree,
            multi_color_flag, multi_color_str,
            amazon_label,
            auctions_num, auctions_num_all, skus_num_creat, skus_num_creat_all
        FROM dwt_flow_asin
        WHERE site_name = '{site_name}' AND date_type = 'month' AND date_info = '{date_info}'
    """
    print(f"sql=\n{sql}")
    df_raw = spark.sql(sqlQuery=sql).repartition(40, 'asin')

    def _norm_dt(col_name):
        """text → DATETIME 容错：先 yyyy-MM-dd HH:mm:ss，失败退到 yyyy-MM-dd"""
        c = F.col(col_name)
        return F.coalesce(
            F.to_timestamp(c, 'yyyy-MM-dd HH:mm:ss'),
            F.to_timestamp(c, 'yyyy-MM-dd'),
        )

    df_save = df_raw.select(
        F.lit(date_info).alias('date_info'),
        F.col('asin'),
        # ===== 核心选品指标 =====
        F.round(F.col('asin_ao_val').cast('double'), 4).cast('decimal(20,4)').alias('ao_val'),
        F.col('asin_zr_counts').cast('int').alias('zr_counts'),
        F.col('asin_sp_counts').cast('int').alias('sp_counts'),
        F.col('asin_sb_counts').cast('int').alias('sb_counts'),
        F.col('asin_vi_counts').cast('int').alias('vi_counts'),
        F.col('asin_bs_counts').cast('int').alias('bs_counts'),
        F.col('asin_ac_counts').cast('int').alias('ac_counts'),
        F.col('asin_tr_counts').cast('int').alias('tr_counts'),
        F.col('asin_er_counts').cast('int').alias('er_counts'),
        F.col('bsr_orders').cast('int').alias('bsr_orders'),
        F.round(F.col('bsr_orders_sale').cast('double'), 2).cast('decimal(20,2)').alias('bsr_orders_sale'),
        # ===== ASIN 基础属性 =====
        F.col('asin_title').alias('title'),
        F.col('asin_title_len').cast('int').alias('title_len'),
        F.round(F.col('asin_price').cast('double'), 2).cast('decimal(20,2)').alias('price'),
        F.round(F.col('asin_rating').cast('double'), 1).cast('decimal(10,1)').alias('rating'),
        F.col('asin_total_comments').cast('int').alias('total_comments'),
        F.col('asin_buy_box_seller_type').cast('tinyint').alias('buy_box_seller_type'),
        F.col('asin_page_inventory').cast('int').alias('page_inventory'),
        F.col('asin_volume').alias('volume'),
        F.round(F.col('asin_weight').cast('double'), 4).cast('decimal(20,4)').alias('weight'),
        F.col('asin_color').alias('color'),
        F.col('asin_size').alias('size'),
        F.col('asin_style').alias('style'),
        F.col('asin_material').alias('material'),
        _norm_dt('asin_launch_time').alias('launch_time'),
        F.col('asin_img_num').cast('int').alias('img_num'),
        F.col('parent_asin'),
        # asin_img_type string "1,2,3" → ARRAY<INT> → 再 to_json 成 "[1,2,3]"（Doris StreamLoad ARRAY<INT> 要求）
        F.to_json(F.expr("transform(split(asin_img_type, ','), x -> cast(x as int))")).alias('img_type'),
        F.col('asin_img_url').alias('img_url'),
        F.col('asin_activity_type').alias('activity_type'),
        F.round(F.col('act_one_two_val').cast('double'), 4).cast('decimal(20,4)').alias('one_two_val'),
        F.round(F.col('act_three_four_val').cast('double'), 4).cast('decimal(20,4)').alias('three_four_val'),
        F.round(F.col('act_five_six_val').cast('double'), 4).cast('decimal(20,4)').alias('five_six_val'),
        F.round(F.col('act_eight_val').cast('double'), 4).cast('decimal(20,4)').alias('eight_val'),
        F.col('asin_brand_name').alias('brand'),
        F.col('variation_num').cast('int').alias('variation_num'),
        F.col('one_star').cast('int').alias('one_star'),
        F.col('two_star').cast('int').alias('two_star'),
        F.col('three_star').cast('int').alias('three_star'),
        F.col('four_star').cast('int').alias('four_star'),
        F.col('five_star').cast('int').alias('five_star'),
        F.col('low_star').cast('int').alias('low_star'),
        F.col('together_asin'),
        F.col('account_name'),
        F.col('account_id'),
        # ===== 环比/同比变化率 =====
        F.col('asin_rank_rise').cast('int').alias('rank_rise'),
        F.round(F.col('asin_rank_mom').cast('double'), 4).cast('decimal(20,4)').alias('rank_mom'),
        F.round(F.col('asin_rank_yoy').cast('double'), 4).cast('decimal(20,4)').alias('rank_yoy'),
        F.round(F.col('asin_ao_rise').cast('double'), 4).cast('decimal(20,4)').alias('ao_rise'),
        F.round(F.col('asin_ao_mom').cast('double'), 4).cast('decimal(20,4)').alias('ao_mom'),
        F.round(F.col('asin_ao_yoy').cast('double'), 4).cast('decimal(20,4)').alias('ao_yoy'),
        F.round(F.col('asin_price_rise').cast('double'), 2).cast('decimal(20,2)').alias('price_rise'),
        F.round(F.col('asin_price_mom').cast('double'), 4).cast('decimal(20,4)').alias('price_mom'),
        F.round(F.col('asin_price_yoy').cast('double'), 4).cast('decimal(20,4)').alias('price_yoy'),
        F.round(F.col('asin_rating_rise').cast('double'), 1).cast('decimal(20,1)').alias('rating_rise'),
        F.round(F.col('asin_rating_mom').cast('double'), 4).cast('decimal(20,4)').alias('rating_mom'),
        F.round(F.col('asin_rating_yoy').cast('double'), 4).cast('decimal(20,4)').alias('rating_yoy'),
        F.col('asin_comments_rise').cast('int').alias('comments_rise'),
        F.round(F.col('asin_comments_mom').cast('double'), 4).cast('decimal(20,4)').alias('comments_mom'),
        F.round(F.col('asin_comments_yoy').cast('double'), 4).cast('decimal(20,4)').alias('comments_yoy'),
        F.col('asin_bsr_orders_rise').cast('int').alias('bsr_orders_rise'),
        F.round(F.col('asin_bsr_orders_mom').cast('double'), 4).cast('decimal(20,4)').alias('bsr_orders_mom'),
        F.round(F.col('asin_bsr_orders_yoy').cast('double'), 4).cast('decimal(20,4)').alias('bsr_orders_yoy'),
        F.round(F.col('asin_sales_rise').cast('double'), 2).cast('decimal(20,2)').alias('sales_rise'),
        F.round(F.col('asin_sales_mom').cast('double'), 4).cast('decimal(20,4)').alias('sales_mom'),
        F.round(F.col('asin_sales_yoy').cast('double'), 4).cast('decimal(20,4)').alias('sales_yoy'),
        F.col('asin_variation_rise').cast('int').alias('variation_rise'),
        F.round(F.col('asin_variation_mom').cast('double'), 4).cast('decimal(20,4)').alias('variation_mom'),
        F.round(F.col('asin_variation_yoy').cast('double'), 4).cast('decimal(20,4)').alias('variation_yoy'),
        F.round(F.col('asin_bought_mom').cast('double'), 4).cast('decimal(20,4)').alias('bought_month_mom'),
        F.round(F.col('asin_bought_yoy').cast('double'), 4).cast('decimal(20,4)').alias('bought_month_yoy'),
        # ===== 类型枚举字段 =====
        F.col('asin_size_type').cast('tinyint').alias('size_type'),
        F.col('asin_rating_type').cast('tinyint').alias('rating_type'),
        F.col('asin_site_name_type').cast('tinyint').alias('site_name_type'),
        F.col('asin_weight_type').cast('tinyint').alias('weight_type'),
        F.col('asin_ao_val_type').cast('tinyint').alias('ao_val_type'),
        F.col('asin_rank_type').cast('tinyint').alias('rank_type'),
        F.col('asin_price_type').cast('tinyint').alias('price_type'),
        F.col('asin_quantity_variation_type').cast('tinyint').alias('quantity_variation_type'),
        F.col('package_quantity').cast('int').alias('package_quantity'),
        F.col('is_movie_label').cast('tinyint').alias('is_movie_label'),
        F.col('is_brand_label').cast('tinyint').alias('is_brand_label'),
        F.col('is_alarm_brand').cast('int').alias('is_alarm_brand'),
        # ===== 分类信息 =====
        _norm_dt('asin_crawl_date').alias('asin_crawl_date'),
        F.col('category_first_id'),
        F.col('category_id'),
        F.col('desc_category_first_id'),
        F.col('first_category_rank').cast('int').alias('first_category_rank'),
        F.col('current_category_rank').cast('int').alias('current_category_rank'),
        F.round(F.col('asin_weight_ratio').cast('double'), 4).cast('decimal(20,4)').alias('asin_weight_ratio'),
        F.col('seller_country_name').alias('site_name'),
        F.col('asin_type').cast('tinyint').alias('asin_type'),
        # ===== 月销及母体信息 =====
        F.col('asin_bought_month').cast('int').alias('asin_bought_month'),
        F.round(F.col('asin_lqs_rating').cast('double'), 1).cast('decimal(20,1)').alias('asin_lqs_rating'),
        F.col('asin_lqs_rating_detail'),
        F.col('asin_lob_info'),
        F.col('is_contains_lob_info').cast('tinyint').alias('is_contains_lob_info'),
        F.col('is_package_quantity_abnormal').cast('tinyint').alias('is_package_quantity_abnormal'),
        F.round(F.col('zr_flow_proportion').cast('double'), 4).cast('decimal(20,4)').alias('zr_flow_proportion'),
        F.round(F.col('matrix_flow_proportion').cast('double'), 4).cast('decimal(20,4)').alias('matrix_flow_proportion'),
        F.round(F.col('matrix_ao_val').cast('double'), 4).cast('decimal(20,4)').alias('matrix_ao_val'),
        F.col('customer_reviews_json').alias('product_features'),
        F.col('img_info'),
        F.col('collapse_asin'),
        F.col('follow_sellers_count').cast('int').alias('follow_sellers_count'),
        F.col('asin_describe'),
        F.round(F.col('asin_fbm_price').cast('double'), 2).cast('decimal(20,2)').alias('fbm_price'),
        F.col('describe_len').cast('int').alias('describe_len'),
        F.round(F.col('title_matching_degree').cast('double'), 4).cast('decimal(20,4)').alias('title_matching_degree'),
        # ===== 颜色组合识别 =====
        F.col('multi_color_flag').cast('tinyint').alias('multi_color_flag'),
        F.col('multi_color_str'),
        # ===== Amazon标签 =====
        F.col('amazon_label'),
        # ===== 竞卖与SKU创建 =====
        F.col('auctions_num').cast('int').alias('auctions_num'),
        F.col('auctions_num_all').cast('int').alias('auctions_num_all'),
        F.col('skus_num_creat').cast('int').alias('skus_num_creat'),
        F.col('skus_num_creat_all').cast('int').alias('skus_num_creat_all'),
        # ===== 变体销量和及环比/同比（新增字段放末尾）=====
        F.col('variation_bought_month').cast('int').alias('variation_bought_month'),
        F.round(F.col('variation_bought_month_mom').cast('double'), 4).cast('decimal(20,4)').alias('variation_bought_month_mom'),
        F.round(F.col('variation_bought_month_yoy').cast('double'), 4).cast('decimal(20,4)').alias('variation_bought_month_yoy'),
    ).cache()

    count = df_save.count()
    print(f"写入数据量：{count:,}")
    df_save.show(5, truncate=False)
    return df_save


# ============================================================
# [Step 3] 写入 Doris dwt 主表
# ============================================================

def write_dwt_month_table(df_save, doris_table):
    """写入 Doris dwt 主表 dwt.{site}_flow_asin_month，写完 unpersist 传入的 df_save"""
    table_columns = (
        "date_info, 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, is_alarm_brand, "
        "asin_crawl_date, category_first_id, category_id, desc_category_first_id, "
        "first_category_rank, current_category_rank, 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, "
        "title_matching_degree, multi_color_flag, multi_color_str, amazon_label, "
        "auctions_num, auctions_num_all, skus_num_creat, skus_num_creat_all, "
        "variation_bought_month, variation_bought_month_mom, variation_bought_month_yoy"
    )
    print(f"[Step 3] 写入 Doris {DORIS_DB}.{doris_table}")
    DorisHelper.spark_export_with_columns(
        df_save=df_save,
        db_name=DORIS_DB,
        table_name=doris_table,
        table_columns=table_columns,
    )
    df_save.unpersist()


# ============================================================
# [Step 4] 维护 dwt.{site}_flow_asin_365day（近12个月ASIN最新快照）
# ============================================================

def build_stg_create_table_sql(site_name):
    """构建基础详情快照中间表 dwt.{site}_flow_asin_365day 建表语句（与 doris_ddl.sql 一致）"""
    return f"""
CREATE TABLE IF NOT EXISTS `dwt`.`{site_name}_flow_asin_365day`
(
    `asin`                        VARCHAR(20)      NOT NULL,
    `date_info`                   DATE             NOT NULL,
    `ao_val`                      DECIMAL(20,4),
    `zr_counts`                   INT,
    `sp_counts`                   INT,
    `sb_counts`                   INT,
    `vi_counts`                   INT,
    `bs_counts`                   INT,
    `ac_counts`                   INT,
    `tr_counts`                   INT,
    `er_counts`                   INT,
    `bsr_orders`                  INT,
    `bsr_orders_sale`             DECIMAL(20,2),
    `title`                       STRING,
    `title_len`                   INT,
    `price`                       DECIMAL(20,2),
    `rating`                      DECIMAL(10,1),
    `total_comments`              INT,
    `buy_box_seller_type`         TINYINT,
    `page_inventory`              INT,
    `volume`                      STRING,
    `weight`                      DECIMAL(20,4),
    `color`                       STRING,
    `size`                        STRING,
    `style`                       STRING,
    `material`                    STRING,
    `launch_time`                 DATETIME,
    `img_num`                     INT,
    `parent_asin`                 STRING,
    `img_type`                    ARRAY<INT>,
    `img_url`                     STRING,
    `activity_type`               STRING,
    `one_two_val`                 DECIMAL(20,4)    DEFAULT "0.0000",
    `three_four_val`              DECIMAL(20,4)    DEFAULT "0.0000",
    `five_six_val`                DECIMAL(20,4)    DEFAULT "0.0000",
    `eight_val`                   DECIMAL(20,4)    DEFAULT "0.0000",
    `brand`                       STRING,
    `variation_num`               INT,
    `one_star`                    INT,
    `two_star`                    INT,
    `three_star`                  INT,
    `four_star`                   INT,
    `five_star`                   INT,
    `low_star`                    INT,
    `together_asin`               STRING,
    `account_name`                STRING,
    `account_id`                  STRING,
    `rank_rise`                   INT,
    `rank_mom`                    DECIMAL(20,4),
    `rank_yoy`                    DECIMAL(20,4),
    `ao_rise`                     DECIMAL(20,4),
    `ao_mom`                      DECIMAL(20,4),
    `ao_yoy`                      DECIMAL(20,4),
    `price_rise`                  DECIMAL(20,2),
    `price_mom`                   DECIMAL(20,4),
    `price_yoy`                   DECIMAL(20,4),
    `rating_rise`                 DECIMAL(20,1),
    `rating_mom`                  DECIMAL(20,4),
    `rating_yoy`                  DECIMAL(20,4),
    `comments_rise`               INT,
    `comments_mom`                DECIMAL(20,4),
    `comments_yoy`                DECIMAL(20,4),
    `bsr_orders_rise`             INT,
    `bsr_orders_mom`              DECIMAL(20,4),
    `bsr_orders_yoy`              DECIMAL(20,4),
    `sales_rise`                  DECIMAL(20,2),
    `sales_mom`                   DECIMAL(20,4),
    `sales_yoy`                   DECIMAL(20,4),
    `variation_rise`              INT,
    `variation_mom`               DECIMAL(20,4),
    `variation_yoy`               DECIMAL(20,4),
    `bought_month_mom`            DECIMAL(20,4),
    `bought_month_yoy`            DECIMAL(20,4),
    `size_type`                   TINYINT,
    `rating_type`                 TINYINT,
    `site_name_type`              TINYINT,
    `weight_type`                 TINYINT,
    `ao_val_type`                 TINYINT,
    `rank_type`                   TINYINT,
    `price_type`                  TINYINT,
    `quantity_variation_type`     TINYINT          DEFAULT "0",
    `package_quantity`            INT              DEFAULT "1",
    `is_movie_label`              TINYINT          DEFAULT "0",
    `is_brand_label`              TINYINT          DEFAULT "0",
    `is_alarm_brand`              INT              DEFAULT "0",
    `asin_crawl_date`             DATETIME,
    `category_first_id`           STRING,
    `category_id`                 STRING,
    `desc_category_first_id`      STRING,
    `first_category_rank`         INT,
    `current_category_rank`       INT,
    `asin_weight_ratio`           DECIMAL(20,4),
    `site_name`                   STRING,
    `asin_type`                   TINYINT,
    `asin_bought_month`           INT,
    `asin_lqs_rating`             DECIMAL(20,1),
    `asin_lqs_rating_detail`      STRING,
    `asin_lob_info`               STRING,
    `is_contains_lob_info`        TINYINT,
    `is_package_quantity_abnormal` TINYINT,
    `zr_flow_proportion`          DECIMAL(20,4),
    `matrix_flow_proportion`      DECIMAL(20,4),
    `matrix_ao_val`               DECIMAL(20,4),
    `product_features`            STRING,
    `img_info`                    STRING,
    `collapse_asin`               STRING,
    `follow_sellers_count`        INT              DEFAULT "-1",
    `asin_describe`               STRING,
    `fbm_price`                   DECIMAL(20,2),
    `describe_len`                INT              DEFAULT "0",
    `title_matching_degree`       DECIMAL(20,4),
    `multi_color_flag`            TINYINT          DEFAULT "0",
    `multi_color_str`             STRING,
    `amazon_label`                STRING,
    `auctions_num`                INT,
    `auctions_num_all`            INT,
    `skus_num_creat`              INT,
    `skus_num_creat_all`          INT,
    `variation_bought_month`      INT,
    `variation_bought_month_mom`  DECIMAL(20,4),
    `variation_bought_month_yoy`  DECIMAL(20,4)
)
ENGINE=OLAP
UNIQUE KEY(`asin`)
DISTRIBUTED BY HASH(`asin`) BUCKETS 32
PROPERTIES (
    "replication_num" = "3",
    "enable_unique_key_merge_on_write" = "true",
    "function_column.sequence_col" = "date_info"
)
"""


# 中间表字段列表（不含 date_info，因为 INSERT INTO 时按 SELECT * FROM dwt.month 顺序写入，date_info 在 SELECT 列表末尾）
STG_INSERT_COLUMNS = """
    asin, date_info, 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, is_alarm_brand,
    asin_crawl_date, category_first_id, category_id, desc_category_first_id,
    first_category_rank, current_category_rank, 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, title_matching_degree,
    multi_color_flag, multi_color_str, amazon_label,
    auctions_num, auctions_num_all, skus_num_creat, skus_num_creat_all,
    variation_bought_month, variation_bought_month_mom, variation_bought_month_yoy
""".strip()


def maintain_stg_table(site_name, months, date_info):
    """维护基础详情快照中间表 dwt.{site}_flow_asin_365day:
       1) 强制重写入参 date_info 当月数据
       2) 探测缺失月份，逐月 INSERT INTO 补齐
       3) 清理 < 近12月窗口起点的过期数据
       :param months: 近12个月列表，需按 yyyy-MM 升序排列
    """
    stg_table = f"{site_name}_flow_asin_365day"

    # 注意：中间表 date_info 为 DATE 类型(月首日,如 '2026-05-01'),
    # 而 dwt 月表 date_info 为 VARCHAR(yyyy-MM,如 '2026-05')。
    # 写入时需用 CAST(CONCAT(date_info, '-01') AS DATE) 把月表的 VARCHAR 转为 DATE。
    stg_select_columns = STG_INSERT_COLUMNS.replace(
        'date_info',
        "CAST(CONCAT(date_info, '-01') AS DATE)",
        1   # 只替换 SELECT 列表中第一次出现的 date_info(列名本身)
    )

    def _build_insert_sql(month_yyyymm):
        return f"""
            INSERT INTO `dwt`.`{stg_table}` ({STG_INSERT_COLUMNS})
            SELECT {stg_select_columns}
            FROM `dwt`.`{site_name}_flow_asin_month`
            WHERE date_info = '{month_yyyymm}'
        """

    print(f"[维护365day] 强制重写入参当月 {date_info} 数据到 dwt.{stg_table}")
    _exec_doris_sql([_build_insert_sql(date_info)])

    existing_rows = _query_doris(
        f"SELECT DISTINCT DATE_FORMAT(date_info, '%Y-%m') FROM `dwt`.`{stg_table}`"
    )
    existing_months = {row[0] for row in existing_rows}
    print(f"[维护365day] 中间表已有月份: {sorted(existing_months)}")

    missing_months = [m for m in months if m not in existing_months]
    if missing_months:
        print(f"[维护365day] 待补齐月份: {missing_months}")
        for m in missing_months:
            print(f"              → INSERT INTO dwt.{stg_table} WHERE date_info='{m}'")
            _exec_doris_sql([_build_insert_sql(m)])
    else:
        print(f"[维护365day] 近12月数据已齐全，无需补齐")

    min_keep_month = months[0]
    min_keep_date = f"{min_keep_month}-01"
    expired_months = [m for m in existing_months if m < min_keep_month]
    if expired_months:
        print(f"[维护365day] 清理过期数据(date_info < {min_keep_date}): {sorted(expired_months)}")
        _exec_doris_sql([f"DELETE FROM `dwt`.`{stg_table}` WHERE date_info < '{min_keep_date}'"])
    else:
        print(f"[维护365day] 无过期数据需要清理")


# ============================================================
# [Step 5] 同步 Hive dwt_flow_asin_year → Doris dwt.{site}_flow_asin_365day_extra
# ============================================================

EXTRA_COLUMNS = (
    "asin, bought_month_total, "
    "bought_month_1, bought_month_2, bought_month_3, bought_month_4, "
    "bought_month_5, bought_month_6, bought_month_7, bought_month_8, "
    "bought_month_9, bought_month_10, bought_month_11, bought_month_12, "
    "bought_month_q1, bought_month_q2, bought_month_q3, bought_month_q4, "
    "total_appear_month, bought_month_peak, peak_month_arr, "
    "is_periodic_flag, is_seasonal_flag, bsr_seen_count_total, nsr_seen_count_total, "
    "variation_bought_month_total, bought_month_total_yoy, variation_bought_month_total_yoy, "
    "variation_bought_month_1, variation_bought_month_2, variation_bought_month_3, variation_bought_month_4, "
    "variation_bought_month_5, variation_bought_month_6, variation_bought_month_7, variation_bought_month_8, "
    "variation_bought_month_9, variation_bought_month_10, variation_bought_month_11, variation_bought_month_12, "
    "variation_bought_month_q1, variation_bought_month_q2, variation_bought_month_q3, variation_bought_month_q4, "
    "variation_bought_month_peak, variation_peak_month_arr"
)


def build_extra_create_table_sql(table_name):
    """构建年度聚合中间表 dwt.{site}_flow_asin_365day_extra 建表语句（每次运行 TRUNCATE 后全量重算，无需 sequence_col）"""
    return f"""
CREATE TABLE IF NOT EXISTS `dwt`.`{table_name}`
(
    `asin`                  VARCHAR(20)  NOT NULL,
    `bought_month_total`    INT,
    `bought_month_1`        INT,
    `bought_month_2`        INT,
    `bought_month_3`        INT,
    `bought_month_4`        INT,
    `bought_month_5`        INT,
    `bought_month_6`        INT,
    `bought_month_7`        INT,
    `bought_month_8`        INT,
    `bought_month_9`        INT,
    `bought_month_10`       INT,
    `bought_month_11`       INT,
    `bought_month_12`       INT,
    `bought_month_q1`       INT,
    `bought_month_q2`       INT,
    `bought_month_q3`       INT,
    `bought_month_q4`       INT,
    `total_appear_month`    ARRAY<INT>,
    `bought_month_peak`     INT,
    `peak_month_arr`        ARRAY<INT>,
    `is_periodic_flag`      INT,
    `is_seasonal_flag`      INT,
    `bsr_seen_count_total`  INT,
    `nsr_seen_count_total`  INT,
    `variation_bought_month_total`      INT,
    `bought_month_total_yoy`            DECIMAL(20,4),
    `variation_bought_month_total_yoy`  DECIMAL(20,4),
    `variation_bought_month_1`          INT,
    `variation_bought_month_2`          INT,
    `variation_bought_month_3`          INT,
    `variation_bought_month_4`          INT,
    `variation_bought_month_5`          INT,
    `variation_bought_month_6`          INT,
    `variation_bought_month_7`          INT,
    `variation_bought_month_8`          INT,
    `variation_bought_month_9`          INT,
    `variation_bought_month_10`         INT,
    `variation_bought_month_11`         INT,
    `variation_bought_month_12`         INT,
    `variation_bought_month_q1`         INT,
    `variation_bought_month_q2`         INT,
    `variation_bought_month_q3`         INT,
    `variation_bought_month_q4`         INT,
    `variation_bought_month_peak`       INT,
    `variation_peak_month_arr`          ARRAY<INT>
) ENGINE=OLAP
UNIQUE KEY(`asin`)
COMMENT '流量选品年度聚合指标'
DISTRIBUTED BY HASH(`asin`) BUCKETS 8
PROPERTIES (
    "replication_num" = "3",
    "enable_unique_key_merge_on_write" = "true"
)
"""


def sync_extra_table(spark, site_name, date_info):
    """从 Hive dwt_flow_asin_year 对应 site_name+date_info 分区读取年度聚合指标，
    同步到 Doris dwt.{site}_flow_asin_365day_extra（TRUNCATE 后全量重算）
    依赖：dwt/dwt_flow_asin_year.py 必须已经算好并写入该分区"""
    extra_tb = f"{site_name}_flow_asin_365day_extra"

    sql = f"""
        SELECT
            asin, bought_month_total,
            bought_month_1, bought_month_2, bought_month_3, bought_month_4,
            bought_month_5, bought_month_6, bought_month_7, bought_month_8,
            bought_month_9, bought_month_10, bought_month_11, bought_month_12,
            bought_month_q1, bought_month_q2, bought_month_q3, bought_month_q4,
            total_appear_month,
            bought_month_peak,
            peak_month_arr,
            is_periodic_flag, is_seasonal_flag, bsr_seen_count_total, nsr_seen_count_total,
            variation_bought_month_total, bought_month_total_yoy, variation_bought_month_total_yoy,
            variation_bought_month_1, variation_bought_month_2, variation_bought_month_3, variation_bought_month_4,
            variation_bought_month_5, variation_bought_month_6, variation_bought_month_7, variation_bought_month_8,
            variation_bought_month_9, variation_bought_month_10, variation_bought_month_11, variation_bought_month_12,
            variation_bought_month_q1, variation_bought_month_q2, variation_bought_month_q3, variation_bought_month_q4,
            variation_bought_month_peak, variation_peak_month_arr
        FROM dwt_flow_asin_year
        WHERE site_name = '{site_name}' AND date_info = '{date_info}'
    """
    df_raw = spark.sql(sqlQuery=sql)

    # total_appear_month / peak_month_arr / variation_peak_month_arr 在 Hive 那边存的是逗号拼接的 STRING
    df_extra = df_raw.withColumn(
        'total_appear_month', F.split(F.col('total_appear_month'), ',')
    ).withColumn(
        'total_appear_month', F.expr("transform(total_appear_month, x -> cast(x as int))")
    ).withColumn(
        'total_appear_month', F.to_json(F.col('total_appear_month'))
    ).withColumn(
        'peak_month_arr', F.split(F.col('peak_month_arr'), ',')
    ).withColumn(
        'peak_month_arr', F.expr("transform(peak_month_arr, x -> cast(x as int))")
    ).withColumn(
        'peak_month_arr', F.to_json(F.col('peak_month_arr'))
    ).withColumn(
        'variation_peak_month_arr', F.split(F.col('variation_peak_month_arr'), ',')
    ).withColumn(
        'variation_peak_month_arr', F.expr("transform(variation_peak_month_arr, x -> cast(x as int))")
    ).withColumn(
        'variation_peak_month_arr', F.to_json(F.col('variation_peak_month_arr'))
    ).repartition(20, 'asin').cache()

    row_count = df_extra.count()
    print(f"[同步365day_extra] 读取 Hive dwt_flow_asin_year[site_name={site_name}, date_info={date_info}]："
          f"{row_count} 条")
    if row_count == 0:
        raise ValueError(
            f"Hive dwt_flow_asin_year[site_name={site_name}, date_info={date_info}] 读到0条，"
            f"请确认 dwt/dwt_flow_asin_year.py 是否已经跑完这个 site_name+date_info"
        )

    _exec_doris_sql([build_extra_create_table_sql(extra_tb)])
    _exec_doris_sql([f"TRUNCATE TABLE `dwt`.`{extra_tb}`"])

    DorisHelper.spark_export_with_columns(
        df_save=df_extra,
        db_name='dwt',
        table_name=extra_tb,
        table_columns=EXTRA_COLUMNS,
    )
    print(f"[同步365day_extra] 年度聚合指标同步至 dwt.{extra_tb} 完成")


# ============================================================
# [Step 6] selection 月表 INSERT OVERWRITE
# ============================================================

def build_month_insert_overwrite_sql(site_name, table_name, date_info):
    """构建 INSERT OVERWRITE 到 selection.{table_name} 的 SQL；
    table_name 由外层拼接：{site}_flow_asin_month_{yyyy_mm}[_test]
    launch_time_type / tracking_since_type 的天数基准统一用 date_info 当月最后一天"""
    type_anchor_date = f"LAST_DAY(CAST(CONCAT('{date_info}', '-01') AS DATE))"
    return f"""
INSERT OVERWRITE TABLE `selection`.`{table_name}`
SELECT
    f.asin,
    f.parent_asin,
    f.collapse_asin,
    f.asin_crawl_date,
    f.title,
    stem_en(f.title) AS title_stem,
    f.title_len,
    f.brand,
    f.asin_describe,
    f.describe_len,
    f.product_features,
    f.together_asin,
    f.price,
    f.fbm_price,
    f.rating,
    f.total_comments,
    f.one_star, f.two_star, f.three_star, f.four_star, f.five_star, f.low_star,
    f.bsr_orders,
    f.bsr_orders_sale,
    f.asin_bought_month,
    f.ao_val,
    f.zr_counts,
    f.sp_counts, f.sb_counts, f.vi_counts, f.bs_counts,
    f.ac_counts, f.tr_counts, f.er_counts,
    f.zr_flow_proportion,
    f.matrix_flow_proportion,
    f.matrix_ao_val,
    f.one_two_val, f.three_four_val, f.five_six_val, f.eight_val,
    f.category_first_id,
    f.category_id,
    f.first_category_rank,
    f.current_category_rank,
    f.weight,
    f.volume,
    f.asin_weight_ratio,
    f.color, f.size, f.style, f.material,
    COALESCE(uma.package_quantity, f.package_quantity) AS package_quantity,
    f.is_package_quantity_abnormal,
    f.variation_num,
    f.page_inventory,
    f.activity_type,
    COALESCE(f.launch_time, kp.keepa_launch_time) AS launch_time,
    CASE
        WHEN COALESCE(f.launch_time, kp.keepa_launch_time) IS NULL THEN 0
        WHEN DATEDIFF({type_anchor_date}, COALESCE(f.launch_time, kp.keepa_launch_time)) <= 30   THEN 1
        WHEN DATEDIFF({type_anchor_date}, COALESCE(f.launch_time, kp.keepa_launch_time)) <= 90   THEN 2
        WHEN DATEDIFF({type_anchor_date}, COALESCE(f.launch_time, kp.keepa_launch_time)) <= 180  THEN 3
        WHEN DATEDIFF({type_anchor_date}, COALESCE(f.launch_time, kp.keepa_launch_time)) <= 360  THEN 4
        WHEN DATEDIFF({type_anchor_date}, COALESCE(f.launch_time, kp.keepa_launch_time)) <= 720  THEN 5
        WHEN DATEDIFF({type_anchor_date}, COALESCE(f.launch_time, kp.keepa_launch_time)) <= 1080 THEN 6
        ELSE 7
    END AS launch_time_type,
    f.img_url,
    f.img_num,
    f.img_type AS img_type_arr,
    f.img_info,
    f.account_id,
    f.account_name,
    f.buy_box_seller_type,
    f.site_name,
    f.follow_sellers_count,
    f.asin_lob_info,
    f.is_contains_lob_info,
    f.asin_lqs_rating,
    f.asin_lqs_rating_detail,
    f.amazon_label,
    f.is_movie_label,
    f.is_brand_label,
    COALESCE(uma.multi_color_flag, f.multi_color_flag) AS multi_color_flag,
    f.multi_color_str,
    f.rank_type, f.ao_val_type, f.price_type, f.rating_type,
    f.size_type, f.weight_type, f.site_name_type, f.quantity_variation_type,
    f.rank_rise,        f.rank_mom         AS rank_change,         f.rank_yoy,
    f.ao_rise,          f.ao_mom           AS ao_change,           f.ao_yoy,
    f.price_rise,       f.price_mom        AS price_change,        f.price_yoy,
    f.rating_rise,      f.rating_mom       AS rating_change,       f.rating_yoy,
    f.comments_rise,    f.comments_mom     AS comments_change,     f.comments_yoy,
    f.bsr_orders_rise,  f.bsr_orders_mom   AS bsr_orders_change,   f.bsr_orders_yoy,
    f.sales_rise,       f.sales_mom        AS sales_change,        f.sales_yoy,
    f.variation_rise,   f.variation_mom    AS variation_change,    f.variation_yoy,
    f.bought_month_mom, f.bought_month_yoy,
    pr.ocean_profit, pr.air_profit,
    FROM_UNIXTIME((CAST(kp.tracking_since AS BIGINT) + 21564000) * 60) AS tracking_since,
    CASE
        WHEN kp.tracking_since IS NULL OR kp.tracking_since <= 0 THEN 0
        WHEN DATEDIFF({type_anchor_date}, FROM_UNIXTIME((CAST(kp.tracking_since AS BIGINT) + 21564000) * 60)) <= 30   THEN 1
        WHEN DATEDIFF({type_anchor_date}, FROM_UNIXTIME((CAST(kp.tracking_since AS BIGINT) + 21564000) * 60)) <= 90   THEN 2
        WHEN DATEDIFF({type_anchor_date}, FROM_UNIXTIME((CAST(kp.tracking_since AS BIGINT) + 21564000) * 60)) <= 180  THEN 3
        WHEN DATEDIFF({type_anchor_date}, FROM_UNIXTIME((CAST(kp.tracking_since AS BIGINT) + 21564000) * 60)) <= 360  THEN 4
        WHEN DATEDIFF({type_anchor_date}, FROM_UNIXTIME((CAST(kp.tracking_since AS BIGINT) + 21564000) * 60)) <= 720  THEN 5
        WHEN DATEDIFF({type_anchor_date}, FROM_UNIXTIME((CAST(kp.tracking_since AS BIGINT) + 21564000) * 60)) <= 1080 THEN 6
        ELSE 7
    END AS tracking_since_type,
    kp.package_length,
    kp.package_width,
    kp.package_height,
    CASE WHEN kp.item_weight > 0 THEN kp.item_weight ELSE kp.package_weight END AS item_weight,
    f.is_alarm_brand,
    CAST(-1 AS SMALLINT) AS bsr_best_orders_type,
    COALESCE(cf.asin_cate_flag, ARRAY(0))                    AS asin_source_flag,
    COALESCE(cf.bsr_latest_date, CAST('1970-01-01' AS DATE)) AS bsr_last_seen_at,
    COALESCE(cf.bsr_30day_count, 0)                          AS bsr_seen_count_30d,
    COALESCE(cf.nsr_latest_date, CAST('1970-01-01' AS DATE)) AS nsr_last_seen_at,
    COALESCE(cf.nsr_30day_count, 0)                          AS nsr_seen_count_30d,
    f.asin_type,
    uma.usr_mask_progress,
    COALESCE(uma.usr_mask_type, umc.usr_mask_type) AS usr_mask_type,
    COALESCE(f.auctions_num,       0) AS auctions_num,
    COALESCE(f.auctions_num_all,   0) AS auctions_num_all,
    COALESCE(f.skus_num_creat,     0) AS skus_num_creat,
    COALESCE(f.skus_num_creat_all, 0) AS skus_num_creat_all,
    f.title_matching_degree,
    bb.brand_badge_reason,
    ARRAY_JOIN(ARRAY_SLICE(SPLIT_BY_STRING(stem_en(f.title), ' '), 1, 15), ' ') AS title_stem_15,
    addr.seller_province,
    addr.seller_city,
    f.variation_bought_month,
    f.variation_bought_month_mom,
    f.variation_bought_month_yoy
FROM `dwt`.`{site_name}_flow_asin_month` f
LEFT JOIN `dwd`.`dwd_asin_profit_rate_latest` pr
    ON f.asin = pr.asin AND f.price = pr.price AND pr.site_name = '{site_name}'
LEFT JOIN `dwd`.`dwd_keepa_asin_detail` kp
    ON f.asin = kp.asin AND kp.site_name = '{site_name}'
LEFT JOIN `selection`.`user_mask_asin` uma ON f.asin = uma.asin
LEFT JOIN `selection`.`user_mask_category` umc ON f.category_id = umc.category_id
LEFT JOIN (
    SELECT asin, asin_cate_flag, bsr_latest_date, bsr_30day_count, nsr_latest_date, nsr_30day_count
    FROM `dwd`.`dwd_asin_source_flag`
    WHERE site_name = '{site_name}' AND date_type = 'month' AND date_info = '{date_info}'
) cf ON f.asin = cf.asin
LEFT JOIN `dwd`.`dwd_st_brand_badge` bb
    ON f.brand = bb.brand AND bb.site_name = '{site_name}' AND bb.date_info = '{date_info}'
-- ===== 卖家地址（按 site_name 查对应分区） =====
LEFT JOIN (
    SELECT seller_id, fb_country_name, seller_province, seller_city
    FROM `dim`.`dim_seller_address`
    WHERE site_name = '{site_name}'
) addr ON f.account_id = addr.seller_id AND f.site_name = addr.fb_country_name
WHERE f.date_info = '{date_info}'
"""


# ============================================================
# [Step 7] selection 年表建表 + INSERT OVERWRITE
# ============================================================

def build_year_create_table_sql(table_name):
    """构建 selection.{table_name} 建表语句：以 selection 月表（build_month_create_table_sql）
    为基准调整而来——
      剔除月度专属字段：asin_source_flag / bsr_last_seen_at /
                       bsr_seen_count_30d / nsr_last_seen_at / nsr_seen_count_30d
      末尾追加：latest_date_info + dwt.{site}_flow_asin_365day_extra 里的年度聚合字段
    旧表已手动清理，直接 CREATE 新表，不再需要 ALTER 兼容旧表结构"""
    return f"""
CREATE TABLE IF NOT EXISTS `selection`.`{table_name}`
(
    `asin`                        VARCHAR(20)      NOT NULL,
    `parent_asin`                 STRING           NULL,
    `collapse_asin`               STRING           NULL,
    `asin_crawl_date`             DATETIME         NULL,
    `title`                       STRING           NULL,
    `title_stem`                  STRING           NULL,
    `title_len`                   INT              NULL,
    `brand`                       STRING           NULL,
    `asin_describe`               STRING           NULL,
    `describe_len`                INT              NULL,
    `product_features`            STRING           NULL,
    `together_asin`               STRING           NULL,
    `price`                       DECIMAL(20,2)    NULL,
    `fbm_price`                   DECIMAL(20,2)    NULL,
    `rating`                      DECIMAL(10,1)    NULL,
    `total_comments`              INT              NULL,
    `one_star`                    INT              NULL,
    `two_star`                    INT              NULL,
    `three_star`                  INT              NULL,
    `four_star`                   INT              NULL,
    `five_star`                   INT              NULL,
    `low_star`                    INT              NULL,
    `bsr_orders`                  INT              NULL,
    `bsr_orders_sale`             DECIMAL(20,2)    NULL,
    `asin_bought_month`           INT              NULL,
    `ao_val`                      DECIMAL(20,4)    NULL,
    `zr_counts`                   INT              NULL,
    `sp_counts`                   INT              NULL,
    `sb_counts`                   INT              NULL,
    `vi_counts`                   INT              NULL,
    `bs_counts`                   INT              NULL,
    `ac_counts`                   INT              NULL,
    `tr_counts`                   INT              NULL,
    `er_counts`                   INT              NULL,
    `zr_flow_proportion`          DECIMAL(20,4)    NULL,
    `matrix_flow_proportion`      DECIMAL(20,4)    NULL,
    `matrix_ao_val`               DECIMAL(20,4)    NULL,
    `one_two_val`                 DECIMAL(20,4)    NULL,
    `three_four_val`              DECIMAL(20,4)    NULL,
    `five_six_val`                DECIMAL(20,4)    NULL,
    `eight_val`                   DECIMAL(20,4)    NULL,
    `category_first_id`           STRING           NULL,
    `category_id`                 STRING           NULL,
    `first_category_rank`         INT              NULL,
    `current_category_rank`       INT              NULL,
    `weight`                      DECIMAL(20,4)    NULL,
    `volume`                      STRING           NULL,
    `asin_weight_ratio`           DECIMAL(20,4)    NULL,
    `color`                       STRING           NULL,
    `size`                        STRING           NULL,
    `style`                       STRING           NULL,
    `material`                    STRING           NULL,
    `package_quantity`            INT              NULL,
    `is_package_quantity_abnormal` TINYINT         NULL,
    `variation_num`               INT              NULL,
    `page_inventory`              INT              NULL,
    `activity_type`               STRING           NULL,
    `launch_time`                 DATETIME         NULL,
    `launch_time_type`            INT              NULL,
    `img_url`                     STRING           NULL,
    `img_num`                     INT              NULL,
    `img_type_arr`                ARRAY<INT>       NULL,
    `img_info`                    STRING           NULL,
    `account_id`                  STRING           NULL,
    `account_name`                STRING           NULL,
    `buy_box_seller_type`         TINYINT          NULL,
    `site_name`                   STRING           NULL,
    `follow_sellers_count`        INT              NULL,
    `asin_lob_info`               STRING           NULL,
    `is_contains_lob_info`        TINYINT          NULL,
    `asin_lqs_rating`             DECIMAL(20,1)    NULL,
    `asin_lqs_rating_detail`      STRING           NULL,
    `amazon_label`                STRING           NULL,
    `is_movie_label`              TINYINT          NULL,
    `is_brand_label`              TINYINT          NULL,
    `multi_color_flag`            TINYINT          NULL,
    `multi_color_str`             STRING           NULL,
    `rank_type`                   TINYINT          NULL,
    `ao_val_type`                 TINYINT          NULL,
    `price_type`                  TINYINT          NULL,
    `rating_type`                 TINYINT          NULL,
    `size_type`                   TINYINT          NULL,
    `weight_type`                 TINYINT          NULL,
    `site_name_type`              TINYINT          NULL,
    `quantity_variation_type`     TINYINT          NULL,
    `rank_rise`                   INT              NULL,
    `rank_change`                 DECIMAL(20,4)    NULL,
    `rank_yoy`                    DECIMAL(20,4)    NULL,
    `ao_rise`                     DECIMAL(20,4)    NULL,
    `ao_change`                   DECIMAL(20,4)    NULL,
    `ao_yoy`                      DECIMAL(20,4)    NULL,
    `price_rise`                  DECIMAL(20,2)    NULL,
    `price_change`                DECIMAL(20,4)    NULL,
    `price_yoy`                   DECIMAL(20,4)    NULL,
    `rating_rise`                 DECIMAL(20,1)    NULL,
    `rating_change`               DECIMAL(20,4)    NULL,
    `rating_yoy`                  DECIMAL(20,4)    NULL,
    `comments_rise`               INT              NULL,
    `comments_change`             DECIMAL(20,4)    NULL,
    `comments_yoy`                DECIMAL(20,4)    NULL,
    `bsr_orders_rise`             INT              NULL,
    `bsr_orders_change`           DECIMAL(20,4)    NULL,
    `bsr_orders_yoy`              DECIMAL(20,4)    NULL,
    `sales_rise`                  DECIMAL(20,2)    NULL,
    `sales_change`                DECIMAL(20,4)    NULL,
    `sales_yoy`                   DECIMAL(20,4)    NULL,
    `variation_rise`              INT              NULL,
    `variation_change`            DECIMAL(20,4)    NULL,
    `variation_yoy`               DECIMAL(20,4)    NULL,
    `bought_month_mom`            DECIMAL(20,4)    NULL,
    `bought_month_yoy`            DECIMAL(20,4)    NULL,
    `ocean_profit`                DECIMAL(20,4)    NULL,
    `air_profit`                  DECIMAL(20,4)    NULL,
    `tracking_since`              STRING           NULL,
    `tracking_since_type`         INT              NULL,
    `package_length`              INT              NULL,
    `package_width`               INT              NULL,
    `package_height`              INT              NULL,
    `item_weight`                 INT              NULL,
    `is_alarm_brand`              INT              NULL,
    `bsr_best_orders_type`        SMALLINT         NULL,
    `asin_type`                   SMALLINT         NULL,
    `usr_mask_progress`           STRING           NULL,
    `usr_mask_type`               STRING           NULL,
    `auctions_num`                INT              NULL,
    `auctions_num_all`            INT              NULL,
    `skus_num_creat`              INT              NULL,
    `skus_num_creat_all`          INT              NULL,
    `title_matching_degree`       DECIMAL(20,4)    NULL,
    `brand_badge_reason`          STRING           NULL,
    `title_stem_15`               STRING           NULL     COMMENT '标题词干前15个词(按空格分词截取)',
    `seller_province`             STRING           NULL     COMMENT '卖家所在省份(dim_seller_address关联,仅国内卖家有值)',
    `seller_city`                 STRING           NULL     COMMENT '卖家所在城市(dim_seller_address关联,仅国内卖家有值)',
    `latest_date_info`            STRING           NULL     COMMENT '最新出现月份(yyyy-MM)',
    `bought_month_total`          INT              NULL,
    `bought_month_1`              INT              NULL,
    `bought_month_2`              INT              NULL,
    `bought_month_3`              INT              NULL,
    `bought_month_4`              INT              NULL,
    `bought_month_5`              INT              NULL,
    `bought_month_6`              INT              NULL,
    `bought_month_7`              INT              NULL,
    `bought_month_8`              INT              NULL,
    `bought_month_9`              INT              NULL,
    `bought_month_10`             INT              NULL,
    `bought_month_11`             INT              NULL,
    `bought_month_12`             INT              NULL,
    `bought_month_q1`             INT              NULL,
    `bought_month_q2`             INT              NULL,
    `bought_month_q3`             INT              NULL,
    `bought_month_q4`             INT              NULL,
    `total_appear_month`          ARRAY<INT>       NULL,
    `bsr_seen_count_total`        INT              NULL,
    `nsr_seen_count_total`        INT              NULL,
    `bought_month_peak`           INT              NULL,
    `peak_month_arr`              ARRAY<INT>       NULL,
    `is_periodic_flag`            INT              NULL,
    `is_seasonal_flag`            INT              NULL,
    `variation_bought_month_total`      INT              NULL,
    `bought_month_total_yoy`            DECIMAL(20,4)    NULL,
    `variation_bought_month_total_yoy`  DECIMAL(20,4)    NULL,
    `variation_bought_month_1`          INT              NULL,
    `variation_bought_month_2`          INT              NULL,
    `variation_bought_month_3`          INT              NULL,
    `variation_bought_month_4`          INT              NULL,
    `variation_bought_month_5`          INT              NULL,
    `variation_bought_month_6`          INT              NULL,
    `variation_bought_month_7`          INT              NULL,
    `variation_bought_month_8`          INT              NULL,
    `variation_bought_month_9`          INT              NULL,
    `variation_bought_month_10`         INT              NULL,
    `variation_bought_month_11`         INT              NULL,
    `variation_bought_month_12`         INT              NULL,
    `variation_bought_month_q1`         INT              NULL,
    `variation_bought_month_q2`         INT              NULL,
    `variation_bought_month_q3`         INT              NULL,
    `variation_bought_month_q4`         INT              NULL,
    `variation_bought_month_peak`       INT              NULL,
    `variation_peak_month_arr`          ARRAY<INT>       NULL,
    `variation_bought_month`            INT              NULL     COMMENT '变体销量和(当月值，非最新月置空)',
    `variation_bought_month_mom`        DECIMAL(20,4)    NULL     COMMENT '变体销量和环比(非最新月置空)',
    `variation_bought_month_yoy`        DECIMAL(20,4)    NULL     COMMENT '变体销量和同比(非最新月置空)',
    INDEX idx_title      (`title`)      USING INVERTED PROPERTIES("parser" = "english") COMMENT '标题倒排索引',
    INDEX idx_title_stem (`title_stem`) USING INVERTED PROPERTIES("parser" = "english") COMMENT '标题词干倒排索引',
    INDEX idx_title_stem_15 (`title_stem_15`) USING INVERTED PROPERTIES("parser" = "english") COMMENT '标题词干前15个词倒排索引',
    INDEX idx_color      (`color`)      USING INVERTED PROPERTIES("parser" = "english") COMMENT '颜色倒排索引',
    INDEX idx_brand      (`brand`)      USING INVERTED PROPERTIES("parser" = "english") COMMENT '品牌倒排索引',
    INDEX idx_account_name (`account_name`) USING INVERTED PROPERTIES("parser" = "english") COMMENT '卖家名称倒排索引'
) ENGINE=OLAP
UNIQUE KEY(`asin`)
COMMENT '流量选品近 12 个月聚合表'
DISTRIBUTED BY HASH(`asin`) BUCKETS 32
PROPERTIES (
    "replication_num" = "3",
    "enable_unique_key_merge_on_write" = "true"
)
"""


def _year_select_from_join_sql(site_name, latest_date_info):
    """年表 SELECT 主体（不含 INSERT 前缀，不含 WHERE），供整表覆盖 / 分批插入两种场景共用：
    - 主体: dwt.{site}_flow_asin_365day (每 asin 最新月快照,字段名沿用 dwt 月表)
    - 年度聚合: dwt.{site}_flow_asin_365day_extra (Hive 算好经 sync_extra_table 同步来的年度指标,
                含周期性/季节性/峰值)
    - 外层 LEFT JOIN: profit_rate / keepa / user_mask / auction 等
    - launch_time_type / tracking_since_type 的天数基准统一用 latest_date_info（本次跑批最新月份）
      当月的最后一天，而不是各批次自身的 f.asin_crawl_date：dwt.{site}_flow_asin_365day 是按 asin 去重
      只保留最新一次快照的表，长期未被重新抓取的 asin 的 asin_crawl_date 会停留在很久之前，用它做基准
      会导致这两个分档字段冻结在过去、不随时间推进
    :param latest_date_info: main() 的 date_info 参数（yyyy-MM），与分批用的历史 date_info_batch 无关
    """
    type_anchor_date = f"LAST_DAY(CAST(CONCAT('{latest_date_info}', '-01') AS DATE))"
    return f"""
SELECT
    f.asin,
    f.parent_asin,
    f.collapse_asin,
    f.asin_crawl_date,
    f.title,
    stem_en(f.title) AS title_stem,
    f.title_len,
    f.brand,
    f.asin_describe,
    f.describe_len,
    f.product_features,
    f.together_asin,
    f.price,
    f.fbm_price,
    f.rating,
    f.total_comments,
    f.one_star, f.two_star, f.three_star, f.four_star, f.five_star, f.low_star,
    f.bsr_orders,
    f.bsr_orders_sale,
    -- asin本月快照没跟上最新月（latest_date_info != 本次跑批月份）时，月销/环比/同比置空
    CASE WHEN DATE_FORMAT(f.date_info, '%Y-%m') = '{latest_date_info}' THEN f.asin_bought_month ELSE NULL END AS asin_bought_month,
    f.ao_val,
    f.zr_counts,
    f.sp_counts, f.sb_counts, f.vi_counts, f.bs_counts,
    f.ac_counts, f.tr_counts, f.er_counts,
    f.zr_flow_proportion,
    f.matrix_flow_proportion,
    f.matrix_ao_val,
    f.one_two_val, f.three_four_val, f.five_six_val, f.eight_val,
    f.category_first_id,
    f.category_id,
    f.first_category_rank,
    f.current_category_rank,
    f.weight,
    f.volume,
    f.asin_weight_ratio,
    f.color, f.size, f.style, f.material,
    COALESCE(uma.package_quantity, f.package_quantity) AS package_quantity,
    f.is_package_quantity_abnormal,
    f.variation_num,
    f.page_inventory,
    f.activity_type,
    COALESCE(f.launch_time, kp.keepa_launch_time) AS launch_time,
    CASE
        WHEN COALESCE(f.launch_time, kp.keepa_launch_time) IS NULL THEN 0
        WHEN DATEDIFF({type_anchor_date}, COALESCE(f.launch_time, kp.keepa_launch_time)) <= 30   THEN 1
        WHEN DATEDIFF({type_anchor_date}, COALESCE(f.launch_time, kp.keepa_launch_time)) <= 90   THEN 2
        WHEN DATEDIFF({type_anchor_date}, COALESCE(f.launch_time, kp.keepa_launch_time)) <= 180  THEN 3
        WHEN DATEDIFF({type_anchor_date}, COALESCE(f.launch_time, kp.keepa_launch_time)) <= 360  THEN 4
        WHEN DATEDIFF({type_anchor_date}, COALESCE(f.launch_time, kp.keepa_launch_time)) <= 720  THEN 5
        WHEN DATEDIFF({type_anchor_date}, COALESCE(f.launch_time, kp.keepa_launch_time)) <= 1080 THEN 6
        ELSE 7
    END AS launch_time_type,
    f.img_url,
    f.img_num,
    f.img_type AS img_type_arr,
    f.img_info,
    f.account_id,
    f.account_name,
    f.buy_box_seller_type,
    f.site_name,
    f.follow_sellers_count,
    f.asin_lob_info,
    f.is_contains_lob_info,
    f.asin_lqs_rating,
    f.asin_lqs_rating_detail,
    f.amazon_label,
    f.is_movie_label,
    f.is_brand_label,
    COALESCE(uma.multi_color_flag, f.multi_color_flag) AS multi_color_flag,
    f.multi_color_str,
    f.rank_type, f.ao_val_type, f.price_type, f.rating_type,
    f.size_type, f.weight_type, f.site_name_type, f.quantity_variation_type,
    f.rank_rise,        f.rank_mom         AS rank_change,         f.rank_yoy,
    f.ao_rise,          f.ao_mom           AS ao_change,           f.ao_yoy,
    f.price_rise,       f.price_mom        AS price_change,        f.price_yoy,
    f.rating_rise,      f.rating_mom       AS rating_change,       f.rating_yoy,
    f.comments_rise,    f.comments_mom     AS comments_change,     f.comments_yoy,
    f.bsr_orders_rise,  f.bsr_orders_mom   AS bsr_orders_change,   f.bsr_orders_yoy,
    f.sales_rise,       f.sales_mom        AS sales_change,        f.sales_yoy,
    f.variation_rise,   f.variation_mom    AS variation_change,    f.variation_yoy,
    CASE WHEN DATE_FORMAT(f.date_info, '%Y-%m') = '{latest_date_info}' THEN f.bought_month_mom ELSE NULL END AS bought_month_mom,
    CASE WHEN DATE_FORMAT(f.date_info, '%Y-%m') = '{latest_date_info}' THEN f.bought_month_yoy ELSE NULL END AS bought_month_yoy,
    pr.ocean_profit, pr.air_profit,
    FROM_UNIXTIME((CAST(kp.tracking_since AS BIGINT) + 21564000) * 60) AS tracking_since,
    CASE
        WHEN kp.tracking_since IS NULL OR kp.tracking_since <= 0 THEN 0
        WHEN DATEDIFF({type_anchor_date}, FROM_UNIXTIME((CAST(kp.tracking_since AS BIGINT) + 21564000) * 60)) <= 30   THEN 1
        WHEN DATEDIFF({type_anchor_date}, FROM_UNIXTIME((CAST(kp.tracking_since AS BIGINT) + 21564000) * 60)) <= 90   THEN 2
        WHEN DATEDIFF({type_anchor_date}, FROM_UNIXTIME((CAST(kp.tracking_since AS BIGINT) + 21564000) * 60)) <= 180  THEN 3
        WHEN DATEDIFF({type_anchor_date}, FROM_UNIXTIME((CAST(kp.tracking_since AS BIGINT) + 21564000) * 60)) <= 360  THEN 4
        WHEN DATEDIFF({type_anchor_date}, FROM_UNIXTIME((CAST(kp.tracking_since AS BIGINT) + 21564000) * 60)) <= 720  THEN 5
        WHEN DATEDIFF({type_anchor_date}, FROM_UNIXTIME((CAST(kp.tracking_since AS BIGINT) + 21564000) * 60)) <= 1080 THEN 6
        ELSE 7
    END AS tracking_since_type,
    kp.package_length,
    kp.package_width,
    kp.package_height,
    CASE WHEN kp.item_weight > 0 THEN kp.item_weight ELSE kp.package_weight END AS item_weight,
    f.is_alarm_brand,
    CAST(-1 AS SMALLINT) AS bsr_best_orders_type,
    f.asin_type,
    uma.usr_mask_progress,
    COALESCE(uma.usr_mask_type, umc.usr_mask_type) AS usr_mask_type,
    COALESCE(f.auctions_num,       0) AS auctions_num,
    COALESCE(f.auctions_num_all,   0) AS auctions_num_all,
    COALESCE(f.skus_num_creat,     0) AS skus_num_creat,
    COALESCE(f.skus_num_creat_all, 0) AS skus_num_creat_all,
    f.title_matching_degree,
    bb.brand_badge_reason,
    ARRAY_JOIN(ARRAY_SLICE(SPLIT_BY_STRING(stem_en(f.title), ' '), 1, 15), ' ') AS title_stem_15,
    addr.seller_province,
    addr.seller_city,
    -- ===== 年度聚合（dwt/dwt_flow_asin_year.py 在 Hive 预计算，经 sync_extra_table 同步到 Doris）=====
    DATE_FORMAT(f.date_info, '%Y-%m') AS latest_date_info,
    COALESCE(agg.bought_month_total, 0) AS bought_month_total,
    agg.bought_month_1, agg.bought_month_2,  agg.bought_month_3,  agg.bought_month_4,
    agg.bought_month_5, agg.bought_month_6,  agg.bought_month_7,  agg.bought_month_8,
    agg.bought_month_9, agg.bought_month_10, agg.bought_month_11, agg.bought_month_12,
    agg.bought_month_q1, agg.bought_month_q2, agg.bought_month_q3, agg.bought_month_q4,
    agg.total_appear_month,
    COALESCE(agg.bsr_seen_count_total, 0) AS bsr_seen_count_total,
    COALESCE(agg.nsr_seen_count_total, 0) AS nsr_seen_count_total,
    agg.bought_month_peak,
    agg.peak_month_arr,
    COALESCE(uma.is_periodic_flag, agg.is_periodic_flag, 0) AS is_periodic_flag,
    COALESCE(uma.is_seasonal_flag, agg.is_seasonal_flag, 0) AS is_seasonal_flag,
    COALESCE(agg.variation_bought_month_total, 0) AS variation_bought_month_total,
    agg.bought_month_total_yoy,
    agg.variation_bought_month_total_yoy,
    agg.variation_bought_month_1, agg.variation_bought_month_2,  agg.variation_bought_month_3,  agg.variation_bought_month_4,
    agg.variation_bought_month_5, agg.variation_bought_month_6,  agg.variation_bought_month_7,  agg.variation_bought_month_8,
    agg.variation_bought_month_9, agg.variation_bought_month_10, agg.variation_bought_month_11, agg.variation_bought_month_12,
    agg.variation_bought_month_q1, agg.variation_bought_month_q2, agg.variation_bought_month_q3, agg.variation_bought_month_q4,
    agg.variation_bought_month_peak,
    agg.variation_peak_month_arr,
    -- asin本月快照没跟上最新月（latest_date_info != 本次跑批月份）时，变体销量和/环比/同比置空，跟asin_bought_month同理
    CASE WHEN DATE_FORMAT(f.date_info, '%Y-%m') = '{latest_date_info}' THEN f.variation_bought_month ELSE NULL END AS variation_bought_month,
    CASE WHEN DATE_FORMAT(f.date_info, '%Y-%m') = '{latest_date_info}' THEN f.variation_bought_month_mom ELSE NULL END AS variation_bought_month_mom,
    CASE WHEN DATE_FORMAT(f.date_info, '%Y-%m') = '{latest_date_info}' THEN f.variation_bought_month_yoy ELSE NULL END AS variation_bought_month_yoy
FROM `dwt`.`{site_name}_flow_asin_365day` f
LEFT JOIN `dwt`.`{site_name}_flow_asin_365day_extra` agg ON f.asin = agg.asin
-- ===== 利润率 =====
LEFT JOIN `dwd`.`dwd_asin_profit_rate_latest` pr
    ON f.asin = pr.asin AND f.price = pr.price AND pr.site_name = '{site_name}'
-- ===== Keepa 详情 =====
LEFT JOIN `dwd`.`dwd_keepa_asin_detail` kp
    ON f.asin = kp.asin AND kp.site_name = '{site_name}'
-- ===== 用户标记 ASIN/品类 =====
LEFT JOIN `selection`.`user_mask_asin`     uma ON f.asin        = uma.asin
LEFT JOIN `selection`.`user_mask_category` umc ON f.category_id = umc.category_id
-- ===== 品牌推荐原因——
LEFT JOIN (
    SELECT brand, brand_badge_reason
    FROM `dwd`.`dwd_st_brand_badge`
    WHERE site_name = '{site_name}'
      AND date_info = (SELECT MAX(date_info) FROM `dwd`.`dwd_st_brand_badge` WHERE site_name = '{site_name}')
) bb ON f.brand = bb.brand
-- ===== 卖家地址（按 site_name 查对应分区） =====
LEFT JOIN (
    SELECT seller_id, fb_country_name, seller_province, seller_city
    FROM `dim`.`dim_seller_address`
    WHERE site_name = '{site_name}'
) addr ON f.account_id = addr.seller_id AND f.site_name = addr.fb_country_name
"""


def build_year_insert_overwrite_sql(site_name, table_name, latest_date_info):
    """整表覆盖版（非分批）：保留供手工排查/小站点场景使用，正式流程走下面的分批版本
    write_year_table_batched，不再直接调用这个函数"""
    return f"INSERT OVERWRITE TABLE `selection`.`{table_name}`\n{_year_select_from_join_sql(site_name, latest_date_info)}"


def build_year_batch_insert_sql(site_name, table_name, date_info_batch, latest_date_info):
    """构造单个 date_info 批次的 INSERT INTO SQL
    :param date_info_batch: dwt.{site}_flow_asin_365day 里的某个 date_info 取值（DATE，如 '2026-05-01'）
    :param latest_date_info: main() 的 date_info 参数（yyyy-MM，本次跑批最新月份），传给
        _year_select_from_join_sql 做 launch_time_type/tracking_since_type 的统一天数基准，
        与 date_info_batch（历史批次键）是两个不同的概念
    """
    return (
        f"INSERT INTO `selection`.`{table_name}`\n"
        f"{_year_select_from_join_sql(site_name, latest_date_info)}"
        f"WHERE f.date_info = '{date_info_batch}'\n"
    )


def get_year_batch_date_infos(site_name):
    """获取 dwt.{site_name}_flow_asin_365day 当前实际存在的 date_info 取值（DATE，如
    '2026-05-01'），升序排列，作为年表分批 INSERT 的批次键（近12个月窗口，最多约12批）"""
    stg_table = f"{site_name}_flow_asin_365day"
    rows = _query_doris(f"SELECT DISTINCT date_info FROM `dwt`.`{stg_table}` ORDER BY date_info")
    return [str(row[0]) for row in rows]


def write_year_table_batched(site_name, live_table, latest_date_info):
    """按 date_info 分批写入 selection 年表，避免一次性 13 路 JOIN 大 SQL 导致的内存爆炸/超时：
       1) DROP + 重建 copy 表（保证 schema 跟当前 DDL 一致）
       2) 按 dwt.{site}_flow_asin_365day 现有的 date_info 取值逐批 INSERT INTO copy 表
       3) 数据量校验：copy 表最终行数 应等于 dwt.{site}_flow_asin_365day 原表行数——驱动表
       4) 校验通过后 ALTER TABLE ... REPLACE WITH TABLE ... PROPERTIES('swap'='true') 原子
          切换成正式表，copy 表则拿到正式表交换前的旧数据（可回滚，不会立刻销毁）
    :param latest_date_info: main() 的 date_info 参数（yyyy-MM，本次跑批最新月份），所有批次统一用它
        算 launch_time_type/tracking_since_type 的天数基准（次月初），而不是各批次自身的历史 date_info_batch
    """
    copy_table = f"{live_table}_copy"
    stg_table  = f"{site_name}_flow_asin_365day"

    print(f"[Step 7] 重建 copy 表 selection.{copy_table}")
    _exec_doris_sql([f"DROP TABLE IF EXISTS `selection`.`{copy_table}`"])
    _exec_doris_sql([build_year_create_table_sql(copy_table)])

    date_infos = get_year_batch_date_infos(site_name)
    print(f"[Step 7] 待分批写入 selection.{copy_table} 的 date_info 批次（共 {len(date_infos)} 批）：{date_infos}")
    for date_info_batch in date_infos:
        print(f"[Step 7] INSERT INTO selection.{copy_table} WHERE f.date_info = '{date_info_batch}'")
        _exec_doris_sql([build_year_batch_insert_sql(site_name, copy_table, date_info_batch, latest_date_info)])

    dwt_count  = _query_doris(f"SELECT COUNT(1) FROM `dwt`.`{stg_table}`")[0][0]
    copy_count = _query_doris(f"SELECT COUNT(1) FROM `selection`.`{copy_table}`")[0][0]
    print(f"[Step 7] 数据量校验：dwt.{stg_table}={dwt_count}，selection.{copy_table}={copy_count}")
    if copy_count == 0:
        # 兜底：即使 dwt_count 恰好也是0导致下面的相等校验通过，也绝不能把空表swap成正式表
        raise ValueError(
            f"[Step 7] copy 表 selection.{copy_table} 写入后行数为0，已中止交换，"
            f"避免把正式表 selection.{live_table} 清空，请检查 dwt.{stg_table} 是否为空或分批写入是否异常"
        )
    if copy_count != dwt_count:
        raise ValueError(
            f"[Step 7] copy 表行数({copy_count})与 dwt.{stg_table} 原表行数({dwt_count})不一致，"
            f"疑似分批写入有遗漏/重复，已中止交换，请人工排查（copy 表已保留，不会自动清理）"
        )

    print(f"[Step 7] 确保正式表 selection.{live_table} 存在")
    _exec_doris_sql([build_year_create_table_sql(live_table)])

    print(f"[Step 7] REPLACE WITH TABLE：selection.{live_table} <-> selection.{copy_table}")
    # 注意：REPLACE WITH TABLE 后面的表名不能带库名前缀，只能是裸表名（同库），
    # 带上 `selection`. 前缀会导致 Doris 解析报错 "mismatched input '.'"
    _exec_doris_sql([
        f"ALTER TABLE `selection`.`{live_table}` REPLACE WITH TABLE `{copy_table}` "
        f"PROPERTIES('swap' = 'false')"
    ])
    print(f"[Step 7] selection.{live_table} 已切换为最新数据")


# ============================================================
# [Step 8] 流程记录表更新（月+年各写一条，仅 formal 模式）
# ============================================================

def modify_mission_record_status(site_name, date_info, result_type):
    """流程记录表更新：仅 formal 模式才入库 mysql workflow_everyday，月/年各写一条记录"""
    if result_type != 'formal':
        print(f"[Step 8] result_type={result_type}，跳过流程记录表更新")
        return
    record_table = 'workflow_everyday'
    engine_mysql = DBUtil.get_db_engine(db_type=DbTypes.mysql.name, site_name='us')

    def _write_record(date_type, table_name_field, status_remark):
        select_sql = (
            f"select id from {record_table} where site_name='{site_name}' and date_type='{date_type}' "
            f"and report_date='{date_info}' and page='流量选品' and status_val=14 and is_end='是'"
        )
        df_is_finished = pd.read_sql(select_sql, engine_mysql)
        if df_is_finished.empty:
            replace_sql = f"""
                replace into {record_table} (site_name, report_date, status, status_val, table_name, date_type, page, is_end, remark, export_db_type)
                VALUES ('{site_name}', '{date_info}', '{status_remark}', 14, '{table_name_field}', '{date_type}', '流量选品', '是', '{status_remark}', 'doris')
            """
            DBUtil.exec_sql('mysql', 'us', replace_sql)
            print(f"[Step 8] 流程记录表 workflow_everyday 已写入：{site_name} {date_info} date_type={date_type}")
        else:
            print(f"[Step 8] 流程记录表已存在该记录，跳过：date_type={date_type}")

    _write_record('month', f'{site_name}_flow_asin_last_month', '流量选品计算完毕')
    _write_record('365_day', f'{site_name}_flow_asin_365day', '流量选品近一年计算完毕')


def main(site_name, date_info, result_type='formal'):
    assert site_name in SUPPORTED_SITES, f"不支持的站点：{site_name}，仅支持 us/uk/de"
    assert result_type in ('formal', 'test'), f"不支持的 result_type：{result_type}，仅支持 formal/test"

    doris_table            = f"{site_name}_flow_asin_month"   # dwt 主表，不区分 test/formal
    date_info_underscore   = date_info.replace('-', '_')
    env_suffix             = '_test' if result_type == 'test' else ''
    month_selection_table  = f"{site_name}_flow_asin_month_{date_info_underscore}{env_suffix}"
    year_selection_table   = f"{site_name}_flow_asin_365day{env_suffix}"
    last_12_month          = sorted(CommonUtil.get_month_offset(date_info, -i) for i in range(0, 12))

    print(f"启动：site={site_name}, date_info={date_info}, result_type={result_type}")
    print(f"  dwt 主表：dwt.{doris_table}（不区分 test/formal）")
    print(f"  selection 月物化表：selection.{month_selection_table}")
    print(f"  selection 年表：selection.{year_selection_table}")
    print(f"  近12月窗口：{last_12_month}")

    spark = SparkUtil.get_spark_session(
        f"DwtFlowAsinMonth: {site_name} {date_info} {result_type}"
    )

    # ===== [Step 1] Doris 建表 selection.{month_selection_table} =====
    print(f"[Step 1] Doris 建表 selection.{month_selection_table}")
    _exec_doris_sql([build_month_create_table_sql(month_selection_table, date_info)])

    # ===== [Step 2] 读 Hive dwt_flow_asin 月数据 + 字段规范化 =====
    print(f"[Step 2] 读 Hive dwt_flow_asin 月数据 + 字段规范化")
    df_save = read_and_normalize_month_data(spark, site_name, date_info)

    # ===== [Step 3] 写入 Doris dwt 主表 =====
    write_dwt_month_table(df_save, doris_table)

    # ===== [Step 4] 写入 Doris dwt 年表 dwt.{site}_flow_asin_365day（自动聚合+清理过期数据）=====
    print(f"[Step 4] 写入 Doris dwt 年表 dwt.{site_name}_flow_asin_365day")
    _exec_doris_sql([build_stg_create_table_sql(site_name)])
    maintain_stg_table(site_name, last_12_month, date_info)

    # ===== [Step 5] 同步年度聚合指标 dwt.{site}_flow_asin_365day_extra =====
    print(f"[Step 5] 同步年度聚合指标 dwt.{site_name}_flow_asin_365day_extra")
    sync_extra_table(spark, site_name, date_info)

    # ===== [Step 6] Doris INSERT OVERWRITE 到 selection 月物化表 =====
    print(f"[Step 6] Doris INSERT OVERWRITE selection.{month_selection_table}")
    _exec_doris_sql([build_month_insert_overwrite_sql(site_name, month_selection_table, date_info)])

    # ===== [Step 7] 按 date_info 分批写入 selection 年表的 copy 表，再原子交换成正式表 =====
    print(f"[Step 7] 分批写入 selection.{year_selection_table} 的 copy 表并原子交换")
    write_year_table_batched(site_name, year_selection_table, date_info)

    # ===== [Step 8] 流程记录表更新（月+年各一条，仅 formal 模式）=====
    modify_mission_record_status(site_name, date_info, result_type)

    print("success!")


if __name__ == "__main__":
    site_name = sys.argv[1]
    date_info = sys.argv[2]
    result_type = sys.argv[3] if len(sys.argv) > 3 else 'formal'
    main(site_name, date_info, result_type)
