feat: create_tables.py加pipeline_core/service幂等建表(IF NOT EXISTS不DROP,固化skill_proposals等表)

scripts/create_tables.py

@@ -6,7 +6,7 @@
 用法：
   py3/bin/python scripts/create_tables.py
 """
-import sys, os, asyncio, subprocess
+import sys, os, re, asyncio, subprocess
 
 SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
 ROOT_DIR = os.path.dirname(SCRIPT_DIR)
@@ -20,6 +20,8 @@ from ahserver.serverenv import ServerEnv
 from ahserver.globalEnv import initEnv
 
 TABLE_MODULES = ['product_management', 'discount', 'pricing', 'unipay', 'smssend']
+# 幂等建表模块（不 DROP，用 IF NOT EXISTS，避免清空运行时数据如 llm/pipelines/skill_proposals/tasks 等）
+IDEMPOTENT_MODULES = ['pipeline_core', 'pipeline-service']
 
 
 # appbase 系统级配置表（幂等建表 + 默认参数，不 DROP 以免清空运行时数据）
@@ -72,6 +74,26 @@ async def main():
                 n += 1
             print(f'  tables: {mod} created ({n} statements)')
 
+        # 幂等建表模块（去 DROP + IF NOT EXISTS，不清理运行时数据）
+        for mod in IDEMPOTENT_MODULES:
+            models_dir = os.path.join(ROOT_DIR, 'pkgs', mod, 'models')
+            if not os.path.isdir(models_dir):
+                print(f'  skip {mod}: no models dir')
+                continue
+            r = subprocess.run([json2ddl, 'mysql', models_dir], capture_output=True, text=True)
+            if r.returncode != 0 or not r.stdout.strip():
+                print(f'  skip {mod}: json2ddl failed')
+                continue
+            n = 0
+            for stmt in split_ddl(r.stdout):
+                s = re.sub(r'(?i)drop\s+table\s+if\s+exists\s+[\w`]+\s*;?', '', stmt)
+                s = re.sub(r'(?i)CREATE\s+TABLE\s+(`?\w+`?)', r'CREATE TABLE IF NOT EXISTS \1', s)
+                if not s.strip():
+                    continue
+                await sor.execute(s, {})
+                n += 1
+            print(f'  tables: {mod} ensured ({n} statements, idempotent)')
+
         # appbase params 表幂等建表 + 默认参数
         try:
             await sor.execute(_PARAMS_DDL, {})