流量选品-周流程计算导出

Pyspark_job/doris_handle/dwt_flow_asin_week.py

+"""
+   @Author      : CT
+   @Description : 周流量选品 Doris 落地脚本
+                  流程：
+                    [Step 1] Doris 建表 dwt.{site}_flow_asin_week、selection.{site}_flow_asin_week_{yyyy_ww}[_test]
+                    [Step 2] 读 Hive dwt_flow_asin_week 周数据 + 字段规范化（字段名与月表 dwt.{site}_flow_asin_month 对齐）
+                    [Step 3] 写入 Doris dwt 周表 dwt.{site}_flow_asin_week
+                    [Step 4] Doris INSERT OVERWRITE 到 selection 周物化表 selection.{site}_flow_asin_week_{yyyy_ww}[_test]
+                    [Step 5] 更新 MySQL workflow_everyday 流程记录表（仅 formal 模式）
+                  支持 us / uk / de 三站点
+                  支持 formal / test 模式：
+                    - formal：selection 表名无后缀，更新流程记录表
+                    - test  ：selection 表名加 _test 后缀，不更新流程记录表（dwt 主表不变）
+执行示例:
+    spark-submit dwt_flow_asin_week.py us 2026-41            # 默认 formal
+    spark-submit dwt_flow_asin_week.py us 2026-41 formal
+    spark-submit dwt_flow_asin_week.py us 2026-41 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.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 get_week_end_date(spark, date_info):
+    """获取本周最后一天，作为上架时间/keepa追踪时间类型的天数基准"""
+    sql = f"SELECT max(`date`) AS week_end_date FROM dim_date_20_to_30 WHERE year_week = '{date_info}'"
+    week_end_date = spark.sql(sql).collect()[0]['week_end_date']
+    assert week_end_date is not None, f"dim_date_20_to_30 中不存在 year_week={date_info}"
+    return str(week_end_date)
+
+
+# ============================================================
+# [Step 1] Doris 建表
+# ============================================================
+
+def build_dwt_week_create_table_sql(site_name):
+    """构建 dwt.{site}_flow_asin_week 建表语句：字段与 dwt.{site}_flow_asin_month 一致，周维度独有字段追加在末尾"""
+    return f"""
+CREATE TABLE IF NOT EXISTS `dwt`.`{site_name}_flow_asin_week`
+(
+    `date_info`                   VARCHAR(20)      NOT NULL COMMENT '年-周(yyyy-ww)',
+    `asin`                        VARCHAR(20)      NOT NULL COMMENT 'ASIN',
+    `ao_val`                      DECIMAL(20,4)    COMMENT 'AO值',
+    `zr_counts`                   INT              COMMENT '自然词数量',
+    `sp_counts`                   INT              COMMENT 'SP广告词数量',
+    `sb_counts`                   INT              COMMENT 'SB广告词数量',
+    `vi_counts`                   INT              COMMENT '视频广告词数量',
+    `bs_counts`                   INT              COMMENT 'BS广告词数量',
+    `ac_counts`                   INT              COMMENT 'AC广告词数量',
+    `tr_counts`                   INT              COMMENT 'TR广告词数量',
+    `er_counts`                   INT              COMMENT 'ER广告词数量',
+    `bsr_orders`                  INT              COMMENT 'BSR月销量估算(取最新month_week)',
+    `bsr_orders_sale`             DECIMAL(20,2)    COMMENT 'BSR月销售额估算',
+    `title`                       STRING           COMMENT '标题(lowercase)',
+    `title_len`                   INT              COMMENT '标题字符长度',
+    `price`                       DECIMAL(20,2)    COMMENT '价格',
+    `rating`                      DECIMAL(10,1)    COMMENT '评分',
+    `total_comments`              INT              COMMENT '评论总数',
+    `buy_box_seller_type`         TINYINT          COMMENT '买家盒卖家类型',
+    `page_inventory`              INT              COMMENT '页面显示库存',
+    `volume`                      STRING           COMMENT '体积描述原文',
+    `weight`                      DECIMAL(20,4)    COMMENT '重量',
+    `color`                       STRING           COMMENT '颜色变体属性',
+    `size`                        STRING           COMMENT '尺寸变体属性',
+    `style`                       STRING           COMMENT '款式变体属性',
+    `material`                    STRING           COMMENT '材质',
+    `launch_time`                 DATETIME         COMMENT '上架时间',
+    `img_num`                     INT              COMMENT '图片数量',
+    `parent_asin`                 STRING           COMMENT '父ASIN',
+    `img_type`                    ARRAY<INT>       COMMENT '图片类型数组',
+    `img_url`                     STRING           COMMENT '主图URL',
+    `activity_type`               STRING           COMMENT '活动类型',
+    `one_two_val`                 DECIMAL(20,4)    DEFAULT "0.0000" COMMENT '1-2位流量占比',
+    `three_four_val`              DECIMAL(20,4)    DEFAULT "0.0000" COMMENT '3-4位流量占比',
+    `five_six_val`                DECIMAL(20,4)    DEFAULT "0.0000" COMMENT '5-6位流量占比',
+    `eight_val`                   DECIMAL(20,4)    DEFAULT "0.0000" COMMENT '8位以上流量占比',
+    `brand`                       STRING           COMMENT '品牌(lowercase)',
+    `variation_num`               INT              COMMENT '变体总数量',
+    `one_star`                    INT              COMMENT '1星评论数',
+    `two_star`                    INT              COMMENT '2星评论数',
+    `three_star`                  INT              COMMENT '3星评论数',
+    `four_star`                   INT              COMMENT '4星评论数',
+    `five_star`                   INT              COMMENT '5星评论数',
+    `low_star`                    INT              COMMENT '低评分评论数',
+    `together_asin`               STRING           COMMENT '组合购买推荐ASIN',
+    `account_name`                STRING           COMMENT '卖家名称',
+    `account_id`                  STRING           COMMENT '卖家ID',
+    `rank_rise`                   INT              COMMENT '排名差值(对比上周)',
+    `rank_mom`                    DECIMAL(20,4)    COMMENT '排名近30天环比(对比4周前)',
+    `rank_yoy`                    DECIMAL(20,4)    COMMENT '排名同比(去年同周)',
+    `ao_rise`                     DECIMAL(20,4)    COMMENT 'AO值差值(对比上周)',
+    `ao_mom`                      DECIMAL(20,4)    COMMENT 'AO值近30天环比',
+    `ao_yoy`                      DECIMAL(20,4)    COMMENT 'AO值同比',
+    `price_rise`                  DECIMAL(20,2)    COMMENT '价格差值(对比上周)',
+    `price_mom`                   DECIMAL(20,4)    COMMENT '价格近30天环比',
+    `price_yoy`                   DECIMAL(20,4)    COMMENT '价格同比',
+    `rating_rise`                 DECIMAL(20,1)    COMMENT '评分差值(对比上周)',
+    `rating_mom`                  DECIMAL(20,4)    COMMENT '评分近30天环比',
+    `rating_yoy`                  DECIMAL(20,4)    COMMENT '评分同比',
+    `comments_rise`               INT              COMMENT '评论数差值(对比上周)',
+    `comments_mom`                DECIMAL(20,4)    COMMENT '评论数近30天环比',
+    `comments_yoy`                DECIMAL(20,4)    COMMENT '评论数同比',
+    `bsr_orders_rise`             INT              COMMENT 'BSR销量差值(对比上周)',
+    `bsr_orders_mom`              DECIMAL(20,4)    COMMENT 'BSR销量近30天环比',
+    `bsr_orders_yoy`              DECIMAL(20,4)    COMMENT 'BSR销量同比',
+    `sales_rise`                  DECIMAL(20,2)    COMMENT '销售额差值(对比上周)',
+    `sales_mom`                   DECIMAL(20,4)    COMMENT '销售额近30天环比',
+    `sales_yoy`                   DECIMAL(20,4)    COMMENT '销售额同比',
+    `variation_rise`              INT              COMMENT '变体数差值(对比上周)',
+    `variation_mom`               DECIMAL(20,4)    COMMENT '变体数近30天环比',
+    `variation_yoy`               DECIMAL(20,4)    COMMENT '变体数同比',
+    `bought_month_mom`            DECIMAL(20,4)    COMMENT '月销近30天环比',
+    `bought_month_yoy`            DECIMAL(20,4)    COMMENT '月销同比',
+    `size_type`                   TINYINT          COMMENT '尺寸类型',
+    `rating_type`                 TINYINT          COMMENT '评分类型',
+    `site_name_type`              TINYINT          COMMENT '卖家地类型',
+    `weight_type`                 TINYINT          COMMENT '重量类型',
+    `ao_val_type`                 TINYINT          COMMENT 'AO值类型',
+    `rank_type`                   TINYINT          COMMENT 'BSR排名类型',
+    `price_type`                  TINYINT          COMMENT '价格类型',
+    `quantity_variation_type`     TINYINT          DEFAULT "0" COMMENT '是否数量变体',
+    `package_quantity`            INT              DEFAULT "1" COMMENT '包装数量',
+    `is_movie_label`              TINYINT          DEFAULT "0" COMMENT '是否影视/书籍类(周维度固定0)',
+    `is_brand_label`              TINYINT          DEFAULT "0" COMMENT '是否有品牌',
+    `is_alarm_brand`              INT              DEFAULT "0" COMMENT '是否预警品牌',
+    `asin_crawl_date`             DATETIME         COMMENT '抓取时间',
+    `category_first_id`           STRING           COMMENT 'BSR一级分类ID',
+    `category_id`                 STRING           COMMENT 'BSR当前分类ID',
+    `desc_category_first_id`      STRING           COMMENT 'category文本截取后匹配得到的一级分类ID',
+    `first_category_rank`         INT              COMMENT 'BSR一级分类排名',
+    `current_category_rank`       INT              COMMENT 'BSR当前分类排名',
+    `asin_weight_ratio`           DECIMAL(20,4)    COMMENT '体积重/毛重比',
+    `site_name`                   STRING           COMMENT '卖家所在国家名称(seller_country_name)',
+    `asin_type`                   SMALLINT         COMMENT 'ASIN类型(0默认/1内部/2书籍影视/3隐藏)',
+    `asin_bought_month`           INT              COMMENT 'Amazon近30天购买量',
+    `asin_lqs_rating`             DECIMAL(20,1)    COMMENT 'LQS评分',
+    `asin_lqs_rating_detail`      STRING           COMMENT 'LQS各维度明细(JSON)',
+    `asin_lob_info`               STRING           COMMENT 'Bundle捆绑ASIN列表',
+    `is_contains_lob_info`        TINYINT          COMMENT '是否含Bundle信息',
+    `is_package_quantity_abnormal` TINYINT         COMMENT '包装数量是否异常',
+    `zr_flow_proportion`          DECIMAL(20,4)    COMMENT '自然流量占比',
+    `matrix_flow_proportion`      DECIMAL(20,4)    COMMENT '母体自然流量占比',
+    `matrix_ao_val`               DECIMAL(20,4)    COMMENT '母体AO值',
+    `product_features`            STRING           COMMENT '产品评论/特色JSON',
+    `img_info`                    STRING           COMMENT '图片详情列表JSON',
+    `collapse_asin`               STRING           COMMENT '聚合ASIN',
+    `follow_sellers_count`        INT              DEFAULT "-1" COMMENT '跟卖数(-1=无数据)',
+    `asin_describe`               STRING           COMMENT '五点描述',
+    `fbm_price`                   DECIMAL(20,2)    COMMENT 'FBM配送费',
+    `describe_len`                INT              DEFAULT "0" COMMENT '五点描述总字符长度',
+    `title_matching_degree`       DECIMAL(20,4)    COMMENT '标题匹配度(周维度固定0)',
+    `multi_color_flag`            TINYINT          DEFAULT "0" COMMENT '多色标识',
+    `multi_color_str`             STRING           COMMENT '识别到的颜色组合字符串',
+    `amazon_label`                STRING           COMMENT 'Amazon标签',
+    `auctions_num`                INT              COMMENT 'ASIN竞卖数',
+    `auctions_num_all`            INT              COMMENT '父asin累计竞卖数',
+    `skus_num_creat`              INT              COMMENT 'ASIN创建SKU数',
+    `skus_num_creat_all`          INT              COMMENT '父asin累计创建SKU数',
+    `variation_bought_month`      INT              COMMENT '变体销量和',
+    `variation_bought_month_mom`  DECIMAL(20,4)    COMMENT '变体销量和近30天环比',
+    `variation_bought_month_yoy`  DECIMAL(20,4)    COMMENT '变体销量和同比',
+    `asin_source_flag`            ARRAY<INT>       COMMENT 'ASIN来源标识(本周7天)',
+    `bsr_last_seen_at`            DATE             COMMENT '本周最后上BSR日期',
+    `bsr_seen_count_30d`          INT              COMMENT '本周上BSR天数',
+    `nsr_last_seen_at`            DATE             COMMENT '本周最后上NSR日期',
+    `nsr_seen_count_30d`          INT              COMMENT '本周上NSR天数',
+    `rank_wow`                    DECIMAL(20,4)    COMMENT '排名近7天环比(对比上周)',
+    `ao_wow`                      DECIMAL(20,4)    COMMENT 'AO值近7天环比',
+    `price_wow`                   DECIMAL(20,4)    COMMENT '价格近7天环比',
+    `rating_wow`                  DECIMAL(20,4)    COMMENT '评分近7天环比',
+    `comments_wow`                DECIMAL(20,4)    COMMENT '评论数近7天环比',
+    `bsr_orders_wow`              DECIMAL(20,4)    COMMENT 'BSR销量近7天环比',
+    `sales_wow`                   DECIMAL(20,4)    COMMENT '销售额近7天环比',
+    `variation_wow`               DECIMAL(20,4)    COMMENT '变体数近7天环比',
+    `bought_month_wow`            DECIMAL(20,4)    COMMENT '月销近7天环比',
+    `variation_bought_month_wow`  DECIMAL(20,4)    COMMENT '变体销量和近7天环比',
+    `bought_month_wow_last_1w`    DECIMAL(20,4)    COMMENT '前1周月销近7天环比',
+    `bought_month_wow_last_2w`    DECIMAL(20,4)    COMMENT '前2周月销近7天环比',
+    `bought_month_wow_last_3w`    DECIMAL(20,4)    COMMENT '前3周月销近7天环比',
+    `bought_month_rise`           INT              COMMENT '月销差值(对比上周)',
+    `variation_bought_month_rise` INT              COMMENT '变体销量和差值(对比上周)'
+)
+UNIQUE KEY(`date_info`, `asin`)
+COMMENT '流量选品周维度主表'
+AUTO PARTITION BY LIST(`date_info`) ()
+DISTRIBUTED BY HASH(`asin`) BUCKETS 32
+PROPERTIES (
+    "replication_num" = "3",
+    "enable_unique_key_merge_on_write" = "true"
+)
+"""
+
+
+def build_week_create_table_sql(table_name, date_info):
+    """构建 selection.{table_name} 建表语句：字段与 selection 月物化表一致，周维度独有字段追加在末尾；
+    table_name 由外层拼接：{site}_flow_asin_week_{yyyy_ww}[_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 '抓取时间',
+    `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     COMMENT '排名差值(对比上周)',
+    `rank_change`                 DECIMAL(20,4)    NULL     COMMENT '排名近30天环比(对比4周前)',
+    `rank_yoy`                    DECIMAL(20,4)    NULL     COMMENT '排名同比(去年同周)',
+    `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     COMMENT '本周上BSR天数',
+    `nsr_last_seen_at`            DATE             NULL,
+    `nsr_seen_count_30d`          INT              NULL     COMMENT '本周上NSR天数',
+    `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 '变体销量和',
+    `variation_bought_month_mom`  DECIMAL(20,4)    NULL     COMMENT '变体销量和近30天环比',
+    `variation_bought_month_yoy`  DECIMAL(20,4)    NULL     COMMENT '变体销量和同比',
+    `keepa_type`                  STRING           NULL     COMMENT 'Keepa type（小写、_转-）',
+    `keepa_item_type`             STRING           NULL     COMMENT 'Keepa itemTypeKeyword（小写、_转-）',
+    `keepa_type_cn`               STRING           NULL     COMMENT 'keepa_type 中文翻译',
+    `keepa_item_type_cn`          STRING           NULL     COMMENT 'keepa_item_type 中文翻译',
+    `rank_wow`                    DECIMAL(20,4)    NULL     COMMENT '排名近7天环比(对比上周)',
+    `ao_wow`                      DECIMAL(20,4)    NULL     COMMENT 'AO值近7天环比',
+    `price_wow`                   DECIMAL(20,4)    NULL     COMMENT '价格近7天环比',
+    `rating_wow`                  DECIMAL(20,4)    NULL     COMMENT '评分近7天环比',
+    `comments_wow`                DECIMAL(20,4)    NULL     COMMENT '评论数近7天环比',
+    `bsr_orders_wow`              DECIMAL(20,4)    NULL     COMMENT 'BSR销量近7天环比',
+    `sales_wow`                   DECIMAL(20,4)    NULL     COMMENT '销售额近7天环比',
+    `variation_wow`               DECIMAL(20,4)    NULL     COMMENT '变体数近7天环比',
+    `bought_month_wow`            DECIMAL(20,4)    NULL     COMMENT '月销近7天环比',
+    `variation_bought_month_wow`  DECIMAL(20,4)    NULL     COMMENT '变体销量和近7天环比',
+    `bought_month_wow_last_1w`    DECIMAL(20,4)    NULL     COMMENT '前1周月销近7天环比',
+    `bought_month_wow_last_2w`    DECIMAL(20,4)    NULL     COMMENT '前2周月销近7天环比',
+    `bought_month_wow_last_3w`    DECIMAL(20,4)    NULL     COMMENT '前3周月销近7天环比',
+    `bought_month_rise`           INT              NULL     COMMENT '月销差值(对比上周)',
+    `variation_bought_month_rise` INT              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 8
+PROPERTIES (
+    "replication_num" = "3",
+    "enable_unique_key_merge_on_write" = "true"
+)
+"""
+
+
+# ============================================================
+# [Step 2] 读 Hive dwt_flow_asin_week 周数据 + 字段规范化
+# ============================================================
+
+def read_and_normalize_week_data(spark, site_name, date_info):
+    """读 Hive dwt_flow_asin_week 周数据，规范化字段（与 dwt.{site}_flow_asin_month 对齐，周独有字段追加在末尾）
+    返回 cache 好的 df_save，调用方用完需自行 unpersist（见 write_dwt_week_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,
+            asin_source_flag, bsr_last_seen_at, bsr_seen_count_30d, nsr_last_seen_at, nsr_seen_count_30d,
+            asin_rank_wow, asin_ao_wow, asin_price_wow, asin_rating_wow, asin_comments_wow,
+            asin_bsr_orders_wow, asin_sales_wow, asin_variation_wow, asin_bought_wow, variation_bought_month_wow,
+            asin_bought_wow_last_1w, asin_bought_wow_last_2w, asin_bought_wow_last_3w,
+            asin_bought_rise, variation_bought_month_rise
+        FROM dwt_flow_asin_week
+        WHERE site_name = '{site_name}' AND date_type = 'week' 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'),
+        )
+
+    def _dec4(col_name, alias):
+        return F.round(F.col(col_name).cast('double'), 4).cast('decimal(20,4)').alias(alias)
+
+    df_save = df_raw.select(
+        F.lit(date_info).alias('date_info'),
+        F.col('asin'),
+        # ===== 核心选品指标 =====
+        _dec4('asin_ao_val', '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'),
+        _dec4('asin_weight', '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'),
+        _dec4('act_one_two_val', 'one_two_val'),
+        _dec4('act_three_four_val', 'three_four_val'),
+        _dec4('act_five_six_val', 'five_six_val'),
+        _dec4('act_eight_val', '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'),
+        # ===== 差值/近30天环比/同比 =====
+        F.col('asin_rank_rise').cast('int').alias('rank_rise'),
+        _dec4('asin_rank_mom', 'rank_mom'),
+        _dec4('asin_rank_yoy', 'rank_yoy'),
+        _dec4('asin_ao_rise', 'ao_rise'),
+        _dec4('asin_ao_mom', 'ao_mom'),
+        _dec4('asin_ao_yoy', 'ao_yoy'),
+        F.round(F.col('asin_price_rise').cast('double'), 2).cast('decimal(20,2)').alias('price_rise'),
+        _dec4('asin_price_mom', 'price_mom'),
+        _dec4('asin_price_yoy', 'price_yoy'),
+        F.round(F.col('asin_rating_rise').cast('double'), 1).cast('decimal(20,1)').alias('rating_rise'),
+        _dec4('asin_rating_mom', 'rating_mom'),
+        _dec4('asin_rating_yoy', 'rating_yoy'),
+        F.col('asin_comments_rise').cast('int').alias('comments_rise'),
+        _dec4('asin_comments_mom', 'comments_mom'),
+        _dec4('asin_comments_yoy', 'comments_yoy'),
+        F.col('asin_bsr_orders_rise').cast('int').alias('bsr_orders_rise'),
+        _dec4('asin_bsr_orders_mom', 'bsr_orders_mom'),
+        _dec4('asin_bsr_orders_yoy', 'bsr_orders_yoy'),
+        F.round(F.col('asin_sales_rise').cast('double'), 2).cast('decimal(20,2)').alias('sales_rise'),
+        _dec4('asin_sales_mom', 'sales_mom'),
+        _dec4('asin_sales_yoy', 'sales_yoy'),
+        F.col('asin_variation_rise').cast('int').alias('variation_rise'),
+        _dec4('asin_variation_mom', 'variation_mom'),
+        _dec4('asin_variation_yoy', 'variation_yoy'),
+        _dec4('asin_bought_mom', 'bought_month_mom'),
+        _dec4('asin_bought_yoy', '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'),
+        _dec4('asin_weight_ratio', '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'),
+        _dec4('zr_flow_proportion', 'zr_flow_proportion'),
+        _dec4('matrix_flow_proportion', 'matrix_flow_proportion'),
+        _dec4('matrix_ao_val', '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'),
+        _dec4('title_matching_degree', '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'),
+        _dec4('variation_bought_month_mom', 'variation_bought_month_mom'),
+        _dec4('variation_bought_month_yoy', 'variation_bought_month_yoy'),
+        # ===== 周维度独有字段 =====
+        # asin_source_flag string "1,2" → ARRAY<INT> json
+        F.to_json(F.expr("transform(split(asin_source_flag, ','), x -> cast(x as int))")).alias('asin_source_flag'),
+        F.to_date(F.col('bsr_last_seen_at')).alias('bsr_last_seen_at'),
+        F.col('bsr_seen_count_30d').cast('int').alias('bsr_seen_count_30d'),
+        F.to_date(F.col('nsr_last_seen_at')).alias('nsr_last_seen_at'),
+        F.col('nsr_seen_count_30d').cast('int').alias('nsr_seen_count_30d'),
+        _dec4('asin_rank_wow', 'rank_wow'),
+        _dec4('asin_ao_wow', 'ao_wow'),
+        _dec4('asin_price_wow', 'price_wow'),
+        _dec4('asin_rating_wow', 'rating_wow'),
+        _dec4('asin_comments_wow', 'comments_wow'),
+        _dec4('asin_bsr_orders_wow', 'bsr_orders_wow'),
+        _dec4('asin_sales_wow', 'sales_wow'),
+        _dec4('asin_variation_wow', 'variation_wow'),
+        _dec4('asin_bought_wow', 'bought_month_wow'),
+        _dec4('variation_bought_month_wow', 'variation_bought_month_wow'),
+        _dec4('asin_bought_wow_last_1w', 'bought_month_wow_last_1w'),
+        _dec4('asin_bought_wow_last_2w', 'bought_month_wow_last_2w'),
+        _dec4('asin_bought_wow_last_3w', 'bought_month_wow_last_3w'),
+        F.col('asin_bought_rise').cast('int').alias('bought_month_rise'),
+        F.col('variation_bought_month_rise').cast('int').alias('variation_bought_month_rise'),
+    ).cache()
+
+    count = df_save.count()
+    print(f"写入数据量：{count:,}")
+    if count == 0:
+        raise ValueError(f"Hive dwt_flow_asin_week[site_name={site_name}, date_info={date_info}] 读到0条，请确认 dwt/dwt_flow_asin_week.py 是否已跑完")
+    df_save.show(5, truncate=False)
+    return df_save
+
+
+# ============================================================
+# [Step 3] 写入 Doris dwt 周表
+# ============================================================
+
+DWT_WEEK_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, "
+    "asin_source_flag, bsr_last_seen_at, bsr_seen_count_30d, nsr_last_seen_at, nsr_seen_count_30d, "
+    "rank_wow, ao_wow, price_wow, rating_wow, comments_wow, bsr_orders_wow, sales_wow, variation_wow, "
+    "bought_month_wow, variation_bought_month_wow, "
+    "bought_month_wow_last_1w, bought_month_wow_last_2w, bought_month_wow_last_3w, "
+    "bought_month_rise, variation_bought_month_rise"
+)
+
+
+def write_dwt_week_table(df_save, doris_table, date_info):
+    """写入 Doris dwt 周表 dwt.{site}_flow_asin_week：先按 date_info 清理该周旧数据，再整批写入；写完 unpersist 传入的 df_save"""
+    print(f"[Step 3] 清理 {DORIS_DB}.{doris_table} 旧数据 WHERE date_info='{date_info}'")
+    _exec_doris_sql([f"DELETE FROM `{DORIS_DB}`.`{doris_table}` WHERE date_info = '{date_info}'"])
+    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=DWT_WEEK_COLUMNS,
+    )
+    df_save.unpersist()
+
+
+# ============================================================
+# [Step 4] selection 周表 INSERT OVERWRITE
+# ============================================================
+
+def build_week_insert_overwrite_sql(site_name, table_name, date_info, week_end_date):
+    """构建 INSERT OVERWRITE 到 selection.{table_name} 的 SQL；
+    launch_time_type / tracking_since_type 的天数基准用本周最后一天；
+    来源标识直接取 dwt 周表（Hive 已按本周7天聚合），品牌徽章取最新一期月数据"""
+    type_anchor_date = f"CAST('{week_end_date}' 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(f.asin_source_flag, ARRAY(0))                    AS asin_source_flag,
+    COALESCE(f.bsr_last_seen_at, CAST('1970-01-01' AS DATE))  AS bsr_last_seen_at,
+    COALESCE(f.bsr_seen_count_30d, 0)                         AS bsr_seen_count_30d,
+    COALESCE(f.nsr_last_seen_at, CAST('1970-01-01' AS DATE))  AS nsr_last_seen_at,
+    COALESCE(f.nsr_seen_count_30d, 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,
+    kp.keepa_type,
+    kp.keepa_item_type,
+    kp.keepa_type_cn,
+    kp.keepa_item_type_cn,
+    f.rank_wow,
+    f.ao_wow,
+    f.price_wow,
+    f.rating_wow,
+    f.comments_wow,
+    f.bsr_orders_wow,
+    f.sales_wow,
+    f.variation_wow,
+    f.bought_month_wow,
+    f.variation_bought_month_wow,
+    f.bought_month_wow_last_1w,
+    f.bought_month_wow_last_2w,
+    f.bought_month_wow_last_3w,
+    f.bought_month_rise,
+    f.variation_bought_month_rise
+FROM `dwt`.`{site_name}_flow_asin_week` 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
+-- ===== 品牌徽章（取最新一期月数据，date_info=yyyy-MM） =====
+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}' AND LENGTH(date_info) = 7
+      )
+) 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
+WHERE f.date_info = '{date_info}'
+"""
+
+
+# ============================================================
+# [Step 5] 流程记录表更新（仅 formal 模式）
+# ============================================================
+
+def modify_mission_record_status(site_name, date_info, result_type):
+    """流程记录表更新：仅 formal 模式才入库 mysql workflow_everyday"""
+    if result_type != 'formal':
+        print(f"[Step 5] result_type={result_type}，跳过流程记录表更新")
+        return
+    record_table = 'workflow_everyday'
+    engine_mysql = DBUtil.get_db_engine(db_type=DbTypes.mysql.name, site_name='us')
+    date_type = 'week'
+    table_name_field = f'{site_name}_flow_asin_last_week'
+    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 5] 流程记录表 workflow_everyday 已写入：{site_name} {date_info} date_type={date_type}")
+    else:
+        print(f"[Step 5] 流程记录表已存在该记录，跳过：date_type={date_type}")
+
+
+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_week"   # dwt 主表，不区分 test/formal
+    env_suffix            = '_test' if result_type == 'test' else ''
+    week_selection_table  = f"{site_name}_flow_asin_week_{date_info.replace('-', '_')}{env_suffix}"
+
+    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.{week_selection_table}")
+
+    spark = SparkUtil.get_spark_session(
+        f"DwtFlowAsinWeek: {site_name} {date_info} {result_type}"
+    )
+    week_end_date = get_week_end_date(spark, date_info)
+    print(f"  本周最后一天：{week_end_date}")
+
+    # ===== [Step 1] Doris 建表 =====
+    print(f"[Step 1] Doris 建表 dwt.{doris_table}、selection.{week_selection_table}")
+    _exec_doris_sql([build_dwt_week_create_table_sql(site_name),
+                     build_week_create_table_sql(week_selection_table, date_info)])
+
+    # ===== [Step 2] 读 Hive dwt_flow_asin_week 周数据 + 字段规范化 =====
+    print(f"[Step 2] 读 Hive dwt_flow_asin_week 周数据 + 字段规范化")
+    df_save = read_and_normalize_week_data(spark, site_name, date_info)
+
+    # ===== [Step 3] 写入 Doris dwt 周表 =====
+    write_dwt_week_table(df_save, doris_table, date_info)
+
+    # ===== [Step 4] Doris INSERT OVERWRITE 到 selection 周物化表 =====
+    print(f"[Step 4] Doris INSERT OVERWRITE selection.{week_selection_table}")
+    _exec_doris_sql([build_week_insert_overwrite_sql(site_name, week_selection_table, date_info, week_end_date)])
+
+    # ===== [Step 5] 流程记录表更新（仅 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)

Pyspark_job/dwt/dwt_flow_asin_week.py

+"""
+   @Author      : CT
+   @Description : 流量选品-周维度
+   @SourceTable :
+                  dim_asin_detail(week)
+                  dim_asin_bs_info(week)
+                  dwd_asin_measure(最新month_week)
+                  dwd_asin_cate_flag(week)
+                  dim_fb_asin_info(最新month)
+                  dim_keepa_asin_info
+                  dim_asin_color_info
+                  dim_bsr_category_tree
+                  asin_auction_sku_stat(mysql)
+                  category_full_name/category_disable_config(mysql)
+   @SinkTable   : dwt_flow_asin_week
+   @CreateTime  : 2026/10/21
+   @UpdateTime  : 2026/10/21
+"""
+import os
+import re
+import sys
+
+sys.path.append(os.path.dirname(sys.path[0]))  # 上级目录
+from utils.templates import Templates
+from pyspark.sql.window import Window
+from datetime import datetime, timedelta
+from pyspark.sql import functions as F
+from utils.db_util import DBUtil
+from utils.spark_util import SparkUtil
+from utils.common_util import CommonUtil
+from utils.hdfs_utils import HdfsUtils
+from pyspark.storagelevel import StorageLevel
+from pyspark.sql.types import *
+from yswg_utils.common_udf import resolve_asin_jump
+
+
+class DwtFlowAsinWeek(Templates):
+    def __init__(self, site_name="us", date_type="week", date_info="2026-41"):
+        super().__init__()
+        self.site_name = site_name
+        self.date_type = date_type
+        self.date_info = date_info
+        self.hive_tb = f"dwt_flow_asin_week"
+        self.partition_dict = {
+            "site_name": site_name,
+            "date_type": date_type,
+            "date_info": date_info
+        }
+        self.hdfs_path = CommonUtil.build_hdfs_path(self.hive_tb, partition_dict=self.partition_dict)
+        self.spark = self.create_spark_object(
+            app_name=f"{self.hive_tb}: {self.site_name}, {self.date_type}, {self.date_info}")
+        # 前1/2/3周、4周前、去年同周、本周最后一天
+        self.last_week, self.last_2_week, self.last_3_week, self.previous_week, self.last_year_week, self.week_end_date = self.get_week_date_info()
+        print(f"前1/2/3周: {self.last_week}/{self.last_2_week}/{self.last_3_week}, 4周前: {self.previous_week}, "
+              f"去年同周: {self.last_year_week}, 本周最后一天: {self.week_end_date}")
+        self.launch_time_interval_dict = self.get_launch_time_interval_dict(self.week_end_date)
+        # 写入、分区初始化
+        self.df_save = self.spark.sql(f"select 1+1;")
+        self.partitions_by = ['site_name', 'date_type', 'date_info']
+        self.reset_partitions(60)
+        # 初始化全局df
+        self.df_asin_detail = self.spark.sql(f"select 1+1;")
+        self.df_asin_measure = self.spark.sql(f"select 1+1;")
+        self.df_asin_bs_category = self.spark.sql(f"select 1+1;")
+        self.df_fb_asin_info = self.spark.sql(f"select 1+1;")
+        self.df_flow_asin_last_week = self.spark.sql(f"select 1+1;")
+        self.df_bought_history = self.spark.sql(f"select 1+1;")
+        self.df_flow_asin_last = self.spark.sql(f"select 1+1;")
+        self.df_flow_asin_last_year = self.spark.sql(f"select 1+1;")
+        self.df_keepa_asin = self.spark.sql(f"select 1+1;")
+        self.df_asin_source_flag = self.spark.sql(f"select 1+1;")
+        self.df_asin_auction = self.spark.sql(f"select 1+1;")
+        self.color_set = set()  # 颜色词表，read_data 阶段填充
+
+    def get_week_date_info(self):
+        """获取前1/2/3周、4周前、去年同周、本周最后一天"""
+        df = self.spark.sql(f"select id, `date`, year_week, week_day from dim_date_20_to_30").toPandas()
+        current_week_id = int(list(df.loc[(df.year_week == self.date_info) & (df.week_day == 1)].id)[0])
+        last_week = list(df.loc[df.id == current_week_id - 7].year_week)[0]
+        last_2_week = list(df.loc[df.id == current_week_id - 14].year_week)[0]
+        last_3_week = list(df.loc[df.id == current_week_id - 21].year_week)[0]
+        previous_week = list(df.loc[df.id == current_week_id - 28].year_week)[0]
+        week_end_date = str(list(df.loc[(df.year_week == self.date_info) & (df.week_day == 7)].date)[0])
+        year, week = self.date_info.split("-")
+        last_year_week = f"{int(year) - 1}-{week}"
+        return last_week, last_2_week, last_3_week, previous_week, last_year_week, week_end_date
+
+    @staticmethod
+    def get_launch_time_interval_dict(base_date):
+        base_date = datetime.strptime(base_date, '%Y-%m-%d')
+        return {
+            "one_month": (base_date + timedelta(days=-30)).strftime('%Y-%m-%d'),
+            "three_month": (base_date + timedelta(days=-90)).strftime('%Y-%m-%d'),
+            "six_month": (base_date + timedelta(days=-180)).strftime('%Y-%m-%d'),
+            "twelve_month": (base_date + timedelta(days=-360)).strftime('%Y-%m-%d'),
+            "twenty_four_month": (base_date + timedelta(days=-720)).strftime('%Y-%m-%d'),
+            "thirty_six_month": (base_date + timedelta(days=-1080)).strftime('%Y-%m-%d')
+        }
+
+    @staticmethod
+    def build_time_interval_type_expr(col_name, interval_dict):
+        """根据列名和时间区间字典生成上架/追踪时间类型 CASE WHEN 表达式"""
+        one_month = interval_dict['one_month']
+        three_month = interval_dict['three_month']
+        six_month = interval_dict['six_month']
+        twelve_month = interval_dict['twelve_month']
+        twenty_four_month = interval_dict['twenty_four_month']
+        thirty_six_month = interval_dict['thirty_six_month']
+        return f"""CASE WHEN {col_name} >= '{one_month}' THEN 1
+                        WHEN {col_name} >= '{three_month}' AND {col_name} < '{one_month}' THEN 2
+                        WHEN {col_name} >= '{six_month}' AND {col_name} < '{three_month}' THEN 3
+                        WHEN {col_name} >= '{twelve_month}' AND {col_name} < '{six_month}' THEN 4
+                        WHEN {col_name} >= '{twenty_four_month}' AND {col_name} < '{twelve_month}' THEN 5
+                        WHEN {col_name} >= '{thirty_six_month}' AND {col_name} < '{twenty_four_month}' THEN 6
+                        WHEN {col_name} < '{thirty_six_month}' THEN 7 ELSE 0 END"""
+
+    @staticmethod
+    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
+
+    def get_latest_date_info(self, table_name, date_type):
+        """获取表指定date_type下最新的date_info"""
+        sql = f"select max(date_info) as max_date_info from {table_name} where site_name='{self.site_name}' and date_type='{date_type}'"
+        latest_date_info = self.spark.sql(sql).collect()[0]['max_date_info']
+        print(f"{table_name} {date_type} 最新分区: {latest_date_info}")
+        return latest_date_info
+
+    def read_data(self):
+        print("1.获取dwd_asin_measure(最新month_week),得到各种类型的统计、ao值、自然流量占比、月销信息、bsr销量信息")
+        measure_date_info = self.get_latest_date_info("dwd_asin_measure", "month_week")
+        sql = f"""
+            select asin, round(asin_ao_val, 3) as asin_ao_val, round(asin_ao_val_matrix, 3) as matrix_ao_val,
+            asin_zr_counts, asin_sp_counts, (asin_sb1_counts + asin_sb2_counts) as asin_sb_counts,
+            asin_sb3_counts as asin_vi_counts, asin_bs_counts, asin_ac_counts, asin_tr_counts, asin_er_counts,
+            asin_bsr_orders as bsr_orders, asin_amazon_orders, round(asin_zr_flow_proportion, 3) as zr_flow_proportion,
+            round(asin_flow_proportion_matrix, 3) as matrix_flow_proportion
+            from dwd_asin_measure where site_name = '{self.site_name}' and date_type = 'month_week' and date_info = '{measure_date_info}'"""
+        print("sql:" + sql)
+        self.df_asin_measure = self.spark.sql(sqlQuery=sql)
+        self.df_asin_measure = self.df_asin_measure.repartition(100).persist(StorageLevel.DISK_ONLY)
+        self.df_asin_measure.show(10, truncate=False)
+        print("2.获取dim_asin_detail,得到asin详情")
+        sql = f"""
+            select asin, asin_img_url, lower(asin_title) as asin_title, asin_title_len, asin_price, asin_rating, asin_total_comments,
+            asin_buy_box_seller_type, asin_page_inventory, asin_category_desc, asin_volume, asin_weight, asin_length, asin_width, asin_height,
+            asin_color, asin_size, asin_style, asin_launch_time, asin_img_num, asin_img_type, asin_material,
+            lower(asin_brand_name) as asin_brand_name, asin_activity_type, act_one_two_val, act_three_four_val, act_five_six_val, act_eight_val,
+            one_star,two_star, three_star, four_star, five_star, low_star, together_asin, variation_num, account_name, account_id, parent_asin,
+            asin_lob_info, is_contains_lob_info, is_package_quantity_abnormal, asin_quantity_variation_type, package_quantity,
+            asin_is_movie as is_movie_label, asin_is_brand as is_brand_label, asin_is_alarm as is_alarm_brand, asin_is_self,
+            date_format(created_time, 'yyyy-MM-dd HH:mm:ss') as asin_crawl_date, asin_bought_month, asin_image_view,
+            case when product_description is not null then 1 else 0 end as is_with_product_description, asin_describe,
+            category_id as top_category_id, category_first_id as top_category_first_id, customer_reviews_json, img_list as img_info,
+            asin_follow_sellers as follow_sellers_count, asin_fbm_price, amazon_label, current_asin
+            from dim_asin_detail where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info='{self.date_info}'"""
+        print("sql:" + sql)
+        self.df_asin_detail = self.spark.sql(sqlQuery=sql)
+        self.df_asin_detail = self.df_asin_detail.repartition(100).persist(StorageLevel.DISK_ONLY)
+        self.df_asin_detail.show(10, truncate=False)
+        print("3.获取dim_asin_bs_info,获取分类排名信息")
+        sql = f"""
+            select asin, asin_bs_cate_1_rank as first_category_rank, asin_bs_cate_current_rank as current_category_rank,
+            asin_bs_cate_1_id as category_first_id, asin_bs_cate_current_id as category_id
+            from dim_asin_bs_info where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info = '{self.date_info}'"""
+        print("sql:" + sql)
+        self.df_asin_bs_category = self.spark.sql(sqlQuery=sql)
+        self.df_asin_bs_category = self.df_asin_bs_category.repartition(100).persist(StorageLevel.DISK_ONLY)
+        self.df_asin_bs_category.show(10, truncate=False)
+        print("4.获取dim_fb_asin_info(最新month),得到卖家相关信息")
+        fb_date_info = self.get_latest_date_info("dim_fb_asin_info", "month")
+        sql = f"""
+            select seller_id as account_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='month' and date_info='{fb_date_info}' and seller_id is not null
+            """
+        print("sql:" + sql)
+        self.df_fb_asin_info = self.spark.sql(sqlQuery=sql)
+        self.df_fb_asin_info = self.df_fb_asin_info.repartition(100).persist(StorageLevel.DISK_ONLY)
+        self.df_fb_asin_info.show(10, truncate=False)
+        print("5.获取上周整合结果(_rise、近7天环比_wow)")
+        sql = f"""
+            select asin, round(asin_ao_val, 3) as lastweek_asin_ao_val, asin_price as lastweek_asin_price,
+            sales as lastweek_sales, variation_num as lastweek_variation_num, asin_rating as lastweek_asin_rating,
+            bsr_orders as lastweek_bsr_orders, asin_total_comments as lastweek_asin_total_comments,
+            first_category_rank as lastweek_first_category_rank, asin_bought_month as lastweek_asin_bought_month,
+            variation_bought_month as lastweek_variation_bought_month, asin_bought_wow as lastweek_asin_bought_wow
+            from {self.hive_tb}
+            where site_name = '{self.site_name}' and date_type = '{self.date_type}' and date_info = '{self.last_week}'
+            """
+        print("sql:" + sql)
+        self.df_flow_asin_last_week = self.spark.sql(sqlQuery=sql)
+        self.df_flow_asin_last_week = self.df_flow_asin_last_week.repartition(100).persist(StorageLevel.DISK_ONLY)
+        self.df_flow_asin_last_week.show(10, truncate=False)
+        print("5.1获取前2/3周月销_wow")
+        sql = f"""
+            select asin,
+            max(case when date_info = '{self.last_2_week}' then asin_bought_wow end) as asin_bought_wow_last_2w,
+            max(case when date_info = '{self.last_3_week}' then asin_bought_wow end) as asin_bought_wow_last_3w
+            from {self.hive_tb}
+            where site_name = '{self.site_name}' and date_type = '{self.date_type}' and date_info in ('{self.last_2_week}', '{self.last_3_week}')
+            group by asin
+            """
+        print("sql:" + sql)
+        self.df_bought_history = self.spark.sql(sqlQuery=sql)
+        self.df_bought_history = self.df_bought_history.repartition(100).persist(StorageLevel.DISK_ONLY)
+        self.df_bought_history.show(10, truncate=False)
+        print("5.2获取4周前整合结果(近30天环比_mom)")
+        sql = f"""
+            select asin, round(asin_ao_val, 3) as previous_asin_ao_val, asin_price as previous_asin_price,
+            sales as previous_sales, variation_num as previous_variation_num, asin_rating as previous_asin_rating,
+            bsr_orders as previous_bsr_orders, asin_total_comments as previous_asin_total_comments,
+            first_category_rank as previous_first_category_rank, asin_bought_month as previous_asin_bought_month,
+            variation_bought_month as previous_variation_bought_month from {self.hive_tb}
+            where site_name = '{self.site_name}' and date_type = '{self.date_type}' and date_info = '{self.previous_week}'
+            """
+        print("sql:" + sql)
+        self.df_flow_asin_last = self.spark.sql(sqlQuery=sql)
+        self.df_flow_asin_last = self.df_flow_asin_last.repartition(100).persist(StorageLevel.DISK_ONLY)
+        self.df_flow_asin_last.show(10, truncate=False)
+        print("6.获取同比去年同周整合结果")
+        sql = f"""
+            select asin, round(asin_ao_val, 3) as lastyear_asin_ao_val, asin_price as lastyear_asin_price,
+            sales as lastyear_sales, variation_num as lastyear_variation_num, asin_rating as lastyear_asin_rating,
+            bsr_orders as lastyear_bsr_orders, asin_total_comments as lastyear_asin_total_comments,
+            first_category_rank as lastyear_first_category_rank, asin_bought_month as lastyear_asin_bought_month,
+            variation_bought_month as lastyear_variation_bought_month from {self.hive_tb}
+            where site_name = '{self.site_name}' and date_type = '{self.date_type}' and date_info = '{self.last_year_week}'
+            """
+        print("sql:" + sql)
+        self.df_flow_asin_last_year = self.spark.sql(sqlQuery=sql)
+        self.df_flow_asin_last_year = self.df_flow_asin_last_year.repartition(100).persist(StorageLevel.DISK_ONLY)
+        self.df_flow_asin_last_year.show(10, truncate=False)
+        print("7.读取最新keepa数据")
+        sql = f"""
+        select asin, tracking_since, keepa_launch_time from dim_keepa_asin_info where site_name = '{self.site_name}'
+        """
+        print("sql:" + sql)
+        self.df_keepa_asin = self.spark.sql(sqlQuery=sql)
+        self.df_keepa_asin = self.df_keepa_asin.repartition(100).persist(StorageLevel.DISK_ONLY)
+        self.df_keepa_asin.show(10, truncate=False)
+        print("8.获取asin不同来源标识(本周7天聚合)")
+        sql = f"""
+        select asin, asin_cate_flag as asin_source_flag, bsr_latest_date as bsr_last_seen_at, bsr_30day_count as bsr_seen_count_30d,
+        nsr_latest_date as nsr_last_seen_at, nsr_30day_count as nsr_seen_count_30d from dwd_asin_cate_flag
+        where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info='{self.date_info}'
+        """
+        print("sql:" + sql)
+        self.df_asin_source_flag = self.spark.sql(sqlQuery=sql)
+        self.df_asin_source_flag = self.df_asin_source_flag.repartition(100).persist(StorageLevel.DISK_ONLY)
+        self.df_asin_source_flag.show(10, truncate=False)
+        print("9.读取颜色词表 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("10.获取asin_auction_sku_stat,得到asin竞卖数、sku创建数")
+        mysql_con = DBUtil.get_connection_info("mysql", "us")
+        sql = """
+            select asin, auctions_num, skus_num_creat
+            from asin_auction_sku_stat
+            WHERE asin REGEXP '^[0-9A-Z]{10}$'
+        """
+        print("sql:" + sql)
+        self.df_asin_auction = SparkUtil.read_jdbc_query(session=self.spark, url=mysql_con['url'], pwd=mysql_con['pwd'],
+                                                          username=mysql_con['username'], query=sql)
+        self.df_asin_auction = self.df_asin_auction.repartition(100).persist(StorageLevel.DISK_ONLY)
+        self.df_asin_auction.show(10, truncate=False)
+
+    # asin跳转处理：判断规则见 resolve_asin_jump；按asin去重保留最新抓取时间
+    def handle_asin_redirect(self):
+        df_old = self.df_asin_detail
+        window_redirect = Window.partitionBy('asin').orderBy(F.col('asin_crawl_date').desc_nulls_last())
+        self.df_asin_detail = df_old \
+            .withColumn('asin', resolve_asin_jump()) \
+            .drop('current_asin') \
+            .withColumn('_rk', F.row_number().over(window_redirect)) \
+            .filter(F.col('_rk') == 1) \
+            .drop('_rk') \
+            .repartition(100) \
+            .persist(StorageLevel.DISK_ONLY)
+        redirect_count = self.df_asin_detail.count()
+        print(f"asin跳转处理完成，去重后数据量: {redirect_count}")
+        df_old.unpersist()
+
+    # 处理asin基础属性信息(体积重量相关)
+    def handle_asin_basic_attribute(self):
+        # 处理重量类型
+        self.df_asin_detail = self.df_asin_detail.withColumn("asin_weight_type", F.expr("""CASE
+                    WHEN asin_weight BETWEEN 0 AND 0.2 THEN 1
+                    WHEN asin_weight BETWEEN 0.2 AND 0.4 THEN 2
+                    WHEN asin_weight BETWEEN 0.4 AND 0.6 THEN 3
+                    WHEN asin_weight BETWEEN 0.6 AND 1 THEN 4
+                    WHEN asin_weight BETWEEN 1 AND 2 THEN 5
+                    WHEN asin_weight >= 2 THEN 6
+                    ELSE 0 END"""))
+        # 处理体积重/毛重相关信息
+        self.df_asin_detail = self.df_asin_detail.withColumn(
+            "asin_weight_ratio",
+            F.when(
+                F.col("asin_length").isNotNull() & (F.col("asin_width").isNotNull()) & (F.col("asin_height").isNotNull()) & (F.col("asin_weight") > 0),
+                F.round(F.col("asin_length") * F.col("asin_width") * F.col("asin_height") * 3.2774128 / (F.col("asin_weight") * 453.59), 4)
+            ).otherwise(F.lit(-1))
+        )
+        # 处理尺寸类型
+        if self.site_name == 'us':
+            expr_str = f"""
+           CASE WHEN asin_weight > 0 AND asin_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 asin_weight > 0 AND asin_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 asin_weight > 0 AND asin_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 asin_weight > 0 AND asin_weight <= 150 AND asin_length > 0 AND asin_length <= 108 AND asin_length + asin_length + (asin_width + asin_height) * 2 <= 130  THEN 4
+           WHEN asin_weight > 0 AND asin_weight <= 150 AND asin_length > 0 AND asin_length <= 108 AND asin_length + asin_length + (asin_width + asin_height) * 2 <= 165  THEN 5
+           WHEN asin_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 asin_weight > 0 AND asin_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 asin_weight > 0 AND asin_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 asin_weight > 0 AND asin_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 asin_weight > 0 AND asin_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 asin_weight > 0 AND asin_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"""
+        self.df_asin_detail = self.df_asin_detail.withColumn("asin_size_type", F.expr(expr_str)).drop("asin_length", "asin_width", "asin_height")
+
+    # 通过ASIN页面信息处理(评分类型、上架时间类型、价格类型)
+    def handle_asin_detail_all_type(self):
+        # 评分类型
+        self.df_asin_detail = self.df_asin_detail.withColumn(
+            "asin_rating_type", F.expr("""CASE WHEN asin_rating >= 4.5 THEN 1 WHEN asin_rating >= 4 AND asin_rating < 4.5 THEN 2 WHEN asin_rating >= 3.5 AND asin_rating < 4 THEN 3
+                             WHEN asin_rating >= 3 AND asin_rating < 3.5 THEN 4 WHEN asin_rating < 3 AND asin_rating >= 0 THEN 5 ELSE 0 END"""))
+        # 关联keepa上架时间
+        self.df_asin_detail = self.df_asin_detail.join(
+            self.df_keepa_asin, on=['asin'], how='left'
+        )
+        self.df_asin_detail = self.df_asin_detail.withColumn(
+            "asin_launch_time", F.coalesce(F.col("asin_launch_time"), F.col("keepa_launch_time"))
+        )
+        # 上架时间类型(以本周最后一天为基准)
+        self.df_asin_detail = self.df_asin_detail.withColumn(
+            "asin_launch_time_type",
+            F.expr(self.build_time_interval_type_expr("asin_launch_time", self.launch_time_interval_dict))
+        )
+        # 价格类型
+        self.df_asin_detail = self.df_asin_detail.withColumn(
+            "asin_price_type", F.expr("""
+            CASE WHEN asin_price IS NOT NULL AND asin_price > 0 AND asin_price < 10 THEN 1
+            WHEN asin_price >= 10 AND asin_price < 15 THEN 2
+            WHEN asin_price >= 15 AND asin_price < 20 THEN 3
+            WHEN asin_price >= 20 AND asin_price < 30 THEN 4
+            WHEN asin_price >= 30 AND asin_price < 50 THEN 5
+            WHEN asin_price >= 50 AND asin_price < 70 THEN 6
+            WHEN asin_price >= 70 AND asin_price < 100 THEN 7
+            WHEN asin_price >= 100 AND asin_price < 150 THEN 8
+            WHEN asin_price >= 150 THEN 9 ELSE 0 END"""))
+        self.df_keepa_asin.unpersist()
+
+    # 处理asin分类、排名、排名类型字段、是否有效排名信息
+    def handle_asin_category_info(self):
+        self.df_asin_detail = self.df_asin_detail.join(self.df_asin_bs_category, on=['asin'], how='left')
+        self.df_asin_detail = self.df_asin_detail.withColumn(
+            "category_id", F.coalesce(F.col("category_id"), F.col("top_category_id"))
+        ).withColumn(
+            "category_first_id", F.coalesce(F.col("category_first_id"), F.col("top_category_first_id"))
+        ).drop("top_category_id", "top_category_first_id")
+        self.df_asin_detail = self.df_asin_detail.withColumn(
+            "asin_rank_type", F.expr("""
+            CASE WHEN first_category_rank IS NOT NULL AND first_category_rank BETWEEN 0 AND 1000 THEN 1
+            WHEN first_category_rank BETWEEN 1000 AND 5000 THEN 2
+            WHEN first_category_rank BETWEEN 5000 AND 10000 THEN 3
+            WHEN first_category_rank BETWEEN 10000 AND 20000 THEN 4
+            WHEN first_category_rank BETWEEN 20000 AND 30000 THEN 5
+            WHEN first_category_rank BETWEEN 30000 AND 50000 THEN 6
+            WHEN first_category_rank BETWEEN 50000 AND 70000 THEN 7
+            WHEN first_category_rank >= 70000 THEN 8 ELSE 0 END"""))
+        self.df_asin_bs_category.unpersist()
+
+    # 处理asin的bsr销量、亚马逊月销信息、ao值、母体ao、自然流量占比、母体自然流量占比、ao值类型
+    def handle_asin_measure(self):
+        self.df_asin_detail = self.df_asin_detail.join(self.df_asin_measure, on=['asin'], how='left')
+        self.df_asin_detail = self.df_asin_detail.withColumn(
+            "sales", F.round(F.col("bsr_orders") * F.col("asin_price"), 2)
+        ).withColumn(
+            "asin_bought_month", F.coalesce(F.col("asin_bought_month"), F.col("asin_amazon_orders"))
+        ).withColumn("asin_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""")).drop("asin_amazon_orders").persist(StorageLevel.DISK_ONLY)
+        # 关联流量asin由于来自广告位，无法计算ao值，填充母体ao值和母体zr占比
+        df_ao_stage = self.df_asin_detail
+        df_parent_asin_ao_val = df_ao_stage.select('parent_asin', 'matrix_ao_val', 'matrix_flow_proportion').groupby(['parent_asin']).agg(
+            F.first('matrix_ao_val').alias('parent_matrix_ao_val'), F.first('matrix_flow_proportion').alias('parent_matrix_flow_proportion')
+        )
+        self.df_asin_detail = df_ao_stage.join(df_parent_asin_ao_val, on=['parent_asin'], how='left').withColumn(
+            "matrix_ao_val", F.coalesce(F.col("matrix_ao_val"), F.col("parent_matrix_ao_val"))
+        ).withColumn(
+            "matrix_flow_proportion", F.coalesce(F.col("matrix_flow_proportion"), F.col("parent_matrix_flow_proportion"))
+        ).drop("parent_matrix_ao_val", "parent_matrix_flow_proportion")
+        self.df_asin_measure.unpersist()
+        df_ao_stage.unpersist()
+
+        # 变体销量和(variation_bought_month)：按parent_asin聚合asin_bought_month
+        df_variation_bought_month = self.df_asin_detail.filter(
+            "parent_asin is not null"
+        ).select("parent_asin", "asin_bought_month")
+        df_variation_bought_month_agg = df_variation_bought_month.groupby(['parent_asin']).agg(
+            F.sum("asin_bought_month").cast("int").alias("variation_bought_month")
+        )
+        self.df_asin_detail = self.df_asin_detail.join(
+            df_variation_bought_month_agg, on=['parent_asin'], how='left'
+        ).withColumn(
+            "variation_bought_month",
+            F.coalesce(F.col("variation_bought_month"), F.col("asin_bought_month").cast("int"))
+        )
+
+    # 处理竞卖数据(auctions_num/skus_num_creat)及母体竞卖数据(auctions_num_all/skus_num_creat_all)
+    def handle_asin_auction(self):
+        self.df_asin_detail = self.df_asin_detail.join(self.df_asin_auction, on=['asin'], how='left')
+        self.df_asin_auction.unpersist()
+        self.df_asin_detail = self.df_asin_detail.withColumn(
+            "auctions_num", F.col("auctions_num").cast("int")
+        ).withColumn(
+            "skus_num_creat", F.col("skus_num_creat").cast("int")
+        ).na.fill({"auctions_num": 0, "skus_num_creat": 0})
+        # 母体竞卖数：按parent_asin聚合子体auctions_num/skus_num_creat
+        df_auction_all = self.df_asin_detail.filter(
+            "parent_asin is not null"
+        ).select("parent_asin", "auctions_num", "skus_num_creat")
+        df_auction_all_agg = df_auction_all.groupby(['parent_asin']).agg(
+            F.sum("auctions_num").cast("int").alias("auctions_num_all"),
+            F.sum("skus_num_creat").cast("int").alias("skus_num_creat_all")
+        )
+        self.df_asin_detail = self.df_asin_detail.join(
+            df_auction_all_agg, on=['parent_asin'], how='left'
+        ).withColumn(
+            "auctions_num_all", F.coalesce(F.col("auctions_num_all"), F.col("auctions_num"))
+        ).withColumn(
+            "skus_num_creat_all", F.coalesce(F.col("skus_num_creat_all"), F.col("skus_num_creat"))
+        )
+
+    # 处理配送方式、卖家所在地以及卖家所在地类型
+    def handle_seller_country(self):
+        # df1: account_id + seller_country_name 去重
+        df_seller_country = self.df_fb_asin_info.select('account_id', 'seller_country_name').dropDuplicates(['account_id'])
+        # df2: asin 去重，保留最新的 account_id + account_name
+        window = Window.partitionBy('asin').orderBy(F.col('fb_crawl_date').desc())
+        df_asin_account = self.df_fb_asin_info.select('asin', 'account_id', 'account_name', 'fb_crawl_date') \
+            .withColumn('rank', F.row_number().over(window)) \
+            .filter(F.col('rank') == 1) \
+            .drop('rank', 'fb_crawl_date') \
+            .withColumnRenamed('account_id', 'fd_account_id') \
+            .withColumnRenamed('account_name', 'fd_account_name')
+        # 1. 关联df2，用于填充dim表中account_id和account_name为空的情况
+        self.df_asin_detail = self.df_asin_detail.join(df_asin_account, on=['asin'], how='left')
+        # 2. 优先使用dim表数据，account_id为空时account_id和account_name整体取fd表数据
+        self.df_asin_detail = self.df_asin_detail.withColumn(
+            "account_name", F.when(F.col("account_id").isNull(), F.col("fd_account_name")).otherwise(F.col("account_name"))
+        ).withColumn(
+            "account_id", F.when(F.col("account_id").isNull(), F.col("fd_account_id")).otherwise(F.col("account_id"))
+        ).drop("fd_account_id", "fd_account_name")
+        # 3. 关联df1获取seller_country_name
+        self.df_asin_detail = self.df_asin_detail.join(df_seller_country, on=['account_id'], how='left')
+        self.df_asin_detail = self.df_asin_detail.withColumn("asin_site_name_type", F.expr("""
+        CASE WHEN asin_buy_box_seller_type = 1  THEN 4
+        WHEN asin_buy_box_seller_type != 1 AND seller_country_name is not null AND seller_country_name like '%US%' THEN 1
+        WHEN asin_buy_box_seller_type != 1 AND seller_country_name is not null AND seller_country_name like '%CN%' THEN 2
+        WHEN asin_buy_box_seller_type != 1 AND seller_country_name is not null AND seller_country_name like '%HK%' THEN 5
+        WHEN asin_buy_box_seller_type != 1 AND seller_country_name is not null AND seller_country_name like '%TW%' THEN 6
+        ELSE 3 END"""))
+        self.df_fb_asin_info.unpersist()
+
+    # 处理asin的lqs评分
+    def handle_asin_lqs_rating(self):
+        self.df_asin_detail = self.df_asin_detail.withColumn(
+            "category_node_rating", F.when(F.col("category_id").isNotNull(), F.lit(1)).otherwise(F.lit(0))
+        ).withColumn(
+            "zr_rating", F.when(F.col("asin_zr_counts") > 0, F.lit(0.5)).otherwise(F.lit(0))
+        ).withColumn(
+            "sp_rating", F.when(F.col("asin_sp_counts") > 0, F.lit(1)).otherwise(F.lit(0))
+        ).withColumn(
+            "a_add_rating", F.when(F.col("asin_img_type").contains("3"), F.lit(1)).otherwise(F.lit(0))
+        ).withColumn(
+            "video_rating", F.when(F.col("asin_img_type").contains("2"), F.lit(0.5)).otherwise(F.lit(0))
+        ).withColumn(
+            "brand_rating", F.when(F.col("is_brand_label") == 1, F.lit(0.2)).otherwise(F.lit(0))
+        ).withColumn(
+            "product_describe_rating", F.when(F.col("is_with_product_description") == 1, F.lit(0.2)).otherwise(F.lit(0))
+        ).withColumn(
+            "highlight_rating",
+            F.when((F.col("asin_describe").isNotNull()) & (F.size(F.split(F.col("asin_describe"), '\\|-\\|')) <= 4),
+                   F.size(F.split(F.col("asin_describe"), '\\|-\\|')) * 0.4)
+            .when((F.col("asin_describe").isNotNull()) & (F.size(F.split(F.col("asin_describe"), '\\|-\\|')) > 4),
+                  F.lit(1.6))
+            .otherwise(F.lit(0))
+        ).withColumn(
+            "title_len_rating",
+            F.when((F.col("asin_title_len") >= 50) & (F.col("asin_title_len") <= 200), F.lit(0.5)).otherwise(F.lit(0))
+        ).withColumn(
+            "title_brand_rating",
+            F.expr(f"""CASE WHEN asin_brand_name is not null AND lower(regexp_replace(asin_title, '[^a-zA-Z0-9\\s]', '')) LIKE lower(regexp_replace(asin_brand_name, '[^a-zA-Z0-9\\s]', '')) || '%' THEN 0.5 ELSE 0 END""")
+        ).withColumn(
+            "img_num_rating", F.when(F.col("asin_img_num") <= 4, F.col("asin_img_num") * 0.5).when(F.col("asin_img_num") > 4, F.lit(2)).otherwise(F.lit(0))
+        ).withColumn(
+            "img_enlarge_rating", F.when(F.col("asin_image_view") == 1, F.lit(0.5)).otherwise(F.lit(0))
+        )
+        self.df_asin_detail = self.df_asin_detail.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"))
+            )
+        )
+        self.df_asin_detail = self.df_asin_detail.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")
+        )
+        self.df_asin_detail = self.df_asin_detail.\
+            drop("is_with_product_description", "asin_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")
+
+    # 处理asin是否隐藏分类信息（US/UK/DE站点通用）以及asin_type信息
+    def handle_asin_is_hide(self):
+        mysql_con = DBUtil.get_connection_info("mysql", "us")
+        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}'
+        """
+        df_hide_category = SparkUtil.read_jdbc_query(session=self.spark, url=mysql_con['url'], pwd=mysql_con['pwd'],
+                                                     username=mysql_con['username'], query=sql
+                                                     ).withColumn("hide_flag", F.lit(1))
+        self.df_asin_detail = self.df_asin_detail.join(df_hide_category, on=['category_id'], how='left')
+        self.df_asin_detail = self.df_asin_detail.withColumn(
+            "asin_is_hide", F.when(F.col("hide_flag") == 1, F.lit(1)).otherwise(F.lit(0))
+        ).drop("hide_flag")
+        # 解析 asin_category_desc 取 › 分隔的第一个元素作为补充分类名称
+        self.df_asin_detail = self.df_asin_detail.withColumn(
+            "desc_category_first_name",
+            F.lower(F.trim(F.split(F.col("asin_category_desc"), "›").getItem(0)))
+        )
+        sql_dim = f"""
+            select lower(trim(en_name)) as desc_category_first_name, category_first_id as desc_category_first_id
+            from dim_bsr_category_tree where site_name = '{self.site_name}' and category_parent_id = 0 and leaf_node = 2
+        """
+        df_bsr_category = F.broadcast(self.spark.sql(sqlQuery=sql_dim))
+        self.df_asin_detail = self.df_asin_detail.join(df_bsr_category, on=['desc_category_first_name'], how='left')
+        # 两个分类ID均在过滤列表中才标记为1
+        need_categories = "('mobile-apps', 'audible', 'books', 'music', 'dmusic', 'digital-text', 'magazines', 'movies-tv', 'software', 'videogames', 'amazon-devices', 'boost', 'us-live-explorations', 'amazon-renewed')"
+        self.df_asin_detail = self.df_asin_detail.withColumn("asin_is_need", F.expr(f"""
+        CASE WHEN category_first_id in {need_categories}
+                  AND desc_category_first_id in {need_categories} THEN 1
+             WHEN asin NOT LIKE 'B0%' THEN 1
+        ELSE 0 END"""))
+        self.df_asin_detail = self.df_asin_detail.drop("desc_category_first_name")
+        self.df_asin_detail = self.df_asin_detail.withColumn("asin_type", F.expr("""
+            CASE WHEN asin_is_self=1 THEN 1 WHEN asin_is_need=1 THEN 2 WHEN asin_is_hide=1 THEN 3 ELSE 0 END"""
+            )).drop("asin_is_self", "asin_is_need", "asin_is_hide")
+
+    # 处理所有变化率相关指标（_rise/_wow对比上周、_mom对比4周前、_yoy对比去年同周）
+    def handle_change_rate(self):
+        self.df_asin_detail = self.df_asin_detail.join(self.df_flow_asin_last_week, on=['asin'], how='left')
+        self.df_asin_detail = self.df_asin_detail.join(self.df_flow_asin_last, on=['asin'], how='left')
+        self.df_asin_detail = self.df_asin_detail.join(self.df_flow_asin_last_year, on=['asin'], how='left')
+        # (current_col, base_col, suffix, rise_round)
+        # base_col 为不带前缀的历史字段名；rise_round: None 表示不计算 _rise；整数表示保留小数位；'int' 表示转 IntegerType
+        columns_to_change = [
+            ("first_category_rank", "first_category_rank", "asin_rank", "int"),
+            ("bsr_orders", "bsr_orders", "asin_bsr_orders", "int"),
+            ("asin_rating", "asin_rating", "asin_rating", 1),
+            ("asin_total_comments", "asin_total_comments", "asin_comments", "int"),
+            ("variation_num", "variation_num", "asin_variation", "int"),
+            ("asin_ao_val", "asin_ao_val", "asin_ao", 4),
+            ("asin_price", "asin_price", "asin_price", 2),
+            ("sales", "sales", "asin_sales", 2),
+            ("asin_bought_month", "asin_bought_month", "asin_bought", "int"),
+            ("variation_bought_month", "variation_bought_month", "variation_bought_month", "int"),
+        ]
+        for current_col, base_col, suffix, rise_round in columns_to_change:
+            rise_col, wow_col = self.calculate_change(current_col, f"lastweek_{base_col}")
+            _, mom_col = self.calculate_change(current_col, f"previous_{base_col}")
+            _, yoy_col = self.calculate_change(current_col, f"lastyear_{base_col}")
+            if rise_round is not None:
+                if rise_round == "int":
+                    self.df_asin_detail = self.df_asin_detail.withColumn(f"{suffix}_rise", rise_col.cast(IntegerType()))
+                else:
+                    self.df_asin_detail = self.df_asin_detail.withColumn(f"{suffix}_rise", F.round(rise_col, rise_round))
+            self.df_asin_detail = self.df_asin_detail \
+                .withColumn(f"{suffix}_wow", F.round(wow_col, 4)) \
+                .withColumn(f"{suffix}_mom", F.round(mom_col, 4)) \
+                .withColumn(f"{suffix}_yoy", F.round(yoy_col, 4))
+        # 前1/2/3周月销_wow，供后端判断连续上升
+        self.df_asin_detail = self.df_asin_detail.join(self.df_bought_history, on=['asin'], how='left') \
+            .withColumn("asin_bought_wow_last_1w", F.col("lastweek_asin_bought_wow"))
+        self.df_flow_asin_last_week.unpersist()
+        self.df_flow_asin_last.unpersist()
+        self.df_flow_asin_last_year.unpersist()
+        self.df_bought_history.unpersist()
+
+    def handle_asin_different_source(self):
+        self.df_asin_detail = self.df_asin_detail.join(
+            self.df_asin_source_flag, on=['asin'], how='left'
+        ).fillna({"asin_source_flag": "0", "bsr_last_seen_at": "1970-01-01", "nsr_last_seen_at": "1970-01-01"})
+        self.df_asin_source_flag.unpersist()
+
+    def handle_multi_color_flag(self):
+        """判断 ASIN 是否为颜色组合产品
+        multi_color_flag:
+          0 = 非多色
+          1 = 从颜色变体属性字段（asin_color）解析为多色
+          2 = 从标题或五点描述解析为多色（降级 fallback）
+        """
+        # 第1层：关键词正则
+        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层：分隔符正则
+        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")"
+        )
+
+        self.df_asin_detail = self.df_asin_detail.withColumn(
+            "_matched_colors_str", udf_matched_colors(F.col("asin_color"))
+        )
+        self.df_asin_detail = self.df_asin_detail.withColumn(
+            "multi_color_flag",
+            F.when(
+                F.col("asin_color").isNotNull() & F.col("asin_color").rlike(KEYWORD_PATTERN), 1
+            ).when(
+                F.col("asin_color").isNotNull() & F.col("asin_color").rlike(SEPARATOR_PATTERN), 1
+            ).when(
+                F.col("asin_color").isNotNull() & F.col("_matched_colors_str").isNotNull(), 1
+            ).when(
+                F.lower(F.concat_ws(" ", F.col("asin_title"), F.col("asin_describe"))).rlike(FALLBACK_PATTERN), 2
+            ).otherwise(0)
+        ).withColumn(
+            "multi_color_str",
+            F.when(
+                F.col("asin_color").isNotNull() & F.col("asin_color").rlike(KEYWORD_PATTERN),
+                F.regexp_extract(F.col("asin_color"), KEYWORD_PATTERN, 1)
+            ).when(
+                F.col("asin_color").isNotNull() & F.col("asin_color").rlike(SEPARATOR_PATTERN),
+                F.col("asin_color")
+            ).when(
+                F.col("asin_color").isNotNull() & F.col("_matched_colors_str").isNotNull(),
+                F.col("_matched_colors_str")
+            ).when(
+                F.lower(F.concat_ws(" ", F.col("asin_title"), F.col("asin_describe"))).rlike(FALLBACK_PATTERN),
+                F.regexp_extract(
+                    F.lower(F.concat_ws(" ", F.col("asin_title"), F.col("asin_describe"))),
+                    FALLBACK_PATTERN, 1
+                )
+            ).otherwise(F.lit(None))
+        ).drop("_matched_colors_str")
+
+    def handle_other_new_col(self):
+        # 处理五点描述长度
+        self.df_asin_detail = self.df_asin_detail.withColumn(
+            "describe_len", F.length(F.regexp_replace(F.col("asin_describe"), "\\|-\\|", ""))
+        )
+        # 处理keepa追踪时间(以本周最后一天为基准)
+        self.df_asin_detail = self.df_asin_detail.withColumn(
+            "tracking_since",
+            F.when(
+                F.col("tracking_since").isNull(), F.lit(None)
+            ).otherwise(
+                F.date_format(F.from_unixtime((F.col("tracking_since") + F.lit(21564000)) * 60), "yyyy-MM-dd")
+            )
+        ).withColumn(
+            "tracking_since_type",
+            F.expr(self.build_time_interval_type_expr("tracking_since", self.launch_time_interval_dict))
+        )
+
+    # 字段标准化：保留月表导出Doris的全部有效字段，周维度新增字段追加在末尾
+    def handle_column(self):
+        self.df_save = self.df_asin_detail.select(
+            "asin",
+            F.col("asin_ao_val").cast("double").alias("asin_ao_val"),
+            F.col("asin_zr_counts").cast("int").alias("asin_zr_counts"),
+            F.col("asin_sp_counts").cast("int").alias("asin_sp_counts"),
+            F.col("asin_sb_counts").cast("int").alias("asin_sb_counts"),
+            F.col("asin_vi_counts").cast("int").alias("asin_vi_counts"),
+            F.col("asin_bs_counts").cast("int").alias("asin_bs_counts"),
+            F.col("asin_ac_counts").cast("int").alias("asin_ac_counts"),
+            F.col("asin_tr_counts").cast("int").alias("asin_tr_counts"),
+            F.col("asin_er_counts").cast("int").alias("asin_er_counts"),
+            F.col("bsr_orders").cast("int").alias("bsr_orders"),
+            F.col("sales").cast("double").alias("sales"),
+            "asin_img_url", "asin_title",
+            F.col("asin_title_len").cast("int").alias("asin_title_len"),
+            F.col("asin_price").cast("double").alias("asin_price"),
+            F.col("asin_rating").cast("double").alias("asin_rating"),
+            F.col("asin_total_comments").cast("int").alias("asin_total_comments"),
+            F.col("asin_buy_box_seller_type").cast("int").alias("asin_buy_box_seller_type"),
+            F.col("asin_page_inventory").cast("int").alias("asin_page_inventory"),
+            F.col("asin_volume").cast("string").alias("asin_volume"),
+            F.col("asin_weight").cast("double").alias("asin_weight"),
+            "asin_color", "asin_size", "asin_style", "asin_material",
+            F.col("asin_launch_time").cast("string").alias("asin_launch_time"),
+            F.col("asin_img_num").cast("int").alias("asin_img_num"),
+            "asin_img_type", "asin_brand_name", "asin_activity_type",
+            F.col("act_one_two_val").cast("double").alias("act_one_two_val"),
+            F.col("act_three_four_val").cast("double").alias("act_three_four_val"),
+            F.col("act_five_six_val").cast("double").alias("act_five_six_val"),
+            F.col("act_eight_val").cast("double").alias("act_eight_val"),
+            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"),
+            "together_asin",
+            F.col("variation_num").cast("int").alias("variation_num"),
+            "account_name", "account_id", "seller_country_name", "parent_asin",
+            F.col("asin_rank_rise").cast("int").alias("asin_rank_rise"),
+            F.col("asin_rank_mom").cast("double").alias("asin_rank_mom"),
+            F.col("asin_rank_yoy").cast("double").alias("asin_rank_yoy"),
+            F.col("asin_ao_rise").cast("double").alias("asin_ao_rise"),
+            F.col("asin_ao_mom").cast("double").alias("asin_ao_mom"),
+            F.col("asin_ao_yoy").cast("double").alias("asin_ao_yoy"),
+            F.col("asin_price_rise").cast("double").alias("asin_price_rise"),
+            F.col("asin_price_mom").cast("double").alias("asin_price_mom"),
+            F.col("asin_price_yoy").cast("double").alias("asin_price_yoy"),
+            F.col("asin_rating_rise").cast("double").alias("asin_rating_rise"),
+            F.col("asin_rating_mom").cast("double").alias("asin_rating_mom"),
+            F.col("asin_rating_yoy").cast("double").alias("asin_rating_yoy"),
+            F.col("asin_comments_rise").cast("int").alias("asin_comments_rise"),
+            F.col("asin_comments_mom").cast("double").alias("asin_comments_mom"),
+            F.col("asin_comments_yoy").cast("double").alias("asin_comments_yoy"),
+            F.col("asin_bsr_orders_rise").cast("int").alias("asin_bsr_orders_rise"),
+            F.col("asin_bsr_orders_mom").cast("double").alias("asin_bsr_orders_mom"),
+            F.col("asin_bsr_orders_yoy").cast("double").alias("asin_bsr_orders_yoy"),
+            F.col("asin_sales_rise").cast("double").alias("asin_sales_rise"),
+            F.col("asin_sales_mom").cast("double").alias("asin_sales_mom"),
+            F.col("asin_sales_yoy").cast("double").alias("asin_sales_yoy"),
+            F.col("asin_variation_rise").cast("int").alias("asin_variation_rise"),
+            F.col("asin_variation_mom").cast("double").alias("asin_variation_mom"),
+            F.col("asin_variation_yoy").cast("double").alias("asin_variation_yoy"),
+            F.col("asin_bought_mom").cast("double").alias("asin_bought_mom"),
+            F.col("asin_bought_yoy").cast("double").alias("asin_bought_yoy"),
+            F.col("asin_size_type").cast("int").alias("asin_size_type"),
+            F.col("asin_rating_type").cast("int").alias("asin_rating_type"),
+            F.col("asin_site_name_type").cast("int").alias("asin_site_name_type"),
+            F.col("asin_weight_type").cast("int").alias("asin_weight_type"),
+            F.col("asin_launch_time_type").cast("int").alias("asin_launch_time_type"),
+            F.col("asin_ao_val_type").cast("int").alias("asin_ao_val_type"),
+            F.col("asin_rank_type").cast("int").alias("asin_rank_type"),
+            F.col("asin_price_type").cast("int").alias("asin_price_type"),
+            F.col("asin_quantity_variation_type").cast("int").alias("asin_quantity_variation_type"),
+            F.col("package_quantity").cast("int").alias("package_quantity"),
+            F.col("is_package_quantity_abnormal").cast("int").alias("is_package_quantity_abnormal"),
+            F.col("is_movie_label").cast("int").alias("is_movie_label"),
+            F.col("is_brand_label").cast("int").alias("is_brand_label"),
+            F.col("is_alarm_brand").cast("int").alias("is_alarm_brand"),
+            F.col("asin_type").cast("int").alias("asin_type"),
+            "asin_crawl_date", "category_first_id", "category_id", "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.col("asin_weight_ratio").cast("double").alias("asin_weight_ratio"),
+            F.col("asin_bought_month").cast("int").alias("asin_bought_month"),
+            F.col("variation_bought_month").cast("int").alias("variation_bought_month"),
+            F.col("variation_bought_month_mom").cast("double").alias("variation_bought_month_mom"),
+            F.col("variation_bought_month_yoy").cast("double").alias("variation_bought_month_yoy"),
+            F.col("asin_lqs_rating").cast("double").alias("asin_lqs_rating"),
+            "asin_lqs_rating_detail", "asin_lob_info",
+            F.col("is_contains_lob_info").cast("int").alias("is_contains_lob_info"),
+            F.col("zr_flow_proportion").cast("double").alias("zr_flow_proportion"),
+            F.col("matrix_flow_proportion").cast("double").alias("matrix_flow_proportion"),
+            F.col("matrix_ao_val").cast("double").alias("matrix_ao_val"),
+            "customer_reviews_json", "img_info",
+            F.col("follow_sellers_count").cast("int").alias("follow_sellers_count"),
+            "asin_describe",
+            F.col("asin_fbm_price").cast("double").alias("asin_fbm_price"),
+            F.col("describe_len").cast("int").alias("describe_len"),
+            F.lit(0.0).cast("double").alias("title_matching_degree"),
+            F.col("multi_color_flag").cast("int").alias("multi_color_flag"),
+            "multi_color_str", "amazon_label",
+            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"),
+            "tracking_since",
+            F.col("tracking_since_type").cast("int").alias("tracking_since_type"),
+            F.col("asin_source_flag").cast("string").alias("asin_source_flag"),
+            F.col("bsr_last_seen_at").cast("string").alias("bsr_last_seen_at"),
+            F.col("bsr_seen_count_30d").cast("int").alias("bsr_seen_count_30d"),
+            F.col("nsr_last_seen_at").cast("string").alias("nsr_last_seen_at"),
+            F.col("nsr_seen_count_30d").cast("int").alias("nsr_seen_count_30d"),
+            F.col("asin_bought_wow").cast("double").alias("asin_bought_wow"),
+            F.col("asin_rank_wow").cast("double").alias("asin_rank_wow"),
+            F.col("asin_price_wow").cast("double").alias("asin_price_wow"),
+            F.col("asin_bsr_orders_wow").cast("double").alias("asin_bsr_orders_wow"),
+            F.col("asin_rating_wow").cast("double").alias("asin_rating_wow"),
+            F.col("asin_comments_wow").cast("double").alias("asin_comments_wow"),
+            F.col("asin_variation_wow").cast("double").alias("asin_variation_wow"),
+            F.col("asin_ao_wow").cast("double").alias("asin_ao_wow"),
+            F.col("asin_sales_wow").cast("double").alias("asin_sales_wow"),
+            F.col("variation_bought_month_wow").cast("double").alias("variation_bought_month_wow"),
+            F.col("asin_bought_wow_last_1w").cast("double").alias("asin_bought_wow_last_1w"),
+            F.col("asin_bought_wow_last_2w").cast("double").alias("asin_bought_wow_last_2w"),
+            F.col("asin_bought_wow_last_3w").cast("double").alias("asin_bought_wow_last_3w"),
+            F.col("asin_bought_rise").cast("int").alias("asin_bought_rise"),
+            F.col("variation_bought_month_rise").cast("int").alias("variation_bought_month_rise"),
+            F.lit(self.site_name).alias("site_name"),
+            F.lit(self.date_type).alias("date_type"),
+            F.lit(self.date_info).alias("date_info")
+        )
+        self.df_save = self.df_save.na.fill(
+            {"asin_zr_counts": 0, "asin_sp_counts": 0, "asin_sb_counts": 0, "asin_vi_counts": 0, "asin_bs_counts": 0,
+             "asin_ac_counts": 0, "asin_tr_counts": 0, "asin_er_counts": 0, "asin_title_len": 0,
+             "asin_total_comments": 0, "variation_num": 0, "asin_img_num": 0, "act_one_two_val": 0.0,
+             "act_three_four_val": 0.0, "act_five_six_val": 0.0, "act_eight_val": 0.0, "one_star": 0,
+             "two_star": 0, "three_star": 0, "four_star": 0, "five_star": 0, "low_star": 0, "asin_size_type": 0,
+             "asin_rating_type": 0, "asin_site_name_type": 0, "asin_weight_type": 0, "asin_launch_time_type": 0,
+             "asin_ao_val_type": 0, "asin_rank_type": 0, "asin_price_type": 0, "asin_quantity_variation_type": 0,
+             "package_quantity": 1, "is_movie_label": 0, "is_brand_label": 0, "is_alarm_brand": 0,
+             "title_matching_degree": 0.0, "asin_lqs_rating": 0.0, "follow_sellers_count": -1, "describe_len": 0,
+             "bsr_seen_count_30d": 0, "nsr_seen_count_30d": 0, "multi_color_flag": 0})
+        self.df_save = self.df_save.repartition(100).persist(StorageLevel.DISK_ONLY)
+        self.df_save = self.df_save.drop_duplicates(['asin']).filter((F.col("asin").isNotNull()) & (F.col("asin") != "") & (F.length(F.col("asin")) <= 10))
+        print("数据量为：", self.df_save.count())
+        self.df_save.show(10, truncate=False)
+
+    # 保存数据
+    def save_data(self):
+        print(f"清除hdfs目录中:{self.hdfs_path}")
+        HdfsUtils.delete_file_in_folder(self.hdfs_path)
+        print(f"当前存储的表名为：{self.hive_tb},分区为{self.partitions_by}")
+        self.df_save.write.saveAsTable(name=self.hive_tb, format='hive', mode='append', partitionBy=self.partitions_by)
+        print("save success")
+
+    def handle_data(self):
+        self.handle_asin_redirect()
+        self.handle_asin_basic_attribute()
+        self.handle_asin_detail_all_type()
+        self.handle_asin_category_info()
+        self.handle_asin_measure()
+        self.handle_asin_auction()
+        # 打断 lazy 链：measure 阶段含自连接，落盘后释放，避免后续 action 触发时计划树过长导致 OOM
+        df_measure_done = self.df_asin_detail
+        self.df_asin_detail = self.df_asin_detail.repartition(100).persist(StorageLevel.DISK_ONLY)
+        cnt = self.df_asin_detail.count()
+        print(f"measure 阶段处理完成，数据量: {cnt}")
+        df_measure_done.unpersist()
+        self.handle_seller_country()
+        self.handle_asin_lqs_rating()
+        self.handle_asin_is_hide()
+        self.handle_change_rate()
+        self.handle_other_new_col()
+        self.handle_multi_color_flag()
+        self.handle_asin_different_source()
+        self.handle_column()
+
+
+if __name__ == '__main__':
+    site_name = sys.argv[1]  # 参数1：站点
+    date_type = sys.argv[2]  # 参数2：类型：week
+    date_info = sys.argv[3]  # 参数3：年-周, 比如: 2026-41
+    assert date_type == 'week', "date_type 只支持 week"
+    handle_obj = DwtFlowAsinWeek(site_name=site_name, date_type=date_type, date_info=date_info)
+    handle_obj.run()