流量选品-增加asin近6月月销变化表

Pyspark_job/doris_handle/dwt_flow_asin_month.py

@@ -9,7 +9,8 @@
                     [Step 5] 同步年度聚合指标 dwt.{site}_flow_asin_365day_extra：从 Hive dwt_flow_asin_year 同步过来
                     [Step 6] Doris INSERT OVERWRITE 到 selection 月物化表 selection.{site}_flow_asin_month_{yyyy_mm}[_test]
                     [Step 7] Doris INSERT 到 selection 年表 selection.{site}_flow_asin_365day[_test]
-                    [Step 8] 更新 MySQL workflow_everyday 流程记录表（月/年各写一条，仅 formal 模式）
+                    [Step 8] 写入 selection 近6个月月销变化表 selection.{site}_flow_asin_bought_trend[_test]
+                    [Step 9] 更新 MySQL workflow_everyday 流程记录表（月/年各写一条，仅 formal 模式）
                   支持 us / uk / de 三站点
                   支持 formal / test 模式：
                     - formal：selection 表名无后缀，更新流程记录表
@@ -1526,13 +1527,126 @@ def write_year_table_batched(site_name, live_table, latest_date_info):
 
 
 # ============================================================
-# [Step 8] 流程记录表更新（月+年各写一条，仅 formal 模式）
+# [Step 8] selection 近6个月月销变化表
+# ============================================================
+
+BOUGHT_TREND_MONTHS = 6
+
+
+def build_bought_trend_create_table_sql(table_name):
+    """构建 selection.{site}_flow_asin_bought_trend 建表语句（与 doris_ddl.sql 一致）"""
+    mom_cols = "\n".join(
+        f"    `bought_month_mom_{i}_month_ago`    DECIMAL(20,4)    COMMENT '{i}个月前月销环比',"
+        for i in range(1, BOUGHT_TREND_MONTHS)
+    )
+    rise_cols = ",\n".join(
+        f"    `bought_month_rise_{i}_month_ago`   INT              COMMENT '{i}个月前月销差值'"
+        for i in range(1, BOUGHT_TREND_MONTHS)
+    )
+    return f"""
+CREATE TABLE IF NOT EXISTS `selection`.`{table_name}`
+(
+    `date_info`                       VARCHAR(20)      NOT NULL COMMENT '基准月(yyyy-MM)',
+    `asin`                            VARCHAR(20)      NOT NULL COMMENT 'ASIN',
+    `asin_bought_month`               INT              COMMENT '基准月月销',
+    `bought_month_mom`                DECIMAL(20,4)    COMMENT '基准月月销环比',
+{mom_cols}
+    `bought_month_rise`               INT              COMMENT '基准月月销差值',
+{rise_cols}
+)
+UNIQUE KEY(`date_info`, `asin`)
+COMMENT '流量选品近6个月月销变化'
+AUTO PARTITION BY LIST(`date_info`) ()
+DISTRIBUTED BY HASH(`asin`) BUCKETS 8
+PROPERTIES (
+    "replication_num" = "3",
+    "enable_unique_key_merge_on_write" = "true"
+)
+"""
+
+
+def build_bought_trend_insert_sql(site_name, table_name, date_info):
+    """从 dwt.{site}_flow_asin_month 读基准月及前6个月，行转列得到近6个月环比/差值；
+    环比直接取月表已算好的值，差值 = 当月 - 上月（负值视为NULL）"""
+    # months[0]=基准月，months[6]=6个月前（仅用于计算 rise_5_month_ago）
+    months = [CommonUtil.get_month_offset(date_info, -i) for i in range(0, BOUGHT_TREND_MONTHS + 1)]
+    bought_cols = ",\n".join(
+        f"        MAX(CASE WHEN date_info = '{m}' AND asin_bought_month >= 0 THEN asin_bought_month END) AS b{i}"
+        for i, m in enumerate(months)
+    )
+    mom_cols = ",\n".join(
+        f"        MAX(CASE WHEN date_info = '{m}' THEN bought_month_mom END) AS mom{i}"
+        for i, m in enumerate(months[:BOUGHT_TREND_MONTHS])
+    )
+    select_mom = ",\n".join(
+        [f"    mom0 AS bought_month_mom"] +
+        [f"    mom{i} AS bought_month_mom_{i}_month_ago" for i in range(1, BOUGHT_TREND_MONTHS)]
+    )
+    select_rise = ",\n".join(
+        [f"    b0 - b1 AS bought_month_rise"] +
+        [f"    b{i} - b{i + 1} AS bought_month_rise_{i}_month_ago" for i in range(1, BOUGHT_TREND_MONTHS)]
+    )
+    insert_cols = ", ".join(
+        ["date_info", "asin", "asin_bought_month", "bought_month_mom"] +
+        [f"bought_month_mom_{i}_month_ago" for i in range(1, BOUGHT_TREND_MONTHS)] +
+        ["bought_month_rise"] +
+        [f"bought_month_rise_{i}_month_ago" for i in range(1, BOUGHT_TREND_MONTHS)]
+    )
+    month_in = ", ".join(f"'{m}'" for m in months)
+    return f"""
+INSERT INTO `selection`.`{table_name}` ({insert_cols})
+SELECT
+    '{date_info}' AS date_info,
+    asin,
+    b0 AS asin_bought_month,
+{select_mom},
+{select_rise}
+FROM (
+    SELECT
+        asin,
+        MAX(CASE WHEN date_info = '{date_info}' THEN 1 END) AS has_m0,
+{bought_cols},
+{mom_cols}
+    FROM `{DORIS_DB}`.`{site_name}_flow_asin_month`
+    WHERE date_info IN ({month_in})
+    GROUP BY asin
+) t
+WHERE has_m0 = 1
+"""
+
+
+def write_bought_trend_table(site_name, table_name, date_info):
+    """建表 → 清理基准月旧数据 → INSERT → 行数校验（应与 dwt 月表基准月行数一致）"""
+    print(f"[Step 8] 确保 selection.{table_name} 存在")
+    _exec_doris_sql([build_bought_trend_create_table_sql(table_name)])
+
+    print(f"[Step 8] 清理 selection.{table_name} 旧数据 WHERE date_info='{date_info}'")
+    _exec_doris_sql([f"DELETE FROM `selection`.`{table_name}` WHERE date_info = '{date_info}'"])
+
+    print(f"[Step 8] INSERT INTO selection.{table_name}（基准月 {date_info}）")
+    _exec_doris_sql([build_bought_trend_insert_sql(site_name, table_name, date_info)])
+
+    month_count = _query_doris(
+        f"SELECT COUNT(1) FROM `{DORIS_DB}`.`{site_name}_flow_asin_month` WHERE date_info = '{date_info}'"
+    )[0][0]
+    trend_count = _query_doris(
+        f"SELECT COUNT(1) FROM `selection`.`{table_name}` WHERE date_info = '{date_info}'"
+    )[0][0]
+    print(f"[Step 8] 数据量校验：dwt.{site_name}_flow_asin_month={month_count}，selection.{table_name}={trend_count}")
+    if trend_count != month_count:
+        raise ValueError(
+            f"[Step 8] selection.{table_name} 行数({trend_count})与 dwt 月表基准月行数({month_count})不一致，请人工排查"
+        )
+
+
+# ============================================================
+# [Step 9] 流程记录表更新（月+年各写一条，仅 formal 模式）
 # ============================================================
 
 def modify_mission_record_status(site_name, date_info, result_type):
     """流程记录表更新：仅 formal 模式才入库 mysql workflow_everyday，月/年各写一条记录"""
     if result_type != 'formal':
