店铺+搜索词增加兜底 历史回刷读取原数据/快照

Pyspark_job/dwt/dwt_aba_st_analytics.py

@@ -362,12 +362,13 @@ class DwtAbaStAnalytics(Templates):
         # self.df_asin_detail.show(10, truncate=True)
 
         # 仅获取 asin和country_name,对country_name进行了聚合处理
+        # dim_fb_asin_info 从 2026-06 开始有分区，历史月读 2026-06
+        asin_info_date = self.date_info if self.date_info >= '2026-06' else '2026-06'
         sql = f"""
-        select
-            asin,
+        select asin,
             concat_ws(\",\",collect_list(cast(fb_country_name as string))) as country_name
         from dim_fb_asin_info
-        where site_name = '{self.site_name}' and date_type = '{self.date_type}' and date_info = '{self.date_info}'
+        where site_name = '{self.site_name}' and date_type = '{self.date_type}' and date_info = '{asin_info_date}'
         group by asin;
         """
         self.df_seller_asin_country = self.spark.sql(sqlQuery=sql)
@@ -375,14 +376,11 @@ class DwtAbaStAnalytics(Templates):
         print("self.df_seller_asin_country:")
         # self.df_seller_asin_country.show(10, truncate=True)
 
-        # 获取 dim_fd_asin_info 表
+        # 获取卖家-asin 信息
         sql = f"""
-        select
-            asin,
-            seller_id as account_id,
-            fb_country_name as country_name
+        select asin, seller_id as account_id, fb_country_name as country_name
         from dim_fb_asin_info
-        where site_name = '{self.site_name}' and date_type = '{self.date_type}' and date_info = '{self.date_info}'
+        where site_name = '{self.site_name}' and date_type = '{self.date_type}' and date_info = '{asin_info_date}'
         """
         self.df_seller_asin_info = self.spark.sql(sqlQuery=sql)
         self.df_seller_asin_info = self.df_seller_asin_info.drop_duplicates(['asin']).repartition(80, 'asin').cache()
@@ -1073,7 +1071,7 @@ class DwtAbaStAnalytics(Templates):
             select 
                 word, 
                 langs 
-            from big_data_selection.tmp_lang_word_frequency;
+            from big_data_selection.lang_word_frequency;
         """
         lang_word_list = self.spark.sql(sql).collect()
         # 转为map

Pyspark_job/dwt/dwt_aba_st_analytics_report.py

@@ -324,7 +324,9 @@ class DwtAbaStAnalyticsReport(Templates):
         self.df_asin_measure = self.spark.sql(sql).repartition(80, 'asin').cache()
         self.df_asin_measure = self.df_asin_measure.na.fill({"asin_bsr_orders": 0})
 
-        print("1.7 读取dim_seller_asin_history_info系列表")
+        # dim_fb_asin_info 从 2026-06 开始有分区，历史月读 2026-06
+        print("1.7 读取dim_fb_asin_info卖家国家信息")
+        asin_info_date = self.date_info if self.date_info >= '2026-06' else '2026-06'
         sql = f"""
             select asin,
             case when upper(fb_country_name) = 'US' then 'US'
@@ -333,7 +335,7 @@ class DwtAbaStAnalyticsReport(Templates):
             when upper(fb_country_name) = 'FR' then 'FR'
             when upper(fb_country_name) = 'DE' then 'DE'
             else 'OTHER' end as seller_name
