320 lines
13 KiB
Python
320 lines
13 KiB
Python
#!/usr/bin/env python3
|
||
"""建表脚本：从各模块 models/*.json 经 json2ddl 生成 DDL，幂等落库。
|
||
|
||
铁律（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, json, asyncio, subprocess, glob
|
||
|
||
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
|
||
ROOT_DIR = os.path.dirname(SCRIPT_DIR)
|
||
sys.path.insert(0, os.path.join(ROOT_DIR, 'py3', 'lib', 'python3.10', 'site-packages'))
|
||
sys.path.insert(0, ROOT_DIR)
|
||
|
||
from sqlor.dbpools import DBPools
|
||
from appPublic.jsonConfig import getConfig
|
||
from appPublic.folderUtils import ProgramPath
|
||
from ahserver.serverenv import ServerEnv
|
||
from ahserver.globalEnv import initEnv
|
||
|
||
# 不建表的模块（无 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_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'"),
|
||
]
|
||
|
||
|
||
def split_ddl(ddl):
|
||
"""按分号分割 DDL，跳过注释行和空语句。
|
||
|
||
引号感知：单引号字符串内的分号不作为分隔符（实测事故 2026-09-07：
|
||
llm_usage 的 COMMENT '记账状态(三态:...;仅status=SUCCEEDED...)' 注释含分号，
|
||
裸 split(';') 把 CREATE TABLE 拦腰截断 → 语法错误 → 该模块建表整段中止）。
|
||
"""
|
||
stmts = []
|
||
cur = []
|
||
in_str = False
|
||
i = 0
|
||
n = len(ddl)
|
||
while i < n:
|
||
ch = ddl[i]
|
||
if in_str:
|
||
cur.append(ch)
|
||
if ch == "'":
|
||
if i + 1 < n and ddl[i + 1] == "'": # '' 转义，仍在串内
|
||
cur.append("'")
|
||
i += 1
|
||
else:
|
||
in_str = False
|
||
elif ch == '\\' and i + 1 < n:
|
||
cur.append(ddl[i + 1]) # 反斜杠转义（MySQL 风格）
|
||
i += 1
|
||
elif ch == "'":
|
||
in_str = True
|
||
cur.append(ch)
|
||
elif ch == ';':
|
||
raw = ''.join(cur)
|
||
lines = [l for l in raw.split('\n') if not l.strip().startswith('--')]
|
||
s = '\n'.join(lines).strip()
|
||
if s:
|
||
stmts.append(s)
|
||
cur = []
|
||
else:
|
||
cur.append(ch)
|
||
i += 1
|
||
raw = ''.join(cur)
|
||
lines = [l for l in raw.split('\n') if not l.strip().startswith('--')]
|
||
s = '\n'.join(lines).strip()
|
||
if s:
|
||
stmts.append(s)
|
||
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)
|
||
initEnv()
|
||
env = ServerEnv()
|
||
env.get_module_dbname = lambda m: 'pipeline' if 'pipeline' in m else 'sage'
|
||
|
||
# 动态扫描 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, models_dir in mods:
|
||
print(f' === {mod} ===')
|
||
await apply_module_ddl(sor, mod, models_dir, stats)
|
||
|
||
# appbase params 默认参数（数据种子）
|
||
try:
|
||
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(' seed: appbase params ensured (workspace_base/task_max_retry/register_open)')
|
||
except Exception as e:
|
||
print(f' WARN: appbase params seed 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]}')
|
||
|
||
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__':
|
||
asyncio.run(main())
|