信息库月流程调整

Pyspark_job/doris_handle/dwt_ai_asin.py

 """
    @Author      : CT
    @Description : 信息库月度整合脚本
-                  - 从 dwt_flow_asin 读取当月符合条件的ASIN及详情字段（含 mom/yoy）
+                  - 从 dwt_flow_asin 读取当月符合条件的ASIN及详情字段
                   - 过滤禁用分类，筛选信息库候选ASIN（月销>=50，asin_type in (0,1)）
-                  - is_new_flag：跟上个月 Hive dwt_ai_asin 比，上月没有的视为新增
-                  - is_ascending_flag：近6月 asin_bought_mom>0 满6次
-                  - 关联店铺（dwt_fb_base_report）、ODS评论（ods_asin_detail）
-                  - Hive dwt_ai_asin 备份；不再写 Doris selection.{site}_ai_asin_detail，
-                    改成写 PG15 {site}_ai_asin_detail_month_{yyyy_mm}（LIKE上月同名表建表），
-                    且只导出增量：跟 Doris selection.{site}_ai_asin_analyze_detail 反连接，
+                  - is_new_flag / is_ascending_flag：已废弃，不再计算，Hive里固定写NULL（保留字段位避免破坏历史表结构）
+                  - 关联ODS评论（ods_asin_detail）
+                  - Hive dwt_ai_asin 备份；PG表 {site}_ai_asin_detail 且只导出增量：跟 Doris selection.{site}_ai_asin_analyze_detail 反连接，
                     已经在里面（已完成AI分析）的不用再导出
 """
 import os
@@ -35,17 +32,9 @@ class DwtAiAsin(object):
         app_name = f"{self.__class__.__name__}:{site_name}:{date_type}:{date_info}"
         self.spark = SparkUtil.get_spark_session(app_name)
 
-        # 近6个月（含当月）用于 is_ascending_flag
-        self.last_6_month = [CommonUtil.get_month_offset(date_info, -i) for i in range(0, 6)]
-        # 上个月，用于 is_new_flag 比对 + PG建表模板月份
-        self.last_month = CommonUtil.get_month_offset(date_info, -1)
-
-        self.df_base_asin      = self.spark.sql("select 1+1")
-        self.df_existing_asin  = self.spark.sql("select 1+1")
-        self.df_ascending_flag = self.spark.sql("select 1+1")
-        self.df_fb_info        = self.spark.sql("select 1+1")
-        self.df_ods_detail     = self.spark.sql("select 1+1")
-        self.df_save           = self.spark.sql("select 1+1")
+        self.df_base_asin = self.spark.sql("select 1+1")
+        self.df_ods_detail = self.spark.sql("select 1+1")
+        self.df_save      = self.spark.sql("select 1+1")
 
     def run(self):
         self.read_data()
@@ -64,20 +53,21 @@ class DwtAiAsin(object):
                 asin_title                                      AS title,
                 asin_brand_name                                 AS brand,
                 account_name,
+                account_id                                      AS seller_id,
+                seller_country_name                              AS fb_country_name,
                 asin_buy_box_seller_type                        AS buy_box_seller_type,
                 asin_launch_time                                AS launch_time,
                 asin_img_num                                    AS img_num,
                 CASE WHEN variation_num > 0 THEN 1 ELSE 0 END  AS variation_flag,
                 variation_num,
                 asin_ao_val                                     AS ao_val,
-                category_first_id                               AS category_id,
+                category_first_id,
                 category_id                                     AS category_current_id,
                 parent_asin,
                 first_category_rank                             AS bsr_rank,
                 asin_price                                      AS price,
                 asin_rating                                     AS rating,
                 asin_total_comments                             AS total_comments,
-                asin_launch_time_type                           AS launch_time_type,
                 asin_describe                                   AS describe,
                 asin_bought_mom                                 AS bought_month_mom,
                 asin_bought_yoy                                 AS bought_month_yoy
@@ -116,41 +106,13 @@ class DwtAiAsin(object):
         )
 
         self.df_base_asin = self.df_base_asin \
-            .join(df_filter_cat_id,   'category_id', 'left_anti') \
-            .join(df_filter_cat_desc, 'category',    'left_anti') \
+            .join(df_filter_cat_id, self.df_base_asin['category_current_id'] == df_filter_cat_id['category_id'], 'left_anti') \
+            .join(df_filter_cat_desc, 'category', 'left_anti') \
             .repartition(40, 'asin') \
             .persist(StorageLevel.DISK_ONLY)
         print(f"当月信息库ASIN：{self.df_base_asin.count()}")
 
-        # ── 2. 存量asin（上个月 Hive dwt_ai_asin），用于 is_new_flag：相比上月是否为新增 ──
-        sql_exist = f"""
-            SELECT asin FROM dwt_ai_asin
-            WHERE site_name = '{self.site_name}' AND date_type = '{self.date_type}' AND date_info = '{self.last_month}'
-        """
-        self.df_existing_asin = self.spark.sql(sql_exist).repartition(40, 'asin').persist(StorageLevel.DISK_ONLY)
-        print(f"上月({self.last_month}) 信息库存量ASIN：{self.df_existing_asin.count()}")
-
-        # ── 3. 近6月 asin_bought_mom，用于 is_ascending_flag ──
-        sql_asc = f"""
-            SELECT asin, asin_bought_mom
-            FROM dwt_flow_asin
-            WHERE site_name = '{self.site_name}'
-              AND date_type = '{self.date_type}'
-              AND date_info IN ({CommonUtil.list_to_insql(self.last_6_month)})
-        """
-        self.df_ascending_flag = self.spark.sql(sql_asc).repartition(40, 'asin').persist(StorageLevel.DISK_ONLY)
-
-        # ── 4. 店铺信息 ──
-        sql_fb = f"""
-            SELECT account_name, seller_id, fb_country_name, business_addr AS account_addr
-            FROM dwt_fb_base_report
-            WHERE site_name = '{self.site_name}'
-              AND date_type = '{self.date_type}'
-              AND date_info = '{self.date_info}'
-        """
-        self.df_fb_info = self.spark.sql(sql_fb).dropDuplicates(['account_name']).persist(StorageLevel.DISK_ONLY)
-
-        # ── 5. ODS 评论/产品详情（取 updated_at 最新一条）──
+        # ── 2. ODS 评论/产品详情（取 updated_at 最新一条）──
         sql_ods = f"""
             SELECT asin, review_json_list, product_json, product_detail_json, updated_at
             FROM ods_asin_detail
@@ -165,27 +127,9 @@ class DwtAiAsin(object):
             .repartition(40, 'asin').persist(StorageLevel.DISK_ONLY)
 
     def handle_data(self):
-        # is_new_flag：不在上月 Hive dwt_ai_asin 存量中的视为新增
-        self.df_base_asin = self.df_base_asin \
-            .join(self.df_existing_asin.withColumn('_exist', F.lit(0)), 'asin', 'left') \
-            .withColumn('is_new_flag', F.when(F.col('_exist').isNull(), F.lit(1)).otherwise(F.lit(0))) \
-            .drop('_exist')
-
-        # is_ascending_flag：近6月 mom 全部 > 0 且出现6次
-        df_asc = self.df_ascending_flag.groupBy('asin').agg(
-            F.count('asin').alias('month_cnt'),
-            F.sum(F.when(F.col('asin_bought_mom') > 0, 1).otherwise(0)).alias('rising_cnt')
-        ).withColumn(
-            'is_ascending_flag',
-            F.when((F.col('month_cnt') == 6) & (F.col('rising_cnt') == 6), F.lit(1)).otherwise(F.lit(0))
-        ).drop('month_cnt', 'rising_cnt')
-
         # 汇总关联
         self.df_save = self.df_base_asin \
-            .join(df_asc,             'asin', 'left') \
-            .join(self.df_fb_info,    'account_name', 'left') \
-            .join(self.df_ods_detail, 'asin', 'left') \
-            .fillna({'is_new_flag': 0, 'is_ascending_flag': 0})
+            .join(self.df_ods_detail, 'asin', 'left')
 
     def save_data(self):
         # 字段对齐 us_ai_asin_detail
@@ -198,14 +142,14 @@ class DwtAiAsin(object):
             F.col('title'),
             F.col('brand'),
             F.col('account_name'),
-            F.col('account_addr'),
+            F.lit(None).alias('account_addr'),   # 废弃字段
             F.col('buy_box_seller_type'),
             F.col('launch_time'),
             F.col('img_num'),
             F.col('variation_flag'),
             F.col('variation_num'),
             F.col('ao_val'),
-            F.col('category_id'),
+            F.col('category_first_id'),
             F.col('category_current_id'),
             F.col('parent_asin'),
             F.col('bsr_rank'),
@@ -215,14 +159,14 @@ class DwtAiAsin(object):
             F.col('seller_id'),
             F.col('fb_country_name'),
             F.col('review_json_list'),
-            F.col('launch_time_type'),
+            F.lit(None).alias('launch_time_type'),  # 废弃字段
             F.col('describe'),
             F.col('product_json'),
             F.col('product_detail_json'),
             F.col('bought_month_mom'),
             F.col('bought_month_yoy'),
-            F.col('is_new_flag'),
-            F.col('is_ascending_flag'),
+            F.lit(None).alias('is_new_flag'),        # 废弃字段
+            F.lit(None).alias('is_ascending_flag'),  # 同上
             F.lit(0).alias('chat_flag'),
             # 分区字段必须在最后
             F.lit(self.site_name).alias('site_name'),
@@ -245,25 +189,23 @@ class DwtAiAsin(object):
             partitionBy=['site_name', 'date_type', 'date_info']
         )
 
-        # ── PG 写入（只导出增量：不在 Doris selection.{site}_ai_asin_analyze_detail 里的才需要）──
+        # ── PG 写入 {site}_ai_asin_detail，只导出增量：不在 Doris selection.{site}_ai_asin_analyze_detail 里的才需要──
         sql_analyzed = f"SELECT asin FROM `selection`.`{self.site_name}_ai_asin_analyze_detail`"
         df_analyzed_asin = DorisHelper.spark_import_with_sql(self.spark, sql_analyzed).repartition(40, 'asin')
         # persist：下面先count()校验数量、再write()真正落库，两次action共用同一份物化结果，
         # 不然anti join(含一次Doris JDBC读取)会被重新算一遍
-        df_to_export = self.df_save.drop('date_type', 'date_info') \
-            .join(df_analyzed_asin, 'asin', 'left_anti') \
+        df_to_export = self.df_save.drop(
+            'date_type', 'date_info', 'account_addr', 'launch_time_type', 'is_new_flag', 'is_ascending_flag'
+        ).join(df_analyzed_asin, 'asin', 'left_anti') \
             .persist(StorageLevel.DISK_ONLY)
         export_count = df_to_export.count()
         print(f"待导出PG的增量ASIN数量：{export_count}")
 
         pg_con = DBUtil.get_connection_info("postgresql", "us")
-        export_pg_tb = f"{self.site_name}_ai_asin_detail_month_{self.date_info.replace('-', '_')}"
-        template_tb = f"{self.site_name}_ai_asin_detail_month_{self.last_month.replace('-', '_')}"
+        export_pg_tb = f"{self.site_name}_ai_asin_detail"
         try:
-            # 即使这个月没有增量，也要把当月空表建出来，避免下个月LIKE这个月当模板建表时找不到表
-            self.create_pg_table_if_absent(export_pg_tb, template_tb)
             if export_count == 0:
-                print("没有需要导出到PG的增量ASIN，空表已建好，跳过写入")
+                print("没有需要导出到PG的增量ASIN，跳过写入")
                 return
             df_to_export.write.format("jdbc") \
                 .option("url", pg_con["url"]) \
@@ -280,23 +222,6 @@ class DwtAiAsin(object):
             CommonUtil.send_wx_msg(['chenyuanjie'], '⚠ ASIN信息库月度数据写入PG失败',
                                     f'详情：{export_pg_tb} {self.site_name} {self.date_type} {self.date_info}')
 
-    @staticmethod
-    def create_pg_table_if_absent(export_pg_tb, template_tb):
-        """LIKE 上月同名表建表（INCLUDING ALL 复制字段类型/约束/索引），并重建独立id序列；表已存在则跳过"""
-        engine = DBUtil.get_db_engine("postgresql", "us")
-        result = DBUtil.engine_exec_sql(engine, f"SELECT to_regclass('{export_pg_tb}')")
-        if list(result)[0][0] is not None:
-            print(f"PostgreSQL 表 {export_pg_tb} 已存在，跳过建表")
-            return
-        sql = f"""
-            CREATE TABLE {export_pg_tb} (LIKE {template_tb} INCLUDING ALL);
-            ALTER TABLE {export_pg_tb} ALTER COLUMN id DROP DEFAULT;
-            CREATE SEQUENCE {export_pg_tb}_id_seq OWNED BY {export_pg_tb}.id;
-            ALTER TABLE {export_pg_tb} ALTER COLUMN id SET DEFAULT nextval('{export_pg_tb}_id_seq')
-        """
-        DBUtil.exec_sql("postgresql", "us", sql)
-        print(f"PostgreSQL 表 {export_pg_tb} 创建完成（LIKE {template_tb}，独立自增序列）")
-
 
 if __name__ == "__main__":
     site_name = sys.argv[1]