-            from dim_fb_asin_info where site_name = '{self.site_name}' and date_type = '{self.date_type}' and date_info = '{self.date_info}';
+            from dim_fb_asin_info where site_name = '{self.site_name}' and date_type = '{self.date_type}' and date_info = '{asin_info_date}';
         """
         print("sql:", sql)
         self.df_seller_asin = self.spark.sql(sql).repartition(80, 'asin').cache()

Pyspark_job/dwt/dwt_fb_base_report.py

@@ -23,7 +23,8 @@ import sys
 sys.path.append(os.path.dirname(sys.path[0]))
 from utils.hdfs_utils import HdfsUtils
 from utils.common_util import CommonUtil
-from pyspark.sql.types import IntegerType
+from pyspark.sql.types import IntegerType, StringType, DoubleType
+from pyspark.sql.window import Window
 from utils.spark_util import SparkUtil
 from pyspark.sql import functions as F
 from yswg_utils.common_udf import udf_new_asin_flag       # top20 新品判断仍需 UDF
@@ -62,13 +63,104 @@ class DwtFbBaseReport(object):
 
         # UDF 初始化
         self.udf_new_asin_flag = F.udf(udf_new_asin_flag, IntegerType())      # top20 新品判断仍需 UDF
+        self.u_get_business_val = F.udf(self.get_business_val, StringType())   # 历史月 ODS 回退用
+
+    @staticmethod
+    def get_business_val(seller_address, key):
+        """解析 seller_address 字段，提取 Business Name / Business Address 等值"""
+        if not seller_address:
+            return None
+        parts = [p.strip() for p in seller_address.split("|-|")]
+        for i, p in enumerate(parts):
+            if p.startswith(key):
+                if key in ("Business Address", "Geschäftsadresse"):
+                    return "  ".join(parts[i + 1:]).strip()
+                elif i + 1 < len(parts):
+                    return parts[i + 1].strip()
+        return None
+
+    def _read_fb_detail_from_ods(self, month):
+        """历史月(< 2026-06)回退读 ods_seller_account_feedback + 重做解析，复刻 dim_fb_detail 的逻辑"""
+        print(f"获取 ods_seller_account_feedback（{month}）")
+        sql = f"""
+        select
+            seller_id,
+            num as fb_web_asin_num,
+            count_30_day as count_30_day_num,
+            count_1_year as count_1_year_num,
+            count_lifetime as count_lifetime_num,
+            country_name as fb_country_name,
+            seller_address,
+            seller_rating,
+            feedback_histogram,
+            metadata_json,
+            created_at
+        from ods_seller_account_feedback
+        where site_name = '{self.site_name}'
+          and date_type = '{self.date_type}'
+          and date_info = '{month}'
+          and length(seller_id) > 2
+        """
+        df = self.spark.sql(sqlQuery=sql)
+        print(sql)
 
+        # 同一 seller_id 按 created_at 倒序去重，保留最新一条
+        w_fb = Window.partitionBy('seller_id').orderBy(F.col('created_at').desc())
+        df = df.withColumn('_rn', F.row_number().over(w_fb)) \
+               .filter(F.col('_rn') == 1) \
+               .withColumn('fb_crawl_date', F.date_format(F.col('created_at'), 'yyyy-MM-dd HH:mm:ss')) \
+               .drop('_rn', 'created_at')
+
+        # 解析 seller_rating: "30天|-|90天|-|1年|-|历史"，-1.0 替换为 0.0
+        _rs = F.split(F.col("seller_rating"), r"\|-\|")
+        df = df \
+            .withColumn("rating_30_day_num",
+                F.when(_rs.getItem(0).cast(DoubleType()) == -1.0, F.lit(0.0))
+                 .otherwise(_rs.getItem(0).cast(DoubleType()))) \
+            .withColumn("rating_90_day_num",
+                F.when(_rs.getItem(1).cast(DoubleType()) == -1.0, F.lit(0.0))
+                 .otherwise(_rs.getItem(1).cast(DoubleType()))) \
+            .withColumn("rating_1_year_num",
+                F.when(_rs.getItem(2).cast(DoubleType()) == -1.0, F.lit(0.0))
+                 .otherwise(_rs.getItem(2).cast(DoubleType()))) \
+            .withColumn("rating_lifetime_num",
+                F.when(_rs.getItem(3).cast(DoubleType()) == -1.0, F.lit(0.0))
+                 .otherwise(_rs.getItem(3).cast(DoubleType())))
+
+        # 解析 feedback_histogram JSON -> 5 个星级占比
+        for star in range(5, 0, -1):
+            col_name = f"fb_star_{star}_pct"
+            json_path = f"$.365d.star{star}"
+            df = df.withColumn(col_name,
+                F.when(F.col("feedback_histogram").isNull(), F.lit(-1.0))
+                 .otherwise(F.coalesce(
+                     F.round(F.get_json_object(F.col("feedback_histogram"), json_path).cast(DoubleType()) / 100, 4),
+                     F.lit(-1.0))))
+
+        # metadata_json.totalResultCount 覆盖 fb_web_asin_num
+        df = df.withColumn("fb_web_asin_num",
+            F.when(
+                F.get_json_object(F.col("metadata_json"), "$.totalResultCount").isNotNull(),
+                F.get_json_object(F.col("metadata_json"), "$.totalResultCount").cast(IntegerType())
+            ).otherwise(F.col("fb_web_asin_num")))
+
+        # 解析 business_name / business_addr
+        if self.site_name in ("us", "uk"):
+            df = df.withColumn("business_name", self.u_get_business_val(F.col("seller_address"), F.lit("Business Name"))) \
+                   .withColumn("business_addr", self.u_get_business_val(F.col("seller_address"), F.lit("Business Address")))
+        elif self.site_name == "de":
+            df = df.withColumn("business_name", self.u_get_business_val(F.col("seller_address"), F.lit("Geschäftsname"))) \
+                   .withColumn("business_addr", self.u_get_business_val(F.col("seller_address"), F.lit("Geschäftsadresse")))
+        else:
+            df = df.withColumn("business_name", F.lit(None).cast(StringType())) \
+                   .withColumn("business_addr", F.lit(None).cast(StringType()))
+
+        df = df.drop('seller_address', 'seller_rating', 'feedback_histogram', 'metadata_json')
+        return df
 
     def read_data(self):
-        # 读取 dim_fb_detail 当月数据
-        # 替代: ods_seller_account_feedback + ods_seller_account_syn + MySQL selection.accounts
-        # dim_fb_detail 已预处理: rating_*_num / fb_star_*_pct / business_name / business_addr /
-        #   is_self_account / fb_web_asin_num(含 metadata 覆盖) / fb_crawl_date
+        if self.date_info >= '2026-06':
+            # 2026-06 起读 dim_fb_detail（已预处理 rating/histogram/metadata/business/is_self）
             print("获取 dim_fb_detail（当月）")
             sql = f"""
             select
