流量选品-每日刷新任务适配小站点

Pyspark_job/doris_handle/dws_flow_asin_refresh.py

@@ -21,11 +21,9 @@
                   4) Spark 聚合中间表全量历史 → 趋势物理表 selection.{site}_asin_profit_rate_trend
                   5) 刷新选品模式结果表 selection.user_selection_pattern
                   【站点范围】
-                  目前只支持 us 站点（这几个每日变化字段目前都是 us 站点专属逻辑）
-                  【无入参】
-                  不接收 sys.argv，所有逻辑动态推断；海豚定时调度直接执行 spark-submit 即可
+                  支持 us / uk / de，通过入参 site_name 指定；最新月份等其余逻辑动态推断
 执行示例:
-    spark-submit dws_flow_asin_refresh.py
+    spark-submit dws_flow_asin_refresh.py us
 """
 import os
 import re
@@ -42,7 +40,7 @@ from utils.db_util import DBUtil, DbTypes
 from utils.DorisHelper import DorisHelper
 
 
-SITE_NAME = 'us'  # 目前只支持 us 站点
+SUPPORTED_SITES = ('us', 'uk', 'de')
 
 # ===== 任务5：选品模式配置 =====
 MYSQL_FILTER_TABLE     = 'flow_increment_filter_sql'  # MySQL 筛选模式日志表
@@ -91,11 +89,12 @@ def _exec_doris_sql(sql_list, use_type='selection'):
         conn.close()
 
 
-def get_latest_month(spark):
-    """通过 MySQL workflow_everyday.MAX(report_date) 探测最新的流量选品月份（只取1个月，不再往前推3月）"""
+def get_latest_month(spark, site_name):
+    """通过 MySQL workflow_everyday.MAX(report_date) 探测最新的流量选品月份（只取1个月，不再往前推3月）
+    workflow_everyday 是跨站点共享表(带 site_name 字段区分)，固定连 us 实例"""
     sql_max_month = (
         "select MAX(report_date) as date_info from workflow_everyday "
-        "where site_name = 'us' and date_type = 'month' and page = '流量选品'"
+        f"where site_name = '{site_name}' and date_type = 'month' and page = '流量选品'"
     )
     print(f"sql_max_month = {sql_max_month}")
     mysql_con = DBUtil.get_connection_info('mysql', 'us')
@@ -293,9 +292,10 @@ def spark_aggregate_and_export_trend(spark, site_name):
     print(f"[完成] selection.{trend_table} 已更新, 总 asin 数 {cnt:,}")
 
 
-# ===== 任务5：刷新选品模式结果表 selection.user_selection_pattern（us 站点刷新型 filter）=====
+# ===== 任务5：刷新选品模式结果表 selection.user_selection_pattern（当前站点刷新型 filter）=====
 
 def _get_filter_mysql_engine():
+    """flow_increment_filter_sql 是跨站点共享的控制库表(带 site 字段区分)，固定连 us 实例"""
     return DBUtil.get_db_engine(db_type=DbTypes.mysql.name, site_name='us')
 
 
@@ -312,9 +312,9 @@ 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_'
+def _get_available_pattern_months(doris_cur, site_name):
+    """返回指定站点已存在的月表对应月份列表，升序，排除 _test 表"""
+    prefix = f'{site_name}_flow_asin_month_'
     doris_cur.execute("""
         SELECT TABLE_NAME FROM information_schema.TABLES
         WHERE TABLE_SCHEMA = %s
@@ -379,9 +379,9 @@ def _cleanup_deleted_pattern_partitions(doris_cur, mysql_conn):
             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):
+def _refresh_latest_month(filter_id, where_sql, site_name, latest_month, doris_cur, mysql_conn):
     """对比最新月新旧匹配结果，不同则按 key(filter_id, asin) 删除旧数据后重新插入"""
