待AI分析asin建表

Pyspark_job/export_es/es_ai_asin_add.py

@@ -200,7 +200,32 @@ class EsAiAsinAdd(object):
             'total_comments', 'uses', 'variation_flag', 'variation_num', 'weight'
         )
 
+    def create_pg_table(self):
+        """
+        根据模板表创建月度 PG 表：
+          1. LIKE INCLUDING ALL：复制所有字段类型、其他列默认值、约束、索引
+          2. 重建 id 列独立序列，避免与模板表共享同一序列
+        """
+        template_tb = "us_ai_asin_detail_month_2026_01"
+        engine = DBUtil.get_db_engine("postgresql", "us")
+        # 表已存在则跳过
+        result = DBUtil.engine_exec_sql(engine, f"SELECT to_regclass('{self.export_pg_tb}')")
+        if list(result)[0][0] is not None:
+            print(f"PostgreSQL 表 {self.export_pg_tb} 已存在，跳过建表")
+            return
+        # 建表 + 为 id 列创建独立序列
+        sql = f"""
+            CREATE TABLE {self.export_pg_tb} (LIKE {template_tb} INCLUDING ALL);
+            ALTER TABLE {self.export_pg_tb} ALTER COLUMN id DROP DEFAULT;
+            CREATE SEQUENCE {self.export_pg_tb}_id_seq OWNED BY {self.export_pg_tb}.id;
+            ALTER TABLE {self.export_pg_tb} ALTER COLUMN id SET DEFAULT nextval('{self.export_pg_tb}_id_seq')
+        """
+        DBUtil.exec_sql("postgresql", "us", sql)
+        print(f"PostgreSQL 表 {self.export_pg_tb} 创建完成（独立自增序列）")
+
     def save_data(self):
+        # 创建月度 PG 表
+        self.create_pg_table()
         # 将新增asin导出给济苍
         try:
             self.df_save_pg.write.format("jdbc") \