371 lines
14 KiB
Python
371 lines
14 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""M1b 表定义规范化工具：把 models/m1b/*.json 归一到 database-table-definition-spec 四段式。
|
||
|
||
背景（QC 退回意见 #8）：M1b 首版 models/m1b/*.json 用了「summary 为长字符串 +
|
||
fields 用 varchar/bigint/longtext 具体类型 + codes 为 dict」的自创格式，既不符合
|
||
database-table-definition-spec（四段式必须都是**数组**、type 必须是**抽象类型**），
|
||
也让 tools/m1b_selfcheck.py 的 B2/B3/B4/B5 检查直接抛 AttributeError。
|
||
|
||
本脚本做三件事，全部以 sql/m1b_ddl.sql 为唯一权威来源（保证 models↔DDL 一一对应）：
|
||
|
||
1. 解析 DDL → {table: {cols:[...], indexes:{name:(cols,unique)}, defaults:{col:default}}}
|
||
2. 把 DDL 的具体类型映射为规范抽象类型（varchar→str / bigint→long / longtext→text ...）
|
||
3. 生成规范四段式：
|
||
- summary : [{"name","title","primary":["id"],"catelog","comment"}]
|
||
- fields : [{"name","title","type","length","dec","nullable","default","comment"}]
|
||
- indexes : [{"name","idxtype":"unique|index","idxfields":[...]}] （PRIMARY 归 summary.primary）
|
||
- codes : [{"field","table":"appcodes_kv","valuefield":"k","textfield":"v",
|
||
"cond":"parentid='<字典组>'"}] （cond 必须 parentid=）
|
||
|
||
旧文件里的中文 label / comment / 字典组名会被完整继承（不丢设计信息），
|
||
字典组名从旧 fields[].comment 的「字典 pbl_xxx」中提取。
|
||
|
||
用法：
|
||
python3 tools/m1b_normalize_models.py # 就地重写 models/m1b/*.json
|
||
python3 tools/m1b_normalize_models.py --check # 只校验不改写，违规 rc=1
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
|
||
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
DDL = os.path.join(REPO, "sql", "m1b_ddl.sql")
|
||
MDIR = os.path.join(REPO, "pbl_blueprint", "models", "m1b")
|
||
|
||
# 表中文名（summary[0].title）与分类（catelog）
|
||
TABLE_META = {
|
||
"pbl_blueprint_template": ("PBL蓝图模板", "entity"),
|
||
"pbl_blueprint_ref": ("PBL蓝图外部关联", "relation"),
|
||
"pbl_subobject_ext": ("PBL子对象扩展属性", "entity"),
|
||
"pbl_ext_field_def": ("PBL扩展字段定义", "dimession"),
|
||
}
|
||
|
||
# DDL 具体类型 → 规范抽象类型（database-table-definition-spec 支持表）
|
||
TYPE_MAP = {
|
||
"varchar": "str",
|
||
"char": "char",
|
||
"tinytext": "text",
|
||
"text": "text",
|
||
"mediumtext": "text",
|
||
"longtext": "text",
|
||
"json": "text",
|
||
"tinyint": "short",
|
||
"smallint": "short",
|
||
"mediumint": "int",
|
||
"int": "int",
|
||
"integer": "int",
|
||
"bigint": "long",
|
||
"float": "float",
|
||
"double": "double",
|
||
"decimal": "double",
|
||
"numeric": "double",
|
||
"date": "date",
|
||
"time": "time",
|
||
"datetime": "datetime",
|
||
"timestamp": "timestamp",
|
||
"blob": "bin",
|
||
"longblob": "bin",
|
||
"varbinary": "bin",
|
||
"binary": "bin",
|
||
}
|
||
|
||
# 需要 length 的抽象类型
|
||
NEED_LENGTH = {"str", "char"}
|
||
# 需要 length + dec 的抽象类型
|
||
NEED_DEC = {"float", "double", "ddouble", "decimal"}
|
||
|
||
|
||
def read_ddl():
|
||
with open(DDL, "r", encoding="utf-8") as f:
|
||
return f.read()
|
||
|
||
|
||
def parse_ddl(body):
|
||
"""解析 DDL → {table: {"cols":[(name,sqltype,len,dec,notnull,default)],
|
||
"indexes":{name:(cols,unique)}, "pk":[cols]}}"""
|
||
out = {}
|
||
pat = re.compile(
|
||
r"CREATE\s+TABLE\s+IF\s+NOT\s+EXISTS\s+`?(\w+)`?\s*\((.*?)\n\)\s*ENGINE[^;]*;",
|
||
re.I | re.S)
|
||
for m in pat.finditer(body):
|
||
tname, inner = m.group(1), m.group(2)
|
||
cols, idx, pk = [], {}, []
|
||
for raw in inner.splitlines():
|
||
s = raw.strip().rstrip(",").strip()
|
||
if not s or s.startswith("--"):
|
||
continue
|
||
up = s.upper()
|
||
if up.startswith("PRIMARY KEY"):
|
||
pm = re.search(r"\(([^)]*)\)", s)
|
||
if pm:
|
||
pk = [c.strip().strip("`") for c in pm.group(1).split(",")]
|
||
continue
|
||
if up.startswith("UNIQUE KEY") or up.startswith("KEY ") \
|
||
or up.startswith("INDEX ") or up.startswith("UNIQUE INDEX"):
|
||
nm = re.match(
|
||
r"(?:UNIQUE\s+)?(?:KEY|INDEX)\s+`?(\w+)`?\s*\(([^)]*)\)", s, re.I)
|
||
if nm:
|
||
cs = [c.strip().strip("`") for c in nm.group(2).split(",")]
|
||
idx[nm.group(1)] = (cs, up.startswith("UNIQUE"))
|
||
continue
|
||
if up.startswith("CONSTRAINT") or up.startswith("FOREIGN KEY"):
|
||
continue
|
||
cm = re.match(r"`(\w+)`\s+([A-Za-z]+)\s*(\(([^)]*)\))?(.*)$", s)
|
||
if not cm:
|
||
continue
|
||
name, sqltype, _paren, args, tail = cm.groups()
|
||
sqltype = sqltype.lower()
|
||
ln, dec = None, None
|
||
if args:
|
||
parts = [x.strip() for x in args.split(",")]
|
||
try:
|
||
ln = int(parts[0])
|
||
except ValueError:
|
||
ln = None
|
||
if len(parts) > 1:
|
||
try:
|
||
dec = int(parts[1])
|
||
except ValueError:
|
||
dec = None
|
||
notnull = "NOT NULL" in tail.upper()
|
||
dm = re.search(r"DEFAULT\s+('(?:[^']*)'|NULL|[\w.+-]+)", tail, re.I)
|
||
default = None
|
||
if dm:
|
||
dv = dm.group(1)
|
||
if dv.upper() == "NULL":
|
||
default = None
|
||
elif dv.startswith("'") and dv.endswith("'"):
|
||
default = dv[1:-1]
|
||
else:
|
||
try:
|
||
default = int(dv)
|
||
except ValueError:
|
||
default = dv
|
||
cols.append((name, sqltype, ln, dec, notnull, default))
|
||
out[tname] = {"cols": cols, "indexes": idx, "pk": pk or ["id"]}
|
||
return out
|
||
|
||
|
||
def abstract_type(sqltype, ln, dec):
|
||
"""DDL 具体类型 → (抽象类型, length, dec)；未知类型抛错，禁止静默降级。"""
|
||
at = TYPE_MAP.get(sqltype)
|
||
if at is None:
|
||
raise ValueError("未登记的 DDL 类型：%s（请在 TYPE_MAP 中补映射）" % sqltype)
|
||
length = None
|
||
d = None
|
||
if at in NEED_LENGTH:
|
||
if not ln or ln <= 0:
|
||
raise ValueError("抽象类型 %s 必须有正整数 length（DDL 类型 %s）" % (at, sqltype))
|
||
length = int(ln)
|
||
if at in NEED_DEC:
|
||
length = int(ln) if ln else 18
|
||
d = int(dec) if dec else 2
|
||
return at, length, d
|
||
|
||
|
||
def load_legacy(table):
|
||
"""读旧 models 文件，返回 {col: {"title":..,"comment":..,"dict":..}} 供继承。"""
|
||
fp = os.path.join(MDIR, table + ".json")
|
||
info = {}
|
||
if not os.path.isfile(fp):
|
||
return info
|
||
try:
|
||
with open(fp, "r", encoding="utf-8") as f:
|
||
d = json.load(f)
|
||
except Exception:
|
||
return info
|
||
fields = d.get("fields")
|
||
if isinstance(fields, dict): # M1a 风格：fields 为 dict
|
||
items = [(k, v) for k, v in fields.items()]
|
||
elif isinstance(fields, list): # M1b 首版：fields 为 list
|
||
items = [(x.get("name"), x) for x in fields if isinstance(x, dict)]
|
||
else:
|
||
items = []
|
||
for name, fd in items:
|
||
if not name:
|
||
continue
|
||
title = fd.get("label") or fd.get("title") or name
|
||
comment = fd.get("comment") or fd.get("summary") or ""
|
||
dm = re.search(r"字典\s+([A-Za-z_][\w]*)", comment)
|
||
info[name] = {"title": title, "comment": comment,
|
||
"dict": dm.group(1) if dm else None}
|
||
return info
|
||
|
||
|
||
def build_model(table, spec):
|
||
legacy = load_legacy(table)
|
||
title, catelog = TABLE_META.get(table, (table, "entity"))
|
||
|
||
fields = []
|
||
for (name, sqltype, ln, dec, notnull, default) in spec["cols"]:
|
||
at, length, d = abstract_type(sqltype, ln, dec)
|
||
lg = legacy.get(name, {})
|
||
fd = {
|
||
"name": name,
|
||
"title": lg.get("title") or name,
|
||
"type": at,
|
||
}
|
||
if length is not None:
|
||
fd["length"] = length
|
||
if d is not None:
|
||
fd["dec"] = d
|
||
if notnull:
|
||
fd["nullable"] = "no"
|
||
if default is not None and default != "":
|
||
fd["default"] = default
|
||
elif default == "":
|
||
fd["default"] = ""
|
||
if lg.get("comment"):
|
||
fd["comment"] = lg["comment"]
|
||
fields.append(fd)
|
||
|
||
indexes = []
|
||
for iname in sorted(spec["indexes"]):
|
||
cols, uniq = spec["indexes"][iname]
|
||
indexes.append({
|
||
"name": iname,
|
||
"idxtype": "unique" if uniq else "index",
|
||
"idxfields": list(cols),
|
||
})
|
||
|
||
codes = []
|
||
seen = set()
|
||
for fd in fields:
|
||
grp = legacy.get(fd["name"], {}).get("dict")
|
||
if not grp or fd["name"] in seen:
|
||
continue
|
||
seen.add(fd["name"])
|
||
codes.append({
|
||
"field": fd["name"],
|
||
"table": "appcodes_kv",
|
||
"valuefield": "k",
|
||
"textfield": "v",
|
||
"cond": "parentid='%s'" % grp,
|
||
})
|
||
|
||
return {
|
||
"summary": [{
|
||
"name": table,
|
||
"title": title,
|
||
"primary": list(spec["pk"]),
|
||
"catelog": catelog,
|
||
"comment": "M1b 表；索引一律 tenant_key 打头（tenant_key=COALESCE(tenant_id,"
|
||
"'__PLATFORM__')，由 m1b_db.tenant_key() 统一计算）；无 FOREIGN KEY；"
|
||
"对 world/scene/entity/script 等复用域基表零改动（Q-OPEN-3）。",
|
||
}],
|
||
"fields": fields,
|
||
"indexes": indexes,
|
||
"codes": codes,
|
||
}
|
||
|
||
|
||
def validate(model, table, spec):
|
||
"""规范符合性自查，返回问题清单（空=通过）。"""
|
||
bad = []
|
||
for seg in ("summary", "fields", "indexes", "codes"):
|
||
if not isinstance(model.get(seg), list):
|
||
bad.append("%s: %s 不是数组" % (table, seg))
|
||
sm = model.get("summary") or []
|
||
if not (sm and isinstance(sm[0], dict)):
|
||
bad.append("%s: summary[0] 缺失" % table)
|
||
else:
|
||
if sm[0].get("name") != table:
|
||
bad.append("%s: summary[0].name=%r != 表名" % (table, sm[0].get("name")))
|
||
if not isinstance(sm[0].get("primary"), list) or not sm[0].get("primary"):
|
||
bad.append("%s: summary[0].primary 必须是非空数组" % table)
|
||
cols_ddl = [c[0] for c in spec["cols"]]
|
||
cols_model = [f.get("name") for f in model.get("fields", [])]
|
||
if cols_ddl != cols_model:
|
||
bad.append("%s: 字段顺序/集合与 DDL 不一致 仅DDL=%s 仅models=%s"
|
||
% (table, sorted(set(cols_ddl) - set(cols_model)),
|
||
sorted(set(cols_model) - set(cols_ddl))))
|
||
for f in model.get("fields", []):
|
||
t = f.get("type")
|
||
if t not in TYPE_MAP.values():
|
||
bad.append("%s.%s: 非法抽象类型 %r" % (table, f.get("name"), t))
|
||
if t in NEED_LENGTH and not (isinstance(f.get("length"), int) and f["length"] > 0):
|
||
bad.append("%s.%s: %s 缺正整数 length" % (table, f.get("name"), t))
|
||
if t in NEED_DEC:
|
||
if not (isinstance(f.get("length"), int) and f["length"] > 0):
|
||
bad.append("%s.%s: %s 缺 length" % (table, f.get("name"), t))
|
||
if not (isinstance(f.get("dec"), int) and f["dec"] > 0):
|
||
bad.append("%s.%s: %s 缺 dec" % (table, f.get("name"), t))
|
||
if f.get("name") == "id" and not (t == "str" and f.get("length", 0) >= 32
|
||
or t == "long"):
|
||
bad.append("%s.id: 主键类型异常 %r" % (table, t))
|
||
idx_model = {i.get("name"): (list(i.get("idxfields") or []), i.get("idxtype"))
|
||
for i in model.get("indexes", [])}
|
||
idx_ddl = {k: (list(v[0]), "unique" if v[1] else "index")
|
||
for k, v in spec["indexes"].items()}
|
||
if sorted(idx_model) != sorted(idx_ddl):
|
||
bad.append("%s: 索引名与 DDL 不一致 仅models=%s 仅DDL=%s"
|
||
% (table, sorted(set(idx_model) - set(idx_ddl)),
|
||
sorted(set(idx_ddl) - set(idx_model))))
|
||
for k in idx_model:
|
||
if k in idx_ddl and idx_model[k] != idx_ddl[k]:
|
||
bad.append("%s.%s: 索引定义不一致 models=%s DDL=%s"
|
||
% (table, k, idx_model[k], idx_ddl[k]))
|
||
if not isinstance(idx_model[k][0], list) or not idx_model[k][0]:
|
||
bad.append("%s.%s: idxfields 必须是非空数组" % (table, k))
|
||
if idx_model[k][0] and idx_model[k][0][0] != "tenant_key":
|
||
bad.append("%s.%s: 索引首列不是 tenant_key（%s）"
|
||
% (table, k, idx_model[k][0][0]))
|
||
for c in model.get("codes", []):
|
||
if c.get("table") == "appcodes_kv" and "parentid=" not in (c.get("cond") or ""):
|
||
bad.append("%s.codes[%s]: appcodes_kv 的 cond 必须用 parentid="
|
||
% (table, c.get("field")))
|
||
if "." in (c.get("table") or ""):
|
||
bad.append("%s.codes[%s]: table 禁用 module.table 点号写法"
|
||
% (table, c.get("field")))
|
||
flds = [c.get("field") for c in model.get("codes", [])]
|
||
if len(flds) != len(set(flds)):
|
||
bad.append("%s: codes 存在重复 field（会触发 Duplicate column name）" % table)
|
||
return bad
|
||
|
||
|
||
def main():
|
||
check_only = "--check" in sys.argv
|
||
body = read_ddl()
|
||
ddl = parse_ddl(body)
|
||
expect = sorted(TABLE_META)
|
||
if sorted(ddl) != expect:
|
||
print("FAIL DDL 表集合=%s 期望=%s" % (sorted(ddl), expect))
|
||
return 1
|
||
if not os.path.isdir(MDIR):
|
||
os.makedirs(MDIR)
|
||
all_bad = []
|
||
for t in expect:
|
||
model = build_model(t, ddl[t])
|
||
all_bad += validate(model, t, ddl[t])
|
||
fp = os.path.join(MDIR, t + ".json")
|
||
if check_only:
|
||
old = None
|
||
if os.path.isfile(fp):
|
||
try:
|
||
with open(fp, "r", encoding="utf-8") as f:
|
||
old = json.load(f)
|
||
except Exception:
|
||
old = None
|
||
if old != model:
|
||
all_bad.append("%s: 磁盘文件与规范四段式不一致（需重写）" % t)
|
||
else:
|
||
with open(fp, "w", encoding="utf-8") as f:
|
||
json.dump(model, f, ensure_ascii=False, indent=2)
|
||
f.write("\n")
|
||
print("WROTE %s fields=%d indexes=%d codes=%d"
|
||
% (os.path.relpath(fp, REPO), len(model["fields"]),
|
||
len(model["indexes"]), len(model["codes"])))
|
||
if all_bad:
|
||
print("VALIDATE_FAIL %d 项：" % len(all_bad))
|
||
for b in all_bad:
|
||
print(" -", b)
|
||
return 1
|
||
print("VALIDATE_OK 4 表 models 四段式全部符合 database-table-definition-spec")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|