uk de不计算属性

Pyspark_job/dwt/dwt_aba_st_analytics.py

@@ -592,33 +592,36 @@ class DwtAbaStAnalytics(Templates):
             # self.df_history_st.show(10, truncate=True)
 
             #  读 dws_st_theme 计算搜索词属性标签 st_attribute_label
-            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'"
-            )
+            # 【为什么加站点判断】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'"
+                )
 
-            sql = f"""
-            select
-                search_term,
-                concat_ws(',', sort_array(collect_set(theme_ch))) as st_attribute_label
-            from dws_st_theme
-            where site_name = '{self.site_name}'
-              and date_type = '{self.date_type}'
-              and date_info = '{self.date_info}'
-              and theme_ch is not null
-            group by search_term
-            """
-            self.df_st_attribute = self.spark.sql(sqlQuery=sql).repartition(80, 'search_term').cache()
-            print("self.df_st_attribute:")
-            # self.df_st_attribute.show(10, truncate=True)
+                sql = f"""
+                select
+                    search_term,
+                    concat_ws(',', sort_array(collect_set(theme_ch))) as st_attribute_label
+                from dws_st_theme
+                where site_name = '{self.site_name}'
+                  and date_type = '{self.date_type}'
+                  and date_info = '{self.date_info}'
+                  and theme_ch is not null
+                group by search_term
+                """
+                self.df_st_attribute = self.spark.sql(sqlQuery=sql).repartition(80, 'search_term').cache()
+                print("self.df_st_attribute:")
+                # self.df_st_attribute.show(10, truncate=True)
 
             # ============================================================
             # 读自身历史 11 个月分区的 rank，用于计算峰值月 peak_month / 常年可卖 all_year_text_flag
@@ -712,8 +715,13 @@ class DwtAbaStAnalytics(Templates):
         if self.date_type == 'month':
             self.handle_first_ever_flag()
             self.handle_peak_month()
-            self.df_save = self.df_save.join(self.df_st_attribute, on='search_term', how='left')
-            self.df_st_attribute.unpersist()
+            # st_attribute_label 只有 us 计算
+            # 非 us（uk/de）不 join、直接填占位 '-1'，与非 month 流程同款（Java 转 null 返前端）
+            if 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_save = self.df_save.join(self.df_st_filter, on='search_term', how='left')
             self.df_st_filter.unpersist()
         else: