店铺分类聚合表处理分类归属

Pyspark_job/dwt/dwt_fb_category_report.py

@@ -7,14 +7,21 @@
                   本次只换数据源(ODS -> DIM)，聚合与三个占比公式沿用原逻辑不动。
    @SourceTable :
                   1.dim_fb_detail / ods_seller_account_feedback  (店铺集合，按月门控)
-                  2.dwt_flow_asin                                (当月：account_id 店铺归属 + category_first_id 一级分类 + asin_is_new 新品 + 市场分母)
-                  3.dim_bsr_category_tree                        (一级分类名称)
+                  2.dwt_flow_asin                                (当月：account_id 店铺归属 + category_first_id 一级分类 + asin_is_new 新品 + 市场分母，
+                                                                  category_id 当前节点 / desc_category_first_id 面包屑一级 用于纠正错标一级)
+                  3.dim_bsr_category_tree                        (一级分类名称 + 有效部门 / 重定向 / 节点所属部门，用于纠正错标一级)
    @SinkTable   :
                   1.dwt_fb_category_report
    @CreateTime  : 2023/07/18 17:33
-   @UpdateTime  : 2026/09/10 00:00
+   @UpdateTime  : 2026/10/11 00:57
+                  2026/09/10 00:00
                   2026-09 按月严格同源改口径：店铺-asin 归属/一级分类/新品/市场分母全部改用 dwt_flow_asin 当月，
                   与主表 dwt_fb_base_report(已改 flow 当月)对齐；原 dim_fb_asin_info(全历史名单)+dim_cal_asin_history_detail(全历史)弃用。
+                  2026/10/11 00:57
+                  纠正错标一级：flow 一级取第一个排名链接，遇到亚马逊隐藏榜/特殊榜(private-brands、book-special-features、
+                  audible 等)会把榜单当一级，分类树里取不到名称，页面出现"ID 有值、名称为无"。新增 fix_bsr_cate_1_id，
+                  每个 asin 按自己的当前节点(节点重定向一级，无效时用节点链接一级) / 面包屑一级换成有效部门，
+                  修不好的归入"无"；分类树读取统一加 delete_time is null(有值=已失效)。
 """
 
 import os
@@ -103,11 +110,15 @@ class DwtFbCategoryReport(object):
         #   新品标记: self.df_flow_asin = spark.sql("select asin, asin_is_new as is_asin_new from dwt_flow_asin where ...").drop_duplicates(['asin'])
         #   分类+市场分母: self.df_asin_history = spark.sql("select asin, category_first_id as bsr_cate_1_id from dim_cal_asin_history_detail where site_name=...").cache()
         # flow 当月：account_id 归属店铺、category_first_id 当月一级分类、asin_is_new 新品标记，一次取全
-        print("获取 dwt_flow_asin（account_id/一级分类/新品，当月）")
+        # 旧代码只取上面 4 列，现多取 category_id(当前节点)、desc_category_first_id(面包屑一级)，给 fix_bsr_cate_1_id 纠正错标一级用：
+        # select account_id as seller_id, asin, asin_is_new as is_asin_new, category_first_id as bsr_cate_1_id from dwt_flow_asin where ...
+        print("获取 dwt_flow_asin（account_id/一级分类/新品/当前节点/面包屑一级，当月）")
         sql = f"""
         select account_id as seller_id, asin,
                asin_is_new as is_asin_new,
-               category_first_id as bsr_cate_1_id
+               category_first_id as bsr_cate_1_id,
+               category_id,
+               desc_category_first_id
         from dwt_flow_asin
         where site_name = '{self.site_name}'
           and date_type = '{self.date_type}'
