119 lines
5.3 KiB
Python
119 lines
5.3 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""DDL 迁移基座执行器（T01/T02）：幂等导入 models JSON 生成的 DDL 至 MariaDB hrs 库。
|
||
特性：
|
||
- CREATE TABLE IF NOT EXISTS（可重跑）；
|
||
- 缺索引自动补建（先查 information_schema.STATISTICS）；
|
||
- 迁移台账写入 hrs.schema_migration（version/表数/时间/校验和），重复执行只记一条最新；
|
||
- 依赖 mysql 客户端或 pymysql，自动探测。
|
||
用法：
|
||
python3 scripts/apply_ddl.py [--host 127.0.0.1] [--user test] [--password test123] [--db hrs] [--models models] [--version T01]
|
||
"""
|
||
import argparse, glob, hashlib, json, os, shutil, subprocess, sys
|
||
|
||
BASE = os.path.join(os.path.dirname(__file__), "..")
|
||
sys.path.insert(0, os.path.join(BASE, "scripts"))
|
||
from json2ddl import gen_table_ddl, gen_index_guard_sql # noqa: E402
|
||
|
||
def run_sql_cli(sql, args):
|
||
cmd = ["mysql", f"-h{args.host}", f"-P{args.port}", f"-u{args.user}", f"-p{args.password}", args.db]
|
||
p = subprocess.run(cmd, input=sql.encode("utf-8"), capture_output=True)
|
||
if p.returncode != 0:
|
||
raise RuntimeError(p.stderr.decode("utf-8", "ignore"))
|
||
return p.stdout.decode("utf-8", "ignore")
|
||
|
||
def exec_sql(sql, args):
|
||
try:
|
||
import pymysql
|
||
conn = pymysql.connect(host=args.host, port=args.port, user=args.user,
|
||
password=args.password, database=args.db, charset="utf8mb4")
|
||
try:
|
||
with conn.cursor() as cur:
|
||
cur.execute(sql)
|
||
conn.commit()
|
||
return None
|
||
finally:
|
||
conn.close()
|
||
except ImportError:
|
||
run_sql_cli(sql, args)
|
||
return None
|
||
|
||
def query_sql(sql, args):
|
||
if shutil.which("mysql"):
|
||
out = run_sql_cli(sql + "\n", args)
|
||
return [ln.split("\t") for ln in out.strip().splitlines() if ln]
|
||
import pymysql
|
||
conn = pymysql.connect(host=args.host, port=args.port, user=args.user,
|
||
password=args.password, database=args.db, charset="utf8mb4")
|
||
try:
|
||
with conn.cursor() as cur:
|
||
cur.execute(sql)
|
||
return [list(r) for r in cur.fetchall()]
|
||
finally:
|
||
conn.close()
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--host", default="127.0.0.1")
|
||
ap.add_argument("--port", default=3306, type=int)
|
||
ap.add_argument("--user", default="test")
|
||
ap.add_argument("--password", default="test123")
|
||
ap.add_argument("--db", default="hrs")
|
||
ap.add_argument("--models", default=os.path.join(BASE, "models"))
|
||
ap.add_argument("--version", default="T01")
|
||
args = ap.parse_args()
|
||
|
||
# 0) 迁移台账表
|
||
exec_sql("""CREATE TABLE IF NOT EXISTS `schema_migration` (
|
||
`version` VARCHAR(32) NOT NULL COMMENT '迁移版本号',
|
||
`applied_at` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '执行时间',
|
||
`table_count` INT NOT NULL DEFAULT 0 COMMENT '本次管理表数',
|
||
`checksum` VARCHAR(64) DEFAULT NULL COMMENT 'models JSON 合并校验和',
|
||
`operator` VARCHAR(64) DEFAULT 'agent.develop' COMMENT '执行人',
|
||
PRIMARY KEY (`version`)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='DDL迁移台账(幂等可重跑)';""", args)
|
||
|
||
# 1) 收集全部 models JSON，逐表建表 + 补索引
|
||
created, index_added, total = [], [], 0
|
||
h = hashlib.sha256()
|
||
jobs = []
|
||
for moddir in sorted(glob.glob(os.path.join(args.models, "*"))):
|
||
if not os.path.isdir(moddir):
|
||
continue
|
||
for p in sorted(glob.glob(os.path.join(moddir, "*.json"))):
|
||
doc = json.load(open(p, encoding="utf-8"))
|
||
h.update(open(p, "rb").read())
|
||
table, ddl = gen_table_ddl(p)
|
||
jobs.append((os.path.basename(moddir), table, ddl, doc))
|
||
total += 1
|
||
|
||
for mod, table, ddl, doc in jobs:
|
||
before = query_sql(f"SELECT COUNT(*) FROM information_schema.TABLES WHERE TABLE_SCHEMA='{args.db}' AND TABLE_NAME='{table}'", args)
|
||
n_exist = int(before[0][0]) if before else 0
|
||
exec_sql(ddl, args)
|
||
if not n_exist:
|
||
created.append(table)
|
||
# 索引补建（幂等）
|
||
rows = query_sql(f"SELECT INDEX_NAME FROM information_schema.STATISTICS WHERE TABLE_SCHEMA='{args.db}' AND TABLE_NAME='{table}'", args)
|
||
existing_idx = {r[0] for r in rows}
|
||
for idx in doc.get("indexes", []):
|
||
if idx["name"] not in existing_idx:
|
||
exec_sql(gen_index_guard_sql(table, idx), args)
|
||
index_added.append(f"{table}.{idx['name']}")
|
||
|
||
checksum = h.hexdigest()[:16]
|
||
# 2) 登记迁移台账（幂等 upsert）
|
||
exec_sql(f"""INSERT INTO schema_migration (version, table_count, checksum)
|
||
VALUES ('{args.version}', {total}, '{checksum}')
|
||
ON DUPLICATE KEY UPDATE applied_at=CURRENT_TIMESTAMP, table_count={total}, checksum='{checksum}';""", args)
|
||
|
||
# 3) 校验
|
||
rows = query_sql(f"SELECT COUNT(*) FROM information_schema.TABLES WHERE TABLE_SCHEMA='{args.db}'", args)
|
||
print(f"[apply_ddl] 管理业务表 {total} 张；新建 {len(created)} 张；补索引 {len(index_added)} 个")
|
||
print(f"[apply_ddl] 新建表: {', '.join(created) if created else '(无，全部已存在，幂等重跑)'}")
|
||
print(f"[apply_ddl] schema_migration version={args.version} table_count={total} checksum={checksum}")
|
||
print(f"[apply_ddl] 库 {args.db} 当前表总数: {rows[0][0]}")
|
||
|
||
if __name__ == "__main__":
|
||
main()
|