-    table = f'us_flow_asin_month_{latest_month.replace("-", "_")}'
+    table = f'{site_name}_flow_asin_month_{latest_month.replace("-", "_")}'
     try:
         doris_cur.execute(
             f"SELECT asin FROM `selection`.`{PATTERN_RESULT_TABLE}` "
@@ -430,8 +430,8 @@ def _refresh_latest_month(filter_id, where_sql, latest_month, doris_cur, mysql_c
     _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：历史缺口纯新增追平（不含最新月，且只追平实际存在的月表），最新月走差异化刷新"""
+def _process_refresh_filter(row, site_name, latest_month, available, doris_cur, mysql_conn):
+    """刷新型 filter：历史缺口纯新增追平（不含最新月，且只追平实际存在的月表），最新月走差异化刷新"""
     filter_id   = row['filter_id']
     base_month  = row['base_month']
     where_sql   = row['where_sql']
@@ -442,7 +442,7 @@ def _process_refresh_filter(row, latest_month, available, doris_cur, mysql_conn)
 
     last_ok_month = latest_done
     for month in backfill_months:
-        table = f'us_flow_asin_month_{month.replace("-", "_")}'
+        table = f'{site_name}_flow_asin_month_{month.replace("-", "_")}'
         sql = f"""
             INSERT INTO `selection`.`{PATTERN_RESULT_TABLE}`
                 (filter_id, asin, date_info)
@@ -463,11 +463,11 @@ def _process_refresh_filter(row, latest_month, available, doris_cur, mysql_conn)
             _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)
+    _refresh_latest_month(filter_id, where_sql, site_name, latest_month, doris_cur, mysql_conn)
 
 
-def refresh_user_selection_pattern(latest_month):
-    """任务5: 清理无效分区 + us 站点刷新型 filter 追平历史缺口 + 最新月差异化刷新（date_info 对齐 latest_month）"""
+def refresh_user_selection_pattern(site_name, latest_month):
+    """任务5: 清理无效分区 + 当前站点刷新型 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()
@@ -477,24 +477,24 @@ def refresh_user_selection_pattern(latest_month):
 
         _cleanup_deleted_pattern_partitions(doris_cur, mysql_conn)
 
-        available = _get_available_pattern_months(doris_cur)
+        available = _get_available_pattern_months(doris_cur, site_name)
         if not available:
-            print("    [SKIP] us 站点无可用月表")
+            print(f"    [SKIP] {site_name} 站点无可用月表")
             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'""",
+                WHERE status = 'active' AND site = '{site_name}'""",
             engine
         )
         refresh_df = df[df['where_sql'].apply(_needs_refresh)].reset_index(drop=True)
-        print(f"us 站点刷新型 filter 共 {len(refresh_df)} 个")
+        print(f"{site_name} 站点刷新型 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)
+                _process_refresh_filter(row, site_name, latest_month, available, doris_cur, mysql_conn)
             except Exception as e:
                 print(f"    [ERROR] 未预期异常：{e}")
     finally:
@@ -502,30 +502,32 @@ def refresh_user_selection_pattern(latest_month):
         mysql_conn.close()
 
 
-def main():
-    spark = SparkUtil.get_spark_session("DwsFlowAsinRefresh")
-    latest_month = get_latest_month(spark)
+def main(site_name):
+    spark = SparkUtil.get_spark_session(f"DwsFlowAsinRefresh: {site_name}")
+    latest_month = get_latest_month(spark, site_name)
 
     # 任务1: 月表 selection.{site}_flow_asin_month_{最新月}
-    refresh_month_table(SITE_NAME, latest_month)
+    refresh_month_table(site_name, latest_month)
 
     # 任务2: 年表 selection.{site}_flow_asin_365day
-    refresh_year_table(SITE_NAME, latest_month)
+    refresh_year_table(site_name, latest_month)
 
     # 任务3: 利润率趋势中间表 dwt.dwt_asin_profit_rate_history
-    affected_rows = refresh_profit_trend_staging(SITE_NAME, latest_month)
+    affected_rows = refresh_profit_trend_staging(site_name, latest_month)
 
     # 任务4: Spark 聚合写入趋势物理表 selection.{site}_asin_profit_rate_trend
     if affected_rows > 0:
-        spark_aggregate_and_export_trend(spark, SITE_NAME)
+        spark_aggregate_and_export_trend(spark, site_name)
     else:
         print(f"\n===== [任务4] dwt_asin_profit_rate_history 本次无变化，跳过趋势物理表重聚合 =====")
 
-    # 任务5: 选品模式结果表 selection.user_selection_pattern（us 站点刷新型 filter）
-    refresh_user_selection_pattern(latest_month)
+    # 任务5: 选品模式结果表 selection.user_selection_pattern（当前站点刷新型 filter）
+    refresh_user_selection_pattern(site_name, latest_month)
 
     print("\nsuccess!")
 
 
 if __name__ == "__main__":
-    main()
+    assert len(sys.argv) > 1 and sys.argv[1] in SUPPORTED_SITES, \
+        f"用法: spark-submit dws_flow_asin_refresh.py <site_name>，site_name 需为 {SUPPORTED_SITES} 之一"
+    main(sys.argv[1])