流量选品30天-同环比基准月逻辑优化

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"""