-        print(f"[Step 8] result_type={result_type}，跳过流程记录表更新")
+        print(f"[Step 9] result_type={result_type}，跳过流程记录表更新")
         return
     record_table = 'workflow_everyday'
     engine_mysql = DBUtil.get_db_engine(db_type=DbTypes.mysql.name, site_name='us')
@@ -1549,9 +1663,9 @@ def modify_mission_record_status(site_name, date_info, result_type):
                 VALUES ('{site_name}', '{date_info}', '{status_remark}', 14, '{table_name_field}', '{date_type}', '流量选品', '是', '{status_remark}', 'doris')
             """
             DBUtil.exec_sql('mysql', 'us', replace_sql)
-            print(f"[Step 8] 流程记录表 workflow_everyday 已写入：{site_name} {date_info} date_type={date_type}")
+            print(f"[Step 9] 流程记录表 workflow_everyday 已写入：{site_name} {date_info} date_type={date_type}")
         else:
-            print(f"[Step 8] 流程记录表已存在该记录，跳过：date_type={date_type}")
+            print(f"[Step 9] 流程记录表已存在该记录，跳过：date_type={date_type}")
 
     _write_record('month', f'{site_name}_flow_asin_last_month', '流量选品计算完毕')
     _write_record('365_day', f'{site_name}_flow_asin_365day', '流量选品近一年计算完毕')
@@ -1566,12 +1680,14 @@ def main(site_name, date_info, result_type='formal'):
     env_suffix             = '_test' if result_type == 'test' else ''
     month_selection_table  = f"{site_name}_flow_asin_month_{date_info_underscore}{env_suffix}"
     year_selection_table   = f"{site_name}_flow_asin_365day{env_suffix}"
+    bought_trend_table     = f"{site_name}_flow_asin_bought_trend{env_suffix}"
     last_12_month          = sorted(CommonUtil.get_month_offset(date_info, -i) for i in range(0, 12))
 
     print(f"启动：site={site_name}, date_info={date_info}, result_type={result_type}")
     print(f"  dwt 主表：dwt.{doris_table}（不区分 test/formal）")
     print(f"  selection 月物化表：selection.{month_selection_table}")
     print(f"  selection 年表：selection.{year_selection_table}")
+    print(f"  selection 月销变化表：selection.{bought_trend_table}")
     print(f"  近12月窗口：{last_12_month}")
 
     spark = SparkUtil.get_spark_session(
@@ -1606,7 +1722,11 @@ def main(site_name, date_info, result_type='formal'):
     print(f"[Step 7] 分批写入 selection.{year_selection_table} 的 copy 表并原子交换")
     write_year_table_batched(site_name, year_selection_table, date_info)
 
-    # ===== [Step 8] 流程记录表更新（月+年各一条，仅 formal 模式）=====
+    # ===== [Step 8] 写入 selection 近6个月月销变化表 =====
+    print(f"[Step 8] 写入 selection.{bought_trend_table}")
+    write_bought_trend_table(site_name, bought_trend_table, date_info)
+
+    # ===== [Step 9] 流程记录表更新（月+年各一条，仅 formal 模式）=====
     modify_mission_record_status(site_name, date_info, result_type)
 
     print("success!")