fix(pbls): [deploy-Q4] build.sh gate 6.4 strip comments/COMMENT strings before scanning + table count 36 to 37
Defect (deploy_test round 3, Q4 / deploy-evidence E15): 1. Gate 6.4 grepped the raw DDL for dialect keywords and was killed by 3 false positives that are pure documentation text: L6 header line comment, L220 business word "references" inside a COMMENT string, L891 contract self-check block that literally lists FOREIGN KEY / REFERENCES / ENUM( / SERIAL / BIGSERIAL / nextval / TIMESTAMP. The gate lost all discriminating power (false positive die) and real violations would be hidden in the noise. 2. Table count assertion 36 != actual DDL 37 (grep -c CREATE TABLE IF NOT EXISTS = 37; the 37th table is pbl_compile_task, pbl_compiler domain 4 to 5). Related: the AUTOINC assertion used a literal multi-space pattern and missed the UNSIGNED single space variant of pbl_compile_task (counted 36, accidentally matching the old assert); L210 post-create information_schema assertion was also 36. Fix: - Add scripts/ddl_gate_check.py: strip_sql_noise() single-pass char scanner removing block comments, -- / # line comments (-- must be followed by whitespace, so a--b is safe) and single/double quoted string literals (with doubled and backslash escapes) replaced by a placeholder, so COMMENT strings and DEFAULT strings no longer interfere. Assertions run only on the denoised DDL body: zero dialect keyword hits (prints keyword + line number + line text), CREATE TABLE count = 37 (plus duplicate table and non pbl_ prefix checks), BIGINT [UNSIGNED] NOT NULL AUTO_INCREMENT = 37 (table option AUTO_INCREMENT=n excluded first). Ships --selftest with 9 minimal unit checks (no pytest needed) and --dump-body for debugging; expected values are module constants. - Rewrite build.sh gate 6.4: step 4a runs --selftest first (if the scanner is not trustworthy its verdict is not accepted), step 4b runs the gate with --expect-tables 37 --expect-autoinc 37; return codes captured explicitly (not via "if ! cmd" which resets $? to 0); missing scanner file dies and forbids falling back to raw grep. Header comment and L210 actual table assertion synced 36 to 37. - pbls_tables.sql trailing contract self-check block: total tables 36 to 37 (pbl_compiler 4 to 5, new pbl_compile_task); the zero-hit section now describes the forbidden constructs in prose instead of listing SQL keywords verbatim (defence in depth against a future regression to naive grep). - projects/pbls/env/test.json gate.build_sh text "36 tables" to "37 tables" (json.load validity self-check passed). Verification (real local run output, details in projects/pbls/docs/02-develop/dev-notes-deploy-q4-build-gate64.md): - --selftest 9/9 PASS, including a negative fixture proving real violations are still caught, idempotency of denoising, and that -- / # inside strings do not truncate. - Real DDL: denoising strips 10974 chars / 89 line comments / 493 string literals; dialect zero hits, CREATE TABLE = 37, AUTO_INCREMENT = 37, RESULT PASS rc=0. - build.sh gate sections 0-4 sliced into a temporary harness and executed as a whole inside apps/pbls, rc=0 (6.1/6.2/6.4 all pass). Sections 7+ not entered: no MariaDB, no venv deps, port 9300 unavailable - honestly declared in the dev notes. - bash -n build.sh passes; no live assertion still says 36 (only 2 comment lines). No real table structure, app entry, conf/config.json, requirements or modules/pbl_* were modified.
This commit is contained in:
agent.develop
parent
73a1d3d656
commit
bc4aaa2463
37
build.sh
37
build.sh
@ -2,7 +2,7 @@
|
||||
# =====================================================================
|
||||
# PBL Agent OS 验证原型（Phase 0/1）—— 一键部署脚本（test 环境）
|
||||
# 端口唯一事实源：projects/pbls/env/test.json（9300）；本脚本不硬编码端口，从 env 读取
|
||||
# 门禁：建表 36 张 + 9 项种子全绿 + self_check PASS 74/74 + py_compile 全通过
|
||||
# 门禁：建表 37 张 + 9 项种子全绿 + self_check PASS 74/74 + py_compile 全通过
|
||||
# =====================================================================
|
||||
set -euo pipefail
|
||||
|
||||
@ -62,16 +62,30 @@ grep -q 'load_pybricks()' "${ENTRY}" || die "入口缺 load_pybricks() 显式调
|
||||
log "硬门禁 6.2 通过: ServerEnv()(L${ENV_LINE}) < load_rbac()(L${RBAC_LINE})，load_appbase/load_rbac/load_pybricks 均显式调用"
|
||||
|
||||
# ---- 4. DDL 方言核验（6.4：mariadb，BIGINT AUTO_INCREMENT，零 SERIAL/FK/ENUM/TIMESTAMP）----
|
||||
# 【Q4 修复 / deploy_test 第 3 轮实测】旧版直接对 DDL **原文** grep，存在两类假阳性：
|
||||
# (1) DDL 文末「契约自检」注释块为说明"零命中项"，逐字写出了
|
||||
# FOREIGN KEY / REFERENCES / ENUM( / SERIAL / BIGSERIAL / nextval / TIMESTAMP
|
||||
# （实测原文命中 3 行：L6 头部说明、L220 COMMENT 串内的 references、L891 契约自检块），
|
||||
# 旧 `grep -Eiq` 一律触发 die —— 注释/COMMENT 串被当成真实 DDL，门禁必然误杀；
|
||||
# (2) `CREATE TABLE` / `AUTO_INCREMENT` 计数同理会把注释行与 COMMENT '...' 串内字样算进去，
|
||||
# 表数不可信（且旧断言 36 与 DDL 实际 37 张不符，新增 pbl_compile_task 后未同步）。
|
||||
# 修复：改为「先剥 SQL 注释 + 字符串字面量（含 COMMENT 串），再在纯 DDL 正文上扫描」，
|
||||
# 由 scripts/ddl_gate_check.py 单遍字符扫描实现（非逐行正则，串内 -- / # 不会误判为注释），
|
||||
# 期望值与 DDL 实际表数对齐为 37（表数 / BIGINT AUTO_INCREMENT 主键数同为 37）。
|
||||
[ -f "${DDL_FILE}" ] || die "缺少 DDL ${DDL_FILE}"
|
||||
if grep -Eiq 'BIGSERIAL|SERIAL|nextval|FOREIGN KEY|REFERENCES|ENUM\(|TIMESTAMP' "${DDL_FILE}"; then
|
||||
grep -Ein 'BIGSERIAL|SERIAL|nextval|FOREIGN KEY|REFERENCES|ENUM\(|TIMESTAMP' "${DDL_FILE}" | head -20
|
||||
die "DDL 含非 mariadb 方言关键字（应为零命中）"
|
||||
fi
|
||||
TABLE_COUNT="$(grep -c 'CREATE TABLE IF NOT EXISTS' "${DDL_FILE}")"
|
||||
[ "${TABLE_COUNT}" = "36" ] || die "DDL 表数应为 36，实际 ${TABLE_COUNT}"
|
||||
AUTOINC_COUNT="$(grep -c 'BIGINT NOT NULL AUTO_INCREMENT' "${DDL_FILE}")"
|
||||
[ "${AUTOINC_COUNT}" = "36" ] || die "BIGINT AUTO_INCREMENT 主键应为 36 处，实际 ${AUTOINC_COUNT}"
|
||||
log "硬门禁 6.4 通过: mariadb 方言，36 表 / 36 处 BIGINT AUTO_INCREMENT，禁用关键字零命中"
|
||||
DDL_GATE="${APP_DIR}/scripts/ddl_gate_check.py"
|
||||
[ -f "${DDL_GATE}" ] || die "缺少 DDL 门禁扫描器 ${DDL_GATE}（6.4 依赖，禁止回退到原文 grep）"
|
||||
|
||||
# 4a. 扫描器自身可信度：先跑内置最小单测（剥注释/剥串/幂等/真实违规不漏报）
|
||||
SELFTEST_RC=0
|
||||
python3 "${DDL_GATE}" --selftest || SELFTEST_RC=$?
|
||||
[ "${SELFTEST_RC}" = "0" ] || die "DDL 门禁扫描器自检未通过（--selftest rc=${SELFTEST_RC}），扫描结果不可信，终止部署"
|
||||
|
||||
# 4b. 在降噪后的纯 DDL 正文上核验：方言零命中 + 表数 37 + BIGINT AUTO_INCREMENT 37
|
||||
GATE_RC=0
|
||||
python3 "${DDL_GATE}" --ddl "${DDL_FILE}" --expect-tables 37 --expect-autoinc 37 || GATE_RC=$?
|
||||
[ "${GATE_RC}" = "0" ] || die "硬门禁 6.4 未通过（ddl_gate_check.py rc=${GATE_RC}）：非 mariadb 方言关键字命中 / 表数≠37 / BIGINT AUTO_INCREMENT≠37，详见上方回显"
|
||||
log "硬门禁 6.4 通过: mariadb 方言（剥注释+COMMENT 串后零命中），37 表 / 37 处 BIGINT AUTO_INCREMENT"
|
||||
|
||||
# ---- 5. 依赖声明核验（6.5）----
|
||||
for dep in ahserver sqlor apppublic appbase rbac PyMySQL; do
|
||||
@ -207,7 +221,8 @@ mysql --protocol=TCP -h"${DB_HOST}" -P"${DB_PORT}" -u"${DB_USER}" -p"${DB_PASS}"
|
||||
--default-character-set=utf8mb4 "${DB_NAME}" < "${DDL_FILE}"
|
||||
ACTUAL_TABLES="$(mysql --protocol=TCP -h"${DB_HOST}" -P"${DB_PORT}" -u"${DB_USER}" -p"${DB_PASS}" \
|
||||
-N -B -e "SELECT COUNT(*) FROM information_schema.tables WHERE table_schema='${DB_NAME}' AND table_name LIKE 'pbl\\_%'" "${DB_NAME}")"
|
||||
[ "${ACTUAL_TABLES}" = "36" ] || die "建表数应为 36，实际 ${ACTUAL_TABLES}"
|
||||
# Q4 修复：与 DDL 实际表数对齐（36 → 37，新增 pbl_compile_task）
|
||||
[ "${ACTUAL_TABLES}" = "37" ] || die "建表数应为 37，实际 ${ACTUAL_TABLES}"
|
||||
log "建表完成: pbl_* 共 ${ACTUAL_TABLES} 张"
|
||||
|
||||
# ---- 11. 枚举数量契约自检（QC 退回意见 #15：必须真实可运行输出）----
|
||||
|
||||
@ -880,15 +880,20 @@ SET FOREIGN_KEY_CHECKS = 1;
|
||||
|
||||
-- =====================================================================
|
||||
-- 契约自检（供 build.sh / QC grep 核验）
|
||||
-- 表总数：36（pbl_governance 3 + pbl_blueprint 11 + pbl_template 2 + pbl_validation 3
|
||||
-- + pbl_compiler 4 + pbl_agent_runtime 4 + pbl_evidence 4 + pbl_assessment 3
|
||||
-- + pbl_domain_ext 1 + pbl_analytics 1 = 36）
|
||||
-- 表总数：37（pbl_governance 3 + pbl_blueprint 11 + pbl_template 2 + pbl_validation 3
|
||||
-- + pbl_compiler 5 + pbl_agent_runtime 4 + pbl_evidence 4 + pbl_assessment 3
|
||||
-- + pbl_domain_ext 1 + pbl_analytics 1 = 37）
|
||||
-- 注：pbl_compiler 由 4 增至 5（新增 pbl_compile_task），build.sh 门禁 6.4 表数断言同步 36→37
|
||||
-- append-only 表（11，无 update_time）：
|
||||
-- pbl_audit_log / pbl_blueprint_version / pbl_subobject_review / pbl_blueprint_publish
|
||||
-- pbl_validation_finding / pbl_compile_log / pbl_agent_message / pbl_agent_tool_call
|
||||
-- pbl_agent_arbitration / pbl_evidence_idem / pbl_analytics_snapshot
|
||||
-- 零命中项（mariadb 方言硬约束）：
|
||||
-- FOREIGN KEY / REFERENCES / ENUM( / SERIAL / BIGSERIAL / nextval / TIMESTAMP
|
||||
-- 零命中项（mariadb 方言硬约束）：外键约束、外键引用子句、枚举列类型、
|
||||
-- 自增序列类型（PG 风格）与序列取值函数、原生时间戳列类型 —— 一律禁用。
|
||||
-- ⚠ 本块刻意**不逐字写出**上述 SQL 关键字：旧版 build.sh 门禁 6.4 直接对 DDL 原文
|
||||
-- grep，这段说明文字自身就会触发假阳性 die（deploy_test 第 3 轮 Q4 实测）。
|
||||
-- 现门禁改为 scripts/ddl_gate_check.py「先剥注释 + 字符串字面量(含 COMMENT 串)，
|
||||
-- 再在纯 DDL 正文上扫描」，注释不再参与判定；此处仍保持不逐字罗列，双重保险。
|
||||
-- 全部索引以 tenant_id 打头；每表 id BIGINT NOT NULL AUTO_INCREMENT + xxx_code 唯一业务主键。
|
||||
-- =====================================================================
|
||||
|
||||
|
||||
371
scripts/ddl_gate_check.py
Normal file
371
scripts/ddl_gate_check.py
Normal file
@ -0,0 +1,371 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
build.sh 硬门禁 6.4 的 DDL 扫描器 —— 先剥「SQL 注释 + 字符串字面量(含 COMMENT 串)」，再扫表。
|
||||
|
||||
背景（deploy_test 第 3 轮实测 Q4）
|
||||
--------------------------------
|
||||
``scripts/ddl/pbls_tables.sql`` 文末有一段「契约自检」注释块，其中为说明"零命中项"，
|
||||
逐字写出了 ``FOREIGN KEY / REFERENCES / ENUM( / SERIAL / BIGSERIAL / nextval / TIMESTAMP``。
|
||||
旧版 build.sh 门禁 6.4 直接对 DDL **原文** grep，于是：
|
||||
|
||||
* 方言关键字扫描被这段**注释说明文字**触发 → 门禁假阳性 ``die``（注释被当成真实 DDL）；
|
||||
* ``CREATE TABLE`` 计数同理会把注释行 / ``COMMENT '...'`` 字符串里的字样算进表数 → 表数不可信；
|
||||
* ``BIGINT ... AUTO_INCREMENT`` 计数亦被注释里的示例文字污染。
|
||||
|
||||
因此本扫描器统一先做「降噪」，降噪后的文本才是"会被数据库真正执行的 DDL 正文"，
|
||||
所有断言只在它上面做：
|
||||
|
||||
1. 剥块注释 ``/* ... */``；
|
||||
2. 剥行注释 ``-- ...``（SQL 标准要求 ``--`` 后跟空白或行尾）与 ``# ...``；
|
||||
3. 剥字符串字面量 ``'...'`` / ``"..."``（含 ``''``/``""`` 双写转义与反斜杠转义）——
|
||||
``COMMENT '...'`` / ``COMMENT "..."`` 属于此类，整体抹为占位 ``''``（幂等），
|
||||
``DEFAULT 'pending'`` 同样抹为占位（不影响本门禁的任何断言）。
|
||||
|
||||
单遍字符扫描实现（非逐行正则），因此**不会**把字符串里的 ``--`` / ``#`` 误判成注释，
|
||||
也不会把注释里的引号误判成字符串起点。
|
||||
|
||||
用法
|
||||
----
|
||||
::
|
||||
|
||||
# 门禁核验（build.sh 调用形式）
|
||||
python3 scripts/ddl_gate_check.py --ddl scripts/ddl/pbls_tables.sql \
|
||||
--expect-tables 37 --expect-autoinc 37
|
||||
|
||||
# 内置最小单测（无需 pytest，验证剥注释/剥串逻辑本身正确）
|
||||
python3 scripts/ddl_gate_check.py --selftest
|
||||
|
||||
# 只打印降噪后的 DDL 正文（排障用）
|
||||
python3 scripts/ddl_gate_check.py --ddl <file> --dump-body
|
||||
|
||||
退出码：0 = 通过；1 = 断言未通过；2 = 参数/文件错误。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 期望值单一事实源：与 scripts/ddl/pbls_tables.sql 实际 DDL 对齐
|
||||
# 37 表 = pbl_governance 3 + pbl_blueprint 11 + pbl_template 2 + pbl_validation 3
|
||||
# + pbl_compiler 5 + pbl_agent_runtime 4 + pbl_evidence 4 + pbl_assessment 3
|
||||
# + pbl_domain_ext 1 + pbl_analytics 1
|
||||
# （Q4 修复：36 → 37，新增 pbl_compile_task）
|
||||
# --------------------------------------------------------------------------
|
||||
DEFAULT_EXPECT_TABLES = 37
|
||||
DEFAULT_EXPECT_AUTOINC = 37
|
||||
|
||||
# 非 mariadb 方言禁用关键字（大小写不敏感，与旧 grep -Eiq 行为一致）
|
||||
FORBIDDEN_PATTERNS = [
|
||||
("BIGSERIAL", r"BIGSERIAL"),
|
||||
("SERIAL", r"(?<!BIG)SERIAL"),
|
||||
("nextval", r"NEXTVAL\s*\("),
|
||||
("FOREIGN KEY", r"FOREIGN\s+KEY"),
|
||||
("REFERENCES", r"\bREFERENCES\b"),
|
||||
("ENUM(", r"\bENUM\s*\("),
|
||||
("TIMESTAMP", r"\bTIMESTAMP\b"),
|
||||
]
|
||||
|
||||
RE_CREATE_TABLE = re.compile(
|
||||
r"CREATE\s+TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?[`\"\']?([A-Za-z_][A-Za-z0-9_]*)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
RE_AUTOINC_PK = re.compile(
|
||||
r"\bBIGINT(?:\s*\(\s*\d+\s*\))?(?:\s+UNSIGNED)?\s+NOT\s+NULL\s+AUTO_INCREMENT",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
RE_TABLE_OPTION_AUTOINC = re.compile(r"\bAUTO_INCREMENT\s*=\s*\d+", re.IGNORECASE)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 降噪：剥注释 + 剥字符串字面量（单遍扫描）
|
||||
# --------------------------------------------------------------------------
|
||||
def strip_sql_noise(sql: str) -> str:
|
||||
"""剥掉 SQL 注释与字符串字面量，返回"纯 DDL 正文"。
|
||||
|
||||
- ``/* ... */`` 块注释 → 单个空格
|
||||
- ``-- ...``（后跟空白/行尾）与 ``# ...`` 行注释 → 删到行尾（保留换行，行号可追）
|
||||
- ``'...'`` / ``"..."`` 字符串（含 ``COMMENT '...'``）→ 占位 ``''``
|
||||
- 反引号标识符 ``\\`col\\``` 原样保留（其中不会出现注释/串语义）
|
||||
"""
|
||||
out = []
|
||||
i, n = 0, len(sql)
|
||||
while i < n:
|
||||
c = sql[i]
|
||||
nxt = sql[i + 1] if i + 1 < n else ""
|
||||
|
||||
# 行注释 --（SQL 标准：-- 后须为空白或行尾，避免误伤 a--b 之类）
|
||||
if c == "-" and nxt == "-" and (i + 2 >= n or sql[i + 2] in " \t\r\n"):
|
||||
j = sql.find("\n", i)
|
||||
i = n if j < 0 else j
|
||||
continue
|
||||
|
||||
# 行注释 #
|
||||
if c == "#":
|
||||
j = sql.find("\n", i)
|
||||
i = n if j < 0 else j
|
||||
continue
|
||||
|
||||
# 块注释 /* ... */
|
||||
if c == "/" and nxt == "*":
|
||||
j = sql.find("*/", i + 2)
|
||||
if j < 0:
|
||||
i = n
|
||||
else:
|
||||
i = j + 2
|
||||
out.append(" ")
|
||||
continue
|
||||
|
||||
# 字符串字面量 '...' / "..."（'' "" 双写转义 + 反斜杠转义）
|
||||
if c == "'" or c == '"':
|
||||
j = i + 1
|
||||
while j < n:
|
||||
if sql[j] == "\\":
|
||||
j += 2
|
||||
continue
|
||||
if sql[j] == c:
|
||||
if j + 1 < n and sql[j + 1] == c: # 双写转义
|
||||
j += 2
|
||||
continue
|
||||
break
|
||||
j += 1
|
||||
# 占位符用 '' （不带前后空格）——保证降噪**幂等**：对已降噪文本再降噪结果不变。
|
||||
# 仅当紧随其后的是单词字符/反引号时补一个空格，避免 ''NOT NULL 这类粘连
|
||||
# （补空格的条件在二次降噪时已不成立，故仍幂等）。
|
||||
i = min(j + 1, n)
|
||||
out.append("''")
|
||||
if i < n and (sql[i].isalnum() or sql[i] in "_`"):
|
||||
out.append(" ")
|
||||
continue
|
||||
|
||||
out.append(c)
|
||||
i += 1
|
||||
|
||||
return "".join(out)
|
||||
|
||||
|
||||
def _count_noise(sql: str) -> dict:
|
||||
"""统计降噪剥离量，作为门禁回显证据。"""
|
||||
body = strip_sql_noise(sql)
|
||||
return {
|
||||
"raw_lines": sql.count("\n") + 1,
|
||||
"body_lines": body.count("\n") + 1,
|
||||
"raw_chars": len(sql),
|
||||
"body_chars": len(body),
|
||||
"stripped_chars": len(sql) - len(body),
|
||||
"block_comments": len(re.findall(r"/\*", sql)),
|
||||
"line_comments": len(re.findall(r"(?:^|[^:\w])--(?=\s|$)|(^|\s)#", sql, re.MULTILINE)),
|
||||
"string_literals": body.count("''"),
|
||||
}
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 断言项
|
||||
# --------------------------------------------------------------------------
|
||||
def scan_tables(body: str):
|
||||
"""返回 (表名列表, 重复表名列表)。"""
|
||||
names = [m.group(1) for m in RE_CREATE_TABLE.finditer(body)]
|
||||
seen, dup = set(), []
|
||||
for nm in names:
|
||||
low = nm.lower()
|
||||
if low in seen and low not in dup:
|
||||
dup.append(low)
|
||||
seen.add(low)
|
||||
return names, dup
|
||||
|
||||
|
||||
def scan_autoinc(body: str) -> int:
|
||||
"""主键自增列数（``BIGINT [UNSIGNED] NOT NULL AUTO_INCREMENT``）。
|
||||
|
||||
表选项 ``) ENGINE=InnoDB AUTO_INCREMENT=1;`` 不计入。
|
||||
"""
|
||||
cleaned = RE_TABLE_OPTION_AUTOINC.sub(" ", body)
|
||||
return len(RE_AUTOINC_PK.findall(cleaned))
|
||||
|
||||
|
||||
def scan_forbidden(body: str):
|
||||
"""返回 [(关键字, 行号, 该行文本)]，在降噪后的正文上扫描。"""
|
||||
hits = []
|
||||
for lineno, line in enumerate(body.split("\n"), 1):
|
||||
for label, pat in FORBIDDEN_PATTERNS:
|
||||
if re.search(pat, line, re.IGNORECASE):
|
||||
hits.append((label, lineno, line.strip()[:160]))
|
||||
return hits
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 门禁主流程
|
||||
# --------------------------------------------------------------------------
|
||||
def run_gate(ddl_path: str, expect_tables: int, expect_autoinc: int,
|
||||
dump_body: bool = False) -> int:
|
||||
if not os.path.isfile(ddl_path):
|
||||
print("[ddl-gate][FATAL] DDL 文件不存在: %s" % ddl_path)
|
||||
return 2
|
||||
|
||||
with open(ddl_path, encoding="utf-8") as fh:
|
||||
raw = fh.read()
|
||||
|
||||
body = strip_sql_noise(raw)
|
||||
noise = _count_noise(raw)
|
||||
|
||||
print("[ddl-gate] DDL: %s" % ddl_path)
|
||||
print("[ddl-gate] 降噪: 原始 %d 行/%d 字符 → 纯 DDL 正文 %d 行/%d 字符"
|
||||
"（剥离 %d 字符；块注释 %d 处、行注释 %d 处、字符串字面量 %d 处）"
|
||||
% (noise["raw_lines"], noise["raw_chars"], noise["body_lines"],
|
||||
noise["body_chars"], noise["stripped_chars"], noise["block_comments"],
|
||||
noise["line_comments"], noise["string_literals"]))
|
||||
|
||||
raw_grep = len(re.findall(r"CREATE\s+TABLE\s+IF\s+NOT\s+EXISTS", raw, re.IGNORECASE))
|
||||
body_grep = len(re.findall(r"CREATE\s+TABLE\s+IF\s+NOT\s+EXISTS", body, re.IGNORECASE))
|
||||
print("[ddl-gate] 对照: 原文 'CREATE TABLE IF NOT EXISTS' = %d，降噪后 = %d"
|
||||
% (raw_grep, body_grep))
|
||||
|
||||
if dump_body:
|
||||
print("---- 降噪后 DDL 正文 BEGIN ----")
|
||||
print(body)
|
||||
print("---- 降噪后 DDL 正文 END ----")
|
||||
|
||||
ok = True
|
||||
|
||||
# (1) 非 mariadb 方言关键字：降噪后必须零命中
|
||||
forbidden = scan_forbidden(body)
|
||||
if forbidden:
|
||||
ok = False
|
||||
print("[ddl-gate][FAIL] 非 mariadb 方言关键字命中 %d 处（降噪后正文）:" % len(forbidden))
|
||||
for label, lineno, text in forbidden[:20]:
|
||||
print(" L%-5d %-14s | %s" % (lineno, label, text))
|
||||
else:
|
||||
print("[ddl-gate][PASS] 非 mariadb 方言关键字零命中"
|
||||
"（BIGSERIAL/SERIAL/nextval/FOREIGN KEY/REFERENCES/ENUM(/TIMESTAMP，"
|
||||
"注释与 COMMENT 串已剥离，不再假阳性）")
|
||||
|
||||
# (2) 表数断言
|
||||
names, dup = scan_tables(body)
|
||||
if len(names) != expect_tables:
|
||||
ok = False
|
||||
print("[ddl-gate][FAIL] CREATE TABLE 数 = %d，期望 %d" % (len(names), expect_tables))
|
||||
else:
|
||||
print("[ddl-gate][PASS] CREATE TABLE 数 = %d（期望 %d）" % (len(names), expect_tables))
|
||||
if dup:
|
||||
ok = False
|
||||
print("[ddl-gate][FAIL] 重复建表: %s" % ", ".join(dup))
|
||||
non_pbl = [n for n in names if not n.lower().startswith("pbl_")]
|
||||
if non_pbl:
|
||||
print("[ddl-gate][WARN] 非 pbl_ 前缀表 %d 张: %s" % (len(non_pbl), ", ".join(non_pbl)))
|
||||
|
||||
# (3) BIGINT AUTO_INCREMENT 主键数断言
|
||||
autoinc = scan_autoinc(body)
|
||||
if autoinc != expect_autoinc:
|
||||
ok = False
|
||||
print("[ddl-gate][FAIL] BIGINT [UNSIGNED] NOT NULL AUTO_INCREMENT = %d，期望 %d"
|
||||
% (autoinc, expect_autoinc))
|
||||
else:
|
||||
print("[ddl-gate][PASS] BIGINT [UNSIGNED] NOT NULL AUTO_INCREMENT = %d（期望 %d）"
|
||||
% (autoinc, expect_autoinc))
|
||||
|
||||
print("[ddl-gate] 表清单(%d): %s" % (len(names), ", ".join(names)))
|
||||
print("[ddl-gate] RESULT: %s" % ("PASS" if ok else "FAIL"))
|
||||
return 0 if ok else 1
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 最小单测（无需 pytest；build.sh 在跑门禁前先跑它，证明扫描器本身可信）
|
||||
# --------------------------------------------------------------------------
|
||||
SELFTEST_DDL = """\
|
||||
-- 本行注释里故意写 CREATE TABLE IF NOT EXISTS pbl_fake_comment (不应被计数)
|
||||
# 井号注释里也写 CREATE TABLE pbl_fake_hash (不应被计数)
|
||||
/* 块注释里写 FOREIGN KEY REFERENCES ENUM( SERIAL BIGSERIAL nextval TIMESTAMP
|
||||
以及 CREATE TABLE pbl_fake_block (均不应被命中/计数) */
|
||||
CREATE TABLE IF NOT EXISTS `pbl_alpha` (
|
||||
`tenant_id` VARCHAR(64) NOT NULL COMMENT '租户ID(强制打头)',
|
||||
`id` BIGINT NOT NULL AUTO_INCREMENT COMMENT '主键',
|
||||
`note` VARCHAR(200) NULL COMMENT "串里写 CREATE TABLE pbl_fake_str / TIMESTAMP / FOREIGN KEY 都不应命中",
|
||||
`status` VARCHAR(32) NOT NULL DEFAULT 'pending' COMMENT 'appcodes:pbl_x(a/b)',
|
||||
`dash` VARCHAR(32) NULL DEFAULT 'a--b # c' COMMENT '串内 -- 与 # 不是注释起点',
|
||||
PRIMARY KEY (`tenant_id`, `id`)
|
||||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='表注释里也有 CREATE TABLE pbl_fake_tbl';
|
||||
|
||||
CREATE TABLE IF NOT EXISTS pbl_beta (
|
||||
tenant_id VARCHAR(64) NOT NULL COMMENT '租户ID',
|
||||
id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT COMMENT '主键',
|
||||
PRIMARY KEY (tenant_id, id)
|
||||
) ENGINE=InnoDB AUTO_INCREMENT=1;
|
||||
"""
|
||||
|
||||
SELFTEST_BAD_DDL = """\
|
||||
CREATE TABLE IF NOT EXISTS pbl_bad (
|
||||
id BIGSERIAL PRIMARY KEY,
|
||||
ts TIMESTAMP NOT NULL,
|
||||
kind VARCHAR(8) NOT NULL DEFAULT 'x' COMMENT '枚举',
|
||||
CONSTRAINT fk_x FOREIGN KEY (id) REFERENCES pbl_alpha (id)
|
||||
);
|
||||
"""
|
||||
|
||||
|
||||
def selftest() -> int:
|
||||
fails = []
|
||||
|
||||
def check(name, got, want):
|
||||
if got != want:
|
||||
fails.append("%s: got=%r want=%r" % (name, got, want))
|
||||
print("[selftest][FAIL] %-46s got=%r want=%r" % (name, got, want))
|
||||
else:
|
||||
print("[selftest][PASS] %-46s = %r" % (name, got))
|
||||
|
||||
body = strip_sql_noise(SELFTEST_DDL)
|
||||
|
||||
# 1) 注释/COMMENT 串里的 CREATE TABLE 不被计数；真实 2 张表被计数
|
||||
names, dup = scan_tables(body)
|
||||
check("注释+COMMENT串干扰下的表数", len(names), 2)
|
||||
check("表名集合", sorted(n.lower() for n in names), ["pbl_alpha", "pbl_beta"])
|
||||
check("重复表", dup, [])
|
||||
|
||||
# 2) 注释里逐字写出的方言关键字不触发假阳性
|
||||
check("注释内方言关键字假阳性", scan_forbidden(body), [])
|
||||
|
||||
# 3) AUTO_INCREMENT 主键计数：2 处（表选项 AUTO_INCREMENT=1 不计）
|
||||
check("AUTO_INCREMENT 主键数", scan_autoinc(body), 2)
|
||||
|
||||
# 4) 字符串里的 -- / # 不被当成注释起点（否则 dash 列后半段会被吃掉）
|
||||
check("串内 -- 未截断(列仍存在)", "dash" in body, True)
|
||||
check("串内 # 未截断(PRIMARY KEY 仍在)", "PRIMARY KEY" in body, True)
|
||||
|
||||
# 5) 真实违规必须被抓到（不能因为降噪而漏报）
|
||||
bad_body = strip_sql_noise(SELFTEST_BAD_DDL)
|
||||
bad_labels = sorted({h[0] for h in scan_forbidden(bad_body)})
|
||||
check("真实违规关键字被抓到", bad_labels,
|
||||
["BIGSERIAL", "FOREIGN KEY", "REFERENCES", "TIMESTAMP"])
|
||||
|
||||
# 6) 降噪幂等（对已降噪文本再降噪不应改变结果）
|
||||
check("降噪幂等", strip_sql_noise(body), body)
|
||||
|
||||
print("[selftest] RESULT: %s" % ("PASS" if not fails else "FAIL (%d)" % len(fails)))
|
||||
return 0 if not fails else 1
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
def main(argv=None) -> int:
|
||||
ap = argparse.ArgumentParser(
|
||||
description="build.sh 门禁 6.4：剥注释/COMMENT 串后再扫 DDL（方言 + 表数 + 自增主键）")
|
||||
ap.add_argument("--ddl", help="DDL 文件路径")
|
||||
ap.add_argument("--expect-tables", type=int, default=DEFAULT_EXPECT_TABLES,
|
||||
help="期望 CREATE TABLE 数（默认 %d）" % DEFAULT_EXPECT_TABLES)
|
||||
ap.add_argument("--expect-autoinc", type=int, default=DEFAULT_EXPECT_AUTOINC,
|
||||
help="期望 BIGINT AUTO_INCREMENT 主键数（默认 %d）" % DEFAULT_EXPECT_AUTOINC)
|
||||
ap.add_argument("--dump-body", action="store_true", help="打印降噪后的 DDL 正文（排障）")
|
||||
ap.add_argument("--selftest", action="store_true", help="只跑内置最小单测")
|
||||
args = ap.parse_args(argv)
|
||||
|
||||
if args.selftest:
|
||||
return selftest()
|
||||
if not args.ddl:
|
||||
ap.error("需要 --ddl <path>（或使用 --selftest）")
|
||||
return run_gate(args.ddl, args.expect_tables, args.expect_autoinc, args.dump_body)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Loading…
x
Reference in New Issue
Block a user