@@ -115,6 +126,9 @@ class DwtFbCategoryReport(object):
         """
         df_flow = self.spark.sql(sqlQuery=sql).drop_duplicates(['asin'])
         print(sql)
+
+        # 纠正错标一级。必须放在下面拆分市场分母/店铺分子之前，两边才用同一套一级
+        df_flow = self.fix_bsr_cate_1_id(df_flow)
         # 市场分母用：全站当月 flow asin（不筛 account_id）
         self.df_flow_all = df_flow.cache()
         # 店铺归属用：account_id 非空 = 当月归属到某店的 asin
@@ -125,16 +139,99 @@ class DwtFbCategoryReport(object):
         self.df_fb_asin_info = self.df_fb_asin_info.drop_duplicates(['seller_id', 'asin']).cache()
 
         # 一级分类名称
+        # 旧代码不过滤 delete_time，已失效的部门(如 uk audible)也能取到名称；现加 delete_time is null，
+        # 和 fix_bsr_cate_1_id 的有效部门同一口径，纠正后的一级一定取得到名称
+        # select category_id as bsr_cate_1_id, en_name as bsr_cate_1_name from big_data_selection.dim_bsr_category_tree
+        # where site_name = '{self.site_name}' and category_parent_id = 0
         print("获取 dim_bsr_category_tree")
         sql = f"""
             select category_id as bsr_cate_1_id, en_name as bsr_cate_1_name
             from big_data_selection.dim_bsr_category_tree
             where site_name = '{self.site_name}'
             and category_parent_id = 0
