常年可卖/峰值月/属性标签时间限制

Pyspark_job/dwt/dwt_aba_last365.py

@@ -779,6 +779,21 @@ class DwtAbaLast365(object):
             "peak_month": ""
         }).cache()
 
+        # ============================================================
+        # 【日期判断】派生字段各自"起算月"（早于此覆盖为默认值，与月表同口径，前端隐藏）：
+        #   peak_month / all_year_text_flag：< '2025-07' 填默认
+        #   st_attribute_label：< '2025-01' 填默认（属性单独口径）
+        # 默认值：peak_month='' / all_year_text_flag=-1 / st_attribute_label='-1'（Java 转 null 返前端）。
+        # 注：year 类型 date_info 为 '2024-12'/'2025-12'，字符串比较对 YYYY-MM 成立。
+        # ============================================================
+        if self.date_info < '2025-07':
+            self.df_base = self.df_base \
+                .withColumn('peak_month', F.lit('')) \
+                .withColumn('all_year_text_flag', F.lit(-1))
+        if self.date_info < '2025-01':
+            self.df_base = self.df_base \
+                .withColumn('st_attribute_label', F.lit('-1'))
+
     def save_data(self):
         # 重新分区
         self.df_base = self.df_base.repartition(20)

Pyspark_job/dwt/dwt_aba_st_analytics.py

@@ -595,18 +595,21 @@ class DwtAbaStAnalytics(Templates):
             # 【为什么加站点判断】st_attribute_label 只有 us 计算 因为属性是基于us统计 业务要求 uk de不展示 这里填充默认值
             #   非 us 直接跳过读取，st_attribute_label 在 handle_data 里统一填占位 '-1'（Java 转 null 返前端）。
             if self.site_name == 'us':
-                has_theme_data = self.spark.sql(f"""
-                    SELECT 1 FROM dws_st_theme
-                    WHERE site_name='{self.site_name}'
-                      AND date_type='{self.date_type}'
-                      AND date_info='{self.date_info}'
-                    LIMIT 1
-                """).take(1)
-                assert len(has_theme_data) > 0, (
-                    f"上游 dws_st_theme 分区无数据 "
-                    f"site_name={self.site_name}/date_type={self.date_type}/date_info={self.date_info}，"
-                    f"请确认dws_st_theme 调度是否完成；若强行跑 dwt_aba_st_analytics 会导致所有词的 st_attribute_label 错填 '-1'"
-                )
+                # ===== 临时注释（补数用）：补数时 dws_st_theme 可能还没跑好，先关掉存在性校验避免阻塞任务 =====
+                # 影响范围仅 st_attribute_label 一个字段：无数据时它填占位 '-1'（Java 转 null），不影响任何其他字段。
+                # dws_st_theme 调度恢复正常后，请取消下面注释、恢复 assert。
+                # has_theme_data = self.spark.sql(f"""
+                #     SELECT 1 FROM dws_st_theme
+                #     WHERE site_name='{self.site_name}'
+                #       AND date_type='{self.date_type}'
+                #       AND date_info='{self.date_info}'
+                #     LIMIT 1
+                # """).take(1)
+                # assert len(has_theme_data) > 0, (
+                #     f"上游 dws_st_theme 分区无数据 "
+                #     f"site_name={self.site_name}/date_type={self.date_type}/date_info={self.date_info}，"
+                #     f"请确认dws_st_theme 调度是否完成；若强行跑 dwt_aba_st_analytics 会导致所有词的 st_attribute_label 错填 '-1'"
+                # )
 
                 sql = f"""
                 select
@@ -714,14 +717,28 @@ class DwtAbaStAnalytics(Templates):
         # ============================================================
         if self.date_type == 'month':
             self.handle_first_ever_flag()
+            # ============================================================
+            # 【日期判断】派生字段各自"起算月"（早于此填默认值，前端隐藏）：
+            #   peak_month / all_year_text_flag：>= '2025-07' 才算
+            #     （2025-07 前历史 dwt 曾被删 / 老逻辑读空自身历史补过，峰值月/常年可卖口径不可靠）
+            #   st_attribute_label：>= '2025-01' 才算（属性数据单独口径，2025-01 起可靠）
+            # 默认值与非 month 流程一致：peak_month='' / all_year_text_flag=-1 / st_attribute_label='-1'。
+            # ============================================================
+            calc_peak = self.date_info >= '2025-07'   # 峰值月/常年可卖 起算月
+            calc_attr = self.date_info >= '2025-01'   # 属性标签 起算月（单独口径）
+            if calc_peak:
                 self.handle_peak_month()
-            # st_attribute_label 只有 us 计算
-            # 非 us（uk/de）不 join、直接填占位 '-1'，与非 month 流程同款（Java 转 null 返前端）
-            if self.site_name == 'us':
+            else:
+                self.df_save = self.df_save \
+                    .withColumn('peak_month', F.lit('')) \
+                    .withColumn('all_year_text_flag', F.lit(-1))
+            # st_attribute_label 只有 us 且 date_info>=2025-01 才 join 计算
+            # 非 us（uk/de）或早于 2025-01：直接填占位 '-1'（Java 转 null 返前端）
+            if calc_attr and self.site_name == 'us':
                 self.df_save = self.df_save.join(self.df_st_attribute, on='search_term', how='left')
-                self.df_st_attribute.unpersist()
             else:
                 self.df_save = self.df_save.withColumn('st_attribute_label', F.lit('-1'))
+            self.df_st_attribute.unpersist()   # 无论走哪支都释放（非 us/未算时是空 df，unpersist 无害）
             self.df_save = self.df_save.join(self.df_st_filter, on='search_term', how='left')
             self.df_st_filter.unpersist()
         else: