#!/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())