流量选品月-固定模式增加推送标记

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
     )