用户固定选品模式纳入每日刷新流程

Pyspark_job/doris_handle/dws_flow_asin_refresh.py

@@ -18,9 +18,7 @@
                   2) 刷新年表：selection.{site}_flow_asin_365day，限定 latest_date_info=最新月 这批 asin
                   3) 刷新利润率趋势中间表：dwt.dwt_asin_profit_rate_history，只处理最新1个月（原来是近3月）
                   4) Spark 聚合中间表全量历史 → 趋势物理表 selection.{site}_asin_profit_rate_trend
-                  1/2/3 都只在"新值和已存值不一致"时才真正写入（Doris 支持 NULL-safe 的 <=>，已在生产
-                  环境验证过 有值<=>NULL 会判定为不相等），避免每天对没有变化的行也触发一次 MOW 合并写；
-                  4 本身就是全量重聚合一遍很小的物理表，不受"只刷新最新月"影响，逻辑不变。
+                  5) 刷新选品模式结果表 selection.user_selection_pattern
                   【站点范围】
                   目前只支持 us 站点（这几个每日变化字段目前都是 us 站点专属逻辑）
                   【无入参】
@@ -29,19 +27,33 @@
     spark-submit dws_flow_asin_refresh.py
 """
 import os
+import re
 import sys
+from datetime import datetime
 
 sys.path.append(os.path.dirname(sys.path[0]))
 
 from pyspark.sql import functions as F, Window
+import pandas as pd
 
 from utils.spark_util import SparkUtil
-from utils.db_util import DBUtil
+from utils.db_util import DBUtil, DbTypes
 from utils.DorisHelper import DorisHelper
 
 
 SITE_NAME = 'us'  # 目前只支持 us 站点
 
+# ===== 任务5：选品模式配置 =====
+MYSQL_FILTER_TABLE     = 'flow_increment_filter_sql'  # MySQL 筛选模式日志表
+PATTERN_RESULT_TABLE   = 'user_selection_pattern'     # selection 库下的选品模式结果表
+PATTERN_DELETE_BATCH   = 500  # DELETE IN 列表单批大小，避开 Doris IN 元素/表达式树数量上限
+PATTERN_REFRESH_FIELDS = {
+    'ocean_profit', 'air_profit',
+    'launch_time', 'launch_time_type',
+    'tracking_since', 'tracking_since_type',
+    'package_length', 'package_width', 'package_height', 'item_weight',
+}
+
 
 # ===== Doris 连接 / 执行 =====
 
@@ -275,6 +287,215 @@ def spark_aggregate_and_export_trend(spark, site_name):
     print(f"[完成] selection.{trend_table} 已更新, 总 asin 数 {cnt:,}")
 
 
+# ===== 任务5：刷新选品模式结果表 selection.user_selection_pattern（us 站点刷新型 filter）=====
+
+def _get_filter_mysql_engine():
+    return DBUtil.get_db_engine(db_type=DbTypes.mysql.name, site_name='us')
+
+
+def _next_month(month_str):
+    """'2026-05' → '2026-06'"""
+    y, m = int(month_str[:4]), int(month_str[5:])
+    m += 1
+    if m > 12:
+        y, m = y + 1, 1
+    return f'{y}-{m:02d}'
+
+
+def _needs_refresh(where_sql):
+    return any(re.search(rf'\b{re.escape(field)}\b', where_sql) for field in PATTERN_REFRESH_FIELDS)
+
+
+def _get_available_pattern_months(doris_cur):
+    """返回 us 站点已存在的月表对应月份列表，升序，排除 _test 表"""
+    prefix = 'us_flow_asin_month_'
+    doris_cur.execute("""
+        SELECT TABLE_NAME FROM information_schema.TABLES
+        WHERE TABLE_SCHEMA = %s
+          AND TABLE_NAME LIKE %s
+          AND TABLE_NAME NOT LIKE '%%\_test'
+        ORDER BY TABLE_NAME
+    """, ('selection', f'{prefix}%'))
+    months = []
+    for (table_name,) in doris_cur.fetchall():
+        suffix = table_name[len(prefix):]    # '2026_05'
+        month  = suffix.replace('_', '-')   # '2026-05'
+        if len(month) == 7:
+            months.append(month)
+    return months
+
+
+def _update_pattern_mysql_log(mysql_conn, filter_id, latest_month, status, msg):
+    now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
+    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
+                     WHERE filter_id = %s"""
+        params = (latest_month, status, now, msg[:500], filter_id)
+    else:
+        # latest_month 为 None：仅记录失败状态，不更新水位线（首月失败 / 刷新失败均适用）
+        sql    = f"""UPDATE `{MYSQL_FILTER_TABLE}`
+                     SET last_run_status = %s,
+                         last_run_at     = %s,
+                         last_run_msg    = %s
+                     WHERE filter_id = %s"""
+        params = (status, now, msg[:500], filter_id)
+    cur = mysql_conn.cursor()
+    cur.execute(sql, params)
+    mysql_conn.commit()
+    cur.close()
+
+
+def _cleanup_deleted_pattern_partitions(doris_cur, mysql_conn):
+    """清理MySQL中status=deleted的filter_id对应的Doris无效分区，避免分区膨胀"""
+    cur = mysql_conn.cursor()
+    cur.execute(f"SELECT filter_id FROM `{MYSQL_FILTER_TABLE}` WHERE status = 'deleted'")
+    deleted_ids = [r[0] for r in cur.fetchall()]
+    cur.close()
+    if not deleted_ids:
+        print("    [CLEANUP] 无已删除filter_id，跳过分区清理")
+        return
+
+    print(f"    [CLEANUP] 检测到 {len(deleted_ids)} 个已删除filter_id，开始清理分区")
+    for filter_id in deleted_ids:
+        doris_cur.execute(f"SELECT auto_partition_name('list', {filter_id})")
+        partition_name = doris_cur.fetchone()[0]
+        try:
+            doris_cur.execute(
+                f"ALTER TABLE `selection`.`{PATTERN_RESULT_TABLE}` "
+                f"DROP PARTITION IF EXISTS `{partition_name}`"
+            )
+            print(f"      [DROP] filter_id={filter_id} → {partition_name}")
+        except Exception as e:
+            print(f"      [DROP FAIL] filter_id={filter_id}  错误：{str(e)[:200]}")
+
+
+def _refresh_latest_month(filter_id, where_sql, latest_month, doris_cur, mysql_conn):
+    """对比最新月新旧匹配结果，不同则按 key(filter_id, asin) 删除旧数据后重新插入"""
+    table = f'us_flow_asin_month_{latest_month.replace("-", "_")}'
+    try:
+        doris_cur.execute(
+            f"SELECT asin FROM `selection`.`{PATTERN_RESULT_TABLE}` "
+            f"WHERE filter_id = {filter_id} AND date_info = '{latest_month}'"
+        )
+        old_asins = {r[0] for r in doris_cur.fetchall()}
+
+        doris_cur.execute(f"SELECT asin FROM `selection`.`{table}` WHERE {where_sql}")
+        new_asins = {r[0] for r in doris_cur.fetchall()}
+
+        if old_asins == new_asins:
+            print(f"    [刷新] {latest_month} 无变化，跳过")
+            msg = f'刷新完成：最新月 {latest_month} 无变化'
+        else:
+            if old_asins:
+                doris_cur.execute(f"SELECT auto_partition_name('list', {filter_id})")
+                partition_name = doris_cur.fetchone()[0]
+                old_asins_list = list(old_asins)
+                for i in range(0, len(old_asins_list), PATTERN_DELETE_BATCH):
+                    batch = old_asins_list[i:i + PATTERN_DELETE_BATCH]
+                    asin_list = ",".join(f"'{a}'" for a in batch)
+                    doris_cur.execute(
+                        f"DELETE FROM `selection`.`{PATTERN_RESULT_TABLE}` PARTITION ({partition_name}) "
+                        f"WHERE filter_id = {filter_id} AND asin IN ({asin_list})"
+                    )
+                print(f"    [刷新] {latest_month} 删除旧数据 {len(old_asins)} 条")
+
+            doris_cur.execute(f"""
+                INSERT INTO `selection`.`{PATTERN_RESULT_TABLE}`
+                    (filter_id, asin, date_info)
+                SELECT
+                    {filter_id}  AS filter_id,
+                    asin,
+                    '{latest_month}' AS date_info
+                FROM `selection`.`{table}`
+                WHERE {where_sql}
+            """)
+            print(f"    [刷新] {latest_month} 写入新数据 {len(new_asins)} 条")
+            msg = f'刷新完成：最新月 {latest_month} 差异化刷新（新增{len(new_asins - old_asins)}，剔除{len(old_asins - new_asins)}）'
+    except Exception as e:
+        err = str(e)[:200]
+        print(f"    [刷新 FAIL] {latest_month}  错误：{err}")
+        _update_pattern_mysql_log(mysql_conn, filter_id, None, 'failed', err)
+        return
+
+    _update_pattern_mysql_log(mysql_conn, filter_id, latest_month, 'success', msg)
+
+
+def _process_refresh_filter(row, latest_month, available, doris_cur, mysql_conn):
+    """us 站点刷新型 filter：历史缺口纯新增追平（不含最新月，且只追平实际存在的月表），最新月走差异化刷新"""
+    filter_id   = row['filter_id']
+    base_month  = row['base_month']
+    where_sql   = row['where_sql']
+    latest_done = row['latest_computed_month']  # None 表示从未计算过
+
+    start_month = _next_month(latest_done) if latest_done else base_month
+    backfill_months = [m for m in available if start_month <= m < latest_month]
+
+    last_ok_month = latest_done
+    for month in backfill_months:
+        table = f'us_flow_asin_month_{month.replace("-", "_")}'
+        sql = f"""
+            INSERT INTO `selection`.`{PATTERN_RESULT_TABLE}`
+                (filter_id, asin, date_info)
+            SELECT
+                {filter_id}  AS filter_id,
+                asin,
+                '{month}'    AS date_info
+            FROM `selection`.`{table}`
+            WHERE {where_sql}
+        """
+        try:
+            doris_cur.execute(sql)
+            print(f"    [BACKFILL OK] {month}")
+            last_ok_month = month
+        except Exception as e:
+            err = str(e)[:200]
+            print(f"    [BACKFILL FAIL] {month}  错误：{err}")
+            _update_pattern_mysql_log(mysql_conn, filter_id, last_ok_month, 'failed', err)
+            return
+
+    _refresh_latest_month(filter_id, where_sql, latest_month, doris_cur, mysql_conn)
+
+
+def refresh_user_selection_pattern(latest_month):
+    """任务5: 清理无效分区 + us 站点刷新型 filter 追平历史缺口 + 最新月差异化刷新（date_info 对齐 latest_month）"""
+    print(f"\n========== [任务5/选品模式] selection.{PATTERN_RESULT_TABLE} (latest_month={latest_month}) ==========")
+    engine     = _get_filter_mysql_engine()
+    doris_conn = _doris_connect()
+    mysql_conn = engine.raw_connection()
+    try:
+        doris_cur = doris_conn.cursor()
+
+        _cleanup_deleted_pattern_partitions(doris_cur, mysql_conn)
+
+        available = _get_available_pattern_months(doris_cur)
+        if not available:
+            print("    [SKIP] us 站点无可用月表")
+            return
+
+        df = pd.read_sql(
+            f"""SELECT filter_id, base_month, where_sql, latest_computed_month
+                FROM `{MYSQL_FILTER_TABLE}`
+                WHERE status = 'active' AND site = 'us'""",
+            engine
+        )
+        refresh_df = df[df['where_sql'].apply(_needs_refresh)].reset_index(drop=True)
+        print(f"us 站点刷新型 filter 共 {len(refresh_df)} 个")
+
+        for i, row in refresh_df.iterrows():
+            print(f"  [{i + 1}/{len(refresh_df)}] filter_id={row['filter_id']}")
+            try:
+                _process_refresh_filter(row, latest_month, available, doris_cur, mysql_conn)
+            except Exception as e:
+                print(f"    [ERROR] 未预期异常：{e}")
+    finally:
+        doris_conn.close()
+        mysql_conn.close()
+
+
 def main():
     spark = SparkUtil.get_spark_session("DwsFlowAsinRefresh")
     latest_month = get_latest_month(spark)
