From 16ab9f73a08d079cf3cd60c1007e82acf2070c52 Mon Sep 17 00:00:00 2001
From: yumoqing <yumoqing@gmail.com>
Date: Wed, 19 Aug 2026 06:00:55 +0800
Subject: [PATCH] =?UTF-8?q?feat:=20migrate=5Ftask=5Fhierarchy.py=20?=
 =?UTF-8?q?=E5=AD=98=E9=87=8F=E7=8E=AF=E5=A2=83=E5=B9=82=E7=AD=89=E8=BF=81?=
 =?UTF-8?q?=E7=A7=BB=20parent=5Fid/depends=5Fon=EF=BC=88=E9=81=BF=E5=BC=80?=
 =?UTF-8?q?=20create=5Ftables.py=20=E5=85=A8=E9=87=8F=20DROP=20=E4=B8=9A?=
 =?UTF-8?q?=E5=8A=A1=E8=A1=A8=EF=BC=89?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 scripts/migrate_task_hierarchy.py | 49 +++++++++++++++++++++++++++++++
 1 file changed, 49 insertions(+)
 create mode 100644 scripts/migrate_task_hierarchy.py

diff --git a/scripts/migrate_task_hierarchy.py b/scripts/migrate_task_hierarchy.py
new file mode 100644
index 0000000..4c3be88
--- /dev/null
+++ b/scripts/migrate_task_hierarchy.py
@@ -0,0 +1,49 @@
+#!/usr/bin/env python3
+"""一次性迁移：pipeline_tasks 加 parent_id + depends_on 列（幂等，information_schema 查列存在性）。
+
+用途：任务分解与编排特性——父子关系(parent_id) + 依赖串行(depends_on)。
+已在 create_tables.py 的 _V2_COLUMNS 固化（供 build.sh 全新部署），本脚本供存量环境单独跑迁移。
+"""
+import sys
+import asyncio
+
+ROOT = '/d/pipeline/pipeline-app'
+sys.path.insert(0, ROOT + '/py3/lib/python3.10/site-packages')
+sys.path.insert(0, ROOT)
+
+from sqlor.dbpools import DBPools
+from appPublic.jsonConfig import getConfig
+from appPublic.folderUtils import ProgramPath
+
+COLS = {
+    'parent_id': 'varchar(32) NULL',
+    'depends_on': 'text NULL',
+}
+
+
+async def main():
+    config = getConfig(ROOT, NS={'workdir': ROOT, 'ProgramPath': ProgramPath()})
+    DBPools(config.databases)
+    async with DBPools().sqlorContext('pipeline') as sor:
+        for col, ddl in COLS.items():
+            r = await sor.sqlExe(
+                "SELECT COUNT(*) as c FROM information_schema.COLUMNS "
+                "WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='pipeline_tasks' "
+                "AND COLUMN_NAME=${col}$", {'col': col})
+            await sor.sqlExe("COMMIT", {})
+            if not r or getattr(r[0], 'c', 0) == 0:
+                await sor.execute("ALTER TABLE pipeline_tasks ADD COLUMN %s %s" % (col, ddl), {})
+                print('  added column:', col)
+            else:
+                print('  exists column:', col)
+        # 校验
+        chk = await sor.sqlExe(
+            "SELECT COLUMN_NAME FROM information_schema.COLUMNS "
+            "WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='pipeline_tasks' "
+            "AND COLUMN_NAME IN ('parent_id','depends_on')", {})
+        await sor.sqlExe("COMMIT", {})
+        print('  verify columns:', [getattr(x, 'COLUMN_NAME', '') for x in (chk or [])])
+    print('migration done')
+
+
+asyncio.run(main())