264 lines
13 KiB
Python
264 lines
13 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""validate_models_json —— database-table-definition-spec 机械校验器（M11b-2b）。
|
||
|
||
QC #4/#7/#9/#10 要求：表定义 JSON 必须是 spec 规定的四段式（summary/fields/indexes/
|
||
codes），类型必须是 spec 声明的抽象类型，主键 id 必须是 str/length>=32，索引必须用
|
||
``idxtype`` + ``idxfields``。本脚本把这些规则变成**可执行的断言**，任何一份 models/*.json
|
||
不合规就非零退出（真源不合规同样 FAIL —— QC #9：真源不合规则引用方不可能合规）。
|
||
|
||
已登记偏离（唯一允许的偏离，需 --allow-registered-deviation 显式声明）：
|
||
DEV-ID-AUTOINCREMENT：主键 id 为 long + auto_increment（分区表物理主键需要），
|
||
spec 要求 str/32。该偏离登记在 COMPLIANCE_DEVIATIONS 中，逐条打印，不静默放过。
|
||
|
||
用法：
|
||
python validate_models_json.py <models目录1> [<目录2> ...]
|
||
python validate_models_json.py --json-report out.json <目录>
|
||
"""
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import sys
|
||
|
||
SPEC_ROOT_KEYS = ("summary", "fields", "indexes", "codes")
|
||
SUMMARY_KEYS = ("name", "title", "primary", "catelog")
|
||
SUMMARY_OPTIONAL = ("comment",)
|
||
FIELD_KEYS = ("name", "title", "type", "length", "dec", "nullable", "default")
|
||
INDEX_KEYS = ("name", "idxtype", "idxfields")
|
||
CODE_KEYS = ("field", "table", "valuefield", "textfield", "cond")
|
||
ABSTRACT_TYPES = {
|
||
"str": ("length",), "char": ("length",),
|
||
"short": (), "int": (), "long": (),
|
||
"float": ("length", "dec"), "double": ("length", "dec"),
|
||
"ddouble": ("length", "dec"), "decimal": ("length", "dec"),
|
||
"date": (), "time": (), "datetime": (), "timestamp": (),
|
||
"text": (), "bin": (),
|
||
}
|
||
# spec 明确禁止出现的数据库原生类型写法（出现在 type 里即违规）
|
||
NATIVE_TYPE_WORDS = {"varchar", "nvarchar", "char(", "bigint", "tinyint", "mediumint",
|
||
"json", "blob", "clob", "datetime(", "timestamp(", "numeric",
|
||
"boolean", "enum", "set", "unsigned"}
|
||
NULLABLE_VALUES = ("yes", "no")
|
||
|
||
# 已登记的偏离：(表名, 列名, 偏离类型) —— 只有登记过的才允许放行，且必须打印
|
||
COMPLIANCE_DEVIATIONS = {
|
||
("pbl_runtime_event", "id", "id-not-str32"): "分区表物理主键 (id, created_at) 需要 BIGINT AUTO_INCREMENT；spec 要求 id str(32)。真源 pbl_runtime_ext 已批准结构，登记为偏离。",
|
||
("pbl_runtime_event", "id", "auto-increment"): "id 由数据库自增生成，INSERT 列集须剔除该列。",
|
||
("pbl_entity_state", "id", "id-not-str32"): "真源 pbl_entity_state 物理主键 id BIGINT AUTO_INCREMENT（已批准 DDL sql/pbl_runtime_ext.sql:34）；spec 要求 id str(32)。登记为偏离。",
|
||
("pbl_entity_state", "id", "auto-increment"): "id 由数据库自增生成，INSERT 列集须剔除该列。",
|
||
}
|
||
|
||
|
||
def _is_pure_ascii_letters(text):
|
||
return all(("a" <= ch.lower() <= "z") for ch in text)
|
||
|
||
|
||
def check_type_value(type_value):
|
||
"""返回该 type 写法的问题列表（spec 只允许抽象类型）。"""
|
||
problems = []
|
||
if not isinstance(type_value, str) or not type_value:
|
||
problems.append("type 必须是非空字符串")
|
||
return problems
|
||
low = type_value.strip().lower()
|
||
if "(" in low or " " in low:
|
||
problems.append("type=%r 是数据库方言具体类型（spec 禁止，应使用抽象类型 + length/dec）" % type_value)
|
||
return problems
|
||
if low not in ABSTRACT_TYPES:
|
||
problems.append("type=%r 不在 spec 抽象类型表内" % type_value)
|
||
for word in NATIVE_TYPE_WORDS:
|
||
if low == word.rstrip("("):
|
||
problems.append("type=%r 是数据库原生类型（spec 无此抽象类型，应改用 long/text/int 等）" % type_value)
|
||
return problems
|
||
|
||
|
||
def validate_model(path, strict_id=True):
|
||
"""校验一份表定义 JSON。返回 (errors, deviations, table_name)。"""
|
||
errors = []
|
||
deviations = []
|
||
try:
|
||
with open(path, "r", encoding="utf-8") as fh:
|
||
data = json.load(fh)
|
||
except ValueError as exc:
|
||
return (["JSON 不可解析: %s" % exc], [], None)
|
||
if not isinstance(data, dict):
|
||
return (["根对象必须是 JSON object"], [], None)
|
||
|
||
extra = [k for k in data if k not in SPEC_ROOT_KEYS]
|
||
missing = [k for k in SPEC_ROOT_KEYS if k not in data]
|
||
if extra:
|
||
errors.append("根键多出 spec 未定义的段: %s（四段式只允许 summary/fields/indexes/codes）"
|
||
% ", ".join(sorted(extra)))
|
||
if missing:
|
||
errors.append("根键缺少 spec 必需段: %s" % ", ".join(missing))
|
||
if errors:
|
||
return (errors, deviations, None)
|
||
|
||
summary = data["summary"]
|
||
if not isinstance(summary, list) or len(summary) != 1 or not isinstance(summary[0], dict):
|
||
errors.append("summary 必须是恰好 1 条记录的数组")
|
||
return (errors, deviations, None)
|
||
s = summary[0]
|
||
table = s.get("name")
|
||
for k in ("name", "title"):
|
||
if not s.get(k):
|
||
errors.append("summary[0].%s 必填" % k)
|
||
unknown = [k for k in s if k not in SUMMARY_KEYS + SUMMARY_OPTIONAL]
|
||
if unknown:
|
||
errors.append("summary[0] 含 spec 未定义键: %s（应并入 summary[0].comment 文本或删除）"
|
||
% ", ".join(sorted(unknown)))
|
||
primary = s.get("primary")
|
||
if not isinstance(primary, list) or not primary:
|
||
errors.append("summary[0].primary 必须是非空数组（字符串会渲染成 'i,d'）")
|
||
primary = []
|
||
|
||
fields = data["fields"]
|
||
if not isinstance(fields, list) or not fields:
|
||
errors.append("fields 必须是非空数组")
|
||
return (errors, deviations, table)
|
||
names = []
|
||
for idx, f in enumerate(fields):
|
||
where = "fields[%d]" % idx
|
||
if not isinstance(f, dict):
|
||
errors.append("%s 必须是 object" % where)
|
||
continue
|
||
name = f.get("name")
|
||
where = "fields[%d](%s)" % (idx, name or "?")
|
||
if not name or not isinstance(name, str):
|
||
errors.append("%s.name 必填" % where)
|
||
continue
|
||
names.append(name)
|
||
if not f.get("title"):
|
||
errors.append("%s.title 必填（DDL COMMENT 用）" % where)
|
||
unknown = [k for k in f if k not in FIELD_KEYS]
|
||
if unknown:
|
||
errors.append("%s 含 spec 未定义键: %s" % (where, ", ".join(sorted(unknown))))
|
||
t = f.get("type")
|
||
errors.extend("%s: %s" % (where, m) for m in check_type_value(t))
|
||
req = ABSTRACT_TYPES.get(str(t).lower(), ())
|
||
for need in req:
|
||
v = f.get(need)
|
||
if not isinstance(v, int) or isinstance(v, bool) or v <= 0:
|
||
errors.append("%s: type=%s 必须带正整数 %s（实测 %r）" % (where, t, need, v))
|
||
nullable = f.get("nullable")
|
||
if nullable is not None and nullable not in NULLABLE_VALUES:
|
||
errors.append('%s.nullable 只能是 "yes"|"no"（实测 %r）' % (where, nullable))
|
||
if name == "id":
|
||
if str(t).lower() != "str" or not (isinstance(f.get("length"), int) and f["length"] >= 32):
|
||
key = (table, "id", "id-not-str32")
|
||
if strict_id and key not in COMPLIANCE_DEVIATIONS:
|
||
errors.append("%s: 主键 id 必须 str 且 length>=32（spec 规则），实测 type=%r length=%r"
|
||
% (where, t, f.get("length")))
|
||
else:
|
||
deviations.append("%s: id 非 str(32)（type=%r length=%r）—— 已登记偏离：%s"
|
||
% (where, t, f.get("length"), COMPLIANCE_DEVIATIONS.get(key, "")))
|
||
if f.get("auto_increment"):
|
||
key = (table, "id", "auto-increment")
|
||
if key in COMPLIANCE_DEVIATIONS:
|
||
deviations.append("%s: id auto_increment —— 已登记偏离：%s"
|
||
% (where, COMPLIANCE_DEVIATIONS[key]))
|
||
else:
|
||
errors.append("%s: auto_increment 未在偏离登记表中（spec 无此键）" % where)
|
||
|
||
dup = [n for n in set(names) if names.count(n) > 1]
|
||
if dup:
|
||
errors.append("字段名重复: %s" % ", ".join(sorted(dup)))
|
||
for p in primary:
|
||
if p not in names:
|
||
errors.append("primary 引用了不存在的列: %s" % p)
|
||
|
||
for idx, i in enumerate(data["indexes"] or []):
|
||
where = "indexes[%d](%s)" % (idx, i.get("name") if isinstance(i, dict) else "?")
|
||
if not isinstance(i, dict):
|
||
errors.append("%s 必须是 object" % where)
|
||
continue
|
||
unknown = [k for k in i if k not in INDEX_KEYS]
|
||
if unknown:
|
||
errors.append("%s 含 spec 未定义键: %s（unique/fields 写法不被 DDL 模板识别）"
|
||
% (where, ", ".join(sorted(unknown))))
|
||
if not i.get("name"):
|
||
errors.append("%s.name 必填" % where)
|
||
if i.get("idxtype") not in ("unique", "index"):
|
||
errors.append('%s.idxtype 必须是 "unique"|"index"（实测 %r）' % (where, i.get("idxtype")))
|
||
idxf = i.get("idxfields")
|
||
if not isinstance(idxf, list) or not idxf:
|
||
errors.append("%s.idxfields 必须是非空数组" % where)
|
||
continue
|
||
for c in idxf:
|
||
if c not in names:
|
||
errors.append("%s.idxfields 引用不存在的列: %s" % (where, c))
|
||
if str(i.get("name")) == "PRIMARY":
|
||
errors.append("%s: 主键不写进 indexes（spec 用 summary[0].primary 声明）" % where)
|
||
idx_names = [i.get("name") for i in data["indexes"] or [] if isinstance(i, dict)]
|
||
dup = [n for n in set(idx_names) if idx_names.count(n) > 1]
|
||
if dup:
|
||
errors.append("索引名重复: %s" % ", ".join(sorted(dup)))
|
||
|
||
for idx, c in enumerate(data["codes"] or []):
|
||
where = "codes[%d]" % idx
|
||
if not isinstance(c, dict):
|
||
errors.append("%s 必须是 object" % where)
|
||
continue
|
||
unknown = [k for k in c if k not in CODE_KEYS]
|
||
if unknown:
|
||
errors.append("%s 含 spec 未定义键: %s" % (where, ", ".join(sorted(unknown))))
|
||
for need in ("field", "table", "valuefield", "textfield"):
|
||
if not c.get(need):
|
||
errors.append("%s.%s 必填" % (where, need))
|
||
if c.get("field") and c["field"] not in names:
|
||
errors.append("%s.field 引用不存在的列: %s" % (where, c["field"]))
|
||
if c.get("table") == "appcodes_kv" and "." in str(c.get("cond", "")):
|
||
errors.append("%s: appcodes_kv 的 cond 必须用 parentid= 分组键" % where)
|
||
if "." in str(c.get("table", "")):
|
||
errors.append("%s: codes.table 禁止 module.table 点号写法" % where)
|
||
|
||
return (errors, deviations, table)
|
||
|
||
|
||
def main(argv=None):
|
||
ap = argparse.ArgumentParser(description="database-table-definition-spec 机械校验")
|
||
ap.add_argument("dirs", nargs="+", help="models 目录（校验其下所有 *.json）")
|
||
ap.add_argument("--files", nargs="*", default=None, help="只校验指定文件名（如 pbl_runtime_event.json）")
|
||
ap.add_argument("--json-report", default=None)
|
||
args = ap.parse_args(argv)
|
||
|
||
total_files = 0
|
||
total_errors = 0
|
||
total_deviations = 0
|
||
report = []
|
||
for d in args.dirs:
|
||
if not os.path.isdir(d):
|
||
print("SKIP(not a dir): %s" % d)
|
||
continue
|
||
for name in sorted(os.listdir(d)):
|
||
if not name.endswith(".json"):
|
||
continue
|
||
if args.files and name not in args.files:
|
||
continue
|
||
path = os.path.join(d, name)
|
||
errors, deviations, table = validate_model(path)
|
||
total_files += 1
|
||
total_errors += len(errors)
|
||
total_deviations += len(deviations)
|
||
report.append({"path": path, "table": table, "errors": errors,
|
||
"deviations": deviations})
|
||
print("---- %s (table=%s)" % (path, table))
|
||
print(" root_keys_ok=%s" % (not errors or "四段式" in " ".join(errors)))
|
||
for e in errors:
|
||
print(" ERROR: %s" % e)
|
||
for v in deviations:
|
||
print(" DEVIATION(registered): %s" % v)
|
||
if not errors:
|
||
print(" OK: 符合 database-table-definition-spec（偏离 %d 条已登记）" % len(deviations))
|
||
print("=" * 60)
|
||
print("files=%d errors=%d registered_deviations=%d" % (total_files, total_errors, total_deviations))
|
||
print("RESULT: %s" % ("PASS" if total_errors == 0 else "FAIL"))
|
||
if args.json_report:
|
||
with open(args.json_report, "w", encoding="utf-8") as fh:
|
||
json.dump(report, fh, ensure_ascii=False, indent=1)
|
||
return 0 if total_errors == 0 else 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|