@@ -294,6 +515,9 @@ def main():
     else:
         print(f"\n===== [任务4] dwt_asin_profit_rate_history 本次无变化，跳过趋势物理表重聚合 =====")
 
+    # 任务5: 选品模式结果表 selection.user_selection_pattern（us 站点刷新型 filter）
+    refresh_user_selection_pattern(latest_month)
+
     print("\nsuccess!")
 
 

Pyspark_job/doris_handle/dws_user_selection_pattern.py → Pyspark_job/doris_handle/dwt_user_selection_pattern.py

 """
    @Author      : CT
-   @Description : 根据用户筛选模式日志批量计算选品结果，写入 Doris selection.user_selection_pattern
-                  - 串行逐 filter_id 执行，减小数据库压力
+   @Description : 按站点增量追平用户筛选模式选品结果，写入 Doris selection.user_selection_pattern
+                  - 传入 site_name，只处理该站点的 active filter
+                  - 纯增量追平：水位线落后于该站点最新月表时补齐缺失月份，不做刷新判断
+                    （us 站点利润率/Keepa 字段每日刷新型 filter，最新月差异化刷新见 dws_flow_asin_refresh.py 第5步）
                   - 支持断点续算：latest_computed_month 作为水位线，跳过已算月份
                   - Doris AGGREGATE KEY(filter_id, asin) + MIN(date_info) 自动保留首次入选月份
-                  - us 站点利润率/Keepa 字段每日刷新：TRUNCATE 分区(p{filter_id}) + 全量重算
-                  - 水位线规则：成功→维护到最新月份；失败→维护到失败的前一个月
+                  - 每月执行一次即可（只有新月表出现时才会产生缺口）
+执行示例: python dwt_user_selection_pattern.py us
 """
 import os
