修复日榜单asin传入date_info时环比计算异常

Pyspark_job/my_kafka/kafka_flow_asin_detail_to_doris.py

@@ -838,11 +838,25 @@ class KafkaFlowAsinDetail(Templates):
                 F.col('asin_bought_month').alias(f'{alias_prefix}_asin_bought_month'),
             ).persist(StorageLevel.DISK_ONLY)
 
-        print(f"1a. 读取上个月维度的flow_asin（date_type=month, date_info={self.date_info_last_month}）")
-        self.df_previous_flow_asin = _load_baseline(self.date_info_last_month, 'previous')
+        # 从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')
         self.df_previous_flow_asin.show(10, truncate=False)
-        print(f"1b. 读取同比去年的flow_asin（date_type=month, date_info={self.date_info_last_year}）")
-        self.df_previous_flow_asin_lastyear = _load_baseline(self.date_info_last_year, 'lastyear')
+        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')
         self.df_previous_flow_asin_lastyear.show(10, truncate=False)
         print("2. 获取店铺相关信息")
         sql = f"""