+            and delete_time is null
         """
         self.df_cate_name = self.spark.sql(sqlQuery=sql)
         print(sql)
 
+    def fix_bsr_cate_1_id(self, df_flow):
+        """
+        纠正 flow 里错标的一级分类，返回 seller_id/asin/is_asin_new/bsr_cate_1_id 四列。
+        flow 的一级取 asin 页面第一个排名链接 bestsellers/ 后的第一段(dim_asin_bs_info)，第一个排名是亚马逊隐藏榜/特殊榜时，
+        会把榜单当成一级，如 us 的 private-brands(Our Brands)、book-special-features，uk/de 的 audible、jewelry、apparel。
+        这些 ID 不在分类树的有效部门里，取不到名称，页面会出现"ID 有值、名称为无"。
+        判断流程(有效部门 = dim_bsr_category_tree 中 category_parent_id = 0 且 delete_time is null)：
+          1. 一级 flow.category_first_id 为空 → 不动，归入"无"
+          2. 一级 flow.category_first_id 是有效部门 → 不动
+          3. 当前节点 flow.category_id 查分类树(按 dim.category_id)，节点每行取：
+             重定向一级 dim.redirect_first_id 是有效部门就用它，否则用链接一级 dim.category_first_id(也须是有效部门)；
+             节点只对应一个有效部门 → 用它
+          4. 面包屑一级 flow.desc_category_first_id 是有效部门 → 用它
+          5. 都不行 → 置空，归入"无"(不舍弃，照样计入 fb_asin_num)
+        """
+        # 有效部门：分类树一级(parent=0)且未失效(delete_time 有值=已失效)，与 read_data 取名称同一口径
+        print("获取 dim_bsr_category_tree（有效部门）")
+        sql = f"""
+            select category_id as dept_id
+            from big_data_selection.dim_bsr_category_tree
+            where site_name = '{self.site_name}'
+              and category_parent_id = 0
+              and delete_time is null
+        """
+        df_dept = self.spark.sql(sqlQuery=sql).distinct().cache()
+        print(sql)
+
+        # 不按错标一级 ID 整体映射(如 book-special-features 全部并入 books)，每个 asin 按自己的当前节点归类
+        # 当前节点 → 所属有效部门，分类树按行判断，一个节点可能有多行：
+        #   重定向一级 redirect_first_id 是有效部门 → 用它
+        #   否则 → 用节点链接一级 category_first_id(dim 从节点 path 解析，bestsellers/ 或 zgbs/ 后那段)，也须是有效部门
+        # 重定向无效时不能直接作废：分类树里有写错的重定向值，如 us 69724(Books on CD) 写成 book-special-features、
+        # de 2873947031(Stools) 写成 bestsellers，而链接一级 books / kitchen 是对的
+        # 不直接用 dim 的 rel_first_id：它按节点取 collect_set 第 0 个，节点挂多个一级时取到哪个不确定
+        print("获取 dim_bsr_category_tree（当前节点所属部门）")
+        sql = f"""
+            with dept as (
+                select distinct category_id as dept_id
+                from big_data_selection.dim_bsr_category_tree
+                where site_name = '{self.site_name}' and category_parent_id = 0 and delete_time is null
+            )
+            select t.category_id, collect_set(coalesce(d1.dept_id, d2.dept_id)) as node_cate_1_ids
+            from big_data_selection.dim_bsr_category_tree t
+            left join dept d1 on t.redirect_first_id = d1.dept_id
+            left join dept d2 on t.category_first_id = d2.dept_id
+            where t.site_name = '{self.site_name}'
+              and t.delete_time is null
+              and coalesce(d1.dept_id, d2.dept_id) is not null
+            group by t.category_id
+        """
+        df_node = self.spark.sql(sqlQuery=sql)
+        print(sql)
+
+        # 关联：原一级是否有效部门、当前节点所属部门、面包屑一级是否有效部门
+        # 几张维表都按 join 键唯一(df_dept 已 distinct，df_node 按 category_id 聚合)，不会让 asin 行数变多
+        df_dept_flag = df_dept.select(F.col('dept_id').alias('bsr_cate_1_id'), F.lit(1).alias('is_dept'))
+        df_desc_flag = df_dept.select(F.col('dept_id').alias('desc_category_first_id'), F.lit(1).alias('is_desc_dept'))
+        df_flow = df_flow.join(F.broadcast(df_dept_flag), on='bsr_cate_1_id', how='left')
+        df_flow = df_flow.join(F.broadcast(df_node), on='category_id', how='left')
+        df_flow = df_flow.join(F.broadcast(df_desc_flag), on='desc_category_first_id', how='left')
+
+        # 按顺序判断，前面命中就不再往后(bsr_cate_1_id = flow.category_first_id)
+        # 修不好的置空，handle_fb_agg 里统一归入"无"，不舍弃。不舍弃的原因：
+        #   店铺 asin 总数 fb_asin_num 要和主表 dwt_fb_base_report 对齐，这些 asin 照样算在总数里；
+        #   如果舍弃，店铺各分类产品数加起来会少于总数、店铺占比合计不到 100%，
+        #   asin 全部修不好的店在页面上会显示"无 / 0 个 / 0%"，和店铺产品数对不上，用户看不懂。
+        #   归入"无"后每家店各分类合计 = 店铺总数，"无"的含义是"当月拿不到有效一级分类"(没有排名，或排名在隐藏榜且认不出部门)
+        # 第 3 步：当前节点为空、在树里查不到、或查到多个部门时 node_cate_1_ids 为 null / size≠1，落到第 4 步
+        df_flow = df_flow.withColumn(
+            'bsr_cate_1_id',
+            F.when(F.col('bsr_cate_1_id').isNull(), F.lit(None).cast('string'))                  # 1. flow.category_first_id 为空(当月没有分类)：不动，归入"无"
+             .when(F.col('is_dept') == 1, F.col('bsr_cate_1_id'))                                 # 2. flow.category_first_id 已是有效部门：不动
+             .when(F.size('node_cate_1_ids') == 1, F.col('node_cate_1_ids').getItem(0))           # 3. 当前节点 flow.category_id 查树只属于一个有效部门(重定向一级 dim.redirect_first_id，无效时用链接一级 dim.category_first_id)，如 private-brands 下女装节点 → fashion
+             .when(F.col('is_desc_dept') == 1, F.col('desc_category_first_id'))                   # 4. 面包屑一级 flow.desc_category_first_id 是有效部门(失效部门不认，如 uk audible)
+             .otherwise(F.lit(None).cast('string'))                                               # 5. 修不好：置空，归入"无"
+        )
+        return df_flow.select('seller_id', 'asin', 'is_asin_new', 'bsr_cate_1_id')
+
     def run(self):
         # 读取数据
         self.read_data()
@@ -152,6 +249,7 @@ class DwtFbCategoryReport(object):
         self.df_fb_cate_asin_cal = self.df_fb_asin_info
 
         # 当月无 BSR 分类的 asin 归到 '无'（按月口径：当月没上榜就不硬塞历史分类）
+        # 也包括 fix_bsr_cate_1_id 修不好、被置空的错标一级(隐藏榜且认不出部门)；店铺没有 flow asin 时这里也是 '无'
         self.df_fb_cate_asin_cal = self.df_fb_cate_asin_cal.na.fill({'bsr_cate_1_id': '无'})
 
         # 按 seller_id + 一级分类聚合: 分类下 asin 数 + 新品数
@@ -196,6 +294,7 @@ class DwtFbCategoryReport(object):
         )
 
         # 未匹配到的分类名称置为无
+        # 纠正后 bsr_cate_1_id 只剩有效部门和 '无' 两种，正常只有 '无' 会走到这里，不会再出现"ID 有值、名称为无"
         df_save = df_save.na.fill({'bsr_cate_1_name':'无'})
 
         df_save = df_save.select(