-import re
 import sys
 from datetime import datetime
 
@@ -23,11 +24,6 @@ MYSQL_FILTER_TABLE = 'flow_increment_filter_sql'  # MySQL 筛选模式日志表
 DORIS_RESULT_DB    = 'selection'
 DORIS_RESULT_TABLE = 'user_selection_pattern'
 SUPPORTED_SITES    = ('us', 'uk', 'de')
-REFRESH_FIELDS   = {
-    'ocean_profit', 'air_profit',
-    'launch_time', 'launch_time_type',
-    'tracking_since', 'tracking_since_type',
-}
 
 
 # ===== 连接工厂 =====
@@ -46,6 +42,7 @@ def _get_doris_conn():
 
 
 def _get_mysql_engine():
+    """flow_increment_filter_sql 是跨站点共享的控制库表(带 site 字段区分)，固定连 us 实例"""
     return DBUtil.get_db_engine(db_type=DbTypes.mysql.name, site_name='us')
 
 
@@ -79,10 +76,6 @@ def _get_available_months(doris_cur, site):
     return months
 
 
-def _needs_refresh(where_sql):
-    return any(re.search(rf'\b{re.escape(field)}\b', where_sql) for field in REFRESH_FIELDS)
-
-
 def _update_mysql_log(mysql_conn, filter_id, latest_month, status, msg):
     now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
     if latest_month is not None:
