228 lines
8.0 KiB
Python
228 lines
8.0 KiB
Python
#!/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())
|