asin信息库流程调整

Pyspark_job/doris_handle/dwt_ai_asin.py

@@ -3,10 +3,13 @@
    @Description : 信息库月度整合脚本
                   - 从 dwt_flow_asin 读取当月符合条件的ASIN及详情字段（含 mom/yoy）
                   - 过滤禁用分类，筛选信息库候选ASIN（月销>=50，asin_type in (0,1)）
-                  - is_new_flag：anti join Doris us_ai_asin_detail 存量asin
+                  - 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 增量写入
+                  - 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 反连接，
+                    已经在里面（已完成AI分析）的不用再导出
 """
 import os
 import sys
@@ -34,6 +37,8 @@ class DwtAiAsin(object):
 
         # 近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")
@@ -115,12 +120,13 @@ class DwtAiAsin(object):
             .persist(StorageLevel.DISK_ONLY)
         print(f"当月信息库ASIN：{self.df_base_asin.count()}")
 
-        # ── 2. 存量asin（Doris us_ai_asin_detail），用于 is_new_flag ──
-        sql_exist = f"SELECT asin FROM `selection`.`{self.site_name}_ai_asin_detail`"
-        self.df_existing_asin = DorisHelper.spark_import_with_sql(
-            self.spark, sql_exist
-        ).repartition(40, 'asin').persist(StorageLevel.DISK_ONLY)
-        print(f"Doris 信息库存量ASIN：{self.df_existing_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"""
@@ -157,7 +163,7 @@ class DwtAiAsin(object):
             .repartition(40, 'asin').persist(StorageLevel.DISK_ONLY)
 
     def handle_data(self):
-        # is_new_flag：不在 Doris 存量中的视为新增
+        # 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))) \
@@ -237,30 +243,57 @@ class DwtAiAsin(object):
             partitionBy=['site_name', 'date_type', 'date_info']
         )
 
-        # ── Doris 写入（us_ai_asin_detail 字段，chat_flag=0 增量）──
-        doris_columns = (
-            "asin, site_name, weight, bought_month, category, img, title, brand, "
-            "account_name, account_addr, buy_box_seller_type, launch_time, img_num, "
-            "variation_flag, variation_num, ao_val, category_id, category_current_id, "
-            "parent_asin, bsr_rank, price, rating, total_comments, seller_id, "
-            "fb_country_name, review_json_list, launch_time_type, `describe`, "
-            "product_json, product_detail_json, bought_month_mom, bought_month_yoy, "
-            "is_new_flag, is_ascending_flag, chat_flag"
-        )
+        # ── PG 写入（只导出增量：不在 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') \
+            .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('-', '_')}"
         try:
-            DorisHelper.spark_export_with_columns(
-                df_save=self.df_save.drop('date_type', 'date_info'),
-                db_name='selection',
-                table_name=f'{self.site_name}_ai_asin_detail',
-                table_columns=doris_columns,
-            )
-            print("success!")
-            CommonUtil.send_wx_msg(['chenyuanjie', 'wujicang'], 'ASIN信息库月度数据写入Doris成功',
-                                    f'详情：{self.site_name}_ai_asin_detail {self.site_name} {self.date_type} {self.date_info}')
+            # 即使这个月没有增量，也要把当月空表建出来，避免下个月LIKE这个月当模板建表时找不到表
+            self.create_pg_table_if_absent(export_pg_tb, template_tb)
+            if export_count == 0:
+                print("没有需要导出到PG的增量ASIN，空表已建好，跳过写入")
+                return
+            df_to_export.write.format("jdbc") \
+                .option("url", pg_con["url"]) \
+                .option("dbtable", export_pg_tb) \
+                .option("user", pg_con["username"]) \
+                .option("password", pg_con["pwd"]) \
+                .mode("append") \
+                .save()
+            print(f"PG {export_pg_tb} 增量写入完成")
+            CommonUtil.send_wx_msg(['chenyuanjie', 'tanbin'], 'ASIN信息库月度数据写入PG成功',
+                                    f'详情：{export_pg_tb} {self.site_name} {self.date_type} {self.date_info}')
         except Exception as e:
-            print("An error occurred while writing to Doris:", str(e))
-            CommonUtil.send_wx_msg(['chenyuanjie'], '⚠ ASIN信息库月度数据写入Doris失败',
-                                    f'详情：{self.site_name}_ai_asin_detail {self.site_name} {self.date_type} {self.date_info}')
+            print("An error occurred while writing to PG:", str(e))
+            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__":