@@ -95,7 +88,7 @@ def _update_mysql_log(mysql_conn, filter_id, latest_month, status, msg):
                      WHERE filter_id = %s"""
         params = (latest_month, status, now, msg[:500], filter_id)
     else:
-        # 首月即失败：不更新水位线，仅记录失败状态
+        # latest_month 为 None：仅记录失败状态，不更新水位线
         sql    = f"""UPDATE `{MYSQL_FILTER_TABLE}`
                      SET last_run_status = %s,
                          last_run_at     = %s,
@@ -110,57 +103,24 @@ def _update_mysql_log(mysql_conn, filter_id, latest_month, status, msg):
 
 # ===== 核心计算 =====
 
-def _compute_one_filter(row, doris_cur, mysql_conn, months_by_site):
+def _backfill_one_filter(row, site_name, available, doris_cur, mysql_conn):
+    """水位线落后于最新月表时，纯新增 INSERT 追平缺失月份（不做刷新判断）"""
     filter_id   = row['filter_id']
-    site        = row['site']
     base_month  = row['base_month']
     where_sql   = row['where_sql']
     latest_done = row['latest_computed_month']  # None 表示从未计算过
 
-    available = months_by_site.get(site, [])
-    if not available:
-        print(f"    [SKIP] 站点 {site} 无可用月表")
-        return
-
-    is_refresh = _needs_refresh(where_sql) and site == 'us'
-
-    if is_refresh:
-        # 刷新模式：TRUNCATE 分区 + 从 base_month 全量重算
-        doris_cur.execute(f"SELECT auto_partition_name('list', {filter_id})")
-        partition_name = doris_cur.fetchone()[0]
-        truncate_sql = (f"TRUNCATE TABLE `{DORIS_RESULT_DB}`.`{DORIS_RESULT_TABLE}` "
-                        f"PARTITION ({partition_name})")
-        try:
-            doris_cur.execute(truncate_sql)
-            print(f"    [TRUNCATE] 已清空分区 {partition_name}")
-        except Exception as e:
-            err_str = str(e)
-            if 'does not exist' in err_str:
-                # 首次计算时分区尚未创建，跳过 TRUNCATE 直接 INSERT
-                print(f"    [TRUNCATE SKIP] 分区 {partition_name} 不存在（首次计算），跳过清空")
-            else:
-                err = err_str[:200]
-                print(f"    [TRUNCATE FAIL] {err}")
-                _update_mysql_log(mysql_conn, filter_id, latest_done, 'failed', f'TRUNCATE失败: {err}')
-                return
-        all_months  = [m for m in available if m >= base_month]
-        mode_label  = '刷新重算'
-        last_ok_month = None
-    else:
-        # 补算模式：水位线增量
+    latest_available = available[-1]
     start_month = _next_month(latest_done) if latest_done else base_month
