Merge branch 'developer' of http://47.106.101.75/abel_cjy/Amazon-Selection-Data into developer

Pyspark_job/doris_handle/dwt_flow_asin_month.py

@@ -3,22 +3,13 @@
    @Description : 月流量选品 Doris 落地脚本，同时关联产出年度流量选品数据
                   流程：
                     [Step 1] Doris 建表 selection.{site}_flow_asin_month_{yyyy_mm}[_test]
-                    [Step 2] 读 Hive dwt_flow_asin 月数据 + 字段规范化（与 dwt.us_flow_asin_30day
-                             DDL 对齐），合并为一步
-                    [Step 3] 写入 Doris dwt 主表 dwt.{site}_flow_asin_month
-                    [Step 4] 写入 Doris dwt 年表 dwt.{site}_flow_asin_365day：强制重写当月数据 +
-                             补齐缺失月份，靠 UNIQUE KEY+MOW+sequence_col(date_info) 自动去重只
-                             保留每个 asin 最新一条快照，并删除近12月窗口之外的过期数据
-                    [Step 5] 同步年度聚合指标 dwt.{site}_flow_asin_365day_extra：从 Hive
-                             dwt_flow_asin_year（由 dwt/dwt_flow_asin_year.py 算好写入）对应
-                             site_name+date_info 分区同步过来，TRUNCATE 后全量重算
-                    [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 2] 读 Hive dwt_flow_asin 月数据 + 字段规范化（与 dwt.us_flow_asin_30day DDL 对齐）
+                    [Step 3] 写入 Doris dwt 月表 dwt.{site}_flow_asin_month
+                    [Step 4] 写入 Doris dwt 年表 dwt.{site}_flow_asin_365day，并删除近12月窗口之外的过期数据
+                    [Step 5] 同步年度聚合指标 dwt.{site}_flow_asin_365day_extra：从 Hive dwt_flow_asin_year 同步过来
+                    [Step 6] Doris INSERT OVERWRITE 到 selection 月物化表 selection.{site}_flow_asin_month_{yyyy_mm}[_test]
+                    [Step 7] Doris INSERT 到 selection 年表 selection.{site}_flow_asin_365day[_test]
                     [Step 8] 更新 MySQL workflow_everyday 流程记录表（月/年各写一条，仅 formal 模式）
-                  依赖：dwt/dwt_flow_asin_year.py 必须已经跑完同一个 site_name+date_info，
-                       Step 5 才能同步到有效的年度聚合数据
                   支持 us / uk / de 三站点
                   支持 formal / test 模式：
                     - formal：selection 表名无后缀，更新流程记录表
@@ -1028,7 +1019,7 @@ WHERE f.date_info = '{date_info}'
 def build_year_create_table_sql(table_name):
     """构建 selection.{table_name} 建表语句：以 selection 月表（build_month_create_table_sql）
     为基准调整而来——
-      剔除月度专属字段：asin_bought_month / asin_source_flag / bsr_last_seen_at /
+      剔除月度专属字段：asin_source_flag / bsr_last_seen_at /
                        bsr_seen_count_30d / nsr_last_seen_at / nsr_seen_count_30d
       末尾追加：latest_date_info + dwt.{site}_flow_asin_365day_extra 里的年度聚合字段
     旧表已手动清理，直接 CREATE 新表，不再需要 ALTER 兼容旧表结构"""
