91 lines
3.5 KiB
Python
91 lines
3.5 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""json2ddl 等价实现（ocai database-table-definition-spec）：
|
||
读取 models/{module}/*.json（summary/fields/indexes/codes），生成幂等 MySQL/MariaDB DDL。
|
||
幂等性：CREATE TABLE IF NOT EXISTS；索引随建表语句一并声明，可整体重跑不报错。
|
||
用法：python3 scripts/json2ddl.py <models根目录> [输出sql文件]
|
||
"""
|
||
import json, os, sys, glob
|
||
|
||
TYPE_MAP = {
|
||
"varchar": lambda f: f"VARCHAR({f.get('length', 64)})",
|
||
"text": lambda f: "TEXT",
|
||
"int": lambda f: "INT",
|
||
"date": lambda f: "DATE",
|
||
"timestamp": lambda f: "TIMESTAMP NULL DEFAULT NULL",
|
||
"decimal": lambda f: f"DECIMAL({f.get('precision', 18)},{f.get('scale', 4)})",
|
||
}
|
||
|
||
def render_field(f):
|
||
t = f["type"]
|
||
if t not in TYPE_MAP:
|
||
raise ValueError(f"未知字段类型 {t} ({f.get('name')})")
|
||
col = f"`{f['name']}` {TYPE_MAP[t](f)}"
|
||
if f.get("notnull"):
|
||
col += " NOT NULL"
|
||
if "default" in f and t != "timestamp":
|
||
dv = f["default"]
|
||
col += f" DEFAULT '{dv}'" if isinstance(dv, str) else f" DEFAULT {dv}"
|
||
title = f["title"]
|
||
if f.get("sensitive"):
|
||
title += "【敏感:sensitive，落库加密/脱敏展示】"
|
||
title = title.replace("'", " ")
|
||
col += f" COMMENT '{title}'"
|
||
return col
|
||
|
||
def render_index_sql(table, idx):
|
||
kind = "UNIQUE KEY" if idx.get("unique") else "KEY"
|
||
cols = ",".join(f"`{c}`" for c in idx["fields"])
|
||
return f" {kind} `{idx['name']}` ({cols})"
|
||
|
||
def gen_table_ddl(path):
|
||
doc = json.load(open(path, encoding="utf-8"))
|
||
s = doc["summary"][0]
|
||
table = s["name"]
|
||
lines = [render_field(f) for f in doc["fields"]]
|
||
pk = ",".join(f"`{c}`" for c in s.get("pk", ["id"]))
|
||
lines.append(f" PRIMARY KEY ({pk})")
|
||
for idx in doc.get("indexes", []):
|
||
lines.append(render_index_sql(table, idx))
|
||
charset = s.get("charset", "utf8mb4")
|
||
collate = s.get("collate", "utf8mb4_unicode_ci")
|
||
title = str(s.get("title", table)).replace("'", " ")
|
||
return table, (f"CREATE TABLE IF NOT EXISTS `{table}` (\n" + ",\n".join(lines) +
|
||
f"\n) ENGINE=InnoDB DEFAULT CHARSET={charset} COLLATE={collate} COMMENT='{title}';")
|
||
|
||
def main():
|
||
src = sys.argv[1] if len(sys.argv) > 1 else "models"
|
||
out = sys.argv[2] if len(sys.argv) > 2 else None
|
||
parts = ["-- 人事系统 hr-web 批次1 DDL（json2ddl 生成，幂等可重跑）",
|
||
"-- 来源：models/*.json（database-design.md v3.0，49 表）；目标库：hrs(MariaDB)",
|
||
"-- 执行：mysql -utest -ptest123 hrs < sql/V001__init_hrs.sql 或 python3 scripts/apply_ddl.py",
|
||
"SET NAMES utf8mb4;",
|
||
"SET FOREIGN_KEY_CHECKS=0;", ""]
|
||
n = 0
|
||
for moddir in sorted(glob.glob(os.path.join(src, "*"))):
|
||
if not os.path.isdir(moddir):
|
||
continue
|
||
files = sorted(glob.glob(os.path.join(moddir, "*.json")))
|
||
if not files:
|
||
continue
|
||
parts.append(f"-- ===== {os.path.basename(moddir)}（{len(files)} 张）=====")
|
||
for p in files:
|
||
table, ddl = gen_table_ddl(p)
|
||
parts.append(f"-- {table}")
|
||
parts.append(ddl)
|
||
parts.append("")
|
||
n += 1
|
||
parts.append("SET FOREIGN_KEY_CHECKS=1;")
|
||
parts.append(f"-- 共 {n} 张表")
|
||
sql = "\n".join(parts) + "\n"
|
||
if out:
|
||
os.makedirs(os.path.dirname(out) or ".", exist_ok=True)
|
||
with open(out, "w", encoding="utf-8") as f:
|
||
f.write(sql)
|
||
print(f"生成 {out}：{n} 张表")
|
||
else:
|
||
sys.stdout.write(sql)
|
||
|
||
if __name__ == "__main__":
|
||
main()
|