#!/usr/bin/env python3 # -*- coding: utf-8 -*- """validate_models_json —— database-table-definition-spec 机械校验器(M11b-2b-A 修复版 rev4)。 QC #4/#7/#9/#10 要求:表定义 JSON 必须是 spec 规定的四段式(summary/fields/indexes/ codes),类型必须是 spec 声明的抽象类型,主键 id 必须是 str/length>=32,索引必须用 ``idxtype`` + ``idxfields``。本脚本把这些规则变成**可执行的断言**,任何一份 models/*.json 不合规就非零退出(真源不合规同样 FAIL —— QC #9:真源不合规则引用方不可能合规)。 本版本修复 QC 退回项(只改本脚本,不改任何 models/*.json): QC #2 抽象类型误报:纯抽象类型名(datetime/timestamp/date/time/text/long/…)一律放行, 只有 type 字符串本身含括号或空格(方言写法 varchar(64)/decimal(15,2)/datetime(3)) 或本身就是 spec 不存在的非抽象类型名(varchar/bigint/json/blob/…)才判违规。 NATIVE_TYPE_WORDS 中已删除 "datetime(" / "timestamp(" 两项。 QC #3 CLI 与文档一致:argparse 真正提供 --allow-registered-deviation(见下方「用法」)。 同一轮 QC #3 追加要求「拦截分布可一条命令复现」:本版本在结尾输出 ``RULE_STATS: <规则标签>=<条数>`` 逐规则计数 + ``RULE_STATS_TOTAL`` 总量, 复核者只需一条命令即可核对分布,不必依赖任何中间文件: python3 modules/world_sync/scripts/validate_models_json.py \ $(ls -d modules/*/models) 2>&1 | grep "^RULE_STATS" QC #4 root_keys_ok:由 validate_model 显式计算 (not missing and not extra) 并随结果返回, main() 直接打印该布尔值,不再用「错误串里是否含『四段式』」猜测。 QC #5 引用式 key/schema:spec 不承认引用式结构,直接判 ERROR 并提示改为内联四段式; COMPLIANCE_DEVIATIONS 中针对 pbl_runtime_event/pbl_entity_state 的 4 条失效登记项 一并删除(引用式在根键检查处即 FAIL,根本到不了 fields 校验,登记形同虚设)。 QC #1(第三轮)放行机制对 auto_increment 生效:字段段「spec 未定义键」检查原先无条件报错, 导致 (表名,"id","auto-increment") 即使已登记、即使传了 --allow-registered-deviation, 仍残留 `fields[N](id) 含 spec 未定义键: auto_increment` 而整体 FAIL —— id-not-str32 可豁免、auto_increment 不可豁免,机制自相矛盾。现在该检查改为**偏离感知**: id 上的 auto_increment 键统一交由 id 分支按 strict/allow 两态处理(已登记 + 传开关 → 降级为 DEVIATION(registered) 明细,不计入 errors;未登记或未传开关 → 照旧 ERROR); 其它字段上的未定义键也支持通用登记键 (表名, 列名, "unknown-key:<键名>"),同样两态。 偏离放行语义(唯一允许的偏离,需 --allow-registered-deviation 显式声明才放行): 1) (表名, "id", "id-not-str32") —— 主键 id 非 str 或 length<32; 2) (表名, "id", "auto-increment") —— id 带 spec 未定义键 auto_increment; 3) (表名, 列名, "unknown-key:") —— 任意字段带 spec 未定义键 (通用形式)。 未传该开关时上述项仍判 ERROR,并打印「需 --allow-registered-deviation 放行」提示;传入时 逐条打印 DEVIATION(registered) 明细(含登记说明),不静默放过。 注意:只有**登记过**的才可能放行;登记表为空时 --allow-registered-deviation 不放行任何项。 典型场景:分区表物理主键需要 long + auto_increment(MySQL 1503:分区键必须进主键)。 若某表确需此类写法,须先在其模块任务里把 (表名,"id","id-not-str32") 与 (表名,"id","auto-increment") 登记进 COMPLIANCE_DEVIATIONS,再配合开关使用 —— 两者缺一不可, 只登记不传开关仍 FAIL,只传开关不登记也 FAIL。 与并行子任务 B 的接口:B 把 pbl_runtime_event/pbl_entity_state 内联为四段式后,若保留 id=long(+auto_increment) 写法,须由 B 在其任务内按上述键格式登记(本任务按 QC #5 选 (a), 登记表保持为空 {},不放行任何历史失效项)。 用法: python validate_models_json.py [<目录2> ...] python validate_models_json.py --allow-registered-deviation python validate_models_json.py --files a.json b.json 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": (), } # QC #2:spec 明确不存在的非抽象类型名(type 本身就是这些词即违规)。 # 注意:这里**不得**出现 datetime/timestamp/date/time/text/long 等 spec 合法抽象类型, # 也不得出现带括号的写法(带括号的情况由「type 含括号/空格 = 方言写法」分支统一拦截)。 NATIVE_TYPE_WORDS = {"varchar", "nvarchar", "bigint", "tinyint", "mediumint", "smallint", "integer", "int2", "int4", "int8", "json", "jsonb", "blob", "clob", "bytea", "varbinary", "numeric", "boolean", "bool", "enum", "set", "unsigned", "serial", "bigserial", "uuid", "money"} NULLABLE_VALUES = ("yes", "no") # QC #5:引用式写法(key=schema + schema 段)不被 spec 承认 —— 直接 FAIL,请改为内联四段式。 # 因此这里不再为引用式文件登记任何偏离项(到不了 fields 校验,登记永不命中)。 # # 已登记的偏离:(表名, 列名, 偏离类型) -> 说明 —— 只有登记过的才允许放行,且必须打印 COMPLIANCE_DEVIATIONS = {} DEVIATION_HINT = "需 --allow-registered-deviation 放行" # QC #3(第三轮):逐规则计数标签 —— 让「真·非抽象类型仍被拦截」的分布不依赖中间文件, # 一条命令即可复现(结尾 RULE_STATS 行)。标签按下列顺序尝试匹配(首个命中即归类)。 RULE_STATS_TAGS = ( ("unreadable_json", ("JSON 不可解析", "文件不可读", "根对象必须是 JSON object")), ("reference_style_key_schema", ("引用式写法",)), ("root_key_extra", ("根键多出 spec 未定义的段",)), ("root_key_missing", ("根键缺少 spec 必需段",)), ("type_db_native", ("是数据库原生类型",)), ("type_db_dialect", ("是数据库方言具体类型",)), ("type_not_in_abstract_table", ("不在 spec 抽象类型表内",)), ("type_not_string", ("type 必须是非空字符串",)), ("pk_id_not_str32", ("主键 id 必须 str 且 length>=32",)), ("id_auto_increment", ("auto_increment",)), ("field_length_dec_missing", ("必须带正整数",)), ("field_nullable_bad", ("nullable 只能是",)), ("field_name_missing", (".name 必填",)), ("field_title_missing", (".title 必填",)), ("field_unknown_key", ("fields[", "含 spec 未定义键")), ("field_dup_name", ("字段名重复",)), ("summary_shape", ("summary 必须是恰好 1 条记录的数组", "summary[0].", "summary[0] 含 spec 未定义键")), ("primary_not_array", ("primary 必须是非空数组",)), ("primary_unknown_col", ("primary 引用了不存在的列",)), ("index_unknown_key", ("indexes[", "含 spec 未定义键")), ("index_type_bad", ("idxtype 必须是",)), ("index_fields_bad", ("idxfields 必须是非空数组",)), ("index_ref_unknown_col", ("idxfields 引用不存在的列",)), ("index_name_primary", ("主键不写进 indexes",)), ("index_dup_name", ("索引名重复",)), ("code_unknown_key", ("codes[", "含 spec 未定义键")), ("code_required_missing", ("codes[", ".必填")), ("code_ref_unknown_col", ("codes[", "引用不存在的列")), ("code_cond_parentid", ("cond 必须用 parentid=",)), ("code_table_dot", ("codes.table 禁止",)), ) def _rule_tag(message): """把一条 ERROR 文本归类到规则标签(QC #3:分布可由一条命令复现)。""" for tag, needles in RULE_STATS_TAGS: if all(n in message for n in needles): return tag return "other" def _is_reference_style(data): """识别引用式结构:{"key": "schema", "schema": {...}} 之类(spec 不承认)。""" if not isinstance(data, dict): return False if str(data.get("key", "")).lower() == "schema" and "schema" in data: return True # 变体:只有 schema 段而无四段式,同样属于引用式(把权威定义指向别处) if "schema" in data and not any(k in data for k in SPEC_ROOT_KEYS): return True return False def _unknown_key_deviation_key(table, field_name, key_name): """字段段 spec 未定义键对应的登记键(QC #1)。 id 上的 auto_increment 复用 (表名,"id","auto-increment"),与 id 分支同一登记项, 避免「登记了 auto-increment 却因 unknown-key 分支无条件报错」的放行失效缺陷; 其余未定义键使用通用形式 (表名, 列名, "unknown-key:<键名>")。 """ if table is None: return None if field_name == "id" and key_name == "auto_increment": return (table, "id", "auto-increment") return (table, field_name, "unknown-key:%s" % key_name) def check_type_value(type_value): """返回该 type 写法的问题列表(spec 只允许抽象类型)。 判定顺序(QC #2): 1) 非字符串/空串 → 违规; 2) 含括号或空格(varchar(64)/decimal(15,2)/datetime(3)/timestamp with time zone) → 方言具体类型,违规; 3) 纯抽象类型名且在 ABSTRACT_TYPES 内(datetime/timestamp/text/long/date/time…) → **一律放行**,不再二次怀疑为原生类型; 4) 名字命中 NATIVE_TYPE_WORDS(varchar/bigint/json/blob/…)→ 违规; 5) 其余不在抽象类型表内的名字 → 违规。 """ problems = [] if not isinstance(type_value, str) or not type_value.strip(): problems.append("type 必须是非空字符串") return problems low = type_value.strip().lower() if "(" in low or ")" in low or " " in low: problems.append("type=%r 是数据库方言具体类型(spec 禁止,应使用抽象类型 + length/dec)" % type_value) return problems if low in ABSTRACT_TYPES: return problems # 合法抽象类型,直接放行 if low in NATIVE_TYPE_WORDS: problems.append("type=%r 是数据库原生类型(spec 无此抽象类型,应改用 long/text/int/str 等)" % type_value) return problems problems.append("type=%r 不在 spec 抽象类型表内" % type_value) return problems def validate_model(path, strict_id=True): """校验一份表定义 JSON。 返回 4-tuple: (errors, deviations, table_name, root_keys_ok) root_keys_ok = (not missing and not extra),在函数内显式计算(QC #4)。 """ 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, False) except OSError as exc: return (["文件不可读: %s" % exc], [], None, False) if not isinstance(data, dict): return (["根对象必须是 JSON object"], [], None, False) # QC #5:引用式 key/schema 写法 —— spec 不承认,直接 FAIL(不做跨模块解引用) if _is_reference_style(data): errors.append("引用式写法(key=schema / schema 指向他处权威定义)不被 database-table-definition-spec " "承认,须在本文件内联四段式 summary/fields/indexes/codes") return (errors, deviations, None, False) 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] root_keys_ok = (not missing and not extra) # QC #4:显式计算,不靠字符串猜测 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, root_keys_ok) 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, root_keys_ok) 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, root_keys_ok) 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) # QC #1:spec 未定义键检查改为偏离感知 —— 已登记且传入 --allow-registered-deviation # 的键不再计入 errors(id.auto_increment 交由下方 id 分支统一打印 DEVIATION 明细, # 避免同一偏离重复计数;其它键在此直接降级为 DEVIATION(registered))。 unknown = [] for k in sorted([k for k in f if k not in FIELD_KEYS]): dkey = _unknown_key_deviation_key(table, name, k) if k == "auto_increment" and name == "id" and dkey in COMPLIANCE_DEVIATIONS: continue # strict/allow 两态由 id 分支处理(同一登记项,只报一次) if dkey is not None and dkey in COMPLIANCE_DEVIATIONS: if strict_id: # 已登记但未传开关:照旧 ERROR,并明确提示如何放行(与 id 分支同一语义) errors.append("%s 含 spec 未定义键: %s —— 该项已登记偏离,%s" % (where, k, DEVIATION_HINT)) continue deviations.append("%s: 含 spec 未定义键 %s —— 已登记偏离:%s" % (where, k, COMPLIANCE_DEVIATIONS[dkey])) continue unknown.append(k) if unknown: errors.append("%s 含 spec 未定义键: %s" % (where, ", ".join(unknown))) t = f.get("type") errors.extend("%s: %s" % (where, m) for m in check_type_value(t)) req = ABSTRACT_TYPES.get(str(t).strip().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).strip().lower() != "str" or not (isinstance(f.get("length"), int) and f["length"] >= 32): key = (table, "id", "id-not-str32") if key in COMPLIANCE_DEVIATIONS: if strict_id: errors.append("%s: 主键 id 必须 str 且 length>=32(spec 规则),实测 type=%r length=%r" " —— 该项已登记偏离,%s" % (where, t, f.get("length"), DEVIATION_HINT)) else: deviations.append("%s: id 非 str(32)(type=%r length=%r)—— 已登记偏离:%s" % (where, t, f.get("length"), COMPLIANCE_DEVIATIONS[key])) else: errors.append("%s: 主键 id 必须 str 且 length>=32(spec 规则),实测 type=%r length=%r" % (where, t, f.get("length"))) if "auto_increment" in f: key = (table, "id", "auto-increment") if key in COMPLIANCE_DEVIATIONS: if strict_id: errors.append("%s: id 带 spec 未定义键 auto_increment(实测 %r)" " —— 该项已登记偏离,%s" % (where, f.get("auto_increment"), DEVIATION_HINT)) else: deviations.append("%s: id 带 spec 未定义键 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, root_keys_ok) def collect_targets(dirs, only_files=None): """展开目录列表为待校验文件路径(目录不存在则 SKIP 提示)。""" targets = [] for d in 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 only_files and name not in only_files: continue targets.append(os.path.join(d, name)) return targets def main(argv=None): ap = argparse.ArgumentParser(description="database-table-definition-spec 机械校验") # nargs="*":`--files a.json /tmp` 这种写法下 argparse 会把目录吞进 --files, # 位置参数可能为空,故这里不强制 nargs="+",改由下方规范化后统一校验非空。 ap.add_argument("dirs", nargs="*", help="models 目录(校验其下所有 *.json)") ap.add_argument("--files", nargs="*", default=None, help="只校验指定文件名(如 --files pbl_runtime_event.json models);" "非 .json 结尾的参数会被当作额外的 models 目录") ap.add_argument("--json-report", default=None) # QC #3:docstring 声明的开关必须真实存在于 argparse,否则按文档用法执行会 unrecognized arguments ap.add_argument("--allow-registered-deviation", dest="allow_registered_deviation", action="store_true", default=False, help="放行 COMPLIANCE_DEVIATIONS 中已登记的偏离项(未传时这类项仍判 ERROR 并提示需本开关)") args = ap.parse_args(argv) # --files 用 nargs="*",argparse 会把后面的目录一并吞进来(如 `--files a.json /tmp`)。 # 规范化:以 .json 结尾的作为文件名过滤,其余作为额外目录,保证文档用法可直接执行。 extra_dirs = [] name_filters = [] for tok in (args.files or []): if str(tok).endswith(".json"): name_filters.append(str(tok)) else: extra_dirs.append(str(tok)) dirs = list(args.dirs or []) + extra_dirs if not dirs: ap.error("至少提供一个 models 目录(或 --files <目录>)") strict_id = not args.allow_registered_deviation if args.allow_registered_deviation: print("MODE: --allow-registered-deviation(已登记偏离项放行,登记项 %d 条)" % len(COMPLIANCE_DEVIATIONS)) total_files = 0 total_errors = 0 total_deviations = 0 rule_counts = {} report = [] for path in collect_targets(dirs, name_filters or None): errors, deviations, table, root_keys_ok = validate_model(path, strict_id=strict_id) total_files += 1 total_errors += len(errors) total_deviations += len(deviations) per_file_rules = {} for e in errors: tag = _rule_tag(e) rule_counts[tag] = rule_counts.get(tag, 0) + 1 per_file_rules[tag] = per_file_rules.get(tag, 0) + 1 report.append({"path": path, "table": table, "root_keys_ok": root_keys_ok, "rule_stats": per_file_rules, "errors": errors, "deviations": deviations}) print("---- %s (table=%s)" % (path, table)) print(" root_keys_ok=%s" % root_keys_ok) # QC #4:打印显式计算的布尔值 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) # QC #3:逐规则计数(不依赖中间文件即可复核「真·非抽象类型仍被拦截、抽象类型未放水」) for tag, cnt in sorted(rule_counts.items(), key=lambda kv: (-kv[1], kv[0])): print("RULE_STATS: %s=%d" % (tag, cnt)) print("RULE_STATS_TOTAL: errors=%d files=%d" % (total_errors, total_files)) 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())