@@ -1059,6 +1050,7 @@ CREATE TABLE IF NOT EXISTS `selection`.`{table_name}`
     `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,
@@ -1236,6 +1228,7 @@ SELECT
     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,
@@ -1491,8 +1484,7 @@ def write_year_table_batched(site_name, live_table):
         f"ALTER TABLE `selection`.`{live_table}` REPLACE WITH TABLE `{copy_table}` "
         f"PROPERTIES('swap' = 'false')"
     ])
-    print(f"[Step 7] selection.{live_table} 已切换为最新数据，"
-          f"selection.{copy_table} 保留的是交换前的旧数据（可用于回滚）")
+    print(f"[Step 7] selection.{live_table} 已切换为最新数据")
 
 
 # ============================================================

Pyspark_job/doris_handle/dwt_user_selection_pattern.py

@@ -7,6 +7,8 @@
                   - 支持断点续算：latest_computed_month 作为水位线，跳过已算月份
                   - Doris AGGREGATE KEY(filter_id, asin) + MIN(date_info) 自动保留首次入选月份
                   - 每月执行一次即可（只有新月表出现时才会产生缺口）
+                  - 新月数据计算成功后，标记 notify_status=0，通知后端可以推送给用户
+                    （后端推送完成后自行改为1；每日刷新脚本不涉及新月数据，不修改此字段）
 执行示例: python dwt_user_selection_pattern.py us
 """
 import os
@@ -76,15 +78,18 @@ def _get_available_months(doris_cur, site):
     return months
 
 
-def _update_mysql_log(mysql_conn, filter_id, latest_month, status, msg):
+def _update_mysql_log(mysql_conn, filter_id, latest_month, status, msg, mark_notify=False):
+    """mark_notify=True：新月数据计算成功，同时标记 notify_status=0，通知后端可以推送给用户
+    （仅在真正新增了月份数据时传 True，失败/无需计算的场景不会传，因此每日刷新等场景不会碰到这个字段）"""
     now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
+    notify_set = ", notify_status = 0" if mark_notify else ""
     if latest_month is not None:
         # 正常情况：同步更新水位线和状态
         sql    = f"""UPDATE `{MYSQL_FILTER_TABLE}`
                      SET latest_computed_month = %s,
                          last_run_status       = %s,
                          last_run_at           = %s,
-                         last_run_msg          = %s
+                         last_run_msg          = %s{notify_set}
                      WHERE filter_id = %s"""
         params = (latest_month, status, now, msg[:500], filter_id)
     else:
@@ -92,7 +97,7 @@ def _update_mysql_log(mysql_conn, filter_id, latest_month, status, msg):
         sql    = f"""UPDATE `{MYSQL_FILTER_TABLE}`
                      SET last_run_status = %s,
                          last_run_at     = %s,
-                         last_run_msg    = %s
+                         last_run_msg    = %s{notify_set}
                      WHERE filter_id = %s"""
         params = (status, now, msg[:500], filter_id)
     cur = mysql_conn.cursor()
@@ -143,7 +148,8 @@ def _backfill_one_filter(row, site_name, available, doris_cur, mysql_conn):
 
     _update_mysql_log(
         mysql_conn, filter_id, last_ok_month, 'success',
-        f'完成：新增 {len(all_months)} 个月（{all_months[0]} ~ {all_months[-1]}）'
+        f'完成：新增 {len(all_months)} 个月（{all_months[0]} ~ {all_months[-1]}）',
+        mark_notify=True
     )
 
 

Pyspark_job/my_kafka/kafka_flow_asin_detail_to_doris.py

@@ -29,21 +29,34 @@ class KafkaFlowAsinDetail(Templates):
         self.date_info = date_info
         self.consumer_type = consumer_type
         self.test_flag = test_flag
-        # day 模式 date_info 形如 2026-05-12，month 模式形如 2026-03；统一替换 - 为 _ 拼 topic
-        self.year_month = str(self.date_info).replace("-", "_")
-        # date_info_month 统一月份维度（day 模式 '2026-05-01' → '2026-05'，month 模式 '2026-05' 保留）
-        # 用于父 ASIN 详情表 / ASIN 最新详情表的 date_info 字段写入，避免日级/月级混存
-        self.date_info_month      = self.date_info[:7]
-        # date_info_last_month 取上个月（previous 基线）；date_info_last_year 取去年同月（yoy 基线）
-        self.date_info_last_month = CommonUtil.get_month_offset(self.date_info_month, -1)
-        self.date_info_last_year  = CommonUtil.get_month_offset(self.date_info_month, -12)
+        # 用于父 ASIN 详情表 / ASIN 最新详情表的 date_info 字段写入
+        self.date_info_month = self.date_info[:7]
+        # actual_last_month：环比实际基线月，兼容重跑历史数据——
+        # day模式：从workflow_everyday取最新已产出月份（实时口径，始终对比最新完成的月）；
+        # month模式：直接取上一个月（按date_info推算），避免重跑历史月份时基线漂移到"当前最新月"
+        if self.date_type == 'day':
+            _engine_mysql = DBUtil.get_db_engine('mysql', 'us')
+            _sql_latest = (
+                f"SELECT MAX(report_date) "
+                f"FROM workflow_everyday "
+                f"WHERE site_name='{self.site_name}' AND date_type='month' "
+                f"AND page='流量选品' AND status_val=14 AND is_end='是'"
+            )
+            self.actual_last_month = list(DBUtil.engine_exec_sql(_engine_mysql, _sql_latest))[0][0]
+        elif self.date_type == 'month':
+            self.actual_last_month = CommonUtil.get_month_offset(self.date_info, -1)
+        else:
+            print(f"不支持的date_type: {self.date_type}，程序退出")
+            sys.exit(1)
+        self.actual_last_year = CommonUtil.get_month_offset(self.actual_last_month, -11)
+        print(f"环比基线月：{self.actual_last_month}，同比基线月：{self.actual_last_year}")
         # spark相关参数
         self.app_name = self.get_app_name()
         self.spark = SparkUtil.get_stream_spark(app_name=self.app_name)
         self.processing_time = 900 if self.site_name == 'us' else 600
         self.repartition_num = 80
         # kafka相关参数（topic 按 date_type 动态：day → {site}_asin_detail_day_{yyyy_MM_dd}，month → {site}_asin_detail_month_{yyyy_MM}）
-        self.topic_name = f"{self.site_name}_asin_detail_{self.date_type}_{self.year_month}"
+        self.topic_name = f"{self.site_name}_asin_detail_{self.date_type}_{str(self.date_info).replace('-', '_')}"
         self.batch_size = batch_size
         self.batch_size_history = 20000
         self.check_path = f"/home/big_data_selection/tmp/kafka_checkpoint/{self.topic_name}_{self.consumer_type}_test" if self.test_flag == 'test' else f"/home/big_data_selection/tmp/kafka_checkpoint/{self.topic_name}_{self.consumer_type}"
@@ -834,25 +847,11 @@ class KafkaFlowAsinDetail(Templates):
                 F.col('asin_bought_month').alias(f'{alias_prefix}_asin_bought_month'),
             ).persist(StorageLevel.DISK_ONLY)
 
-        # 从MySQL流程记录表获取最新已产出月份，避免查Hive慢
-        _engine_mysql = DBUtil.get_db_engine('mysql', 'us')
-        _sql_latest = (
-            f"SELECT MAX(report_date) "
-            f"FROM workflow_everyday "
-            f"WHERE site_name='{self.site_name}' AND date_type='month' "
-            f"AND page='流量选品' AND status_val=14 AND is_end='是'"
-        )
-        _result = list(DBUtil.engine_exec_sql(_engine_mysql, _sql_latest))
-        actual_last_month = (_result[0][0] if _result and _result[0][0]
-                             else self.date_info_last_month)
-        actual_last_year  = CommonUtil.get_month_offset(actual_last_month, -11)
-        print(f"环比基线月：{actual_last_month}（原始推算：{self.date_info_last_month}），同比基线月：{actual_last_year}")
-
-        print(f"1a. 读取上个月维度的flow_asin（date_type=month, date_info={actual_last_month}）")
-        self.df_previous_flow_asin = _load_baseline(actual_last_month, 'previous')
+        print(f"1a. 读取上个月维度的flow_asin（date_type=month, date_info={self.actual_last_month}）")
+        self.df_previous_flow_asin = _load_baseline(self.actual_last_month, 'previous')
         self.df_previous_flow_asin.show(10, truncate=False)
-        print(f"1b. 读取同比去年的flow_asin（date_type=month, date_info={actual_last_year}）")
-        self.df_previous_flow_asin_lastyear = _load_baseline(actual_last_year, 'lastyear')
+        print(f"1b. 读取同比去年的flow_asin（date_type=month, date_info={self.actual_last_year}）")
+        self.df_previous_flow_asin_lastyear = _load_baseline(self.actual_last_year, 'lastyear')
         self.df_previous_flow_asin_lastyear.show(10, truncate=False)
         print("2. 获取店铺相关信息")
         sql = f"""