349 lines
15 KiB
Python
349 lines
15 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""pbl_runtime_authority —— M11b-2b：表定义 JSON 的列契约解析层（列名唯一入口）。
|
||
|
||
**结构现状（M11b-2b-B 起，QC #1/#2/#3 整改）**
|
||
PM 裁决 B：``pbl_runtime_event`` / ``pbl_entity_state`` 的**结构权威源**是
|
||
``modules/pbl_runtime_ext/models/*.json``（M11b-1 已批准产物，禁止反向修改）。
|
||
但 QC 明确要求 world_sync 侧 ``models/*.json`` 必须是 **database-table-definition-spec
|
||
四段式（summary/fields/indexes/codes）内联副本**，而不是 ``key=schema`` 引用式
|
||
——校验器 ``scripts/validate_models_json.py`` 对引用式写法直接判 FAIL。
|
||
因此本模块现在解析的是「按真源逐字段内联、并按 spec 归一（抽象类型、去私有键）的
|
||
world_sync 侧副本」，它同时就是 world_sync 消费方的列契约（列名与真源一致）。
|
||
|
||
本模块是列契约的**机械导出层**，不是人工抄写层：
|
||
|
||
* :func:`load_model` —— 读本模块 ``models/<table>.json``（四段式）并校验结构；
|
||
* :func:`columns` / :func:`insert_columns` / :func:`primary` / :func:`indexes`
|
||
—— 暴露列契约（``insert_columns`` 自动剔除自增列，兼容真源私有 ``auto_increment`` 键）；
|
||
* :func:`authority_path` —— 反查结构权威源（pbl_runtime_ext）在磁盘上的真实路径，
|
||
供审计与 QC 取证（写入结果的 ``authority`` 字段）；
|
||
* :func:`ddl` —— 由列契约渲染本地建表语句（仅供测试/本地起库；生产 DDL、分区与
|
||
append-only 触发器归 pbl_runtime_ext 的 ``scripts/pbl_runtime_event_ddl.py``
|
||
与 ``scripts/migrations/*.sql`` 所有，本模块不复制那套逻辑）；
|
||
* :func:`column_diff` —— 消费方用到的列 vs 列契约的双向 diff（fail-closed 的判据）。
|
||
|
||
**fail-closed**：文件缺失、JSON 不可解析、不是四段式、fields 为空 —— 一律抛
|
||
:class:`AuthorityResolutionError`，绝不退化成「猜列名」或回落到旧的手写常量。
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
|
||
__all__ = [
|
||
"AuthorityResolutionError",
|
||
"SPEC_ROOT_KEYS",
|
||
"SPEC_SUMMARY_KEYS",
|
||
"SPEC_FIELD_KEYS",
|
||
"SPEC_INDEX_KEYS",
|
||
"SPEC_CODE_KEYS",
|
||
"AUTHORITY_MODULE",
|
||
"load_model",
|
||
"authority_path",
|
||
"authority_model",
|
||
"columns",
|
||
"insert_columns",
|
||
"auto_increment_columns",
|
||
"field_map",
|
||
"primary",
|
||
"indexes",
|
||
"table_title",
|
||
"ddl",
|
||
"column_diff",
|
||
]
|
||
|
||
# world_sync 侧 models/<table>.json 必须是 spec 四段式（database-table-definition-spec）
|
||
SPEC_ROOT_KEYS = ("summary", "fields", "indexes", "codes")
|
||
# 各段允许的键（白名单，自测脚本按此做机械断言；comment 为 spec 允许的可选说明键）
|
||
SPEC_SUMMARY_KEYS = ("name", "title", "primary", "catelog", "comment")
|
||
SPEC_FIELD_KEYS = ("name", "title", "type", "length", "dec", "nullable", "default")
|
||
SPEC_INDEX_KEYS = ("name", "idxtype", "idxfields")
|
||
SPEC_CODE_KEYS = ("field", "table", "valuefield", "textfield", "cond")
|
||
|
||
# 结构权威源所在模块（PM 裁决 B；本模块只反查路径，不消费其私有键）
|
||
AUTHORITY_MODULE = "pbl_runtime_ext"
|
||
|
||
MODULE_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
MODELS_DIR = os.path.join(MODULE_ROOT, "models")
|
||
# modules/world_sync -> modules -> 机构工作空间根
|
||
WORKSPACE_ROOT = os.path.dirname(os.path.dirname(MODULE_ROOT))
|
||
|
||
_SQLITE_TYPES = {
|
||
"str": "TEXT", "char": "TEXT", "text": "TEXT", "json": "TEXT",
|
||
"bin": "BLOB", "short": "INTEGER", "int": "INTEGER", "long": "INTEGER",
|
||
"bigint": "INTEGER", "float": "REAL", "double": "REAL", "decimal": "REAL",
|
||
"ddouble": "REAL", "date": "TEXT", "time": "TEXT", "datetime": "TEXT",
|
||
"timestamp": "TEXT",
|
||
}
|
||
_MYSQL_TYPES = {
|
||
"str": "VARCHAR({length})", "char": "CHAR({length})", "text": "LONGTEXT",
|
||
"json": "JSON", "bin": "LONGBLOB", "short": "SMALLINT", "int": "INT",
|
||
"long": "BIGINT", "bigint": "BIGINT", "float": "FLOAT",
|
||
"double": "DECIMAL({length},{dec})", "decimal": "DECIMAL({length},{dec})",
|
||
"ddouble": "DECIMAL({length},{dec})", "date": "DATE", "time": "TIME",
|
||
"datetime": "DATETIME(3)", "timestamp": "TIMESTAMP",
|
||
}
|
||
|
||
|
||
class AuthorityResolutionError(Exception):
|
||
"""模型缺失/不可解析/非四段式（fail-closed，调用方不得吞掉）。"""
|
||
|
||
def __init__(self, message, **details):
|
||
Exception.__init__(self, message)
|
||
self.message = message
|
||
self.details = details
|
||
|
||
def __str__(self):
|
||
if not self.details:
|
||
return self.message
|
||
parts = ", ".join("%s=%r" % (k, v) for k, v in sorted(self.details.items()))
|
||
return "%s (%s)" % (self.message, parts)
|
||
|
||
|
||
def _models_dir():
|
||
return MODELS_DIR
|
||
|
||
|
||
def _read_json(path):
|
||
if not os.path.isfile(path):
|
||
raise AuthorityResolutionError("model file missing", path=path)
|
||
try:
|
||
with open(path, "r", encoding="utf-8") as fh:
|
||
data = json.load(fh)
|
||
except ValueError as exc:
|
||
raise AuthorityResolutionError("model not valid JSON: %s" % exc, path=path)
|
||
if not isinstance(data, dict):
|
||
raise AuthorityResolutionError("model must be a JSON object", path=path)
|
||
return data
|
||
|
||
|
||
_MODEL_CACHE = {}
|
||
|
||
|
||
def load_model(table):
|
||
"""读取并校验本模块 ``models/<table>.json``（spec 四段式内联副本）。
|
||
|
||
校验项（fail-closed）：根键含 summary/fields/indexes/codes；summary 恰好 1 条；
|
||
fields 非空且每列有 name。列集与真源的一致性由 scripts/m11b2b_column_diff.py
|
||
与 scripts/m11b2b_selftest.py 机械核对，不在运行期重复解析真源私有键。
|
||
"""
|
||
if not table or not isinstance(table, str):
|
||
raise AuthorityResolutionError("table name is required", table=table)
|
||
if table in _MODEL_CACHE:
|
||
return _MODEL_CACHE[table]
|
||
path = os.path.join(_models_dir(), table + ".json")
|
||
model = _read_json(path)
|
||
missing = [k for k in SPEC_ROOT_KEYS if k not in model]
|
||
if missing:
|
||
raise AuthorityResolutionError(
|
||
"model is not database-table-definition-spec four-section",
|
||
path=path, missing=missing,
|
||
hint="world_sync 侧必须是内联四段式副本（PM 裁决 B + QC #1/#5），不得用 key=schema 引用式")
|
||
summary = model.get("summary")
|
||
if not isinstance(summary, list) or len(summary) != 1 or not isinstance(summary[0], dict):
|
||
raise AuthorityResolutionError("summary must be an array with exactly 1 record",
|
||
path=path, got=type(summary).__name__)
|
||
if not str(summary[0].get("name") or "").strip():
|
||
raise AuthorityResolutionError("summary[0].name is required", path=path)
|
||
fields = model.get("fields")
|
||
if not isinstance(fields, list) or not fields:
|
||
raise AuthorityResolutionError("model has no fields", path=path)
|
||
for f in fields:
|
||
if not isinstance(f, dict) or not str(f.get("name") or "").strip():
|
||
raise AuthorityResolutionError("field without name", path=path, field=f)
|
||
_MODEL_CACHE[table] = model
|
||
return model
|
||
|
||
|
||
# 兼容旧调用名：本模块过去暴露 authority_model（解析引用指向的真源），现在列契约
|
||
# 就是内联副本本身，语义等价。
|
||
def authority_model(table):
|
||
"""列契约模型（= world_sync 侧内联四段式副本，逐字段抄自 pbl_runtime_ext 真源）。"""
|
||
return load_model(table)
|
||
|
||
|
||
def authority_path(table):
|
||
"""反查结构权威源（pbl_runtime_ext）在磁盘上的真实路径，供审计取证。
|
||
|
||
只用于写入结果里的 ``authority`` 字段与 QC 核对，**不参与列集解析**（列集来自
|
||
本模块内联副本）。找不到真源不抛错，返回声明式相对路径（真源缺失属部署期问题，
|
||
不应让运行时写入失败）。
|
||
"""
|
||
declared = os.path.join("modules", AUTHORITY_MODULE, "models", table + ".json")
|
||
candidates = [
|
||
os.path.normpath(os.path.join(WORKSPACE_ROOT, declared)),
|
||
os.path.normpath(os.path.join(MODULE_ROOT, os.pardir, AUTHORITY_MODULE,
|
||
"models", table + ".json")),
|
||
os.path.join(WORKSPACE_ROOT, "apps", "scense", "pkgs", AUTHORITY_MODULE,
|
||
"models", table + ".json"),
|
||
]
|
||
tried = []
|
||
for cand in candidates:
|
||
if cand in tried:
|
||
continue
|
||
tried.append(cand)
|
||
if os.path.isfile(cand):
|
||
return cand
|
||
return declared
|
||
|
||
|
||
def field_map(table):
|
||
"""``{列名: 列定义 dict}``（保持模型顺序）。"""
|
||
out = {}
|
||
for f in load_model(table)["fields"]:
|
||
out[f["name"]] = f
|
||
return out
|
||
|
||
|
||
def columns(table):
|
||
"""权威全列集（有序元组）—— 消费方 SQL 里允许出现的列名以此为唯一白名单。"""
|
||
return tuple(field_map(table).keys())
|
||
|
||
|
||
def auto_increment_columns(table):
|
||
"""由数据库自增生成的列（INSERT 时不得出现）。
|
||
|
||
内联副本按 spec 剥离了 ``auto_increment`` 私有键，故通常为空集；保留该逻辑是为了
|
||
当列契约指向真源（其 id 带 auto_increment）时仍能正确剔除该列。
|
||
"""
|
||
return tuple(n for n, f in field_map(table).items()
|
||
if f.get("auto_increment") or str(f.get("default", "")).upper()
|
||
in ("AUTO_INCREMENT", "IDENTITY"))
|
||
|
||
|
||
def insert_columns(table):
|
||
"""可显式写入的列 = 全列 - 自增列（消费方 INSERT 的列契约）。"""
|
||
gen = set(auto_increment_columns(table))
|
||
return tuple(n for n in columns(table) if n not in gen)
|
||
|
||
|
||
def primary(table):
|
||
"""summary[0].primary（非空数组）。主键不写进 indexes（spec 约定）。"""
|
||
summary = load_model(table).get("summary") or [{}]
|
||
pk = summary[0].get("primary") or summary[0].get("key")
|
||
if isinstance(pk, str):
|
||
pk = [p.strip() for p in pk.split(",") if p.strip()]
|
||
if not isinstance(pk, list) or not pk:
|
||
raise AuthorityResolutionError("model summary[0].primary must be a non-empty array",
|
||
table=table)
|
||
cols = set(columns(table))
|
||
unknown = [p for p in pk if p not in cols]
|
||
if unknown:
|
||
raise AuthorityResolutionError("primary references unknown column",
|
||
table=table, unknown=unknown)
|
||
return tuple(pk)
|
||
|
||
|
||
def indexes(table):
|
||
"""二级索引/唯一键列表：``[{name, idxtype, idxfields}]``（兼容真源 unique/fields 写法）。"""
|
||
out = []
|
||
for idx in (load_model(table).get("indexes") or []):
|
||
if not isinstance(idx, dict):
|
||
raise AuthorityResolutionError("index must be an object", table=table, index=idx)
|
||
fields = idx.get("idxfields") or idx.get("fields") or []
|
||
if isinstance(fields, str):
|
||
fields = [f.strip() for f in fields.split(",") if f.strip()]
|
||
idxtype = idx.get("idxtype") or ("unique" if idx.get("unique") else "index")
|
||
out.append({"name": idx.get("name"), "idxtype": idxtype, "idxfields": list(fields)})
|
||
return out
|
||
|
||
|
||
def table_title(table):
|
||
summary = load_model(table).get("summary") or [{}]
|
||
return summary[0].get("title") or summary[0].get("name") or table
|
||
|
||
|
||
def _render_type(f, dialect):
|
||
t = str(f.get("type") or "").lower()
|
||
table = _MYSQL_TYPES if dialect == "mysql" else _SQLITE_TYPES
|
||
tpl = table.get(t)
|
||
if tpl is None:
|
||
raise AuthorityResolutionError("unknown abstract type in authority model",
|
||
field=f.get("name"), type=t)
|
||
if "{" in tpl:
|
||
length = f.get("length") or (255 if t == "str" else 10)
|
||
dec = f.get("dec") or 2
|
||
return tpl.format(length=length, dec=dec)
|
||
return tpl
|
||
|
||
|
||
def _nullable(f):
|
||
if "nullable" in f:
|
||
return str(f.get("nullable")).lower() != "no"
|
||
if "null" in f:
|
||
v = f.get("null")
|
||
return v is True or str(v).lower() == "true"
|
||
return True
|
||
|
||
|
||
def ddl(table, dialect="sqlite"):
|
||
"""由列契约渲染 CREATE TABLE（测试/本地起库用）。
|
||
|
||
生产库的 DDL（分区、append-only 触发器、存储过程）归 pbl_runtime_ext 所有，
|
||
本函数**只保证列集与主键/唯一键一致**，不复制分区与触发器逻辑。
|
||
"""
|
||
if dialect not in ("sqlite", "mysql"):
|
||
raise AuthorityResolutionError("unsupported ddl dialect", dialect=dialect)
|
||
model = load_model(table)
|
||
gen = set(auto_increment_columns(table))
|
||
lines = []
|
||
for f in model["fields"]:
|
||
fname = f.get("name")
|
||
col = ("`%s`" % fname) if dialect == "mysql" else fname
|
||
piece = [" %s %s" % (col, _render_type(f, dialect))]
|
||
# QC #5 修正：自增豁免必须**按列**判断（旧写法 `not gen` 是表级判断，
|
||
# 只要表里存在任一自增列，本表所有列的 NOT NULL 都会被整体跳过，
|
||
# 导致渲染出的本地/测试 DDL 丢失 NOT NULL 约束）。
|
||
is_gen = fname in gen
|
||
if not _nullable(f) and not is_gen:
|
||
piece.append(" NOT NULL")
|
||
default = f.get("default")
|
||
if default is not None and not is_gen and str(default) != "":
|
||
piece.append(" DEFAULT %s" % ("CURRENT_TIMESTAMP"
|
||
if str(default).upper() == "CURRENT_TIMESTAMP"
|
||
else "'%s'" % default))
|
||
if dialect == "mysql" and f.get("title"):
|
||
piece.append(" COMMENT '%s'" % str(f["title"]).replace("'", "''"))
|
||
lines.append("".join(piece))
|
||
pk = primary(table)
|
||
if pk:
|
||
pk_sql = ", ".join(("`%s`" % c) if dialect == "mysql" else c for c in pk)
|
||
lines.append(" PRIMARY KEY (%s)" % pk_sql)
|
||
for idx in indexes(table):
|
||
if not idx["idxfields"] or idx["name"] in (None, "PRIMARY"):
|
||
continue
|
||
cols = ", ".join(("`%s`" % c) if dialect == "mysql" else c
|
||
for c in idx["idxfields"])
|
||
if idx["idxtype"] == "unique":
|
||
lines.append(" UNIQUE KEY `%s` (%s)" % (idx["name"], cols)
|
||
if dialect == "mysql"
|
||
else " UNIQUE (%s)" % cols)
|
||
else:
|
||
if dialect != "mysql":
|
||
# M11b-2c（QC #8）：表内 `KEY name (cols)` 是 MySQL 专有的内联二级索引
|
||
# 语法，sqlite 解析到 KEY 直接 syntax error。本地/测试库建表只要求
|
||
# 「列集 + 主键 + 唯一键」与真源一致（见本函数 docstring），二级索引
|
||
# 对 sqlite 用例无断言价值，故非 MySQL 方言一律跳过渲染。
|
||
continue
|
||
lines.append(" KEY `%s` (%s)" % (idx["name"], cols))
|
||
body = ",\n".join(lines)
|
||
if dialect == "mysql":
|
||
return ("CREATE TABLE IF NOT EXISTS `%s` (\n%s\n) "
|
||
"ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='%s'"
|
||
% (table, body, table_title(table)))
|
||
return "CREATE TABLE IF NOT EXISTS %s (\n%s\n)" % (table, body)
|
||
|
||
|
||
def column_diff(table, used_columns):
|
||
"""双向列集 diff：返回 (未知列, 未覆盖的非自增列)。
|
||
|
||
``used_columns`` 是消费方实际写入/读取用到的列名集合；空集合表示「不检查覆盖」。
|
||
"""
|
||
used = set(used_columns or ())
|
||
all_cols = set(columns(table))
|
||
unknown = sorted(used - all_cols)
|
||
uncovered = []
|
||
if used:
|
||
covered = used | set(auto_increment_columns(table))
|
||
uncovered = sorted(all_cols - covered)
|
||
return unknown, uncovered
|