# -*- 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())