@@ -87,11 +179,12 @@ class DwtFbBaseReport(object):
             df_cur = self.spark.sql(sqlQuery=sql)
             print(sql)
 
-        # 上月 dim_fb_detail，只取 count 字段用于计算环比
+            # 上月 dim_fb_detail，只取 count 字段算环比
+            last_month_src = 'dim_fb_detail' if self.last_month >= '2026-06' else None
+            if last_month_src:
                 print("获取 dim_fb_detail（上月）")
                 sql_last = f"""
-        select
-            seller_id,
+                select seller_id,
                     count_30_day_num as last_30_day_num,
                     count_1_year_num as last_1_year_num,
                     count_lifetime_num as last_lifetime_num
@@ -102,9 +195,79 @@ class DwtFbBaseReport(object):
                 """
                 df_last = self.spark.sql(sqlQuery=sql_last)
                 print(sql_last)
+            else:
+                # 上月无 dim_fb_detail 分区，回退读 ODS
+                print(f"获取 ods_seller_account_feedback（上月 {self.last_month}，仅 count 字段）")
+                sql_last = f"""
+                select seller_id,
+                    count_30_day as last_30_day_num,
+                    count_1_year as last_1_year_num,
+                    count_lifetime as last_lifetime_num,
+                    created_at
+                from ods_seller_account_feedback
+                where site_name = '{self.site_name}'
+                  and date_type = '{self.date_type}'
+                  and date_info = '{self.last_month}'
+                  and length(seller_id) > 2
+                """
+                df_last = self.spark.sql(sqlQuery=sql_last)
+                w_last = Window.partitionBy('seller_id').orderBy(F.col('created_at').desc())
+                df_last = df_last.withColumn('_rn', F.row_number().over(w_last)) \
+                                 .filter(F.col('_rn') == 1) \
+                                 .drop('_rn', 'created_at')
+                print(sql_last)
+        else:
+            # 历史月(< 2026-06): 读 ods_seller_account_feedback + 重做解析
+            df_cur = self._read_fb_detail_from_ods(self.date_info)
+
+            # account_name 从 ods_seller_account_syn 取（快照表，覆盖全量有名卖家，不受 feedback 月分区限制）
+            print("获取 ods_seller_account_syn（account_name）")
+            sql_syn = f"""
+            select seller_id, account_name
+            from ods_seller_account_syn
+            where site_name = '{self.site_name}'
+            """
+            df_syn = self.spark.sql(sqlQuery=sql_syn).drop_duplicates(['seller_id'])
+            print(sql_syn)
+
+            # is_self_fb 从 dim_fb_detail 2026-06 取（内部店铺标记，极少变动）
+            print("获取 dim_fb_detail 2026-06（is_self_fb）")
+            sql_self = f"""
+            select seller_id, is_self_account as is_self_fb
+            from dim_fb_detail
+            where site_name = '{self.site_name}'
+              and date_type = '{self.date_type}'
+              and date_info = '2026-06'
+            """
+            df_self = self.spark.sql(sqlQuery=sql_self)
+            print(sql_self)
+
+            df_cur = df_cur.join(df_syn, on='seller_id', how='left') \
+                           .join(df_self, on='seller_id', how='left')
+            df_cur = df_cur.na.fill({"is_self_fb": 0})
+
+            # 上月也读 ODS（历史月的上月必然也 < 2026-06）
+            print(f"获取 ods_seller_account_feedback（上月 {self.last_month}，仅 count 字段）")
+            sql_last = f"""
+            select seller_id,
+                count_30_day as last_30_day_num,
+                count_1_year as last_1_year_num,
+                count_lifetime as last_lifetime_num,
+                created_at
+            from ods_seller_account_feedback
+            where site_name = '{self.site_name}'
+              and date_type = '{self.date_type}'
+              and date_info = '{self.last_month}'
+              and length(seller_id) > 2
+            """
+            df_last = self.spark.sql(sqlQuery=sql_last)
+            w_last = Window.partitionBy('seller_id').orderBy(F.col('created_at').desc())
+            df_last = df_last.withColumn('_rn', F.row_number().over(w_last)) \
+                             .filter(F.col('_rn') == 1) \
+                             .drop('_rn', 'created_at')
+            print(sql_last)
 
         # 当月 LEFT JOIN 上月，计算三个 Feedback 环比变化率
