From 72a17f7b845fd26467fb5b058871b83cb38dcaa4 Mon Sep 17 00:00:00 2001 From: yumoqing Date: Tue, 8 Sep 2026 15:28:50 +0800 Subject: [PATCH] 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 --- build.sh | 75 ++--- scripts/create_tables.py | 598 +++++++++++++++------------------------ 2 files changed, 252 insertions(+), 421 deletions(-) diff --git a/build.sh b/build.sh index 2670c2e..25719d6 100644 --- a/build.sh +++ b/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 diff --git a/scripts/create_tables.py b/scripts/create_tables.py index ff41e8d..3173253 100644 --- a/scripts/create_tables.py +++ b/scripts/create_tables.py @@ -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__':