#!/usr/bin/env python3 # -*- coding: utf-8 -*- """DDL 补丁工具：保证每张表 tenant_id 为第一个业务列（多租户强制打头）。 背景 ---- docs/01-design/data-model.md 硬约束：36 表全部 mariadb 方言、无 FK/ENUM/TIMESTAMP， 且 **tenant_id 强制打头**（紧随自增主键 id 之后，作为第一个业务列），以便 pbl_common 的租户上下文在 CRUD 工厂层统一注入 WHERE tenant_id = ?。 历史 DDL（含引用模块导出的 subobjects.sql 片段）存在 tenant_id 位置漂移 （写在表尾或夹在中间），本脚本做**幂等**重排： * 已在首位 → 不动（幂等） * 不在首位 → 抽出该行，移动到第一个业务列位置 * 完全缺失 → 按标准定义补插（BIGINT NOT NULL DEFAULT 0 + 注释） 用法 ---- # 干跑（只报告，不写盘） python3 apps/pbls/scripts/patch_ddl_tenant_first.py --dry-run # 就地修补（自动备份 .bak） python3 apps/pbls/scripts/patch_ddl_tenant_first.py --inplace # 指定文件 python3 apps/pbls/scripts/patch_ddl_tenant_first.py \ --ddl apps/pbls/scripts/ddl/pbls_tables.sql --inplace # 只校验不修补（CI 门禁用，发现漂移退出码 1） python3 apps/pbls/scripts/patch_ddl_tenant_first.py --check """ import argparse import os import re import shutil import sys DEFAULT_DDL = os.path.join( os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "scripts", "ddl", "pbls_tables.sql", ) TENANT_COL = "tenant_id" TENANT_DEF = " `%s` BIGINT NOT NULL DEFAULT 0 COMMENT '租户ID（多租户强制打头）'," % TENANT_COL RE_CREATE = re.compile( r"^\s*CREATE\s+TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?[`\"]?(\w+)[`\"]?", re.I ) RE_COL = re.compile(r"^\s*[`\"]?(\w+)[`\"]?\s+[A-Za-z]") RE_CONSTRAINT = re.compile( r"^\s*(PRIMARY\s+KEY|UNIQUE\s+KEY|UNIQUE\s+INDEX|KEY|INDEX|CONSTRAINT|FOREIGN\s+KEY|CHECK)\b", re.I, ) RE_FORBIDDEN = re.compile( r"\b(BIGSERIAL|SERIAL|nextval|FOREIGN\s+KEY|REFERENCES|ENUM\s*\(|TIMESTAMP)\b", re.I ) class TableBlock(object): """一张表的 DDL 行块（含起止行号，0-based）。""" def __init__(self, name, start): self.name = name self.start = start # CREATE TABLE 行 self.end = None # 收尾 `)` 行（含） self.body = [] # (lineno, text) 列/约束行 def first_business_index(self): """返回 body 中第一个业务列（非约束行）的下标，无则 None。""" for i, (_ln, text) in enumerate(self.body): if RE_CONSTRAINT.match(text): continue if RE_COL.match(text): return i return None def tenant_index(self): for i, (_ln, text) in enumerate(self.body): m = RE_COL.match(text) if m and m.group(1) == TENANT_COL: return i return None def parse_blocks(lines): """把 DDL 文本切成 TableBlock 列表。""" blocks = [] cur = None for i, text in enumerate(lines): m = RE_CREATE.match(text) if m: cur = TableBlock(m.group(1), i) blocks.append(cur) continue if cur is None: continue stripped = text.strip() if stripped.startswith(")"): cur.end = i cur = None continue if not stripped or stripped.startswith("--") or stripped.startswith("#"): continue cur.body.append((i, text)) return blocks def audit(lines, blocks): """返回 (漂移表清单, 缺失表清单, 禁用方言命中清单)。""" drifted, missing, forbidden = [], [], [] for b in blocks: ti = b.tenant_index() fi = b.first_business_index() if ti is None: missing.append(b.name) elif fi is not None and ti != fi: drifted.append((b.name, ti, fi)) for i, text in enumerate(lines, 1): m = RE_FORBIDDEN.search(text) if m and not text.strip().startswith("--"): forbidden.append((i, m.group(1), text.strip()[:90])) return drifted, missing, forbidden def patch(lines, blocks): """就地重排/补插 tenant_id，返回 (新行列表, 改动计数)。""" out = list(lines) changes = 0 # 从后往前处理，避免行号位移 for b in sorted(blocks, key=lambda x: x.start, reverse=True): ti = b.tenant_index() fi = b.first_business_index() if ti is None: anchor = b.body[fi][0] if fi is not None else (b.start + 1) out.insert(anchor, TENANT_DEF) changes += 1 continue if fi is None or ti == fi: continue # 幂等：已在首位 src_line = b.body[ti][0] dst_line = b.body[fi][0] moved = out.pop(src_line) if not moved.rstrip().endswith(","): moved = moved.rstrip() + "," out.insert(dst_line, moved) changes += 1 return out, changes def main(): ap = argparse.ArgumentParser(description="DDL tenant_id 打头幂等补丁") ap.add_argument("--ddl", default=DEFAULT_DDL, help="DDL 文件路径") ap.add_argument("--inplace", action="store_true", help="就地写回（自动 .bak 备份）") ap.add_argument("--dry-run", action="store_true", help="只报告不写盘") ap.add_argument("--check", action="store_true", help="CI 门禁模式：发现漂移/缺失/禁用方言则退出码 1") args = ap.parse_args() if not os.path.isfile(args.ddl): print("FAIL: DDL 文件不存在：%s" % args.ddl) return 2 with open(args.ddl, "r", encoding="utf-8") as fp: text = fp.read() lines = text.splitlines() blocks = parse_blocks(lines) drifted, missing, forbidden = audit(lines, blocks) n_autoinc = len([1 for ln in lines if re.search(r"AUTO_INCREMENT", ln, re.I)]) n_bigint = len([1 for ln in lines if re.search(r"\bBIGINT\b", ln, re.I)]) print("=" * 74) print("patch_ddl_tenant_first :: %s" % args.ddl) print("=" * 74) print("CREATE TABLE = %d" % len(blocks)) print("AUTO_INCREMENT = %d" % n_autoinc) print("BIGINT = %d" % n_bigint) print("tenant_id 打头 = %d/%d" % (len(blocks) - len(drifted) - len(missing), len(blocks))) if drifted: print("\n[DRIFT] tenant_id 位置漂移 %d 表：" % len(drifted)) for name, ti, fi in drifted: print(" - %-28s 当前 body[%d] 应在 body[%d]" % (name, ti, fi)) if missing: print("\n[MISS] 缺 tenant_id 列 %d 表：%s" % (len(missing), ", ".join(missing))) if forbidden: print("\n[FORBIDDEN] 禁用方言命中 %d 处：" % len(forbidden)) for ln, kw, snippet in forbidden[:20]: print(" - L%d %s :: %s" % (ln, kw, snippet)) else: print("\n[FORBIDDEN] 7 类禁用方言零命中（BIGSERIAL/SERIAL/nextval/" "FOREIGN KEY/REFERENCES/ENUM(/TIMESTAMP）") if args.check: bad = len(drifted) + len(missing) + len(forbidden) print("\nCHECK RESULT: %s" % ("PASS" if bad == 0 else "FAIL(%d)" % bad)) return 0 if bad == 0 else 1 if not drifted and not missing: print("\nNOOP: 全部表 tenant_id 已打头，无需修补（幂等）") return 0 new_lines, changes = patch(lines, blocks) print("\n[PATCH] 改动 %d 处" % changes) if args.dry_run or not args.inplace: print("[DRY-RUN] 未写盘。加 --inplace 落盘。") return 0 bak = args.ddl + ".bak" shutil.copy2(args.ddl, bak) with open(args.ddl, "w", encoding="utf-8") as fp: fp.write("\n".join(new_lines) + "\n") print("[WRITE] 已写回 %s（备份 %s）" % (args.ddl, bak)) # 复验：修补后必须零漂移 blocks2 = parse_blocks(new_lines) d2, m2, _f2 = audit(new_lines, blocks2) if d2 or m2: print("[RECHECK] FAIL 仍有漂移 %d / 缺失 %d" % (len(d2), len(m2))) return 1 print("[RECHECK] PASS tenant_id 打头 %d/%d" % (len(blocks2), len(blocks2))) return 0 if __name__ == "__main__": sys.exit(main())