-        # 新店铺（上月无记录）left join 后 last_*_num = null -> rate = null（无法计算）
         self.df_fb_feedback = df_cur.join(df_last, on='seller_id', how='left')
         self.df_fb_feedback = self.df_fb_feedback \
             .withColumn('count_30_day_rate',
@@ -116,15 +279,15 @@ class DwtFbBaseReport(object):
             .drop('last_30_day_num', 'last_1_year_num', 'last_lifetime_num') \
             .cache()
 
-        # 读取 dim_fb_asin_info 店铺-ASIN 关系表（替代 ods_seller_asin_account）
-        # 按月度分区读取，不再用 created_at 过滤
-        print("获取 dim_fb_asin_info")
+        # 读取店铺-ASIN 关系表（dim_fb_asin_info 从 2026-06 开始有分区，历史月读 2026-06）
+        asin_info_date = self.date_info if self.date_info >= '2026-06' else '2026-06'
+        print(f"获取 dim_fb_asin_info（date_info={asin_info_date}）")
         sql = f"""
         select seller_id, account_name, asin
         from dim_fb_asin_info
         where site_name = '{self.site_name}'
           and date_type = '{self.date_type}'
-          and date_info = '{self.date_info}'
+          and date_info = '{asin_info_date}'
         """
         self.df_fb_asin = self.spark.sql(sqlQuery=sql)
         self.df_fb_asin = self.df_fb_asin.drop_duplicates(['seller_id', 'asin'])

Pyspark_job/dwt/dwt_fb_top20_asin_info.py

@@ -76,13 +76,15 @@ class DwtFbTop20Info(object):
         print(sql)
 
         # account_name 从 dim_fb_detail 取(替代 ods_seller_account_syn)
-        print("获取 dim_fb_detail (account_name)")
+        # dim_fb_detail 从 2026-06 开始有分区，历史月读 2026-06
+        detail_date = self.date_info if self.date_info >= '2026-06' else '2026-06'
+        print(f"获取 dim_fb_detail (account_name, date_info={detail_date})")
         sql = f"""
             select seller_id, account_name
             from dim_fb_detail
             where site_name = '{self.site_name}'
               and date_type = '{self.date_type}'
-              and date_info = '{self.date_info}'
+              and date_info = '{detail_date}'
         """
         self.df_seller_account = self.spark.sql(sqlQuery=sql)
         print(sql)