diff --git a/scripts/validate_models_json.py b/scripts/validate_models_json.py index 15f1c5c..423ea66 100644 --- a/scripts/validate_models_json.py +++ b/scripts/validate_models_json.py @@ -1,19 +1,34 @@ #!/usr/bin/env python3 # -*- coding: utf-8 -*- -"""validate_models_json —— database-table-definition-spec 机械校验器（M11b-2b）。 +"""validate_models_json —— database-table-definition-spec 机械校验器（M11b-2b-A 修复版）。 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 中，逐条打印，不静默放过。 +本版本修复 QC 退回项（只改本脚本，不改任何 models/*.json）： + QC #2 抽象类型误报：纯抽象类型名（datetime/timestamp/date/time/text/long/…）一律放行， + 只有 type 字符串本身含括号或空格（方言写法 varchar(64)/decimal(15,2)/datetime(3)） + 或本身就是 spec 不存在的非抽象类型名（varchar/bigint/json/blob/…）才判违规。 + QC #3 CLI 与文档一致：argparse 真正提供 --allow-registered-deviation（见下方「用法」）。 + 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 校验，登记形同虚设）。 + +已登记偏离（唯一允许的偏离，需 --allow-registered-deviation 显式声明才放行）： + 主键 id 非 str(32) 或 id 带 auto_increment，且已登记在 COMPLIANCE_DEVIATIONS 中 + （典型场景：分区表物理主键需要 long + auto_increment）。未传该开关时这类项仍判 ERROR， + 并打印「需 --allow-registered-deviation 放行」提示；传入时逐条打印 DEVIATION(registered) + 明细，不静默放过。当前登记表为空（引用式失效项已删除），后续若真源确需偏离再登记。 用法： python validate_models_json.py [<目录2> ...] - python validate_models_json.py --json-report out.json <目录> + 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 @@ -35,69 +50,105 @@ ABSTRACT_TYPES = { "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"} +# 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") -# 已登记的偏离：(表名, 列名, 偏离类型) —— 只有登记过的才允许放行，且必须打印 -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 列集须剔除该列。", -} +# QC #5：引用式写法（key=schema + schema 段）不被 spec 承认 —— 直接 FAIL，请改为内联四段式。 +# 因此这里不再为引用式文件登记任何偏离项（到不了 fields 校验，登记永不命中）。 +# +# 已登记的偏离：(表名, 列名, 偏离类型) -> 说明 —— 只有登记过的才允许放行，且必须打印 +COMPLIANCE_DEVIATIONS = {} + +DEVIATION_HINT = "需 --allow-registered-deviation 放行" -def _is_pure_ascii_letters(text): - return all(("a" <= ch.lower() <= "z") for ch in text) +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 check_type_value(type_value): - """返回该 type 写法的问题列表（spec 只允许抽象类型）。""" + """返回该 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: + 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: + if "(" in low or ")" 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) + 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。返回 (errors, deviations, table_name)。""" + """校验一份表定义 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) + return (["JSON 不可解析: %s" % exc], [], None, False) + except OSError as exc: + return (["文件不可读: %s" % exc], [], None, False) if not isinstance(data, dict): - return (["根对象必须是 JSON object"], [], None) + 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) + 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) + return (errors, deviations, None, root_keys_ok) s = summary[0] table = s.get("name") for k in ("name", "title"): @@ -115,7 +166,7 @@ def validate_model(path, strict_id=True): fields = data["fields"] if not isinstance(fields, list) or not fields: errors.append("fields 必须是非空数组") - return (errors, deviations, table) + return (errors, deviations, table, root_keys_ok) names = [] for idx, f in enumerate(fields): where = "fields[%d]" % idx @@ -135,7 +186,7 @@ def validate_model(path, strict_id=True): 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(), ()) + 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: @@ -144,19 +195,28 @@ def validate_model(path, strict_id=True): 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): + if str(t).strip().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: + 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"))) - 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])) + if strict_id: + errors.append("%s: id auto_increment —— 该项已登记偏离，%s" + % (where, DEVIATION_HINT)) + else: + deviations.append("%s: id auto_increment —— 已登记偏离：%s" + % (where, COMPLIANCE_DEVIATIONS[key])) else: errors.append("%s: auto_increment 未在偏离登记表中（spec 无此键）" % where) @@ -212,44 +272,77 @@ def validate_model(path, strict_id=True): if "." in str(c.get("table", "")): errors.append("%s: codes.table 禁止 module.table 点号写法" % where) - return (errors, deviations, table) + return (errors, deviations, table, root_keys_ok) -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: +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 args.files and name not in args.files: + if only_files and name not in only_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)) + 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 + 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) + report.append({"path": path, "table": table, "root_keys_ok": root_keys_ok, + "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) print("files=%d errors=%d registered_deviations=%d" % (total_files, total_errors, total_deviations)) print("RESULT: %s" % ("PASS" if total_errors == 0 else "FAIL"))