fix(deploy): create_tables.py rewrite - all DDL via json2ddl from models/*.json, never DROP (2026-09-07 wipe incident), generic column sync, idempotent indexes by column-set; build.sh remove 8b DROP section, uapi seed after tables
This commit is contained in:
parent
86bd282be8
commit
72a17f7b84
75
build.sh
75
build.sh
@ -123,53 +123,10 @@ done
|
||||
# pipeline_task 兜底：模块缺失时保留空目录防菜单 404（正常路径已由上方软链覆盖）
|
||||
[ -e wwwroot/pipeline_task ] || mkdir -p wwwroot/pipeline_task
|
||||
|
||||
# 8b. uapi + rag 模块建表 + 导入外部服务配置种子
|
||||
# uapi 表(upapp/uapi/uapiio...)必须先建，rag 的 uapi_seed.sql 才有表可插
|
||||
# rag 表统一 rag_ 前缀，与产线业务表隔离；数据独立于 ragserver（同代码不同宿主，各自的数据）
|
||||
for m in uapi rag dingdingflow account_resource storage_resource filemgr supplychain; do
|
||||
if [ -d "pkgs/$m/models" ] && ls pkgs/$m/models/*.json >/dev/null 2>&1; then
|
||||
cd "$cdir/pkgs/$m/models"
|
||||
"$cdir/py3/bin/json2ddl" mysql . > "/tmp/pipeline_${m}_ddl.sql" || {
|
||||
echo "ERROR: $m json2ddl failed" >&2; cd "$cdir"; exit 1; }
|
||||
if [ ! -s "/tmp/pipeline_${m}_ddl.sql" ]; then
|
||||
echo "ERROR: $m DDL empty — check models/*.json (need summary+fields)" >&2
|
||||
cd "$cdir"; exit 1
|
||||
fi
|
||||
cd "$cdir"
|
||||
fi
|
||||
done
|
||||
if [ -f /tmp/pipeline_uapi_ddl.sql ] || [ -f /tmp/pipeline_rag_ddl.sql ]; then
|
||||
"$cdir/py3/bin/python" - <<'PYEOF'
|
||||
import sys, os, subprocess
|
||||
sys.path.insert(0, os.getcwd())
|
||||
from appPublic.jsonConfig import getConfig
|
||||
from appPublic.aes import aes_decode_b64
|
||||
cfg = getConfig('.', {'workdir': '.'})
|
||||
kw = cfg.databases['pipeline'].kwargs
|
||||
pwd = aes_decode_b64(cfg.password_key, kw.password)
|
||||
# 顺序关键：uapi 建表 → rag 建表 → rag 的 uapi 配置种子
|
||||
for sqlfile in ['/tmp/pipeline_uapi_ddl.sql', '/tmp/pipeline_rag_ddl.sql',
|
||||
'/tmp/pipeline_dingdingflow_ddl.sql',
|
||||
'/tmp/pipeline_account_resource_ddl.sql',
|
||||
'/tmp/pipeline_storage_resource_ddl.sql',
|
||||
'/tmp/pipeline_filemgr_ddl.sql',
|
||||
'/tmp/pipeline_supplychain_ddl.sql',
|
||||
'pkgs/rag/init/uapi_seed.sql']:
|
||||
if not os.path.exists(sqlfile):
|
||||
print(' SKIP (missing): %s' % sqlfile)
|
||||
continue
|
||||
with open(sqlfile, 'rb') as f:
|
||||
r = subprocess.run(['mysql', '-h', str(kw.host), '-P', str(kw.port),
|
||||
'-u', str(kw.user), '-p%s' % pwd, str(kw.db)],
|
||||
stdin=f, capture_output=True)
|
||||
tag = os.path.basename(sqlfile)
|
||||
if r.returncode == 0:
|
||||
print(' rag/uapi: %s applied' % tag)
|
||||
else:
|
||||
err = r.stderr.decode('utf-8', 'replace').strip().split(chr(10))[0]
|
||||
print(' WARN %s: %s' % (tag, err[:160]))
|
||||
PYEOF
|
||||
fi
|
||||
# 8b. uapi/rag 等模块建表已统一交给 create_tables.py（步骤 10.6，动态扫描 pkgs/*/models，幂等不 DROP）。
|
||||
# 历史事故（2026-09-07）：本段曾对这 7 个模块跑含 `drop table if exists` 的 json2ddl DDL，
|
||||
# 每次部署清空 upapp/uapi/storres_spec/supplychain 等表——与 TABLE_MODULES 清库同源隐患。
|
||||
# rag 的 uapi_seed.sql 数据种子（INSERT IGNORE，依赖 uapi 表先建）移到步骤 10.65 执行。
|
||||
|
||||
# 9. Sync password_key from Sage (not databases — those are app-specific)
|
||||
if [ -f /d/apitest/sage/conf/config.json ]; then
|
||||
@ -246,6 +203,30 @@ else
|
||||
echo " WARN: scripts/create_tables.py not found"
|
||||
fi
|
||||
|
||||
# 10.65 rag 模块 uapi 外部服务配置种子（INSERT IGNORE 幂等；依赖 uapi 表已由 10.6 建好）
|
||||
echo "=== rag uapi seed ==="
|
||||
if [ -f "$cdir/pkgs/rag/init/uapi_seed.sql" ]; then
|
||||
"$cdir/py3/bin/python" - <<'PYEOF'
|
||||
import sys, os, subprocess
|
||||
sys.path.insert(0, os.getcwd())
|
||||
from appPublic.jsonConfig import getConfig
|
||||
from appPublic.aes import aes_decode_b64
|
||||
cfg = getConfig('.', {'workdir': '.'})
|
||||
kw = cfg.databases['pipeline'].kwargs
|
||||
pwd = aes_decode_b64(cfg.password_key, kw.password)
|
||||
sqlfile = 'pkgs/rag/init/uapi_seed.sql'
|
||||
with open(sqlfile, 'rb') as f:
|
||||
r = subprocess.run(['mysql', '-h', str(kw.host), '-P', str(kw.port),
|
||||
'-u', str(kw.user), '-p%s' % pwd, str(kw.db)],
|
||||
stdin=f, capture_output=True)
|
||||
if r.returncode == 0:
|
||||
print(' rag uapi_seed applied')
|
||||
else:
|
||||
err = r.stderr.decode('utf-8', 'replace').strip().split(chr(10))[0]
|
||||
print(' WARN uapi_seed: %s' % err[:160])
|
||||
PYEOF
|
||||
fi
|
||||
|
||||
# 10.7 Import module init data (appcodes dictionaries)
|
||||
echo "=== Module init data import ==="
|
||||
if [ -f "$cdir/scripts/import_init.py" ]; then
|
||||
|
||||
@ -1,12 +1,21 @@
|
||||
#!/usr/bin/env python3
|
||||
"""建表脚本：从各模块 models/*.json 生成 DDL 并在 pipeline 库执行。
|
||||
"""建表脚本：从各模块 models/*.json 经 json2ddl 生成 DDL，幂等落库。
|
||||
|
||||
对每个模块，调用 json2ddl 生成 DDL，按分号分割后逐条执行（幂等：DDL 含 DROP TABLE IF EXISTS）。
|
||||
铁律（2026-09-07 用户定夺）：
|
||||
1. DDL 一律由 json2ddl 从 models/*.json 生成——models JSON 是表结构唯一事实源，
|
||||
禁止手写 CREATE TABLE（历史手写段已全部移除，含 pipeline-llm mysql.ddl.sql 依赖）。
|
||||
2. 全程幂等：CREATE TABLE IF NOT EXISTS，**绝不 DROP**——2026-09-07 14:08 事故：
|
||||
原 TABLE_MODULES 走 DROP 重建，一次部署清空 pipeline 库 product/pricing/discount/sms
|
||||
全部数据（13 个模型定价丢失、27 个产品映射丢失、全平台记账 failed）。
|
||||
3. 存量表 schema 演进靠「通用列同步器」：解析 DDL 的期望列 vs information_schema 实际列，
|
||||
缺列 ALTER TABLE ADD COLUMN 补齐。加字段只改 models JSON，部署自动补列。
|
||||
4. 索引幂等：按 (表, 列集合, 唯一性) 判重后 CREATE（不按索引名——手写时代的旧索引名
|
||||
与 json2ddl 生成名不同，按名判重会重复建同字段索引）。
|
||||
|
||||
用法：
|
||||
py3/bin/python scripts/create_tables.py
|
||||
"""
|
||||
import sys, os, re, asyncio, subprocess
|
||||
import sys, os, re, json, asyncio, subprocess, glob
|
||||
|
||||
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
|
||||
ROOT_DIR = os.path.dirname(SCRIPT_DIR)
|
||||
@ -19,190 +28,28 @@ from appPublic.folderUtils import ProgramPath
|
||||
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 等）
|
||||
# appbase 也放这里：svgicon/appcodes/appcodes_kv/params 是系统级配置表，含种子数据，幂等建（params 另在下方手动 ensure 默认参数）
|
||||
# dapi 放这里：downapp/downapikey 是下位系统接入表，幂等建（不 DROP，避免误删已注册的下位系统）
|
||||
# accounting 放这里：subject/account_config/accounting_config 是记账配置，account/ledger 是运行时数据，幂等建（不 DROP）
|
||||
# 产线模块 pipeline-bidding(bid_*) / pipeline-opportunity(opp_*) 放这里：走 json2ddl 从 models/*.json 生成
|
||||
# （已验证无 DEFAULT ''N'' bug），项目里含运行时数据（标书章节/商机报告），必须幂等不 DROP
|
||||
IDEMPOTENT_MODULES = ['pipeline_core', 'pipeline-service', 'appbase', 'dapi', 'accounting',
|
||||
'pipeline-bidding', 'pipeline-opportunity']
|
||||
|
||||
|
||||
# appbase 系统级配置表（幂等建表 + 默认参数，不 DROP 以免清空运行时数据）
|
||||
_PARAMS_DDL = """
|
||||
CREATE TABLE IF NOT EXISTS params (
|
||||
`id` VARCHAR(32) comment 'id',
|
||||
`params_name` VARCHAR(255) comment '参数名称',
|
||||
`params_value` VARCHAR(4000) comment '参数值',
|
||||
primary key(id)
|
||||
) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci engine=innodb comment '系统参数表';
|
||||
"""
|
||||
# 不建表的模块（无 models/ 或表由宿主/其他机制管理）
|
||||
SKIP_MODULES = set()
|
||||
|
||||
# appbase params 默认参数（数据种子，非 schema；params 表结构走 models JSON）
|
||||
_PARAMS_INIT = [
|
||||
("workspace_base", "/d/pipeline/workspaces"),
|
||||
("task_max_retry", "3"),
|
||||
("register_open", "1"),
|
||||
]
|
||||
|
||||
# sd_features 功能/需求表 DDL（手写，避开 json2ddl 的 DEFAULT ''N'' bug；索引内联避免重复建 Duplicate key name）
|
||||
_SD_FEATURES_DDL = """
|
||||
CREATE TABLE IF NOT EXISTS sd_features (
|
||||
`id` VARCHAR(32) NOT NULL comment '主键ID',
|
||||
`project_id` VARCHAR(32) NOT NULL comment '项目ID',
|
||||
`iteration_id` VARCHAR(32) comment '迭代ID',
|
||||
`feature_name` VARCHAR(200) NOT NULL comment '功能名称',
|
||||
`description` text comment '功能描述',
|
||||
`feature_type` VARCHAR(20) NOT NULL DEFAULT 'new_feature' comment '功能类型',
|
||||
`priority` VARCHAR(10) NOT NULL DEFAULT 'P2' comment '优先级',
|
||||
`status` VARCHAR(20) NOT NULL DEFAULT 'proposed' comment '功能状态',
|
||||
`acceptance_criteria` text comment '验收标准',
|
||||
`task_id` VARCHAR(32) comment '关联Pipeline任务ID',
|
||||
`created_by` VARCHAR(32) comment '创建人',
|
||||
`created_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL comment '创建时间',
|
||||
`updated_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP comment '更新时间',
|
||||
primary key(id),
|
||||
KEY idx_sd_features_project(project_id),
|
||||
KEY idx_sd_features_iteration(iteration_id),
|
||||
KEY idx_sd_features_status(status)
|
||||
) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci engine=innodb comment '功能/需求表';
|
||||
"""
|
||||
|
||||
# pipeline-service v2 运行时表（原 init.py::_init_v2_tables 启动时建，现迁到部署期）。
|
||||
# 手写幂等 DDL：避开 json2ddl 的 DEFAULT ''N'' bug；索引内联避免重复建 Duplicate key name。
|
||||
# 铁律：运行期不做任何 schema 变更，DB 全部变化必须在部署期（build.sh → create_tables.py）完成。
|
||||
_V2_RUNTIME_TABLES = [
|
||||
"""
|
||||
CREATE TABLE IF NOT EXISTS pipeline_user_memory (
|
||||
`id` varchar(32) NOT NULL,
|
||||
`memory_key` varchar(64) NOT NULL,
|
||||
`content` text NOT NULL,
|
||||
`category` varchar(32) NOT NULL DEFAULT 'memory',
|
||||
`scope` varchar(32) NOT NULL DEFAULT 'global',
|
||||
`scope_id` varchar(64) NOT NULL DEFAULT '',
|
||||
`priority` int(11) NOT NULL DEFAULT 0,
|
||||
`access_count` int(11) NOT NULL DEFAULT 0,
|
||||
`created_at` datetime NOT NULL DEFAULT current_timestamp(),
|
||||
`updated_at` datetime NOT NULL DEFAULT current_timestamp() ON UPDATE current_timestamp(),
|
||||
PRIMARY KEY (`id`),
|
||||
UNIQUE KEY `uk_memory` (`memory_key`,`category`),
|
||||
KEY `idx_category_priority` (`category`,`priority`)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;
|
||||
""",
|
||||
"""
|
||||
CREATE TABLE IF NOT EXISTS sd_deploy_accounts (
|
||||
`id` varchar(32) NOT NULL,
|
||||
`user_id` varchar(32) NOT NULL,
|
||||
`account_name` varchar(64) NOT NULL,
|
||||
`deploy_dir` varchar(500) NOT NULL,
|
||||
`status` varchar(20) NOT NULL DEFAULT 'active',
|
||||
`sandbox_config` text,
|
||||
`created_at` datetime NOT NULL DEFAULT current_timestamp(),
|
||||
`updated_at` datetime NOT NULL DEFAULT current_timestamp() ON UPDATE current_timestamp(),
|
||||
PRIMARY KEY (`id`),
|
||||
UNIQUE KEY `uk_account_name` (`account_name`),
|
||||
UNIQUE KEY `uk_user_id` (`user_id`),
|
||||
KEY `idx_status` (`status`)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
||||
""",
|
||||
"""
|
||||
CREATE TABLE IF NOT EXISTS sd_work_envs (
|
||||
`id` varchar(32) NOT NULL,
|
||||
`owner_type` varchar(10) NOT NULL,
|
||||
`owner_id` varchar(32) NOT NULL,
|
||||
`mode` varchar(10) NOT NULL DEFAULT 'local',
|
||||
`remote_host` varchar(200),
|
||||
`remote_port` int(11) DEFAULT 22,
|
||||
`remote_user` varchar(100),
|
||||
`remote_key_path` varchar(500),
|
||||
`remote_dir` varchar(500),
|
||||
`created_at` datetime NOT NULL DEFAULT current_timestamp(),
|
||||
`updated_at` datetime NOT NULL DEFAULT current_timestamp() ON UPDATE current_timestamp(),
|
||||
PRIMARY KEY (`id`),
|
||||
UNIQUE KEY `uk_owner` (`owner_type`,`owner_id`)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
||||
""",
|
||||
"""
|
||||
CREATE TABLE IF NOT EXISTS pipeline_session_settings (
|
||||
`id` varchar(32) NOT NULL,
|
||||
`session_id` varchar(64) NOT NULL,
|
||||
`user_id` varchar(32) NOT NULL,
|
||||
`current_project_id` varchar(32) NOT NULL DEFAULT '',
|
||||
`current_iteration_id` varchar(32) NOT NULL DEFAULT '',
|
||||
`created_at` datetime NOT NULL DEFAULT current_timestamp(),
|
||||
`updated_at` datetime NOT NULL DEFAULT current_timestamp() ON UPDATE current_timestamp(),
|
||||
PRIMARY KEY (`id`),
|
||||
UNIQUE KEY `uk_user_session` (`user_id`,`session_id`),
|
||||
KEY `idx_session` (`session_id`)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
||||
""",
|
||||
"""
|
||||
CREATE TABLE IF NOT EXISTS pipeline_llm_tokens (
|
||||
`id` varchar(32) NOT NULL,
|
||||
`token` varchar(80) NOT NULL,
|
||||
`org_id` varchar(32) NOT NULL DEFAULT '0',
|
||||
`project_id` varchar(32) NOT NULL DEFAULT '',
|
||||
`task_id` varchar(32) NOT NULL DEFAULT '',
|
||||
`model_name` varchar(100) NOT NULL DEFAULT '',
|
||||
`purpose` varchar(100) NOT NULL DEFAULT '',
|
||||
`status` varchar(20) NOT NULL DEFAULT 'active',
|
||||
`expires_at` datetime NOT NULL,
|
||||
`max_calls` int(11) NOT NULL DEFAULT 500,
|
||||
`call_count` int(11) NOT NULL DEFAULT 0,
|
||||
`prompt_tokens` bigint(20) NOT NULL DEFAULT 0,
|
||||
`completion_tokens` bigint(20) NOT NULL DEFAULT 0,
|
||||
`created_by` varchar(64) NOT NULL DEFAULT '',
|
||||
`created_at` datetime NOT NULL DEFAULT current_timestamp(),
|
||||
`last_used_at` datetime NULL,
|
||||
PRIMARY KEY (`id`),
|
||||
UNIQUE KEY `uk_token` (`token`),
|
||||
KEY `idx_org` (`org_id`),
|
||||
KEY `idx_project` (`project_id`),
|
||||
KEY `idx_status_exp` (`status`,`expires_at`)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
||||
""",
|
||||
# sd_iterations 存量数据修复（数据层，非 schema；seq_no 列由通用列同步器补）
|
||||
_DATA_FIXUPS = [
|
||||
# 「当前迭代」改为显式 status='in_progress'，迭代按 seq_no 顺序编号推进
|
||||
("sd_iterations seq_no 回填",
|
||||
"UPDATE sd_iterations t JOIN ("
|
||||
"SELECT id, ROW_NUMBER() OVER (PARTITION BY project_id ORDER BY created_at ASC, id ASC) AS rn "
|
||||
"FROM sd_iterations WHERE seq_no = 0) x ON x.id = t.id "
|
||||
"SET t.seq_no = x.rn"),
|
||||
("sd_iterations 脏状态修复 active→in_progress",
|
||||
"UPDATE sd_iterations SET status='in_progress', updated_at=NOW() WHERE status='active'"),
|
||||
]
|
||||
|
||||
# v2 引擎列迁移（原 init.py::_init_v2_tables 启动时 ALTER 迁移，现迁到部署期，information_schema 幂等）
|
||||
_V2_COLUMNS = {
|
||||
"pipelines": {"agent_config": "text"},
|
||||
"sd_org_settings": {"agent_config": "text"},
|
||||
"pipeline_tasks": {"retry_count": "int NOT NULL DEFAULT 0", "last_error": "text",
|
||||
"parent_id": "varchar(32) NULL", "depends_on": "text NULL"},
|
||||
}
|
||||
|
||||
# app_audit 审计日志表（原 app_audit/init.py::_init_audit 启动时建，现迁到部署期）
|
||||
_AUDIT_LOGS_DDL = """
|
||||
CREATE TABLE IF NOT EXISTS sd_audit_logs (
|
||||
`id` varchar(32) NOT NULL,
|
||||
`user_id` varchar(32),
|
||||
`username` varchar(100),
|
||||
`action` varchar(50) NOT NULL,
|
||||
`target` varchar(200),
|
||||
`detail` text,
|
||||
`result` varchar(10),
|
||||
`client_ip` varchar(64),
|
||||
`created_at` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
PRIMARY KEY (`id`),
|
||||
KEY `idx_user` (`user_id`),
|
||||
KEY `idx_action` (`action`),
|
||||
KEY `idx_created` (`created_at`)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
||||
"""
|
||||
|
||||
# git 级串行锁表：多 worker（含跨主机）对同一 repo 的 git 操作互斥，带 TTL 自动释放
|
||||
# (expires_at 过期可被下一个申请者原子接管)。lock_key = sha256(repo_abs_path)。
|
||||
_GIT_LOCKS_DDL = """
|
||||
CREATE TABLE IF NOT EXISTS pipeline_git_locks (
|
||||
`lock_key` varchar(64) NOT NULL,
|
||||
`token` varchar(64) NOT NULL,
|
||||
`expires_at` datetime NOT NULL,
|
||||
PRIMARY KEY (`lock_key`),
|
||||
KEY `idx_expires` (`expires_at`)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
||||
"""
|
||||
|
||||
|
||||
def split_ddl(ddl):
|
||||
"""按分号分割 DDL，跳过注释行和空语句。
|
||||
@ -221,15 +68,13 @@ def split_ddl(ddl):
|
||||
if in_str:
|
||||
cur.append(ch)
|
||||
if ch == "'":
|
||||
# '' 是转义的单引号，仍在串内
|
||||
if i + 1 < n and ddl[i + 1] == "'":
|
||||
if i + 1 < n and ddl[i + 1] == "'": # '' 转义，仍在串内
|
||||
cur.append("'")
|
||||
i += 1
|
||||
else:
|
||||
in_str = False
|
||||
elif ch == '\\' and i + 1 < n:
|
||||
# 反斜杠转义（MySQL 风格）
|
||||
cur.append(ddl[i + 1])
|
||||
cur.append(ddl[i + 1]) # 反斜杠转义（MySQL 风格）
|
||||
i += 1
|
||||
elif ch == "'":
|
||||
in_str = True
|
||||
@ -252,6 +97,171 @@ def split_ddl(ddl):
|
||||
return stmts
|
||||
|
||||
|
||||
def parse_create_table(stmt):
|
||||
"""从 CREATE TABLE 语句解析 (表名, {列名: 列定义DDL})。
|
||||
|
||||
列定义直接取自 json2ddl 输出——类型映射逻辑单一来源（sqlor 模板），不在此重复实现。
|
||||
"""
|
||||
m = re.search(r'(?is)CREATE\s+TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?`?(\w+)`?\s*\(', stmt)
|
||||
if not m:
|
||||
return None, {}
|
||||
table = m.group(1)
|
||||
body = stmt[m.end():]
|
||||
# 截到最后一个 ) 之前（表选项在括号外）
|
||||
depth = 1
|
||||
end = -1
|
||||
in_str = False
|
||||
for i, ch in enumerate(body):
|
||||
if ch == "'" and (i == 0 or body[i - 1] != '\\'):
|
||||
in_str = not in_str
|
||||
elif not in_str:
|
||||
if ch == '(':
|
||||
depth += 1
|
||||
elif ch == ')':
|
||||
depth -= 1
|
||||
if depth == 0:
|
||||
end = i
|
||||
break
|
||||
if end < 0:
|
||||
return table, {}
|
||||
cols = {}
|
||||
for line in body[:end].split('\n'):
|
||||
line = line.strip().rstrip(',')
|
||||
cm = re.match(r'`(\w+)`\s+(.+)$', line)
|
||||
if cm:
|
||||
cols[cm.group(1)] = cm.group(2).strip()
|
||||
return table, cols
|
||||
|
||||
|
||||
def parse_create_index(stmt):
|
||||
"""解析 CREATE [UNIQUE] INDEX 语句 → (索引名, 表名, 唯一?, 列列表)。"""
|
||||
m = re.match(r'(?is)CREATE\s+(UNIQUE\s+)?INDEX\s+`?(\w+)`?\s+ON\s+`?(\w+)`?\s*\(([^)]+)\)', stmt)
|
||||
if not m:
|
||||
return None
|
||||
unique = bool(m.group(1))
|
||||
name, table = m.group(2), m.group(3)
|
||||
cols = [c.strip().strip('`') for c in m.group(4).split(',')]
|
||||
return name, table, unique, cols
|
||||
|
||||
|
||||
async def get_existing_indexes(sor, table):
|
||||
"""现网索引集合：{(唯一?, (列...)): 索引名}——按列集合判重，免疫索引改名。"""
|
||||
recs = await sor.sqlExe(
|
||||
"SELECT index_name, non_unique, column_name, seq_in_index "
|
||||
"FROM information_schema.statistics "
|
||||
"WHERE table_schema=DATABASE() AND table_name=${t}$ "
|
||||
"ORDER BY index_name, seq_in_index", {"t": table})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
acc = {}
|
||||
for r in (recs or []):
|
||||
name = getattr(r, 'index_name', '')
|
||||
acc.setdefault((name, str(getattr(r, 'non_unique', ''))), []).append(
|
||||
getattr(r, 'column_name', ''))
|
||||
return {(nu == '0', tuple(cols)): name for (name, nu), cols in acc.items()}
|
||||
|
||||
|
||||
async def get_existing_columns(sor, table):
|
||||
recs = await sor.sqlExe(
|
||||
"SELECT column_name FROM information_schema.columns "
|
||||
"WHERE table_schema=DATABASE() AND table_name=${t}$", {"t": table})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
return {getattr(r, 'column_name', '') for r in (recs or [])}
|
||||
|
||||
|
||||
async def table_exists(sor, table):
|
||||
recs = await sor.sqlExe(
|
||||
"SELECT COUNT(*) AS c FROM information_schema.tables "
|
||||
"WHERE table_schema=DATABASE() AND table_name=${t}$", {"t": table})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
return bool(recs) and int(getattr(recs[0], 'c', 0) or 0) > 0
|
||||
|
||||
|
||||
async def apply_module_ddl(sor, mod, models_dir, stats):
|
||||
"""单模块：json2ddl → 幂等建表 → 列同步 → 索引幂等。"""
|
||||
json2ddl = os.path.join(ROOT_DIR, 'py3', 'bin', 'json2ddl')
|
||||
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 ({(r.stderr or "").strip()[:120]})')
|
||||
return
|
||||
|
||||
creates, indexes = [], []
|
||||
for stmt in split_ddl(r.stdout):
|
||||
head = stmt.lstrip('(').lstrip().split(None, 2)
|
||||
kw = (head[0] + ' ' + (head[1] if len(head) > 1 else '')).upper()
|
||||
if re.match(r'(?i)^\s*DROP\s+TABLE', stmt):
|
||||
continue # 铁律：绝不 DROP（2026-09-07 清库事故根治）
|
||||
if re.match(r'(?i)^\s*CREATE\s+TABLE', stmt):
|
||||
creates.append(stmt)
|
||||
elif re.match(r'(?i)^\s*CREATE\s+(UNIQUE\s+)?INDEX', stmt):
|
||||
indexes.append(stmt)
|
||||
elif re.match(r'(?i)^\s*ALTER\s+TABLE', stmt):
|
||||
pass # json2ddl 不产 ALTER；存量演进走列同步器
|
||||
else:
|
||||
stats['other'] += 1
|
||||
|
||||
# 1) 建表（IF NOT EXISTS）+ 收集期望列
|
||||
expected_cols = {}
|
||||
for stmt in creates:
|
||||
table, cols = parse_create_table(stmt)
|
||||
if not table:
|
||||
continue
|
||||
expected_cols[table] = cols
|
||||
s = re.sub(r'(?i)CREATE\s+TABLE\s+', 'CREATE TABLE IF NOT EXISTS ', stmt, count=1)
|
||||
existed = await table_exists(sor, table)
|
||||
try:
|
||||
await sor.execute(s, {})
|
||||
if not existed:
|
||||
stats['created'] += 1
|
||||
print(f' + table {table} (新建)')
|
||||
except Exception as e:
|
||||
print(f' ! table {table} 建表失败: {str(e)[:160]}')
|
||||
stats['errors'] += 1
|
||||
|
||||
# 2) 列同步（存量表补新列——models JSON 是唯一事实源）
|
||||
for table, cols in expected_cols.items():
|
||||
if not await table_exists(sor, table):
|
||||
continue
|
||||
have = await get_existing_columns(sor, table)
|
||||
for col, colddl in cols.items():
|
||||
if col in have:
|
||||
continue
|
||||
try:
|
||||
await sor.execute(
|
||||
f'ALTER TABLE `{table}` ADD COLUMN `{col}` {colddl}', {})
|
||||
stats['cols_added'] += 1
|
||||
print(f' + column {table}.{col} (存量表补列)')
|
||||
except Exception as e:
|
||||
print(f' ! column {table}.{col} 补列失败: {str(e)[:160]}')
|
||||
stats['errors'] += 1
|
||||
|
||||
# 3) 索引幂等（按 列集合+唯一性 判重，免疫手写时代的旧索引名）
|
||||
idx_cache = {}
|
||||
for stmt in indexes:
|
||||
parsed = parse_create_index(stmt)
|
||||
if not parsed:
|
||||
continue
|
||||
name, table, unique, cols = parsed
|
||||
if not await table_exists(sor, table):
|
||||
continue
|
||||
if table not in idx_cache:
|
||||
idx_cache[table] = await get_existing_indexes(sor, table)
|
||||
if (unique, tuple(cols)) in idx_cache[table]:
|
||||
stats['idx_skipped'] += 1
|
||||
continue
|
||||
try:
|
||||
await sor.execute(stmt, {})
|
||||
idx_cache[table][(unique, tuple(cols))] = name
|
||||
stats['idx_created'] += 1
|
||||
print(f' + index {table}({",".join(cols)}){" UNIQUE" if unique else ""}')
|
||||
except Exception as e:
|
||||
msg = str(e)[:160]
|
||||
if 'Duplicate key name' in msg:
|
||||
stats['idx_skipped'] += 1
|
||||
else:
|
||||
print(f' ! index {name} 失败: {msg}')
|
||||
stats['errors'] += 1
|
||||
|
||||
|
||||
async def main():
|
||||
config = getConfig(ROOT_DIR, NS={'workdir': ROOT_DIR, 'ProgramPath': ProgramPath()})
|
||||
DBPools(config.databases)
|
||||
@ -259,210 +269,50 @@ async def main():
|
||||
env = ServerEnv()
|
||||
env.get_module_dbname = lambda m: 'pipeline' if 'pipeline' in m else 'sage'
|
||||
|
||||
json2ddl = os.path.join(ROOT_DIR, 'py3', 'bin', 'json2ddl')
|
||||
# 动态扫描 pkgs/*/models——不再维护静态清单
|
||||
#（历史教训：清单遗漏 → pricing 定价管理 500、product/pricing 走 DROP 清库）
|
||||
mods = []
|
||||
for d in sorted(glob.glob(os.path.join(ROOT_DIR, 'pkgs', '*', 'models'))):
|
||||
mod = os.path.basename(os.path.dirname(d))
|
||||
if mod in SKIP_MODULES:
|
||||
continue
|
||||
if not glob.glob(os.path.join(d, '*.json')):
|
||||
continue
|
||||
mods.append((mod, d))
|
||||
print(f' modules with models/: {len(mods)}')
|
||||
|
||||
stats = {'created': 0, 'cols_added': 0, 'idx_created': 0, 'idx_skipped': 0,
|
||||
'errors': 0, 'other': 0}
|
||||
async with DBPools().sqlorContext('pipeline') as sor:
|
||||
for mod in TABLE_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):
|
||||
await sor.execute(stmt, {})
|
||||
n += 1
|
||||
print(f' tables: {mod} created ({n} statements)')
|
||||
for mod, models_dir in mods:
|
||||
print(f' === {mod} ===')
|
||||
await apply_module_ddl(sor, mod, models_dir, stats)
|
||||
|
||||
# 幂等建表模块（去 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
|
||||
# 跳过索引/约束语句（表已存在时索引也已存在，重复建会 Duplicate key name）
|
||||
if re.match(r'(?i)\s*(CREATE\s+(UNIQUE\s+)?INDEX|ALTER\s+TABLE)', s):
|
||||
continue
|
||||
await sor.execute(s, {})
|
||||
n += 1
|
||||
print(f' tables: {mod} ensured ({n} statements, idempotent)')
|
||||
|
||||
# appbase params 表幂等建表 + 默认参数
|
||||
# appbase params 默认参数（数据种子）
|
||||
try:
|
||||
await sor.execute(_PARAMS_DDL, {})
|
||||
for pname, pval in _PARAMS_INIT:
|
||||
await sor.execute(
|
||||
"INSERT INTO params (id, params_name, params_value) "
|
||||
"VALUES (${id}$, ${n}$, ${v}$) "
|
||||
"ON DUPLICATE KEY UPDATE params_name=params_name",
|
||||
{"id": pname, "n": pname, "v": pval})
|
||||
print(' tables: appbase params ensured (workspace_base)')
|
||||
print(' seed: appbase params ensured (workspace_base/task_max_retry/register_open)')
|
||||
except Exception as e:
|
||||
print(f' WARN: appbase params ensure failed: {e}')
|
||||
print(f' WARN: appbase params seed failed: {e}')
|
||||
|
||||
# pipeline_agent_questions 团队沟通列幂等迁移（显式路由版：current_handler 两列替代 escalation 路径快照）
|
||||
try:
|
||||
_add_cols = {
|
||||
"from_agentid": "VARCHAR(64) NULL COMMENT '提问Agent标识'",
|
||||
"problem_type": "VARCHAR(50) NULL COMMENT '问题类型'",
|
||||
"current_handler_role": "VARCHAR(32) NULL COMMENT '当前处理角色'",
|
||||
"current_handler_agentid": "VARCHAR(64) NULL COMMENT '当前处理Agent标识(空=该角色任意)'",
|
||||
}
|
||||
for col, ddl in _add_cols.items():
|
||||
r = await sor.sqlExe(
|
||||
"SELECT COUNT(*) as c FROM information_schema.COLUMNS "
|
||||
"WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='pipeline_agent_questions' "
|
||||
"AND COLUMN_NAME=${col}$", {"col": col})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
if not r or getattr(r[0], 'c', 0) == 0:
|
||||
await sor.execute(
|
||||
f"ALTER TABLE pipeline_agent_questions ADD COLUMN {col} {ddl}", {})
|
||||
# 废弃列清理（早期路径快照方案，已改为显式 handler）
|
||||
for col in ["escalation_path", "escalation_pos"]:
|
||||
r = await sor.sqlExe(
|
||||
"SELECT COUNT(*) as c FROM information_schema.COLUMNS "
|
||||
"WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='pipeline_agent_questions' "
|
||||
"AND COLUMN_NAME=${col}$", {"col": col})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
if r and getattr(r[0], 'c', 0) > 0:
|
||||
await sor.execute(
|
||||
f"ALTER TABLE pipeline_agent_questions DROP COLUMN {col}", {})
|
||||
print(' migrate: pipeline_agent_questions team-communication columns ensured')
|
||||
except Exception as e:
|
||||
print(f' WARN: pipeline_agent_questions migrate failed: {e}')
|
||||
# 存量数据修复（非 schema）
|
||||
for label, sql in _DATA_FIXUPS:
|
||||
try:
|
||||
await sor.execute(sql, {})
|
||||
print(f' fixup: {label}')
|
||||
except Exception as e:
|
||||
print(f' WARN: {label} failed: {str(e)[:120]}')
|
||||
|
||||
# pipeline_human_tasks SDLC 项目级人类任务列迁移（项目/迭代/bug 归属 + QC + bug 验收）
|
||||
try:
|
||||
_pht_cols = {
|
||||
"project_id": "VARCHAR(32) NULL COMMENT '项目ID'",
|
||||
"iteration_id": "VARCHAR(32) NULL COMMENT '迭代ID'",
|
||||
"bug_id": "VARCHAR(32) NULL COMMENT '关联BugID'",
|
||||
"qc_status": "VARCHAR(20) NULL DEFAULT 'pending' COMMENT 'QC状态'",
|
||||
"qc_comment": "VARCHAR(500) NULL COMMENT 'QC意见'",
|
||||
"title": "VARCHAR(500) NULL COMMENT '任务标题'",
|
||||
"description": "TEXT NULL COMMENT '任务描述'",
|
||||
}
|
||||
for col, ddl in _pht_cols.items():
|
||||
r = await sor.sqlExe(
|
||||
"SELECT COUNT(*) as c FROM information_schema.COLUMNS "
|
||||
"WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='pipeline_human_tasks' "
|
||||
"AND COLUMN_NAME=${col}$", {"col": col})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
if not r or getattr(r[0], 'c', 0) == 0:
|
||||
await sor.execute(
|
||||
f"ALTER TABLE pipeline_human_tasks ADD COLUMN {col} {ddl}", {})
|
||||
for idx_name, idx_col in [("pipeline_human_tasks_idx_pht_project", "project_id"),
|
||||
("pipeline_human_tasks_idx_pht_iteration", "iteration_id")]:
|
||||
r = await sor.sqlExe(
|
||||
"SELECT COUNT(*) as c FROM information_schema.STATISTICS "
|
||||
"WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='pipeline_human_tasks' "
|
||||
"AND INDEX_NAME=${idx}$", {"idx": idx_name})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
if not r or getattr(r[0], 'c', 0) == 0:
|
||||
await sor.execute(
|
||||
f"CREATE INDEX {idx_name} ON pipeline_human_tasks({idx_col})", {})
|
||||
print(' migrate: pipeline_human_tasks project/iteration/bug/qc columns ensured')
|
||||
except Exception as e:
|
||||
print(f' WARN: pipeline_human_tasks migrate failed: {e}')
|
||||
|
||||
# sd_features 功能/需求表（pipeline-sdlc 业务表）幂等建表——手写 DDL 避开 json2ddl 的 DEFAULT ''N'' bug
|
||||
try:
|
||||
await sor.execute(_SD_FEATURES_DDL, {})
|
||||
print(' tables: sd_features ensured (idempotent)')
|
||||
except Exception as e:
|
||||
print(f' WARN: sd_features ensure failed: {e}')
|
||||
|
||||
# pipeline-llm 模型治理表（8 张，llm_ 前缀）——模块自带手写幂等 DDL（mysql.ddl.sql），
|
||||
# 不走 json2ddl（模型定义含 'active' 等带引号默认值，会触发 DEFAULT ''N'' bug）
|
||||
try:
|
||||
_llm_ddl_path = os.path.join(ROOT_DIR, 'pkgs', 'pipeline-llm', 'mysql.ddl.sql')
|
||||
if os.path.isfile(_llm_ddl_path):
|
||||
with open(_llm_ddl_path, 'r', encoding='utf-8') as _f:
|
||||
# 先剔除注释行（-- 开头），避免头部注释与首条 CREATE 粘连被整段跳过
|
||||
_llm_ddl = '\n'.join(
|
||||
_l for _l in _f.read().splitlines() if not _l.strip().startswith('--'))
|
||||
_n = 0
|
||||
for _stmt in split_ddl(_llm_ddl):
|
||||
await sor.execute(_stmt, {})
|
||||
_n += 1
|
||||
print(f' tables: pipeline-llm ensured ({_n} statements, idempotent)')
|
||||
else:
|
||||
print(' WARN: pkgs/pipeline-llm/mysql.ddl.sql not found (skip llm_ tables)')
|
||||
except Exception as e:
|
||||
print(f' WARN: pipeline-llm tables ensure failed: {e}')
|
||||
|
||||
# pipeline-service v2 运行时表（原启动时建，现迁到部署期幂等建表）
|
||||
try:
|
||||
for ddl in _V2_RUNTIME_TABLES:
|
||||
await sor.execute(ddl, {})
|
||||
print(' tables: pipeline-service v2 runtime tables ensured (pipeline_user_memory/sd_deploy_accounts/sd_work_envs)')
|
||||
except Exception as e:
|
||||
print(f' WARN: v2 runtime tables ensure failed: {e}')
|
||||
|
||||
# v2 引擎列迁移（原启动时 ALTER，现迁到部署期，information_schema 查列存在性幂等）
|
||||
try:
|
||||
for tbl, cols in _V2_COLUMNS.items():
|
||||
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=${tbl}$ "
|
||||
"AND COLUMN_NAME=${col}$", {"tbl": tbl, "col": col})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
if not r or getattr(r[0], 'c', 0) == 0:
|
||||
await sor.execute(f"ALTER TABLE {tbl} ADD COLUMN {col} {ddl}", {})
|
||||
print(' migrate: v2 engine columns ensured (agent_config/retry_count/last_error)')
|
||||
except Exception as e:
|
||||
print(f' WARN: v2 engine column migrate failed: {e}')
|
||||
|
||||
# app_audit 审计日志表（原启动时建，现迁到部署期幂等建表）
|
||||
try:
|
||||
await sor.execute(_AUDIT_LOGS_DDL, {})
|
||||
print(' tables: sd_audit_logs ensured (idempotent)')
|
||||
except Exception as e:
|
||||
print(f' WARN: sd_audit_logs ensure failed: {e}')
|
||||
|
||||
# git 级串行锁表（跨主机多 worker 对同 repo git 操作互斥）
|
||||
try:
|
||||
await sor.execute(_GIT_LOCKS_DDL, {})
|
||||
print(' tables: pipeline_git_locks ensured (idempotent)')
|
||||
except Exception as e:
|
||||
print(f' WARN: pipeline_git_locks ensure failed: {e}')
|
||||
|
||||
# sd_iterations 迭代序号 seq_no 列 + 存量回填 + 脏状态修复（active→in_progress）
|
||||
# 「当前迭代」改为显式 status='in_progress' 标记，迭代按 seq_no 顺序编号推进，
|
||||
# 不再靠 created_at 推断当前迭代。
|
||||
try:
|
||||
r = await sor.sqlExe(
|
||||
"SELECT COUNT(*) as c FROM information_schema.COLUMNS "
|
||||
"WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='sd_iterations' "
|
||||
"AND COLUMN_NAME='seq_no'", {})
|
||||
await sor.sqlExe("COMMIT", {})
|
||||
if not r or getattr(r[0], 'c', 0) == 0:
|
||||
await sor.execute("ALTER TABLE sd_iterations ADD COLUMN seq_no INT NOT NULL DEFAULT 0", {})
|
||||
# 存量回填：seq_no=0 的迭代按每项目 created_at ASC 顺序赋 1,2,3...
|
||||
await sor.sqlExe(
|
||||
"UPDATE sd_iterations t JOIN ("
|
||||
"SELECT id, ROW_NUMBER() OVER (PARTITION BY project_id ORDER BY created_at ASC, id ASC) AS rn "
|
||||
"FROM sd_iterations WHERE seq_no = 0) x ON x.id = t.id "
|
||||
"SET t.seq_no = x.rn", {})
|
||||
# 脏状态修复：迭代状态合法值 planning/in_progress/completed/cancelled，无 active
|
||||
await sor.sqlExe(
|
||||
"UPDATE sd_iterations SET status='in_progress', updated_at=NOW() WHERE status='active'", {})
|
||||
print(' migrate: sd_iterations seq_no column + backfill + active→in_progress ensured')
|
||||
except Exception as e:
|
||||
print(f' WARN: sd_iterations migrate failed: {e}')
|
||||
print(f' DONE: 新建表={stats["created"]} 补列={stats["cols_added"]} '
|
||||
f'新建索引={stats["idx_created"]} 索引已存在跳过={stats["idx_skipped"]} '
|
||||
f'错误={stats["errors"]}')
|
||||
if stats['errors']:
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user