流量选品-隐藏分类表优化

Pyspark_job/dwd/dwd_nsr_bsr_keepa_asin.py

@@ -133,8 +133,19 @@ class DwdNsrBsrKeepaAsin(Templates):
         self.df_nsr_tree = self.spark.sql(sqlQuery=sql_nsr).cache()
         self.df_nsr_tree.show(10, truncate=False)
         print(f"2.3 从mysql读取隐藏分类")
-        sql_hide = "select category_id_base as category_id, 1 as hide_flag from us_bs_category_hide group by category_id_base"
-        pdf_hide_cate = self.engine_mysql.read_sql(sql_hide)
+        # 流量选品模块隐藏分类：category_full_name + category_disable_config 按 id_path 前缀匹配
+        # （对齐 dwt.handle_asin_is_hide）；配置表集中存在 us(selection) 库，固定用 us 连接，
+        # 实际站点靠 site 字段过滤，不能用 self.engine_mysql（按 self.site_name 连到对应站点库）
+        engine_mysql_us = get_remote_engine(site_name="us", db_type='mysql')
+        sql_hide = f"""
+            SELECT DISTINCT category_id, 1 as hide_flag FROM category_full_name a
+            WHERE EXISTS (
+                SELECT 1 FROM category_disable_config b
+                WHERE b.site = a.site AND b.module = '流量选品:名称前缀筛选'
+                  AND a.id_path LIKE CONCAT(b.id_path, '%')
+            ) AND a.site = '{self.site_name}'
+        """
+        pdf_hide_cate = engine_mysql_us.read_sql(sql_hide)
         if pdf_hide_cate.shape[0]:
             schema = StructType([
                 StructField('category_id', StringType(), True),
@@ -173,9 +184,10 @@ class DwdNsrBsrKeepaAsin(Templates):
             "hide_flag",
             F.coalesce(F.col("hide_flag"), F.lit(0))
         )
+        # 数字虚拟类目(hide_first，对齐dwt的asin_is_need)优先判为2，隐藏分类配置(hide_flag，对齐asin_is_hide)判为3
         df_union_cate = df_union_cate.withColumn(
             "asin_type",
-            F.when(F.col("hide_first")==1, 1).when(F.col('hide_first')==1, 2).otherwise(0)
+            F.when(F.col("hide_first") == 1, 2).when(F.col("hide_flag") == 1, 3).otherwise(0)
         )
         df_union_cate.show(10, truncate=False)
         df_union_cate.groupby(["asin_type"]).agg(

Pyspark_job/dwt/dwt_flow_asin.py

@@ -556,16 +556,23 @@ class DwtFlowAsin(Templates):
     # 处理asin是否隐藏分类信息（US/UK/DE站点通用）以及asin_type信息
     def handle_asin_is_hide(self):
         mysql_con = DBUtil.get_connection_info("mysql", "us")
+        # 流量选品模块隐藏分类：category_full_name 取 asin 所属分类的完整 id_path，
+        # 按 module 区分场景，匹配 category_disable_config 里配置的 id_path 前缀，命中即隐藏
         sql = f"""
-            select category_id_base as category_id, 1 as hide_flag from us_bs_category_hide group by category_id_base
+            SELECT DISTINCT category_id FROM category_full_name a
+            WHERE EXISTS (
+                SELECT 1 FROM category_disable_config b
+                WHERE b.site = a.site AND b.module = '流量选品:名称前缀筛选'
+                  AND a.id_path LIKE CONCAT(b.id_path, '%')
+            ) AND a.site = '{self.site_name}'
         """
         df_hide_category = SparkUtil.read_jdbc_query(session=self.spark, url=mysql_con['url'], pwd=mysql_con['pwd'],
-                                                     username=mysql_con['username'], query=sql)
+                                                     username=mysql_con['username'], query=sql
+                                                     ).withColumn("hide_flag", F.lit(1))
         self.df_asin_detail = self.df_asin_detail.join(df_hide_category, on=['category_id'], how='left')
-        self.df_asin_detail = self.df_asin_detail.withColumn("asin_is_hide", F.expr("""
-      CASE WHEN hide_flag = 1 THEN 1 WHEN category_first_id = 'grocery' and category_id != '6492272011' THEN 1 
-      WHEN category_id in ('21393128011', '21377129011', '21377127011', '21377130011', '21388218011', '21377132011') THEN 1
-      ELSE 0 END""")).drop("hide_flag")
+        self.df_asin_detail = self.df_asin_detail.withColumn(
+            "asin_is_hide", F.when(F.col("hide_flag") == 1, F.lit(1)).otherwise(F.lit(0))
+        ).drop("hide_flag")
         # 解析 asin_category_desc 取 › 分隔的第一个元素作为补充分类名称
         self.df_asin_detail = self.df_asin_detail.withColumn(
             "desc_category_first_name",

Pyspark_job/script/export_asin_without_keepa.py

@@ -44,12 +44,6 @@ NEED_FILTER_CATEGORIES = (
     'amazon-devices', 'boost', 'us-live-explorations', 'amazon-renewed'
 )
 
-# asin_is_hide 中 category_id 级别的隐藏节点
-HIDE_CATEGORY_IDS = (
-    '21393128011', '21377129011', '21377127011',
-    '21377130011', '21388218011', '21377132011'
-)
-
 
 class ExportAsinWithoutKeepa(object):
 
@@ -149,14 +143,21 @@ class ExportAsinWithoutKeepa(object):
         """
         df_bsr_category = F.broadcast(self.spark.sql(sqlQuery=sql))
 
-        # ⑤ us_bs_category_hide → 隐藏分类（用于 asin_type 计算）
-        print("5. 读取 us_bs_category_hide (隐藏分类)")
-        mysql_con = DBUtil.get_connection_info("mysql", self.site_name)
-        sql = "select category_id_base as category_id, 1 as hide_flag from us_bs_category_hide group by category_id_base"
+        # ⑤ category_full_name + category_disable_config → 隐藏分类（对齐 dwt.handle_asin_is_hide）
+        print("5. 读取隐藏分类（流量选品模块，id_path前缀匹配）")
+        mysql_con = DBUtil.get_connection_info("mysql", "us")
+        sql = f"""
+            SELECT DISTINCT category_id FROM category_full_name a
+            WHERE EXISTS (
+                SELECT 1 FROM category_disable_config b
+                WHERE b.site = a.site AND b.module = '流量选品:名称前缀筛选'
+                  AND a.id_path LIKE CONCAT(b.id_path, '%')
+            ) AND a.site = '{self.site_name}'
+        """
         df_hide = SparkUtil.read_jdbc_query(
             session=self.spark, url=mysql_con['url'],
             pwd=mysql_con['pwd'], username=mysql_con['username'], query=sql
-        )
+        ).withColumn("hide_flag", F.lit(1))
 
         # ⑥ 组装
         print("6. 组装主DataFrame")
@@ -192,12 +193,7 @@ class ExportAsinWithoutKeepa(object):
         need_categories = NEED_FILTER_CATEGORIES
         df = df.withColumn(
             "asin_is_hide",
-            F.expr(f"""
-                CASE WHEN hide_flag = 1 THEN 1
-                     WHEN category_first_id = 'grocery' AND category_id != '6492272011' THEN 1
-                     WHEN category_id IN {HIDE_CATEGORY_IDS} THEN 1
-                     ELSE 0 END
-            """)
+            F.when(F.col("hide_flag") == 1, F.lit(1)).otherwise(F.lit(0))
         ).withColumn(
             "asin_is_need",
             F.expr(f"""