358 lines
18 KiB
Python
358 lines
18 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""M11b-1c-B T-1 补丁应用器：对同目录 scripts/pbl_runtime_event_ddl.py
|
||
做**精确串替换**（每处断言命中且仅命中 1 次），修复 QC #1 的 % 转义缺陷并重引入
|
||
PBL_DDL_DB_HOST/PBL_DDL_DB_PORT 沙箱库开关（凭据仍只来自 env/*.json）。
|
||
|
||
不改 models/*.json、不改任何其他 .py。
|
||
|
||
用法：
|
||
python3 modules/pbl_runtime_ext/scripts/m11b1cb_patch_generator.py
|
||
# 应用到模块仓库工作树（默认目标=同目录 pbl_runtime_event_ddl.py）
|
||
python3 .../m11b1cb_patch_generator.py --check
|
||
# 只试跑，报告每处命中次数，不落盘
|
||
python3 .../m11b1cb_patch_generator.py --target /tmp/base.py
|
||
# 重放到指定文件副本（取证用）
|
||
|
||
幂等（QC #2 要求）：重复运行不会产出重复函数定义——检测到修复已全部在位则 SKIP 并返回 0；
|
||
检测到「只应用了一半」的部分状态则 FAIL 并返回 1（要求先还原到基线再重放）。
|
||
|
||
落点（QC #11 整改）：本脚本从工作空间根 tools/ 迁入模块仓库
|
||
modules/pbl_runtime_ext/scripts/（与它服务的 pbl_runtime_event_ddl.py、以及已入库的
|
||
m11b1cb_verify_sandbox_db.py 同仓同目录），纳入 git 版本控制，可被部署/复现阶段 git pull 取到。
|
||
工作空间根 tools/m11b1cb_patch_generator.py 为迁移前的旧副本，自本次提交起**作废**，
|
||
唯一权威版本是本文件；旧副本仅作历史留痕，不再维护（清理动作见 dev-notes 第 5 节）。
|
||
|
||
取证方式（QC #8 整改：基线引用改为固定 commit + blob，不再用易漂移的 HEAD~1）：
|
||
基线 = modules/pbl_runtime_ext 仓库 commit 4a7c7b9，其 scripts/pbl_runtime_event_ddl.py
|
||
的 blob = 7b9c991（由 M11b-1c-A 还原任务确立）。本补丁所有 old 串以该基线为准。
|
||
|
||
git -C modules/pbl_runtime_ext show 4a7c7b9:scripts/pbl_runtime_event_ddl.py > /tmp/base.py
|
||
python3 modules/pbl_runtime_ext/scripts/m11b1cb_patch_generator.py --target /tmp/base.py
|
||
diff /tmp/base.py modules/pbl_runtime_ext/scripts/pbl_runtime_event_ddl.py
|
||
|
||
⚠ 预期：上面最后一条 diff **非空属正常，不代表补丁有问题**。仓库 HEAD 在本补丁（T-1）
|
||
之外还叠加了 M11b-1c 其它子任务的改动——具体三处：
|
||
(a) _with_delimiter 改为按行扫描重写；
|
||
(b) _norm_part 额外多一个 .strip("'")（单引号剥离）；
|
||
(c) 离线自检 need_map 新增段。
|
||
这三处不在本任务授权范围内，故「REPL 里的 new 串与仓库最终代码逐字一致」这一 stronger
|
||
断言在当前仓库历史下**不成立**，本文件不再声称它成立（旧版 docstring 的该声明已被 QC 实测证伪）。
|
||
|
||
本补丁可机械核验的断言收敛为以下三条（都不依赖 HEAD 的其它改动，可复现）：
|
||
(1) 基线重放：对 commit 4a7c7b9 的副本执行，10/10 处 REPL 各命中且仅命中 1 次，rc=0；
|
||
(2) 幂等：对已应用结果（即仓库 HEAD 版本）重跑 → SKIP，rc=0；
|
||
(3) 部分应用：缺任一 APPLIED_MARKERS 标记 → FAIL，rc=1（QC #8 记录的
|
||
`git show 7f0f560:...`（=HEAD~1）重放即触发此分支，rc=1，属守卫按设计生效）。
|
||
"""
|
||
import argparse
|
||
import io
|
||
import os
|
||
import sys
|
||
|
||
# 迁移后（QC #11）：默认目标与本脚本同目录，不再依赖工作空间根的相对层级
|
||
DEFAULT_TARGET = os.path.join(os.path.dirname(os.path.abspath(__file__)),
|
||
"pbl_runtime_event_ddl.py")
|
||
|
||
# 修复已在位 / 部分在位 的判定标记（幂等守卫用）
|
||
APPLIED_MARKERS = (
|
||
"def _norm_part(name):",
|
||
"def _render(sql, args=None, placeholders=None):",
|
||
'"PBL_DDL_DB_HOST"',
|
||
"part_norm = set(_norm_part(n) for n in part_names)",
|
||
)
|
||
|
||
# ---------------------------------------------------------------- 替换清单
|
||
REPL = []
|
||
|
||
# --- P1: _db_conf 重新实现「仅覆盖 host/port」的沙箱库开关（基线 @@ -531/@@ -539 区段） ---
|
||
REPL.append((
|
||
'''def _db_conf():
|
||
"""读项目唯一事实源 env/test.json（禁止硬编码连接串）。"""
|
||
for env_name in ("test", "prod"):
|
||
path = os.path.join(WORKSPACE, "projects", "pbls", "env", "%s.json" % env_name)
|
||
if os.path.exists(path):
|
||
with open(path, encoding="utf-8") as f:
|
||
conf = json.load(f)
|
||
db = conf.get("db") or {}
|
||
if db.get("host"):
|
||
return env_name, db
|
||
return None, None
|
||
''',
|
||
'''def _db_conf():
|
||
"""读项目唯一事实源 env/test.json（禁止硬编码连接串）。
|
||
|
||
沙箱库开关（harness，QC #1 配套的验证隔离能力）：环境变量 PBL_DDL_DB_HOST /
|
||
PBL_DDL_DB_PORT **只覆盖 host/port**，供 CI / 验证 harness 把连库分支指向一次性沙箱库
|
||
——破坏性判定（DROP PARTITION、探针 INSERT、guard break-glass）不能拿共享 test/prod 库做实验。
|
||
· user / password / dbname 一律仍只来自 env/*.json：本函数不接受任何环境变量覆盖，
|
||
也不内置默认主机、默认端口、默认账号（禁止硬编码）；
|
||
· 两个变量都不设置时，返回值与引入该开关之前逐字一致（零行为变化）；
|
||
· 覆盖生效时 env_name 追加 "+harness" 后缀，日志与 --verify-db 输出可一眼看出连的是沙箱库。
|
||
"""
|
||
for env_name in ("test", "prod"):
|
||
path = os.path.join(WORKSPACE, "projects", "pbls", "env", "%s.json" % env_name)
|
||
if os.path.exists(path):
|
||
with open(path, encoding="utf-8") as f:
|
||
conf = json.load(f)
|
||
db = conf.get("db") or {}
|
||
if db.get("host"):
|
||
host = (os.environ.get("PBL_DDL_DB_HOST") or "").strip()
|
||
port = (os.environ.get("PBL_DDL_DB_PORT") or "").strip()
|
||
if host:
|
||
db = dict(db, host=host)
|
||
env_name = "%s+harness" % env_name
|
||
if port:
|
||
if not port.isdigit():
|
||
raise RuntimeError("PBL_DDL_DB_PORT 必须是数字端口号: %r" % port)
|
||
db = dict(db, port=int(port))
|
||
if "+harness" not in env_name:
|
||
env_name = "%s+harness" % env_name
|
||
return env_name, db
|
||
return None, None
|
||
'''))
|
||
|
||
# --- P2: _explain_partitions docstring 去掉 row[3] 字样（A5 静态断言）+ 入参转义自检 ---
|
||
REPL.append((
|
||
''' 不再用 PARTITIONS 扩展语法（EXPLAIN PARTITIONS）：其分区列固定在结果 row[3]，列序随 MariaDB/MySQL
|
||
版本变化，且该扩展语法在新版 MySQL 已废弃。JSON 计划里的 partitions 字段两版一致。
|
||
取不到（引擎不支持/解析失败）返回 []，由调用方判失败，不猜。
|
||
"""
|
||
try:
|
||
cur.execute("EXPLAIN FORMAT=JSON " + sql, tuple(params or ()))
|
||
row = cur.fetchone()
|
||
''',
|
||
''' 不再用旧的 PARTITIONS 扩展语法（EXPLAIN 后跟 PARTITIONS 关键字那种）：那种写法把分区列固定在
|
||
结果集第 4 列（下标 3），列序随 MariaDB/MySQL 版本变化，且该扩展语法在新版 MySQL 已废弃。
|
||
JSON 计划里的 partitions 字段两版一致。
|
||
取不到（引擎不支持/解析失败）返回 []，由调用方判失败，不猜。
|
||
"""
|
||
# QC #1(转义)：传进来的 sql 必须已完成标识符渲染，只剩 DBAPI 占位符 %s；残留 '%%' 说明
|
||
# 调用方转义写错，直接抛出让它响，别丢给驱动报 "unsupported format character"。
|
||
if "%%" in sql:
|
||
raise RuntimeError("EXPLAIN SQL 仍残留 '%%'，转义写法有误: " + sql[:200])
|
||
args = tuple(params) if params else None
|
||
if args is not None and sql.count("%s") != len(args):
|
||
raise RuntimeError("EXPLAIN SQL 占位符 %d 个与参数 %d 个不符: %s"
|
||
% (sql.count("%s"), len(args), sql[:200]))
|
||
try:
|
||
cur.execute("EXPLAIN FORMAT=JSON " + sql, args)
|
||
row = cur.fetchone()
|
||
'''))
|
||
|
||
# --- P3: _explain_partitions 结果分区名统一口径 ---
|
||
REPL.append((
|
||
''' if key == "partitions" and isinstance(val, list):
|
||
for p in val:
|
||
if isinstance(p, str) and p and p not in found:
|
||
found.append(p)
|
||
''',
|
||
''' if key == "partitions" and isinstance(val, list):
|
||
for p in val:
|
||
name = _norm_part(p)
|
||
if name and name not in found:
|
||
found.append(name)
|
||
'''))
|
||
|
||
# --- P4: 新增 _norm_part / _render 两个私有工具（插在 _explain_partitions 之前） ---
|
||
REPL.append((
|
||
'''def _explain_partitions(cur, sql, params=None):
|
||
''',
|
||
'''def _norm_part(name):
|
||
"""分区名统一口径（QC #1 附带缺陷：两处来源比对方式不一致）。
|
||
|
||
information_schema.PARTITIONS 的 PARTITION_NAME 与 EXPLAIN FORMAT=JSON 计划里的
|
||
partitions 元素，跨引擎/版本可能带反引号、双引号、空白或大小写差异。两边都过这个
|
||
归一化再比对，避免「分区确实存在但判定说不属于清单」的假失败。
|
||
"""
|
||
if not name:
|
||
return ""
|
||
text = name.decode("utf-8", "replace") if isinstance(name, (bytes, bytearray)) else str(name)
|
||
return text.strip().strip("`").strip('"').lower()
|
||
|
||
|
||
def _render(sql, args=None, placeholders=None):
|
||
"""SQL 的 % 转义统一出口：可选渲染标识符 + fail-fast 自检（QC #1 的根因处置）。
|
||
|
||
约定：模板中由 Python 侧填入的标识符（表名/分区名）写 %s；要留给 DBAPI 绑参的
|
||
百分号占位符一律写 %%s，标识符渲染后即变回 %s。
|
||
· args 不为 None：先执行 sql % args 完成标识符渲染；
|
||
· args 为 None：视为调用方已完成渲染（模板就地 % 过），本函数只做转义自检。
|
||
自检两条，任一不满足立即抛 RuntimeError（而不是丢给驱动报
|
||
"unsupported format character"，那会在 verify_db 里被 except 吞成一条模糊失败）：
|
||
· 结果仍残留 '%%' → 转义写错；
|
||
· '%s' 个数与 placeholders 不符 → 漏写/多写绑参占位符。
|
||
渲染单独成行、括号显式定界，不再依赖「相邻字面量先拼接、再整体 %」的隐式优先级。
|
||
"""
|
||
if args is not None:
|
||
sql = sql % tuple(args)
|
||
if "%%" in sql:
|
||
raise RuntimeError("SQL 渲染后仍残留 '%%'，转义写法有误: " + sql[:200])
|
||
if placeholders is not None and sql.count("%s") != placeholders:
|
||
raise RuntimeError("SQL 渲染后 DBAPI 占位符应为 %d 个，实际 %d 个: %s"
|
||
% (placeholders, sql.count("%s"), sql[:200]))
|
||
return sql
|
||
|
||
|
||
def _explain_partitions(cur, sql, params=None):
|
||
'''))
|
||
|
||
# --- P5: _row_in_partition 转义重构（基线 @@ -641 区段） ---
|
||
REPL.append((
|
||
''' try:
|
||
cur.execute("SELECT id FROM `%s` PARTITION (`%s`)"
|
||
" WHERE tenant_id='t_probe' AND event_uid=%%s" % (TABLE, part), (uid,))
|
||
except Exception: # noqa: BLE001
|
||
return False
|
||
''',
|
||
''' sql = _render("SELECT id FROM `%s` PARTITION (`%s`)"
|
||
" WHERE tenant_id='t_probe' AND event_uid=%%s" % (TABLE, _norm_part(part)),
|
||
placeholders=1)
|
||
try:
|
||
cur.execute(sql, (uid,))
|
||
except Exception: # noqa: BLE001
|
||
return False
|
||
'''))
|
||
|
||
# --- P6: _locate_row_partition 复用 _row_in_partition + 转义/口径统一（基线 @@ -660 区段） ---
|
||
REPL.append((
|
||
''' for part in part_names:
|
||
if _row_in_partition(cur, part, uid):
|
||
return part
|
||
planned = _explain_partitions(cur,
|
||
"SELECT id FROM `%s` WHERE tenant_id='t_probe'"
|
||
" AND event_uid=%%s AND created_at=%%s" % TABLE,
|
||
(uid, ts))
|
||
if len(planned) == 1:
|
||
return planned[0]
|
||
return ",".join(planned) or "?"
|
||
''',
|
||
''' # 1) 逐个 PARTITION(p) 物理反查（复用 _row_in_partition，SQL 只在其内拼装一次，
|
||
# 避免同一份转义逻辑在两处各写一遍、改一处漏一处）
|
||
for part in part_names:
|
||
if _row_in_partition(cur, part, uid):
|
||
return _norm_part(part)
|
||
# 2) 反查不中才退回 EXPLAIN FORMAT=JSON；标识符渲染与转义自检统一走 _render
|
||
probe_sql = _render("SELECT id FROM `%s` WHERE tenant_id='t_probe'"
|
||
" AND event_uid=%%s AND created_at=%%s" % TABLE,
|
||
placeholders=2)
|
||
planned = _explain_partitions(cur, probe_sql, (uid, ts))
|
||
if len(planned) == 1:
|
||
return planned[0]
|
||
return ",".join(planned) or "?"
|
||
'''))
|
||
|
||
# --- P7: verify_db 探针 INSERT 转义自检（基线 @@ -694 区段） ---
|
||
REPL.append((
|
||
''' try:
|
||
for i, (uid, ts) in enumerate(probes):
|
||
cur.execute("INSERT INTO `%s` (tenant_id,event_uid,event_type,scene_id,seq,created_at)"
|
||
" VALUES ('t_probe',%%s,'probe',0,%%s,%%s)" % TABLE, (uid, i + 1, ts))
|
||
except Exception as exc: # noqa: BLE001
|
||
''',
|
||
''' ins_sql = _render("INSERT INTO `%s` (tenant_id,event_uid,event_type,scene_id,seq,created_at)"
|
||
" VALUES ('t_probe',%%s,'probe',0,%%s,%%s)" % TABLE,
|
||
placeholders=3)
|
||
try:
|
||
for i, (uid, ts) in enumerate(probes):
|
||
cur.execute(ins_sql, (uid, i + 1, ts))
|
||
except Exception as exc: # noqa: BLE001
|
||
'''))
|
||
|
||
# --- P8: verify_db 分区清单归一化口径（供 P9/P10 比对） ---
|
||
REPL.append((
|
||
''' results.append(("MAXVALUE 兜底", PARTITION_MAXVALUE in part_names, ",".join(part_names)))
|
||
''',
|
||
''' results.append(("MAXVALUE 兜底", PARTITION_MAXVALUE in part_names, ",".join(part_names)))
|
||
# 分区名比对统一走归一化口径（information_schema 与 EXPLAIN JSON 计划两来源一致）
|
||
part_norm = set(_norm_part(n) for n in part_names)
|
||
'''))
|
||
|
||
# --- P9: 跨月落不同分区判定用统一口径 ---
|
||
REPL.append((
|
||
''' located = [_locate_row_partition(cur, uid, ts, part_names) for uid, ts in probes]
|
||
distinct = len(set(located)) == 2
|
||
in_list = all(p in part_names for p in located)
|
||
''',
|
||
''' located = [_locate_row_partition(cur, uid, ts, part_names) for uid, ts in probes]
|
||
distinct = len(set(located)) == 2
|
||
in_list = all(_norm_part(x) in part_norm
|
||
for p in located for x in str(p).split(","))
|
||
'''))
|
||
|
||
# --- P10: 分区裁剪判定：SQL 渲染单独成行 + 统一口径（基线 @@ -710 区段） ---
|
||
REPL.append((
|
||
''' # 旧写法（PARTITIONS 扩展语法 + 取 row[3]）：列序跨版本不稳定，新版 MySQL 已废弃该语法。
|
||
pruned = _explain_partitions(cur,
|
||
"SELECT id FROM `%s` WHERE created_at>='2026-01-01'"
|
||
" AND created_at<'2026-02-01'" % TABLE)
|
||
results.append(("分区裁剪",
|
||
bool(pruned) and all(p in part_names for p in pruned)
|
||
and len(pruned) < max(1, len(part_names)),
|
||
''',
|
||
''' # 旧写法（PARTITIONS 扩展语法 + 取结果集下标 3 那一列）：列序跨版本不稳定，新版 MySQL 已废弃。
|
||
prune_sql = _render("SELECT id FROM `%s` WHERE created_at>='2026-01-01'"
|
||
" AND created_at<'2026-02-01'" % TABLE,
|
||
placeholders=0)
|
||
pruned = _explain_partitions(cur, prune_sql)
|
||
results.append(("分区裁剪",
|
||
bool(pruned) and all(p in part_norm for p in pruned)
|
||
and len(pruned) < max(1, len(part_names)),
|
||
'''))
|
||
|
||
|
||
def _already_applied(src):
|
||
"""幂等守卫：返回 (全已在位, 在位标记列表, 缺失标记列表)。"""
|
||
hit = [m for m in APPLIED_MARKERS if m in src]
|
||
return len(hit) == len(APPLIED_MARKERS), hit, [m for m in APPLIED_MARKERS if m not in src]
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser(description="M11b-1c-B T-1 生成器补丁应用器")
|
||
ap.add_argument("--target", default=DEFAULT_TARGET, help="待打补丁的脚本路径")
|
||
ap.add_argument("--check", action="store_true", help="只试跑并报告命中次数，不落盘")
|
||
opts = ap.parse_args()
|
||
|
||
with io.open(opts.target, encoding="utf-8") as f:
|
||
src = f.read()
|
||
orig = src
|
||
|
||
# ---- 幂等守卫（QC #2）：修复已在位就 SKIP，绝不重复插入函数定义 ----
|
||
done, hit, missing = _already_applied(src)
|
||
if done:
|
||
print("SKIP 补丁已全部在位（%d/%d 标记命中），无需重复应用: %s"
|
||
% (len(hit), len(APPLIED_MARKERS), opts.target))
|
||
return 0
|
||
if hit:
|
||
print("FAIL 目标处于「部分应用」状态（%d 个标记已在位，%d 个缺失），"
|
||
"请先还原到基线 4a7c7b9(blob 7b9c991) 再重放。缺失: %s"
|
||
% (len(hit), len(missing), " / ".join(m[:40] for m in missing)))
|
||
return 1
|
||
|
||
# ---- 逐处替换：断言命中且仅命中 1 次，任一处不符立即中止且**不落盘** ----
|
||
for idx, (old, new) in enumerate(REPL, 1):
|
||
n = src.count(old)
|
||
if n != 1:
|
||
print("FAIL P%d 命中 %d 次（要求恰好 1 次），中止且不写盘: %s"
|
||
% (idx, n, opts.target))
|
||
return 1
|
||
src = src.replace(old, new, 1)
|
||
print("OK P%d 命中 1 次" % idx)
|
||
|
||
print("HIT %d/%d 处替换全部命中且唯一" % (len(REPL), len(REPL)))
|
||
|
||
if opts.check:
|
||
print("CHECK --check 模式：只试跑，未写盘")
|
||
return 0
|
||
if src == orig:
|
||
print("FAIL 替换后内容与原文件一致，疑似未生效，拒绝写盘")
|
||
return 1
|
||
with io.open(opts.target, "w", encoding="utf-8") as f:
|
||
f.write(src)
|
||
print("WRITE 已写盘: %s" % opts.target)
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|