bstop100 历史月数据使用新的新品判断规则回刷

Pyspark_job/dwt/dwt_bs_top100.py

@@ -16,6 +16,7 @@ from pyspark.storagelevel import StorageLevel
 from utils.templates import Templates
 from utils.hdfs_utils import HdfsUtils          # 落表前清 HDFS 分区目录（幂等重跑）
 from utils.common_util import CommonUtil        # 构造 HDFS 分区路径
+from yswg_utils.common_udf import udf_new_asin_flag  # 新品判断 UDF（与 dim 同一个，用于按新规则重算 asin_is_new）
 # from ..utils.templates import Templates
 # from AmazonSpider.pyspark_job.utils.templates_test import Templates
 from pyspark.sql.types import StringType, IntegerType, DoubleType, LongType
@@ -51,6 +52,12 @@ class DwtBsTop100(Templates):
         self.virtual_node_ids = []                          # 虚拟节点 cat_id 列表（运行时算出，拆层时剔除）
         self.df_level_concat = self.spark.sql("select 1+1;")  # DF 对象占位符
         self.partitions_by = ['site_name', 'date_type', 'date_info']  # 三个分区
+        self.udf_new_asin_flag = F.udf(udf_new_asin_flag, IntegerType())  # 新品判断 UDF（与 dim 同一个）
+        # 仅月表、且落在历史回刷窗口(24-07~26-06)才按新规则重算 asin_is_new。
+        # 起点 24-07=同比基准月(让 25-07 起的新品同比不至于"新规则当月÷旧规则基准")；终点 26-06=历史末月。
+        # 窗口外(26-07 及以后)新月份：dim 已是新规则、dwt_flow 冻结值就是对的，不重算直接用。week/4_week 不动。
+        # 本次只回刷 us；uk/de 的 keepa 口径未验证，不在范围（未加 site 判断，靠此注释约束运行）。
+        self.is_recalc_new = (self.date_type == 'month' and '2024-07' <= self.date_info <= '2026-06')
         self.reset_partitions(partitions_num=1)  # 分区数量
 
     def read_data(self):
@@ -64,6 +71,14 @@ class DwtBsTop100(Templates):
         print(f"1. 读取dwt_flow_asin数据: sql -- {sql}")
         # 下游线性链只读一次，去掉 cache
         self.df_bs_data = self.spark.sql(sqlQuery=sql)
+        # 新品口径 2026-06 迭代为 greatest(上架时间, keepa追踪时间) 判 180 天内；历史 dwt_flow 存的还是旧规则值。
+        # 这里旁路读 dim 原始 launch + keepa 快照按新规则重算，对齐到重定向 asin 后覆盖 asin_is_new（详见 recompute_asin_is_new）。
+        # coalesce 兜底：极少数只在 measure、dim 无的 asin 退回旧值，绝不写 null（下游对 asin_is_new 做 F.sum，null 会少算）。
+        if self.is_recalc_new:
+            df_new_flag = self.recompute_asin_is_new()
+            self.df_bs_data = self.df_bs_data.join(df_new_flag, on='asin', how='left') \
+                .withColumn('asin_is_new', F.coalesce(F.col('asin_is_new_recalc'), F.col('asin_is_new'))) \
+                .drop('asin_is_new_recalc')
         # self.df_bs_data.show(10, truncate=False)
         # print("df_bs_data数量", self.df_bs_data.count())
         import pandas as pd
@@ -156,6 +171,43 @@ class DwtBsTop100(Templates):
         print(f"2. 读取利润率数据: sql -- {sql_pr}")
         self.df_profit_rate = self.spark.sql(sqlQuery=sql_pr).dropDuplicates(['asin', 'asin_price'])
 
+    def recompute_asin_is_new(self):
+        """按新规则重算 asin_is_new，产出 (asin=重定向后 target, asin_is_new_recalc)，供 read_data 覆盖。
+
+        新规则 = greatest(上架时间, keepa追踪时间) 距当月最后一天 <=180 天为新品；
+        历史 dwt_flow 存的是旧规则（只看上架时间）值。这里从 dim 原始 launch + keepa 快照重算，
+        并复刻 dwt_flow 的跳转（coalesce+created_time 最新去重）对齐到重定向 asin，和 df_bs_data 同 grain。
+        """
+        # 读 dim 原始 launch 及跳转所需列（与 df_node 同一张 dim，单独取以带上 asin_launch_time）
+        sql_dim = f"""select asin, current_asin, asin_launch_time, created_time
+                      from dim_asin_detail
+                      where site_name='{self.site_name}' and date_type='{self.date_type}' and date_info='{self.date_info}'"""
+        print(f"读取 dim_asin_detail 重算新品: sql -- {sql_dim}")
+        df_dim = self.spark.sql(sql_dim)
+        # keepa 追踪起始时间：全量快照不分月，与 dim 新规则同一段口径（tracking_since 分钟级偏移转 yyyy-MM-dd）
+        sql_keepa = f"""select asin,
+                          date_format(from_unixtime((cast(tracking_since as bigint) + 21564000) * 60), 'yyyy-MM-dd') as keepa_tracking_since
+                        from dim_keepa_asin_info
+                        where site_name='{self.site_name}' and tracking_since is not null"""
+        df_keepa = self.spark.sql(sql_keepa)
+        # 按原始 asin 关联 keepa，用新规则重算（与 dim_asin_detail.handle_asin_flag 完全一致）
+        cal_date = CommonUtil.get_calDay_by_dateInfo(self.spark, self.date_type, self.date_info)
+        df_dim = df_dim.join(df_keepa, on='asin', how='left')
+        # keepa 护栏：回刷历史月用的是"今天的 keepa"，其中含追踪时间晚于目标月的记录。
+        # 新品 UDF 只有上界(offset<=180)、无下界，未来 keepa 会算出负 offset 被误判成新品(实时算当月时不会遇到、故原 UDF 没这判断)。
+        # 这里把晚于当月末的 keepa_tracking_since 置空 → greatest 退回只看上架时间，挡掉误判。
+        df_dim = df_dim.withColumn(
+            'keepa_tracking_since',
+            F.when(F.col('keepa_tracking_since') <= F.lit(cal_date), F.col('keepa_tracking_since')))
+        df_dim = df_dim.withColumn(
+            "asin_is_new_recalc",
+            self.udf_new_asin_flag(F.greatest(F.col('asin_launch_time'), F.col('keepa_tracking_since')), F.lit(cal_date)))
+        # 复刻 dwt_flow 跳转：target=coalesce(current_asin,asin)，同 target 按 created_time 最新留一条
+        df_dim = df_dim.withColumn("asin_eff", F.coalesce(F.col("current_asin"), F.col("asin")))
+        w = Window.partitionBy("asin_eff").orderBy(F.col("created_time").desc_nulls_last())
+        df_dim = df_dim.withColumn("_rk", F.row_number().over(w)).filter(F.col("_rk") == 1)
+        return df_dim.select(F.col("asin_eff").alias("asin"), F.col("asin_is_new_recalc"))
+
     def clean_column(self, df, col_name):
         """统一处理 ID 列：去除 .0 后缀，处理空值"""
         return df.withColumn(