流量选品30天-asin_type字段spark计算+隐藏分类处理

Pyspark_job/my_kafka/kafka_flow_asin_detail_to_doris.py

@@ -20,6 +20,13 @@ from utils.DorisHelper import DorisHelper
 from yswg_utils.common_df import get_node_first_id_df, get_first_id_from_category_desc_df
 from yswg_utils.common_udf import udf_parse_bs_category, parse_weight_str, udf_extract_volume_dimensions, udf_get_package_quantity_with_flag as udf_get_package_quantity, udf_parse_seller_json, udf_parse_amazon_orders, resolve_asin_jump
 
+# asin_is_need 判断中需要过滤的类目（对齐 dwt.handle_asin_is_hide）
+NEED_FILTER_CATEGORIES = (
+    'mobile-apps', 'audible', 'books', 'music', 'dmusic', 'digital-text',
+    'magazines', 'movies-tv', 'software', 'videogames',
+    'amazon-devices', 'boost', 'us-live-explorations', 'amazon-renewed'
+)
+
 
 class KafkaFlowAsinDetail(Templates):
     def __init__(self, site_name='us', date_type="month", date_info='2026-03', consumer_type='history', test_flag='test', batch_size=100000):
@@ -105,6 +112,8 @@ class KafkaFlowAsinDetail(Templates):
         self.df_bs_report = self.spark.sql("select 1+1;")
         self.df_asin_new_cate = self.spark.sql("select 1+ 1;")
         self.df_asin_category = self.spark.sql("select 1+1;")
+        self.df_self_asin = self.spark.sql("select 1+1;")
+        self.df_hide_category = self.spark.sql("select 1+1;")
         self.color_set = set()
         # udf函数注册
         package_schema = StructType([
@@ -624,6 +633,26 @@ class KafkaFlowAsinDetail(Templates):
             "is_amazon_new", F.when(F.expr("exists(badge_list, x -> x like '%new on amazon%')"), F.lit(1)).otherwise(F.lit(0))
         )
         df = df.drop("asin_label_list", "badge_list")
+        # 4. asin_type（对齐 dwt.handle_asin_is_hide：内部asin=1 / 数字虚拟类目=2 / 隐藏分类=3 / 默认0）
+        df = df.join(self.df_self_asin, on=['asin'], how='left')
+        df = df.join(self.df_hide_category, on=['asin_bs_cate_current_id'], how='left')
+        df = df.withColumn(
+            "asin_is_need",
+            F.when(
+                F.col("asin_bs_cate_1_id").isin(*NEED_FILTER_CATEGORIES) &
+                F.col("desc_category_first_id").isin(*NEED_FILTER_CATEGORIES),
+                F.lit(1)
+            ).when(
+                ~F.col("asin").like("B0%"), F.lit(1)
+            ).otherwise(F.lit(0))
+        )
+        df = df.withColumn(
+            "asin_type",
+            F.when(F.col("asin_is_self") == 1, F.lit(1))
+             .when(F.col("asin_is_need") == 1, F.lit(2))
+             .when(F.col("hide_flag") == 1, F.lit(3))
+             .otherwise(F.lit(0)).cast("tinyint")
+        ).drop("asin_is_self", "asin_is_need", "hide_flag")
         return df
 
     # 12. 处理变化率相关字段（环比_mom / 同比_yoy 统一处理）
@@ -931,6 +960,29 @@ class KafkaFlowAsinDetail(Templates):
         ).collect()
         self.color_set = {row.en_name for row in color_rows}
         print(f"颜色词表共 {len(self.color_set)} 条")
+        print("9. 读取内部asin信息(ods_self_asin)，用于asin_type计算")
+        sql = f"select asin, 1 as asin_is_self from ods_self_asin where site_name='{self.site_name}'"
+        print("sql=", sql)
+        self.df_self_asin = self.spark.sql(sqlQuery=sql).dropDuplicates(['asin']).persist(StorageLevel.DISK_ONLY)
+        self.df_self_asin.show(10, truncate=False)
+        print("10. 读取隐藏分类(流量选品模块，id_path前缀匹配)，用于asin_type计算")
+        # category_full_name + category_disable_config 按 id_path 前缀匹配（对齐 dwt.handle_asin_is_hide）
+        # 配置表集中存在 us(selection) 库，固定用 us 连接，实际站点靠 site 字段过滤
+        # category_id 列直接别名成 asin_bs_cate_current_id，跟这个阶段 df 里还没改名前的当前分类列名对齐
+        mysql_con = DBUtil.get_connection_info("mysql", "us")
+        sql = f"""
+            SELECT DISTINCT category_id AS asin_bs_cate_current_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}'
+        """
+        self.df_hide_category = F.broadcast(SparkUtil.read_jdbc_query(
+            session=self.spark, url=mysql_con['url'], pwd=mysql_con['pwd'],
+            username=mysql_con['username'], query=sql
+        ))
+        self.df_hide_category.show(10, truncate=False)
 
     # 字段处理逻辑综合
     def handle_all_field(self, df):
@@ -1027,8 +1079,7 @@ class KafkaFlowAsinDetail(Templates):
             "asin_weight_str",
             "best_sellers_rank",
             "best_sellers_herf",
-            # asin_type kafka 消费阶段统一写 0，业务实际类型由视图层 mv 综合判断
-            F.lit(0).cast('tinyint').alias("asin_type"),
+            "asin_type",
             "is_amazon_new",
         )
         table_columns = """asin, asin_ao_val, asin_title, asin_title_len, asin_category_desc, asin_volume,
@@ -1069,7 +1120,7 @@ class KafkaFlowAsinDetail(Templates):
             price_type, quantity_variation_type, package_quantity,
             is_movie_label, is_brand_label, asin_crawl_date,
             category_first_id, category_id, first_category_rank, current_category_rank, desc_category_first_id, asin_weight_ratio,
-            site_name, asin_bought_month, asin_lqs_rating, asin_lqs_rating_detail, asin_lob_info,
+            site_name, asin_type, asin_bought_month, asin_lqs_rating, asin_lqs_rating_detail, asin_lob_info,
             is_contains_lob_info, is_package_quantity_abnormal, zr_flow_proportion, matrix_flow_proportion,
             matrix_ao_val, product_features, img_info, collapse_asin, follow_sellers_count, asin_describe,
             fbm_price, describe_len, multi_color_flag, multi_color_str, amazon_label, is_amazon_new"""