340 lines
16 KiB
Python
340 lines
16 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
gen_base_ddl.py —— 基座模块（appbase / rbac）表结构 DDL 生成器
|
||
|
||
【为什么需要它 / Bug 4SxqQAulkH9eAzEys_h-e 根因】
|
||
POST /main/rbac/user/up_login.dspy 返回 500 "Server got itself in trouble"。
|
||
该端点（pkgs/rbac/wwwroot/user/up_login.dspy）第一条 SQL 是：
|
||
|
||
dbname = get_module_dbname('rbac') # conf/config.json → 'pbl'
|
||
async with db.sqlorContext(dbname) as sor:
|
||
r = await sor.sqlExe('select * from users where username=${username}$', ns)
|
||
|
||
而 apps/pbls/build.sh 的建表段（第 10 / 10a 节）只覆盖两类表：
|
||
· scripts/ddl/pbls_tables.sql —— 32 张 pbl_* 业务表
|
||
· modules/pbl_*/models/*.json —— 6 张治理/模板域表（10a 打印 "6/6"）
|
||
基座模块 appbase（appcodes/appcodes_kv/params/svgicon）与 rbac
|
||
（users/role/permission/rolepermission/userrole/organization/orgtypes/audit_log）
|
||
的 models 链**从未被建表**。部署机实测 `SHOW TABLES`=41、其中 pbl_ 前缀 38 张，
|
||
非 pbl_ 前缀仅 3 张（appcodes/appcodes_kv/pbl_tenant 类共享表由种子段顺带建出），
|
||
`users` 不在其列 → 登录端点 sqlExe 抛 ERROR 1146 (42S02) Table 'pbl.users' doesn't exist
|
||
→ ahserver 兜底 500，会话 cookie 不下发 → 42 条业务用例持续 blocked。
|
||
|
||
同一口径差还解释了 Bug 描述第 6 点的 ERROR 1146 群：
|
||
pbl_role / pbl_appcode / pbl_validation_rule / pbl_agent_tool 均不存在——
|
||
因为 rbac 的角色表真名是 `role`（见 pkgs/rbac/models/role.json），
|
||
appbase 的编码表真名是 `appcodes`/`appcodes_kv`，
|
||
校验规则集真名是 `pbl_validation_ruleset`，Agent 工具白名单落在 appcodes 组内（零表）。
|
||
即：探针用的表名是"声明口径"，实库是"models 单一真源口径"，两者不一致。
|
||
本脚本以 models/*.json 为唯一真源生成 DDL，并额外产出一份
|
||
「声明表名 → 真源表名」对照表（scripts/ddl/base_table_alias_map.json），
|
||
供 test/deploy 侧探针改用真名，消除同类 1146 误报。
|
||
|
||
【设计约束】
|
||
· 不硬编码基座包路径：三段式探测（环境变量 BASE_PKGS_ROOT → 候选根逐个存在性
|
||
校验 → 全部落空则 die），每条派生路径都打印解析出的绝对路径（web-application-spec
|
||
铁律 5「路径推导必须自校验」）。
|
||
· 不硬编码 DB 连接参数：本脚本只生成 SQL 文件，不连库；连接参数由 build.sh
|
||
运行时从 projects/pbls/env/{env}.json 读取（project-directory-spec §8.6）。
|
||
· 方言 mariadb：BIGINT AUTO_INCREMENT 口径不适用（基座表主键全是 VARCHAR(32)），
|
||
统一 utf8mb4 / utf8mb4_unicode_ci，全部 CREATE TABLE IF NOT EXISTS（幂等）。
|
||
· 抽象类型 → 原生类型映射严格按 database-table-definition-spec：
|
||
只用 models 里的抽象类型，禁止把原生类型写回 models。
|
||
|
||
用法：
|
||
python3 scripts/gen_base_ddl.py \
|
||
--out scripts/ddl/base_tables.sql \
|
||
--alias-map scripts/ddl/base_table_alias_map.json
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import sys
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 抽象类型 → MariaDB 原生类型映射（database-table-definition-spec 口径）
|
||
# --------------------------------------------------------------------------
|
||
_INT_TYPES = {
|
||
"short": "SMALLINT",
|
||
"int": "INT",
|
||
"long": "BIGINT",
|
||
}
|
||
|
||
# 基座包候选根（探测顺序即优先级）。全部为**代码位置**，非 env 连接参数；
|
||
# 仍逐条打印存在性，避免"路径算错"被误报成"基座包缺失"。
|
||
_BASE_PKG_CANDIDATES = (
|
||
os.environ.get("BASE_PKGS_ROOT", ""),
|
||
os.path.join(os.environ.get("WS_DIR", ""), "pkgs"),
|
||
"/d/pipeline/pipeline-app/pkgs",
|
||
os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "pkgs"),
|
||
os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "venv", "lib"),
|
||
)
|
||
|
||
# 需要建表的基座模块（appbase 提供共享编码表，rbac 提供登录/鉴权表）
|
||
_BASE_MODULES = ("appbase", "rbac")
|
||
|
||
# rbac 登录流程额外依赖的列，来源 = pkgs/rbac/migration/add_login_tracking.sql
|
||
# 与 migration/fix_users_schema.sql（up_login.dspy 的锁定检查读 login_fail_count /
|
||
# last_login_fail，成功登录写 last_login）。models/users.json 未含这三列，
|
||
# 故在生成 DDL 时补齐，并保证可空（fix_users_schema.sql 明确：NOT NULL DEFAULT
|
||
# current_timestamp() 会让 "SET last_login_fail=NULL" 被转成当前时间戳）。
|
||
_USERS_LOGIN_TRACKING_COLUMNS = (
|
||
("login_fail_count", "INT NOT NULL DEFAULT 0", "连续登录失败次数"),
|
||
("last_login_fail", "DATETIME NULL DEFAULT NULL", "最近一次登录失败时间(NULL=从未失败)"),
|
||
("last_login", "DATETIME NULL DEFAULT NULL", "最近一次登录成功时间"),
|
||
)
|
||
_EXTRA_COLUMNS = {"users": _USERS_LOGIN_TRACKING_COLUMNS}
|
||
|
||
# 「声明口径表名」→「models 单一真源表名」对照（消除 ERROR 1146 同类误报）
|
||
_DECLARED_TO_REAL = {
|
||
"pbl_role": {"real": "role", "module": "rbac",
|
||
"note": "rbac 角色表真名为 role，非 pbl_role；7 类角色种子写 role + userrole"},
|
||
"pbl_appcode": {"real": "appcodes", "module": "appbase",
|
||
"note": "共享编码父表真名 appcodes（子表 appcodes_kv），非 pbl_appcode"},
|
||
"pbl_validation_rule": {"real": "pbl_validation_ruleset", "module": "pbl_validation",
|
||
"note": "14 维校验规则集真名 pbl_validation_ruleset"},
|
||
"pbl_agent_tool": {"real": None, "module": "pbl_agent_runtime",
|
||
"note": "Agent 工具白名单 13 启用/9 禁用落在 appcodes 组内（零表设计），无独立表"},
|
||
"pbl_role_binding": {"real": "pbl_role_binding", "module": "pbl_governance",
|
||
"note": "治理域角色绑定表，真名一致；行数 0 属种子未注入（见 seed_rbac_admin.py）"},
|
||
}
|
||
|
||
|
||
def _log(msg: str) -> None:
|
||
sys.stderr.write("[gen_base_ddl] %s\n" % msg)
|
||
|
||
|
||
def _die(msg: str) -> None:
|
||
sys.stderr.write("[gen_base_ddl][FATAL] %s\n" % msg)
|
||
raise SystemExit(2)
|
||
|
||
|
||
def resolve_base_pkgs_root() -> str:
|
||
"""三段式探测基座包根目录，逐条打印解析结果与存在性。"""
|
||
for idx, cand in enumerate(_BASE_PKG_CANDIDATES):
|
||
if not cand:
|
||
continue
|
||
abspath = os.path.abspath(cand)
|
||
ok = os.path.isdir(abspath) and all(
|
||
os.path.isdir(os.path.join(abspath, m, "models")) for m in _BASE_MODULES
|
||
)
|
||
_log("candidate[%d] BASE_PKGS_ROOT=%s exists=%s models_ok=%s"
|
||
% (idx, abspath, os.path.isdir(abspath), ok))
|
||
if ok:
|
||
_log("resolved BASE_PKGS_ROOT=%s" % abspath)
|
||
return abspath
|
||
_die("未找到含 appbase/models 与 rbac/models 的基座包根；"
|
||
"请导出 BASE_PKGS_ROOT=<绝对路径> 或 WS_DIR=<机构工作空间> 后重试")
|
||
return "" # unreachable
|
||
|
||
|
||
def _sql_type(field: dict) -> str:
|
||
"""抽象类型 + length/dec → MariaDB 原生类型。"""
|
||
t = (field.get("type") or "").strip().lower()
|
||
length = field.get("length")
|
||
dec = field.get("dec")
|
||
|
||
if t in ("str", "varchar"):
|
||
n = int(length) if length else 255
|
||
return "VARCHAR(%d)" % n
|
||
if t == "char":
|
||
n = int(length) if length else 1
|
||
return "CHAR(%d)" % n
|
||
if t in _INT_TYPES:
|
||
return _INT_TYPES[t]
|
||
if t in ("float", "decimal"):
|
||
p = int(length) if length else 15
|
||
s = int(dec) if dec else 2
|
||
return "DECIMAL(%d,%d)" % (p, s)
|
||
if t in ("double", "ddouble"):
|
||
return "DOUBLE"
|
||
if t == "date":
|
||
return "DATE"
|
||
if t == "time":
|
||
return "TIME"
|
||
if t == "datetime":
|
||
return "DATETIME"
|
||
if t == "timestamp":
|
||
# 显式 NULL DEFAULT NULL：MariaDB 对表内首个 TIMESTAMP 列会自动加
|
||
# DEFAULT CURRENT_TIMESTAMP ON UPDATE，会让幂等种子每次重写时间戳。
|
||
return "TIMESTAMP NULL DEFAULT NULL"
|
||
if t == "text":
|
||
n = int(length) if length else 0
|
||
if n and n <= 65535:
|
||
return "TEXT"
|
||
return "LONGTEXT"
|
||
if t == "bin":
|
||
return "LONGBLOB"
|
||
_log("WARN 未知抽象类型 %r（字段 %s），按 VARCHAR(255) 兜底" % (t, field.get("name")))
|
||
return "VARCHAR(255)"
|
||
|
||
|
||
def _col_ddl(field: dict) -> str:
|
||
name = field["name"]
|
||
parts = [" `%s` %s" % (name, _sql_type(field))]
|
||
nullable = field.get("nullable")
|
||
# 规范：只有字符串 "no" 渲染 NOT NULL；null:false / 布尔 False 均不识别，
|
||
# 这里同时兼容遗留写法，避免静默按可空列生成。
|
||
if nullable == "no" or nullable is False or field.get("null") is False:
|
||
parts.append("NOT NULL")
|
||
default = field.get("default")
|
||
if default is not None and default != "":
|
||
dv = str(default)
|
||
if dv.upper() in ("CURRENT_TIMESTAMP", "NULL"):
|
||
parts.append("DEFAULT %s" % dv)
|
||
else:
|
||
parts.append("DEFAULT '%s'" % dv.replace("'", "''"))
|
||
title = (field.get("title") or "").replace("'", "''")
|
||
if title:
|
||
parts.append("COMMENT '%s'" % title)
|
||
return " ".join(parts)
|
||
|
||
|
||
def load_table_defs(pkgs_root: str):
|
||
"""读取 appbase/rbac 的 models/*.json，返回 [(module, table_name, def_dict)]。"""
|
||
out = []
|
||
for mod in _BASE_MODULES:
|
||
mdir = os.path.join(pkgs_root, mod, "models")
|
||
for fn in sorted(os.listdir(mdir)):
|
||
if not fn.endswith(".json"):
|
||
continue
|
||
path = os.path.join(mdir, fn)
|
||
try:
|
||
with open(path, "r", encoding="utf-8") as fh:
|
||
d = json.load(fh)
|
||
except Exception as exc: # noqa: BLE001
|
||
_die("models JSON 解析失败 %s: %s" % (path, exc))
|
||
summary = d.get("summary") or []
|
||
if not summary:
|
||
_die("%s 缺 summary 段（四段式硬约束）" % path)
|
||
s0 = summary[0]
|
||
tname = s0.get("name") or fn[:-5]
|
||
primary = s0.get("primary") or ["id"]
|
||
if isinstance(primary, str):
|
||
# 规范铁律：primary 必须是数组；字符串会让 DDL 拼出 "i,d"
|
||
_die("%s summary.primary 是字符串 %r，应为数组" % (path, primary))
|
||
out.append((mod, tname, d, primary, path))
|
||
_log("loaded %s/models/%s → table=%s primary=%s fields=%d"
|
||
% (mod, fn, tname, ",".join(primary), len(d.get("fields") or [])))
|
||
if not out:
|
||
_die("基座 models 一张表都没读到，拒绝生成空 DDL")
|
||
return out
|
||
|
||
|
||
def render_ddl(defs) -> str:
|
||
lines = [
|
||
"-- =====================================================================",
|
||
"-- base_tables.sql —— 基座模块（appbase / rbac）表结构，由 scripts/gen_base_ddl.py",
|
||
"-- 从各模块 models/*.json（单一真源）生成，禁止手工编辑本文件。",
|
||
"-- 方言: mariadb / utf8mb4 / utf8mb4_unicode_ci；全部 IF NOT EXISTS（幂等）。",
|
||
"-- 修复: Bug 4SxqQAulkH9eAzEys_h-e（up_login.dspy 500 —— users 表缺失）",
|
||
"-- Bug 0CQZ7O6ftDRHbxih59upr（种子 4 项 0 行 —— 依赖 role/userrole 就位）",
|
||
"-- =====================================================================",
|
||
"SET NAMES utf8mb4;",
|
||
"",
|
||
]
|
||
for mod, tname, d, primary, path in defs:
|
||
fields = d.get("fields") or []
|
||
if not fields:
|
||
_die("%s 无 fields 段" % path)
|
||
names = [f["name"] for f in fields]
|
||
if len(set(names)) != len(names):
|
||
_die("%s 字段名重复: %s" % (path, names))
|
||
for p in primary:
|
||
if p not in names:
|
||
_die("%s 主键 %r 不在 fields 中" % (path, p))
|
||
|
||
lines.append("-- [%s] %s （源: %s）" % (mod, tname, path))
|
||
lines.append("CREATE TABLE IF NOT EXISTS `%s` (" % tname)
|
||
body = [_col_ddl(f) for f in fields]
|
||
|
||
# 补齐登录跟踪列（仅 users）
|
||
for extra_name, extra_type, extra_comment in _EXTRA_COLUMNS.get(tname, ()):
|
||
if extra_name in names:
|
||
continue
|
||
body.append(" `%s` %s COMMENT '%s'" % (extra_name, extra_type, extra_comment))
|
||
names.append(extra_name)
|
||
lines.append("-- + 补列 %s（源: rbac/migration/add_login_tracking.sql）" % extra_name)
|
||
|
||
body.append(" PRIMARY KEY (%s)" % ", ".join("`%s`" % p for p in primary))
|
||
|
||
seen_idx = set()
|
||
for idx in d.get("indexes") or []:
|
||
iname = idx.get("name")
|
||
ifields = idx.get("idxfields")
|
||
if not iname or not ifields:
|
||
continue
|
||
if isinstance(ifields, str):
|
||
ifields = [ifields]
|
||
if not all(c in names for c in ifields):
|
||
_log("WARN %s 索引 %s 含未知列 %s，跳过" % (tname, iname, ifields))
|
||
continue
|
||
# 索引名表内唯一：基座 models 里普遍叫 idx1/idx2，跨表重名在 MySQL 合法，
|
||
# 但为可读性统一加表名前缀。
|
||
uniq = "idx_%s_%s" % (tname, iname)
|
||
if uniq in seen_idx:
|
||
continue
|
||
seen_idx.add(uniq)
|
||
kind = "UNIQUE KEY" if (idx.get("idxtype") or "index") == "unique" else "KEY"
|
||
body.append(" %s `%s` (%s)"
|
||
% (kind, uniq, ", ".join("`%s`" % c for c in ifields)))
|
||
|
||
lines.append(",\n".join(body))
|
||
lines.append(") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;")
|
||
lines.append("")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def render_alias_map(defs) -> dict:
|
||
real_tables = sorted({t for _, t, _, _, _ in defs})
|
||
return {
|
||
"_comment": "声明口径表名 → models 单一真源表名对照；test/deploy 探针请改用 real 名，"
|
||
"避免 ERROR 1146 误报为『表缺失』。real=null 表示该声明项本就零表设计。",
|
||
"generated_by": "scripts/gen_base_ddl.py",
|
||
"base_tables": real_tables,
|
||
"mapping": _DECLARED_TO_REAL,
|
||
}
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser(description="生成基座模块（appbase/rbac）DDL")
|
||
here = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
ap.add_argument("--out", default=os.path.join(here, "scripts", "ddl", "base_tables.sql"))
|
||
ap.add_argument("--alias-map",
|
||
default=os.path.join(here, "scripts", "ddl", "base_table_alias_map.json"))
|
||
args = ap.parse_args()
|
||
|
||
pkgs_root = resolve_base_pkgs_root()
|
||
defs = load_table_defs(pkgs_root)
|
||
ddl = render_ddl(defs)
|
||
|
||
for target in (args.out, args.alias_map):
|
||
d = os.path.dirname(os.path.abspath(target))
|
||
if d and not os.path.isdir(d):
|
||
os.makedirs(d, exist_ok=True)
|
||
|
||
with open(args.out, "w", encoding="utf-8") as fh:
|
||
fh.write(ddl)
|
||
with open(args.alias_map, "w", encoding="utf-8") as fh:
|
||
json.dump(render_alias_map(defs), fh, ensure_ascii=False, indent=2)
|
||
fh.write("\n")
|
||
|
||
n_tables = ddl.count("CREATE TABLE IF NOT EXISTS")
|
||
_log("OK 写出 %s（%d 张表，%d 字节）" % (args.out, n_tables, len(ddl.encode("utf-8"))))
|
||
_log("OK 写出 %s" % args.alias_map)
|
||
if n_tables < 8:
|
||
_die("生成表数 %d < 8，基座 models 读取不完整，拒绝交付" % n_tables)
|
||
# 硬门禁：users 必须在内，否则登录端点仍会 1146
|
||
if "CREATE TABLE IF NOT EXISTS `users`" not in ddl:
|
||
_die("生成的 DDL 不含 users 表 —— up_login.dspy 500 根因未消除")
|
||
_log("SELF_CHECK base_ddl: PASS tables=%d users=present" % n_tables)
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|