-        all_months  = [m for m in available if start_month <= m <= available[-1]]
-        mode_label  = '新增'
-        last_ok_month = None
-
+    all_months  = [m for m in available if start_month <= m <= latest_available]
     if not all_months:
-        print(f"    [SKIP] 无需计算（latest_computed={latest_done}，refresh={is_refresh}）")
+        print(f"    [SKIP] 无需计算（latest_computed={latest_done}，已追平至最新月 {latest_available}）")
         return
 
-    print(f"    [{mode_label}] 计算范围：{all_months[0]} ~ {all_months[-1]}，共 {len(all_months)} 个月")
-
+    print(f"    [新增] 计算范围：{all_months[0]} ~ {all_months[-1]}，共 {len(all_months)} 个月")
+    last_ok_month = None
     for month in all_months:
-        table = f'{site}_flow_asin_month_{month.replace("-", "_")}'
+        table = f'{site_name}_flow_asin_month_{month.replace("-", "_")}'
         sql = f"""
             INSERT INTO `{DORIS_RESULT_DB}`.`{DORIS_RESULT_TABLE}`
                 (filter_id, asin, date_info)
@@ -183,40 +143,43 @@ def _compute_one_filter(row, doris_cur, mysql_conn, months_by_site):
 
     _update_mysql_log(
         mysql_conn, filter_id, last_ok_month, 'success',
-        f'完成：{mode_label} {len(all_months)} 个月（{all_months[0]} ~ {all_months[-1]}）'
+        f'完成：新增 {len(all_months)} 个月（{all_months[0]} ~ {all_months[-1]}）'
     )
 
 
 # ===== 入口 =====
 
 def main():
+    assert len(sys.argv) == 2 and sys.argv[1] in SUPPORTED_SITES, \
+        f"用法: python dwt_user_selection_pattern.py <site_name>，site_name 需为 {SUPPORTED_SITES} 之一"
+    site_name = sys.argv[1]
+
     doris_conn = _get_doris_conn()
     engine     = _get_mysql_engine()
     mysql_conn = engine.raw_connection()
     try:
         doris_cur = doris_conn.cursor()
 
-        # 预加载各站点可用月份
-        months_by_site = {site: _get_available_months(doris_cur, site) for site in SUPPORTED_SITES}
-        for site, months in months_by_site.items():
-            rng = f"{months[0]} ~ {months[-1]}" if months else "无"
-            print(f"[站点 {site}] 可用月份：{rng}（共 {len(months)} 个）")
+        available = _get_available_months(doris_cur, site_name)
+        if not available:
+            print(f"[SKIP] 站点 {site_name} 无可用月表")
+            return
+        print(f"[站点 {site_name}] 可用月份：{available[0]} ~ {available[-1]}（共 {len(available)} 个）")
 
-        # 读取全部 active 筛选模式
         df = pd.read_sql(
-            f"""SELECT filter_id, site, base_month, where_sql, latest_computed_month
+            f"""SELECT filter_id, base_month, where_sql, latest_computed_month
                 FROM `{MYSQL_FILTER_TABLE}`
-                WHERE status = 'active'
+                WHERE status = 'active' AND site = '{site_name}'
                 ORDER BY filter_id""",
             engine
         )
         total = len(df)
-        print(f"\n共 {total} 条 active 筛选模式，开始串行计算...\n")
+        print(f"\n共 {total} 条 active 筛选模式（站点={site_name}），开始串行计算...\n")
 
         for i, row in df.iterrows():
-            print(f"[{i + 1}/{total}] filter_id={row['filter_id']}  site={row['site']}")
+            print(f"[{i + 1}/{total}] filter_id={row['filter_id']}")
             try:
-                _compute_one_filter(row, doris_cur, mysql_conn, months_by_site)
+                _backfill_one_filter(row, site_name, available, doris_cur, mysql_conn)
             except Exception as e:
                 print(f"    [ERROR] 未预期异常：{e}")