279 lines
11 KiB
Python
279 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""M11b-2b 双向列集 diff：world_sync 消费方代码引用的列 vs pbl_runtime_ext 权威列集。
|
||
|
||
PM 裁决 B 的执行要求：「引用后跑双向列集 diff 留证（world_sync 消费方代码若依赖旧结构，
|
||
适配到权威结构）」。
|
||
|
||
方向定义：
|
||
方向 1（代码 -> 权威）：SQL 字面量中出现的列名必须全部存在于权威列集（缺则 FAIL）；
|
||
方向 2（权威 -> 代码）：权威列集中未被任何 SQL 引用的列，打印为 informational
|
||
（由自增/默认值/写入方填充，不判失败）。
|
||
|
||
M11b-2b-B1 修正（标识符误判）：旧版把「文档字符串里叙述的旧列名/类名/函数名」以及
|
||
DB-API 属性、索引名等 Python 侧标识符当成 SQL 列名，导致方向 1 误报 FAIL
|
||
（AuthorityResolutionError、ConcurrentStateConflict、write_event_with_state、
|
||
_assert_contract、execute、format、json、paramstyle、pyformat、qmark、rowcount、
|
||
expected_version、uk_es、event_id/updated_by_event 的文档叙述等）。修正为四层过滤：
|
||
L1 语句级：字面量必须是真 SQL 语句（以 SQL 动词开头或含 INSERT INTO ... / SELECT ... FROM 签名），
|
||
且排除 ast 识别出的 docstring（文档叙述不是代码引用）；
|
||
L2 黑名单：KNOWN_NON_COLUMN_IDENTIFIERS（类名/函数名/DB-API 属性/驱动 paramstyle/索引名/内置模块名）；
|
||
L3 停用词：STOPWORDS（SQL 语法词、函数词、驱动方法名）；
|
||
L4 形态规则：首字母大写=类名、下划线前缀=私有成员、uk_/ix_/idx_/uq_ 前缀=索引名。
|
||
|
||
用法：python3 modules/world_sync/scripts/m11b2b_column_diff.py
|
||
"""
|
||
|
||
import ast
|
||
import io
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
import tokenize
|
||
|
||
CONSUMERS = [
|
||
"modules/world_sync/world_sync/pbl_runtime_sql.py",
|
||
"modules/world_sync/world_sync/pbl_runtime_tx.py",
|
||
"modules/world_sync/world_sync/pbl_runtime_schema.py",
|
||
]
|
||
|
||
AUTHORITY = {
|
||
"pbl_runtime_event": "modules/pbl_runtime_ext/models/pbl_runtime_event.json",
|
||
"pbl_entity_state": "modules/pbl_runtime_ext/models/pbl_entity_state.json",
|
||
}
|
||
|
||
SQL_HINT = re.compile(r"\b(select|insert\s+into|update|delete\s+from|on\s+duplicate)\b", re.IGNORECASE)
|
||
|
||
# L1a：真 SQL 语句的形态签名
|
||
SQL_STATEMENT_START = re.compile(
|
||
r"^\s*(select|insert\s+into|update|delete\s+from|with|for\s+update|lock\s+in\s+share|"
|
||
r"begin|commit|rollback|set\s+transaction|create\s+table|alter\s+table)\b",
|
||
re.IGNORECASE)
|
||
SQL_STATEMENT_SIGNATURE = re.compile(
|
||
r"(\binsert\s+into\s+[\w`.]+\s*\()|(\bselect\b[\s\S]+\bfrom\b)|(\bupdate\s+[\w`.]+\s+set\b)|"
|
||
r"(\bdelete\s+from\b)",
|
||
re.IGNORECASE)
|
||
|
||
# SQL 语法词/函数词/驱动方法名，避免误判为列名
|
||
STOPWORDS = {
|
||
"select", "from", "where", "insert", "into", "values", "update", "set", "delete",
|
||
"and", "or", "not", "null", "is", "in", "on", "as", "by", "order", "group", "limit",
|
||
"join", "left", "right", "inner", "outer", "distinct", "count", "max", "min", "sum",
|
||
"now", "current_timestamp", "true", "false", "begin", "commit", "rollback", "with",
|
||
"returning", "if", "exists", "duplicate", "key", "auto_increment", "using", "lock",
|
||
"share", "for", "transaction", "level", "read", "committed",
|
||
# 驱动/内置模块与游标方法名（f-string 残留或文档叙述会被抓成标识符）
|
||
"json", "format", "execute", "executemany", "cursor", "connection", "fetchall",
|
||
"fetchone", "description", "interval", "utc_timestamp", "date_add", "rowcount",
|
||
}
|
||
|
||
# L2：已确认的非列标识符（Python 类名 / 函数名 / DB-API 属性 / 索引名 / 内置模块名）。
|
||
KNOWN_NON_COLUMN_IDENTIFIERS = {
|
||
# 类名 / 异常名
|
||
"AuthorityResolutionError", "ConcurrentStateConflict", "ConcurrentConflictError",
|
||
"WorldSyncError", "RuntimeContractError",
|
||
# 函数 / 方法名
|
||
"write_event_with_state", "_assert_contract", "execute", "executemany", "format",
|
||
"new_event_id", "new_event_uid", "get_module_dbname", "load_world_sync",
|
||
"columns_for", "event_columns", "state_columns",
|
||
# 内置模块 / 对象名
|
||
"json", "os", "sys", "re", "time", "datetime", "ast", "io",
|
||
# DB-API 2.0 属性与参数风格
|
||
"paramstyle", "pyformat", "qformat", "qmark", "numeric", "named", "string_format",
|
||
"expected_version", "rowcount", "lastrowid",
|
||
# 索引名（前缀形态之外的显式登记）
|
||
"uk_es", "uk_pre", "ix_es", "ix_tx_group", "ix_created_at",
|
||
}
|
||
|
||
# L4：形态规则
|
||
INDEX_NAME_PREFIX = re.compile(r"^(uk_|ix_|idx_|uq_)")
|
||
|
||
TABLE_WORDS = {"pbl_runtime_event", "pbl_entity_state"}
|
||
|
||
|
||
def find_root():
|
||
here = os.path.dirname(os.path.abspath(__file__))
|
||
cur = here
|
||
for _ in range(10):
|
||
if os.path.isdir(os.path.join(cur, "modules")) and os.path.isdir(os.path.join(cur, "apps")):
|
||
return cur
|
||
parent = os.path.dirname(cur)
|
||
if parent == cur:
|
||
break
|
||
cur = parent
|
||
return os.path.abspath(os.path.join(here, os.pardir, os.pardir, os.pardir))
|
||
|
||
|
||
def authority_columns(root, table):
|
||
with open(os.path.join(root, AUTHORITY[table]), "r", encoding="utf-8") as fh:
|
||
data = json.load(fh)
|
||
return [f.get("name") for f in (data.get("fields") or []) if isinstance(f, dict) and f.get("name")]
|
||
|
||
|
||
def docstrings_of(path):
|
||
"""L1b：收集模块/类/函数的 docstring 文本，文档叙述不参与列集比对。"""
|
||
try:
|
||
with open(path, "r", encoding="utf-8") as fh:
|
||
tree = ast.parse(fh.read(), filename=path)
|
||
except Exception: # noqa: BLE001
|
||
return set()
|
||
out = set()
|
||
for node in ast.walk(tree):
|
||
if isinstance(node, (ast.Module, ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)):
|
||
doc = ast.get_docstring(node, clean=False)
|
||
if doc:
|
||
out.add(doc.strip())
|
||
return out
|
||
|
||
|
||
def looks_like_sql_statement(val):
|
||
"""L1a：必须以 SQL 动词开头，或具备完整语句签名（排除散文式叙述）。"""
|
||
if not SQL_HINT.search(val):
|
||
return False
|
||
if SQL_STATEMENT_START.match(val):
|
||
return True
|
||
return bool(SQL_STATEMENT_SIGNATURE.search(val))
|
||
|
||
|
||
def sql_literals(path):
|
||
"""用 tokenize 取出文件中的字符串字面量，仅保留真 SQL 语句（排除 docstring）。"""
|
||
with open(path, "rb") as fh:
|
||
raw = fh.read()
|
||
docs = docstrings_of(path)
|
||
out = []
|
||
try:
|
||
for tok in tokenize.tokenize(io.BytesIO(raw).readline):
|
||
if tok.type != tokenize.STRING:
|
||
continue
|
||
try:
|
||
val = ast_literal_eval(tok.string)
|
||
except Exception: # noqa: BLE001
|
||
continue
|
||
if not isinstance(val, str):
|
||
continue
|
||
if val.strip() in docs: # L1b 文档叙述
|
||
continue
|
||
if not looks_like_sql_statement(val): # L1a 非语句形态
|
||
continue
|
||
out.append(val)
|
||
except (tokenize.TokenError, IndentationError, SyntaxError):
|
||
pass
|
||
return out
|
||
|
||
|
||
def ast_literal_eval(src):
|
||
"""轻量解析字符串字面量（含 f-string 前缀），失败返回 None。"""
|
||
s = src
|
||
m = re.match(r"^([a-zA-Z]*)", s)
|
||
prefix = m.group(1).lower() if m else ""
|
||
body = s[len(prefix):]
|
||
if "f" in prefix: # f-string：把 {expr} 替换为占位，仅取字面部分
|
||
body = re.sub(r"\{[^{}]*\}", " ", body)
|
||
if "b" in prefix:
|
||
body = body.encode("latin-1", "ignore").decode("unicode_escape")
|
||
try:
|
||
import ast as _ast
|
||
return _ast.literal_eval(body) if body[:1] in "\"'" else body
|
||
except Exception: # noqa: BLE001
|
||
return body
|
||
try:
|
||
import ast as _ast
|
||
return _ast.literal_eval(body)
|
||
except Exception: # noqa: BLE001
|
||
return None
|
||
|
||
|
||
def is_non_column_identifier(tok):
|
||
"""L2/L3/L4 标识符过滤：黑名单 -> 停用词 -> 形态规则。"""
|
||
if not tok:
|
||
return True
|
||
if tok in KNOWN_NON_COLUMN_IDENTIFIERS:
|
||
return True
|
||
if tok.lower() in STOPWORDS:
|
||
return True
|
||
if tok[0].isupper(): # 类名/异常名（列名规范为小写下划线）
|
||
return True
|
||
if tok.startswith("_"): # 私有成员
|
||
return True
|
||
if INDEX_NAME_PREFIX.match(tok.lower()): # 索引名
|
||
return True
|
||
return False
|
||
|
||
|
||
def columns_from_sql(sql):
|
||
cols = set()
|
||
cols.update(re.findall(r"`([a-zA-Z_][a-zA-Z_0-9]*)`", sql))
|
||
for grp in re.findall(r"INSERT\s+INTO\s+[\w`.]+\s*\(([^)]*)\)", sql, re.IGNORECASE):
|
||
for tok in grp.split(","):
|
||
tok = tok.strip().strip("`")
|
||
if re.fullmatch(r"[a-zA-Z_][a-zA-Z_0-9]*", tok or ""):
|
||
cols.add(tok)
|
||
for grp in re.findall(r"\bSET\s+(.+?)(?:\bWHERE\b|$)", sql, re.IGNORECASE | re.DOTALL):
|
||
cols.update(re.findall(r"([a-zA-Z_][a-zA-Z_0-9]*)\s*=", grp))
|
||
for grp in re.findall(r"\bWHERE\s+(.+?)$", sql, re.IGNORECASE | re.DOTALL):
|
||
for tok in re.findall(r"([a-zA-Z_][a-zA-Z_0-9]*)\s*(?:=|<|>|!|in\b|is\b)", grp, re.IGNORECASE):
|
||
cols.add(tok)
|
||
for grp in re.findall(r"\b(?:ORDER\s+BY|GROUP\s+BY)\s+([\w`,\s.]+)", sql, re.IGNORECASE):
|
||
for tok in grp.split(","):
|
||
tok = tok.strip().strip("`").split()[-1] if tok.strip() else ""
|
||
if re.fullmatch(r"[a-zA-Z_][a-zA-Z_0-9]*", tok or ""):
|
||
cols.add(tok)
|
||
for tok in re.findall(r"\b(?:pbl_runtime_event|pbl_entity_state)\.([a-zA-Z_][a-zA-Z_0-9]*)", sql):
|
||
cols.add(tok)
|
||
return {c for c in cols if c not in TABLE_WORDS and not is_non_column_identifier(c)}
|
||
|
||
|
||
def main():
|
||
root = find_root()
|
||
print("workspace root: %s" % root)
|
||
|
||
sql_total = []
|
||
ident = set()
|
||
scanned = []
|
||
for rel in CONSUMERS:
|
||
path = os.path.join(root, rel)
|
||
if not os.path.isfile(path):
|
||
print(" SKIP (absent): %s" % rel)
|
||
continue
|
||
scanned.append(rel)
|
||
lits = sql_literals(path)
|
||
sql_total.extend(lits)
|
||
for s in lits:
|
||
ident |= columns_from_sql(s)
|
||
print("consumer files scanned: %d -> %s" % (len(scanned), scanned))
|
||
print("SQL statement literals kept: %d, distinct column identifiers: %d" % (len(sql_total), len(ident)))
|
||
print("identifier filter layers: L1 statement-gate+docstring-exclusion, "
|
||
"L2 known_non_column=%d, L3 stopwords=%d, L4 shape_rules=3"
|
||
% (len(KNOWN_NON_COLUMN_IDENTIFIERS), len(STOPWORDS)))
|
||
|
||
auth_cols = {}
|
||
for table in AUTHORITY:
|
||
auth_cols[table] = authority_columns(root, table)
|
||
print("\nauthority columns [%s] (%d cols): %s" % (table, len(auth_cols[table]), auth_cols[table]))
|
||
|
||
print("\n--- 方向 1: 代码 SQL 引用列 ⊆ 权威列集（必须全部成立）---")
|
||
bad = []
|
||
for name in sorted(ident):
|
||
hit = [t for t in auth_cols if name in auth_cols[t]]
|
||
print(" %-22s %s %s" % (name, "OK " if hit else "MISS", ("->" + ",".join(hit)) if hit else "(不在任何权威列集)"))
|
||
if not hit:
|
||
bad.append(name)
|
||
|
||
print("\n--- 方向 2: 权威列集中未被 SQL 显式引用者（informational）---")
|
||
for table in AUTHORITY:
|
||
unused = [c for c in auth_cols[table] if c not in ident]
|
||
print(" [%s] unreferenced=%d %s" % (table, len(unused), unused))
|
||
|
||
print("\ncode columns checked: %d, missing from authority: %d" % (len(ident), len(bad)))
|
||
if bad:
|
||
print("MISSING: %s" % bad)
|
||
print("RESULT: FAIL")
|
||
return 1
|
||
print("RESULT: PASS")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|