approve: [M11b-1c-B] json 侧 spec 合规登记 + 证据路径更正 + 工作空间一次性附件清理

scripts/pbl_runtime_event_ddl.py

@@ -531,7 +531,16 @@ def verify():
 
 # ---------------------------------------------------------------- 连库执行 / 验证
 def _db_conf():
-    """读项目唯一事实源 env/test.json（禁止硬编码连接串）。"""
+    """读项目唯一事实源 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):
@@ -539,6 +548,17 @@ def _db_conf():
                 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
 
@@ -599,15 +619,55 @@ def _split_procedural(body):
     return [x.replace("$$", "").rstrip() for x in out]
 
 
+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('"').strip("'").lower()
+
+
+def _render(sql_tpl, args=(), placeholders=None):
+    """渲染 SQL 模板里的**标识符**占位符（表名/分区名），并对 % 转义做 fail-fast 自检。
+
+    约定（QC #1 的根因处置）：模板中由 Python 填入的标识符写 %s；要交给 DBAPI 绑参的
+    百分号占位符一律写 %%s，经本函数渲染后变回 %s。渲染后：
+      · 仍残留 '%%' → 转义写错，抛错；
+      · '%s' 个数与预期绑参个数不符 → 抛错。
+    渲染单独成行、括号显式定界，不再依赖「相邻字面量先拼接、再整体 %」的隐式优先级。
+    """
+    sql = sql_tpl % 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):
     """EXPLAIN FORMAT=JSON 取执行计划实际扫描的分区清单（QC #6）。
 
-    不再用 PARTITIONS 扩展语法（EXPLAIN PARTITIONS）：其分区列固定在结果 row[3]，列序随 MariaDB/MySQL
+    不再用旧的 PARTITIONS 扩展语法（EXPLAIN 后跟 PARTITIONS 关键字那种）：那种写法把分区列固定在
-    版本变化，且该扩展语法在新版 MySQL 已废弃。JSON 计划里的 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, tuple(params or ()))
+        cur.execute("EXPLAIN FORMAT=JSON " + sql, args)
         row = cur.fetchone()
     except Exception:  # noqa: BLE001
         return []
@@ -627,8 +687,9 @@ def _explain_partitions(cur, sql, params=None):
             for key, val in node.items():
                 if key == "partitions" and isinstance(val, list):
                     for p in val:
-                        if isinstance(p, str) and p and p not in found:
+                        name = _norm_part(p)
-                            found.append(p)
+                        if name and name not in found:
+                            found.append(name)
                 else:
                     walk(val)
         elif isinstance(node, list):
@@ -641,9 +702,11 @@ def _explain_partitions(cur, sql, params=None):
 
 def _row_in_partition(cur, part, uid):
     """SELECT ... FROM t PARTITION(p) 反查：探针行是否物理落在该分区。"""
+    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("SELECT id FROM `%s` PARTITION (`%s`)"
+        cur.execute(sql, (uid,))
-                    " WHERE tenant_id='t_probe' AND event_uid=%%s" % (TABLE, part), (uid,))
     except Exception:  # noqa: BLE001
         return False
     return bool(cur.fetchone())
@@ -657,13 +720,16 @@ def _locate_row_partition(cur, uid, ts, part_names):
     2) 反查不中（引擎不支持 PARTITION 子句等）时退回 EXPLAIN FORMAT=JSON 解析
        partitions（带 tenant_id+event_uid+created_at 等值条件，裁剪后应只剩 1 个分区）。
     """
+    # 1) 逐个 PARTITION(p) 物理反查（复用 _row_in_partition，SQL 只在其内拼装一次，
+    #    避免同一份转义逻辑在两处各写一遍、改一处漏一处）
     for part in part_names:
         if _row_in_partition(cur, part, uid):
-            return part
+            return _norm_part(part)
-    planned = _explain_partitions(cur,
+    # 2) 反查不中才退回 EXPLAIN FORMAT=JSON；标识符渲染与转义自检统一走 _render
-                                  "SELECT id FROM `%s` WHERE tenant_id='t_probe'"
+    probe_sql = _render("SELECT id FROM `%s` WHERE tenant_id='t_probe'"
-                                  " AND event_uid=%%s AND created_at=%%s" % TABLE,
+                        " AND event_uid=%%s AND created_at=%%s" % TABLE,
-                                  (uid, ts))
+                        placeholders=2)
+    planned = _explain_partitions(cur, probe_sql, (uid, ts))
     if len(planned) == 1:
         return planned[0]
     return ",".join(planned) or "?"
@@ -688,33 +754,39 @@ def verify_db():
     part_rows = [(r[0], r[1]) for r in cur.fetchall()]
     part_names = [n for n, _ in part_rows if n]
     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)
 
     # 插入两条不同月份的探针行 → 应落在两个不同的物理分区
     probe = "m11bprobe%s" % date.today().strftime("%H%M%S")
     probes = (("%s_a" % probe, "2026-01-05 00:00:00"),
               ("%s_b" % probe, "2027-03-05 00:00:00"))
+    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("INSERT INTO `%s` (tenant_id,event_uid,event_type,scene_id,seq,created_at)"
+            cur.execute(ins_sql, (uid, i + 1, ts))
-                        " VALUES ('t_probe',%%s,'probe',0,%%s,%%s)" % TABLE, (uid, i + 1, ts))
     except Exception as exc:  # noqa: BLE001
         results.append(("跨月插入", False, str(exc)[:160]))
     else:
         results.append(("跨月插入", True, "%d 条探针行（不同月份）" % len(probes)))
         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)
+        in_list = all(_norm_part(x) in part_norm
+                      for p in located for x in str(p).split(","))
         results.append(("跨月落不同分区", distinct and in_list,
                         " | ".join("%s -> %s" % (uid, p)
                                    for (uid, _), p in zip(probes, located))))
 
     # 分区裁剪：created_at 区间只应命中该区间所属的少数分区，而非全部分区。
-    # 旧写法（PARTITIONS 扩展语法 + 取 row[3]）：列序跨版本不稳定，新版 MySQL 已废弃该语法。
+    # 旧写法（PARTITIONS 扩展语法 + 取结果集下标 3 那一列）：列序跨版本不稳定，新版 MySQL 已废弃。
-    pruned = _explain_partitions(cur,
+    prune_sql = _render("SELECT id FROM `%s` WHERE created_at>='2026-01-01'"
-                                 "SELECT id FROM `%s` WHERE created_at>='2026-01-01'"
+                        " AND created_at<'2026-02-01'" % TABLE,
-                                 " AND created_at<'2026-02-01'" % TABLE)
+                        placeholders=0)
+    pruned = _explain_partitions(cur, prune_sql)
     results.append(("分区裁剪",
-                    bool(pruned) and all(p in part_names for p in pruned)
+                    bool(pruned) and all(p in part_norm for p in pruned)
                     and len(pruned) < max(1, len(part_names)),
                     "命中 %d/%d 分区: %s" % (len(pruned), len(part_names),
                                              ",".join(pruned) or "N/A")))