466 lines
20 KiB
Python
466 lines
20 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""M1b 内容级自检工具（QC 取证用）。
|
||
|
||
用途：把 QC 退回意见 #3/#4/#6 要求的「内容级核查」变成可重复执行、可留证据的
|
||
机械检查。运行：
|
||
|
||
cd modules/pbl_blueprint && python3 tools/m1b_selfcheck.py
|
||
|
||
退出码 0 = 全部通过；非 0 = 有 FAIL 项（打印明细）。
|
||
|
||
检查项：
|
||
A. DDL 合规（sql/m1b_ddl.sql）
|
||
A1 恰好 4 条 CREATE TABLE IF NOT EXISTS，且为 4 张目标表
|
||
A2 零 FOREIGN KEY
|
||
A3 零 PostgreSQL 方言（BIGSERIAL / nextval / SERIAL）
|
||
A4 零对既有基表的 DDL/DML 动词（ALTER/UPDATE/DELETE/DROP/TRUNCATE/INSERT）
|
||
A5 每个 UNIQUE KEY 首列为 tenant_key（规避 NULL 不去重）
|
||
A6 mariadb 方言要素齐备（InnoDB + utf8mb4 + AUTO_INCREMENT）
|
||
B. 表模型合规（models/m1b/*.json）
|
||
B1 4 个文件均可 json.load
|
||
B2 四段式：summary(数组,含 primary) / fields(数组) / indexes(数组) / codes(数组)
|
||
B3 models 与 DDL 表名一一对应（双向）
|
||
B4 models 字段集合与 DDL 列集合一一对应（双向，逐表）
|
||
B5 models 索引集合与 DDL 索引集合一一对应（逐表，含 unique 标记）
|
||
C. CRUD 定义合规（json/m1b/*.json）
|
||
C1 4 个文件均可 json.load
|
||
C2 含 tblname 且指向本模块 4 表之一
|
||
C3 含 params.editable（QC#3 指出的格式偏差）
|
||
C4 new/update/delete_data_url 在 params 顶层（非仅嵌套在 editable 内）
|
||
C5 json/ 下不含表定义特征键（summary/fields/indexes）——即 json/ 只放 CRUD
|
||
D. Python 源码内容级核查（pbl_blueprint/m1b_*.py）
|
||
D1 零硬编码库名（无 DBNAME = 'xxx' / dbname = 'xxx' 字面量赋值）
|
||
D2 取库名走 get_module_dbname
|
||
D3 只用 sqlor 标准 API（sor.C/U/D/R/I/sqlExe/sqlPaging），无编造 API
|
||
D4 全部 py_compile 通过
|
||
D5 每个 .py 有效语句数 > 0（非空壳）
|
||
E. 三处同步注册核验（QC#4）
|
||
E1 m1b_api.py 定义 register_m1b_routes
|
||
E2 __init__.py 导出 register_m1b_routes（import 行）
|
||
E3 init.py 在 load_pbl_blueprint 内引用 register_m1b_routes（env 注册/调用）
|
||
E4 M1b 每个对外函数在三处齐备（实现 / __init__ 导出 / init.py 注册）
|
||
F. M1a 既有代码零越权改动（QC#6）
|
||
F1 git diff 中 M1a 文件（非 m1b_* / 非 models|json|sql|tests|docs/tools 的
|
||
既有 .py）只允许「追加式」改动：新增行不得删除既有 M1a 函数定义行
|
||
"""
|
||
|
||
import ast
|
||
import json
|
||
import os
|
||
import re
|
||
import subprocess
|
||
import sys
|
||
|
||
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
M1B_TABLES = [
|
||
"pbl_blueprint_template",
|
||
"pbl_blueprint_ref",
|
||
"pbl_subobject_ext",
|
||
"pbl_ext_field_def",
|
||
]
|
||
|
||
DDL_REL = os.path.join("sql", "m1b_ddl.sql")
|
||
MODELS_REL = os.path.join("pbl_blueprint", "models", "m1b")
|
||
JSON_REL = os.path.join("pbl_blueprint", "json", "m1b")
|
||
|
||
# 编造的 sqlor API（module-development-spec 明确：只有 C/U/D/R/I/sqlExe）
|
||
FAKE_SQLOR = [
|
||
"sqlor.save", "sor.save", "sqlor.list", "sor.list", "sqlor.one", "sor.one",
|
||
"sqlor.delete", "sor.delete", "sqlor.insert", "sor.insert", "sqlor.query",
|
||
"sor.query", "sqlor.update", "sor.update", "sqlor.create", "sor.create",
|
||
"sqlor.select", "sor.select", "sqlor.find", "sor.find", "sqlor.get",
|
||
"sor.get_one", "sor.fetch", "sqlor.fetch",
|
||
]
|
||
|
||
# 允许的 sqlor 调用（白名单，用于 D3 正向取证）
|
||
OK_SQLOR = re.compile(r"\bsor\.(C|U|D|R|I|sqlExe|sqlPaging|sqlorContext)\s*\(")
|
||
|
||
results = []
|
||
|
||
|
||
def record(check_id, ok, detail):
|
||
results.append((check_id, bool(ok), detail))
|
||
flag = "PASS" if ok else "FAIL"
|
||
print("[%s] %s :: %s" % (flag, check_id, detail))
|
||
|
||
|
||
def p(*parts):
|
||
return os.path.join(REPO, *parts)
|
||
|
||
|
||
def read(rel):
|
||
with open(p(rel), "r", encoding="utf-8") as f:
|
||
return f.read()
|
||
|
||
|
||
def strip_sql_comments(sql):
|
||
"""去掉 -- 行注释与 /* */ 块注释，避免注释文本干扰动词扫描。"""
|
||
sql = re.sub(r"/\*.*?\*/", " ", sql, flags=re.S)
|
||
out = []
|
||
for line in sql.splitlines():
|
||
idx = line.find("--")
|
||
if idx >= 0:
|
||
line = line[:idx]
|
||
out.append(line)
|
||
return "\n".join(out)
|
||
|
||
|
||
# ---------------------------------------------------------------- A. DDL
|
||
def check_ddl():
|
||
if not os.path.exists(p(DDL_REL)):
|
||
record("A0", False, "sql/m1b_ddl.sql 不存在")
|
||
return None
|
||
raw = read(DDL_REL)
|
||
body = strip_sql_comments(raw)
|
||
|
||
creates = re.findall(
|
||
r"CREATE\s+TABLE\s+IF\s+NOT\s+EXISTS\s+`?(\w+)`?", body, flags=re.I)
|
||
record("A1", len(creates) == 4 and sorted(creates) == sorted(M1B_TABLES),
|
||
"CREATE TABLE IF NOT EXISTS 共 %d 条，表=%s" % (len(creates), creates))
|
||
|
||
fk = len(re.findall(r"FOREIGN\s+KEY", body, flags=re.I))
|
||
record("A2", fk == 0, "FOREIGN KEY 出现 %d 次（应为 0）" % fk)
|
||
|
||
pg = re.findall(r"\b(BIGSERIAL|SERIAL|nextval)\b", body, flags=re.I)
|
||
record("A3", not pg, "PostgreSQL 方言命中 %s（应为空）" % (pg or "无"))
|
||
|
||
# A4：任何针对既有基表的 DDL/DML 动词；同时全文（去注释）不得出现这些动词
|
||
verbs = re.findall(
|
||
r"\b(ALTER|UPDATE|DELETE|DROP|TRUNCATE|INSERT|REPLACE)\b", body, flags=re.I)
|
||
base_hits = re.findall(
|
||
r"\b(CREATE|ALTER|UPDATE|DELETE|DROP|TRUNCATE|INSERT)\b[^;]{0,120}?"
|
||
r"\b(world|scene|entity|script|world_snapshot|scense)\b",
|
||
body, flags=re.I)
|
||
record("A4", not verbs and not base_hits,
|
||
"去注释后 DDL/DML 动词命中 %s；基表语句命中 %s（均应为空）"
|
||
% (sorted(set(v.upper() for v in verbs)) or "无", base_hits or "无"))
|
||
|
||
# A5：每个 UNIQUE KEY 首列必须是 tenant_key
|
||
bad_uk = []
|
||
uks = re.findall(r"UNIQUE\s+KEY\s+`?(\w+)`?\s*\(([^)]*)\)", body, flags=re.I)
|
||
for name, cols in uks:
|
||
first = cols.split(",")[0].strip().strip("`").strip()
|
||
if first != "tenant_key":
|
||
bad_uk.append("%s(首列=%s)" % (name, first))
|
||
record("A5", uks and not bad_uk,
|
||
"UNIQUE KEY 共 %d 个，首列非 tenant_key 的：%s" % (len(uks), bad_uk or "无"))
|
||
|
||
need = ["ENGINE=InnoDB", "utf8mb4", "AUTO_INCREMENT"]
|
||
miss = [n for n in need if n.lower() not in body.lower()]
|
||
record("A6", not miss, "mariadb 方言要素缺失 %s（应为空）" % (miss or "无"))
|
||
|
||
return body
|
||
|
||
|
||
# ------------------------------------------------------- B. models 四段式
|
||
def parse_ddl_tables(body):
|
||
"""从 DDL 正文解析 {table: {"cols": [...], "indexes": {name: (cols, unique)}}}"""
|
||
tables = {}
|
||
for m in re.finditer(
|
||
r"CREATE\s+TABLE\s+IF\s+NOT\s+EXISTS\s+`?(\w+)`?\s*\((.*?)\n\)\s*ENGINE",
|
||
body, flags=re.I | re.S):
|
||
tname, inner = m.group(1), m.group(2)
|
||
cols, idx = [], {}
|
||
for line in inner.splitlines():
|
||
s = line.strip().rstrip(",").strip()
|
||
if not s:
|
||
continue
|
||
up = s.upper()
|
||
if up.startswith("PRIMARY KEY"):
|
||
continue
|
||
if up.startswith("UNIQUE KEY") or up.startswith("KEY ") or up.startswith("INDEX "):
|
||
nm = re.match(r"(?:UNIQUE\s+)?(?:KEY|INDEX)\s+`?(\w+)`?\s*\(([^)]*)\)",
|
||
s, flags=re.I)
|
||
if nm:
|
||
cs = [c.strip().strip("`") for c in nm.group(2).split(",")]
|
||
idx[nm.group(1)] = (cs, up.startswith("UNIQUE"))
|
||
continue
|
||
cm = re.match(r"`(\w+)`", s)
|
||
if cm:
|
||
cols.append(cm.group(1))
|
||
tables[tname] = {"cols": cols, "indexes": idx}
|
||
return tables
|
||
|
||
|
||
def check_models(ddl_tables):
|
||
mdir = p(MODELS_REL)
|
||
if not os.path.isdir(mdir):
|
||
record("B1", False, "models/m1b/ 目录不存在")
|
||
return
|
||
files = sorted(f for f in os.listdir(mdir) if f.endswith(".json"))
|
||
record("B1", len(files) == 4, "models/m1b/ JSON 文件 %d 个：%s" % (len(files), files))
|
||
|
||
model_tables = {}
|
||
four_ok, four_bad = True, []
|
||
for fn in files:
|
||
try:
|
||
with open(os.path.join(mdir, fn), "r", encoding="utf-8") as f:
|
||
d = json.load(f)
|
||
except Exception as e:
|
||
four_ok = False
|
||
four_bad.append("%s: json.load 失败 %s" % (fn, e))
|
||
continue
|
||
for seg in ("summary", "fields", "indexes", "codes"):
|
||
if not isinstance(d.get(seg), list):
|
||
four_ok = False
|
||
four_bad.append("%s: 缺 %s 数组" % (fn, seg))
|
||
sm = d.get("summary") or []
|
||
if not (sm and isinstance(sm[0], dict) and sm[0].get("primary")):
|
||
four_ok = False
|
||
four_bad.append("%s: summary[0].primary 缺失" % fn)
|
||
if sm:
|
||
model_tables[sm[0].get("table")] = d
|
||
record("B2", four_ok, "四段式(summary/fields/indexes/codes) 违规：%s" % (four_bad or "无"))
|
||
|
||
only_model = sorted(set(model_tables) - set(ddl_tables))
|
||
only_ddl = sorted(set(ddl_tables) - set(model_tables))
|
||
record("B3", not only_model and not only_ddl,
|
||
"models↔DDL 表名对应：仅 models 有 %s，仅 DDL 有 %s（均应为空）"
|
||
% (only_model or "无", only_ddl or "无"))
|
||
|
||
col_bad, idx_bad = [], []
|
||
for t, d in model_tables.items():
|
||
if t not in ddl_tables:
|
||
continue
|
||
mcols = [f.get("name") for f in d.get("fields", [])]
|
||
dcols = ddl_tables[t]["cols"]
|
||
if sorted(mcols) != sorted(dcols):
|
||
col_bad.append("%s: 仅models=%s 仅DDL=%s"
|
||
% (t, sorted(set(mcols) - set(dcols)),
|
||
sorted(set(dcols) - set(mcols))))
|
||
midx = {}
|
||
for ix in d.get("indexes", []):
|
||
midx[ix.get("name")] = (ix.get("fields"), bool(ix.get("unique")))
|
||
didx = ddl_tables[t]["indexes"]
|
||
if sorted(midx) != sorted(didx):
|
||
idx_bad.append("%s: 索引名不一致 仅models=%s 仅DDL=%s"
|
||
% (t, sorted(set(midx) - set(didx)),
|
||
sorted(set(didx) - set(midx))))
|
||
else:
|
||
for k in midx:
|
||
if list(midx[k][0]) != list(didx[k][0]) or midx[k][1] != didx[k][1]:
|
||
idx_bad.append("%s.%s: models=%s DDL=%s" % (t, k, midx[k], didx[k]))
|
||
record("B4", not col_bad, "models↔DDL 列一一对应违规：%s" % (col_bad or "无"))
|
||
record("B5", not idx_bad, "models↔DDL 索引一一对应违规：%s" % (idx_bad or "无"))
|
||
|
||
|
||
# ---------------------------------------------------------- C. json CRUD
|
||
def check_crud():
|
||
jdir = p(JSON_REL)
|
||
if not os.path.isdir(jdir):
|
||
record("C1", False, "json/m1b/ 目录不存在")
|
||
return
|
||
files = sorted(f for f in os.listdir(jdir) if f.endswith(".json"))
|
||
record("C1", len(files) == 4, "json/m1b/ JSON 文件 %d 个：%s" % (len(files), files))
|
||
|
||
bad_tbl, bad_edit, bad_url, bad_model = [], [], [], []
|
||
for fn in files:
|
||
try:
|
||
with open(os.path.join(jdir, fn), "r", encoding="utf-8") as f:
|
||
d = json.load(f)
|
||
except Exception as e:
|
||
bad_tbl.append("%s: json.load 失败 %s" % (fn, e))
|
||
continue
|
||
if d.get("tblname") not in M1B_TABLES:
|
||
bad_tbl.append("%s: tblname=%r" % (fn, d.get("tblname")))
|
||
prm = d.get("params") or {}
|
||
if not isinstance(prm.get("editable"), dict):
|
||
bad_edit.append("%s: 缺 params.editable" % fn)
|
||
for k in ("new_data_url", "update_data_url", "delete_data_url"):
|
||
if not prm.get(k):
|
||
bad_url.append("%s: params 顶层缺 %s" % (fn, k))
|
||
for seg in ("summary", "fields", "indexes"):
|
||
if seg in d:
|
||
bad_model.append("%s: 含表定义段 %s（json/ 只放 CRUD）" % (fn, seg))
|
||
record("C2", not bad_tbl, "tblname 违规：%s" % (bad_tbl or "无"))
|
||
record("C3", not bad_edit, "params.editable 违规：%s" % (bad_edit or "无"))
|
||
record("C4", not bad_url, "params 顶层 *_data_url 违规：%s" % (bad_url or "无"))
|
||
record("C5", not bad_model, "json/ 混入表定义：%s" % (bad_model or "无"))
|
||
|
||
|
||
# ------------------------------------------------------- D. Python 源码
|
||
def m1b_py_files():
|
||
pkg = p("pbl_blueprint")
|
||
return sorted(f for f in os.listdir(pkg)
|
||
if f.startswith("m1b_") and f.endswith(".py"))
|
||
|
||
|
||
def check_python():
|
||
files = m1b_py_files()
|
||
if not files:
|
||
record("D1", False, "未找到 pbl_blueprint/m1b_*.py")
|
||
return {}
|
||
|
||
hard, no_dbname, fake, compile_bad, empty = [], [], [], [], []
|
||
funcs = {}
|
||
hard_re = re.compile(
|
||
r"""^\s*(?:DBNAME|dbname|db_name|DATABASE|database)\s*=\s*['"][^'"]+['"]""",
|
||
re.M)
|
||
for fn in files:
|
||
src = read(os.path.join("pbl_blueprint", fn))
|
||
if hard_re.search(src):
|
||
hard.append(fn)
|
||
if "get_module_dbname" not in src and "dbname" in src.lower():
|
||
no_dbname.append(fn)
|
||
for api in FAKE_SQLOR:
|
||
if api in src:
|
||
fake.append("%s: %s" % (fn, api))
|
||
try:
|
||
compile(src, fn, "exec")
|
||
except SyntaxError as e:
|
||
compile_bad.append("%s: %s" % (fn, e))
|
||
try:
|
||
tree = ast.parse(src)
|
||
except SyntaxError:
|
||
continue
|
||
stmts = [n for n in ast.walk(tree)
|
||
if isinstance(n, ast.stmt) and not isinstance(n, (ast.Pass,))]
|
||
if len(stmts) < 3:
|
||
empty.append("%s: 有效语句 %d" % (fn, len(stmts)))
|
||
for n in tree.body:
|
||
if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
||
funcs.setdefault(n.name, []).append(fn)
|
||
|
||
record("D1", not hard, "硬编码库名文件：%s（应为空）" % (hard or "无"))
|
||
record("D2", not no_dbname, "未走 get_module_dbname 的文件：%s" % (no_dbname or "无"))
|
||
record("D3", not fake, "编造 sqlor API：%s（应为空；白名单 sor.C/U/D/R/I/sqlExe/sqlPaging）"
|
||
% (fake or "无"))
|
||
record("D4", not compile_bad, "py_compile 失败：%s" % (compile_bad or "无"))
|
||
record("D5", not empty, "疑似空壳文件：%s" % (empty or "无"))
|
||
return funcs
|
||
|
||
|
||
# --------------------------------------------------- E. 三处同步注册
|
||
def check_registration(funcs):
|
||
api = read(os.path.join("pbl_blueprint", "m1b_api.py")) if os.path.exists(
|
||
p("pbl_blueprint", "m1b_api.py")) else ""
|
||
initpkg = read(os.path.join("pbl_blueprint", "__init__.py"))
|
||
initpy = read(os.path.join("pbl_blueprint", "init.py"))
|
||
|
||
record("E1", "def register_m1b_routes" in api,
|
||
"m1b_api.py 定义 register_m1b_routes：%s"
|
||
% ("是" if "def register_m1b_routes" in api else "否"))
|
||
record("E2", "register_m1b_routes" in initpkg,
|
||
"__init__.py 导出 register_m1b_routes：%s"
|
||
% ("是" if "register_m1b_routes" in initpkg else "否"))
|
||
record("E3", "register_m1b_routes" in initpy,
|
||
"init.py 引用/注册 register_m1b_routes：%s"
|
||
% ("是" if "register_m1b_routes" in initpy else "否"))
|
||
|
||
# E4：M1b 对外函数（m1b_*.py 顶层定义）三处齐备
|
||
# ① 实现：m1b_*.py 中 def/async def
|
||
# ② 导出：__init__.py 出现该名（from .m1b_x import ... 或 __all__）
|
||
# ③ 注册：init.py 出现该名（import + env.xxx= 赋值 或 直接调用）
|
||
PUBLIC_PREFIX = ("m1b_", "init_m1b", "register_m1b", "ensure_m1b",
|
||
"load_m1b", "seed_platform", "ensure_platform")
|
||
public = sorted(n for n in funcs if n.startswith(PUBLIC_PREFIX))
|
||
miss_exp = [n for n in public if not re.search(r"\b%s\b" % re.escape(n), initpkg)]
|
||
miss_reg = [n for n in public if not re.search(r"\b%s\b" % re.escape(n), initpy)]
|
||
# 注册形态取证：env.xxx = 赋值 或 函数调用 或 import 行
|
||
reg_form = {}
|
||
for n in public:
|
||
forms = []
|
||
if re.search(r"env\.%s\s*=" % re.escape(n), initpy):
|
||
forms.append("env赋值")
|
||
if re.search(r'["\']%s["\']' % re.escape(n), initpy) and \
|
||
re.search(r"M1B_ENV_EXPORTS|setattr\(env", initpy):
|
||
forms.append("M1B_ENV_EXPORTS声明式env注册")
|
||
if re.search(r"\b%s\s*\(" % re.escape(n), initpy):
|
||
forms.append("调用")
|
||
if re.search(r"^\s*from\s+\.m1b_\w*\s+import[^\n]*\b%s\b" % re.escape(n),
|
||
initpy, re.M) or re.search(r"^\s*from\s+\.\s+import[^\n]*\b%s\b"
|
||
% re.escape(n), initpy, re.M):
|
||
forms.append("import")
|
||
reg_form[n] = forms or ["未注册"]
|
||
record("E4", not miss_exp and not miss_reg,
|
||
"M1b 对外函数 %d 个；__init__ 未导出 %s；init.py 未注册 %s"
|
||
% (len(public), miss_exp or "无", miss_reg or "无"))
|
||
for n in public:
|
||
record("E4.%s" % n, reg_form[n] != ["未注册"],
|
||
"init.py 注册形态=%s" % "/".join(reg_form[n]))
|
||
|
||
|
||
def check_m1a_untouched():
|
||
"""F1/F2：M1b 改动不得越权破坏 M1a 既有代码。
|
||
|
||
界定：
|
||
* M1a 业务文件 = pbl_blueprint/ 下非 m1b_* 的 .py，且排除共享入口
|
||
__init__.py / init.py（这两个文件是 module-development-spec 规定的
|
||
「三处同步」注册点，M1b 必须在此追加导出/注册，属合规追加而非越权）。
|
||
* 判定：M1a 业务文件不得被修改/删除（git diff 无 M/D）；
|
||
共享入口只允许「追加式」改动——既有 def 行零删除。
|
||
"""
|
||
SHARED = ("__init__.py", "init.py")
|
||
try:
|
||
out = subprocess.run(["git", "diff", "--name-status", "HEAD", "--", "pbl_blueprint/"],
|
||
cwd=REPO, capture_output=True, text=True, timeout=30)
|
||
lines = [l for l in out.stdout.splitlines() if l.strip()]
|
||
except Exception as e:
|
||
lines = []
|
||
print(" (git diff 不可用：%s)" % e)
|
||
|
||
m1a_biz_bad, shared_del = [], []
|
||
for l in lines:
|
||
parts = l.split("\t")
|
||
if len(parts) < 2:
|
||
continue
|
||
st, path = parts[0], parts[-1]
|
||
base = os.path.basename(path)
|
||
if base.startswith("m1b_") or "/m1b/" in path or not base.endswith(".py"):
|
||
continue
|
||
if base in SHARED:
|
||
# 共享入口：只查「既有 def 行是否被删除」
|
||
try:
|
||
d = subprocess.run(["git", "diff", "-U0", "HEAD", "--", path],
|
||
cwd=REPO, capture_output=True, text=True, timeout=30)
|
||
removed_defs = [x for x in d.stdout.splitlines()
|
||
if x.startswith("-") and not x.startswith("---")
|
||
and re.match(r"-\s*(async\s+)?def\s+\w+", x)]
|
||
if removed_defs:
|
||
shared_del.append("%s 删除既有函数 %d 个: %s"
|
||
% (base, len(removed_defs), removed_defs[:3]))
|
||
except Exception:
|
||
pass
|
||
continue
|
||
if st.startswith("M") or st.startswith("D"):
|
||
m1a_biz_bad.append("%s %s" % (st, path))
|
||
|
||
# F2 静态核验：M1a 关键入口与 11 表定义仍在
|
||
initpy = read(os.path.join("pbl_blueprint", "init.py"))
|
||
initpkg = read(os.path.join("pbl_blueprint", "__init__.py"))
|
||
m1a_ok = ("load_pbl_blueprint" in initpy and "load_pbl_blueprint" in initpkg
|
||
and "TABLE_NAMES" in initpy)
|
||
record("F1", not m1a_biz_bad and not shared_del and m1a_ok,
|
||
"M1a 业务文件被改/删：%s；共享入口删除既有函数：%s；"
|
||
"load_pbl_blueprint+TABLE_NAMES 完好：%s"
|
||
% (m1a_biz_bad or "无", shared_del or "无", m1a_ok))
|
||
|
||
|
||
def main():
|
||
print("=" * 72)
|
||
print("M1b 内容级自检 repo=%s" % REPO)
|
||
print("=" * 72)
|
||
body = check_ddl()
|
||
ddl_tables = parse_ddl_tables(body) if body else {}
|
||
check_models(ddl_tables)
|
||
check_crud()
|
||
funcs = check_python()
|
||
check_registration(funcs)
|
||
check_m1a_untouched()
|
||
|
||
total = len(results)
|
||
passed = sum(1 for _, ok, _ in results if ok)
|
||
print("-" * 72)
|
||
print("结果：%d/%d PASS" % (passed, total))
|
||
for cid, ok, detail in results:
|
||
if not ok:
|
||
print(" FAIL %s :: %s" % (cid, detail))
|
||
print("-" * 72)
|
||
return 0 if passed == total else 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|