#!/usr/bin/env python3 # -*- coding: utf-8 -*- """pbl_blueprint M1a 自查门禁脚本。 用法（机构工作空间根执行）:: python3 modules/pbl_blueprint/scripts/selfcheck.py echo "rc=$?" 退出码约定（**本轮修复项 B**）： 0 —— 全部检查通过（或仅 SKIP，无 FAIL） 1 —— 存在 FAIL 项（门禁不通过，QC/CI 据此判定） 2 —— 脚本自身无法运行（仓库结构缺失等致命错误） 检查分组： A. 目录结构 —— 包目录/init.py/pyproject.toml/scripts/skill 真实存在 B. 表定义 —— models/*.json 11 个，四段式齐备、primary=["id"]、id str32、 tenant_id 为首业务字段、金额 double(18,2) C. CRUD 契约 —— json/*.json 11 个，根键 tblname+params、editable 三个 .dspy URL、 browserfields 非空、无自创 table/list 格式 D. 代码质量 —— py_compile 全量、禁硬编码库名、sqlor API 白名单 E. 注册同步 —— 函数三处注册（定义 / __init__ 导出 / init.py env 注册） F. 契约运行 —— import pbl_blueprint 与 errors.py 14 符号 G. 租户 fail-closed —— 缺 tenant_id 一律拒绝 H. DDL 生成 —— sql/*.sql 与 models 表集合一致 本脚本**不依赖** sqlor/ahserver/sage 运行时；缺失运行时依赖的项按 SKIP 处理并标注， 不计入 FAIL（避免 CI 环境假失败），但 SKIP 数会在结论中明示。 """ from __future__ import print_function import json import os import re import subprocess import sys # -------------------------------------------------------------------------- # 路径解析：脚本位于 modules/pbl_blueprint/scripts/，仓库根 = 上一级 # -------------------------------------------------------------------------- HERE = os.path.dirname(os.path.abspath(__file__)) REPO_ROOT = os.path.dirname(HERE) # modules/pbl_blueprint PKG_DIR = os.path.join(REPO_ROOT, "pbl_blueprint") # modules/pbl_blueprint/pbl_blueprint MODELS_DIR = os.path.join(PKG_DIR, "models") JSON_DIR = os.path.join(PKG_DIR, "json") SQL_DIR = os.path.join(PKG_DIR, "sql") EXPECTED_TABLES = [ "pbl_blueprint", "pbl_blueprint_node", "pbl_blueprint_edge", "pbl_blueprint_version", "pbl_blueprint_version_delta", "pbl_blueprint_template", "pbl_blueprint_fork", "pbl_blueprint_lock", "pbl_blueprint_publish", "pbl_blueprint_offline", "pbl_blueprint_audit", ] ERROR_SYMBOLS = [ "PblBlueprintError", "ERR_OK", "ERR_TENANT_MISSING", "ERR_PARAM_INVALID", "ERR_NOT_FOUND", "ERR_DUPLICATE", "ERR_STATE_INVALID", "ERR_LOCKED", "ERR_FORBIDDEN", "ERR_DB", "ERR_INTERNAL", "ok", "fail", "err", ] SQLOR_WHITELIST = {"C", "U", "D", "R", "I", "sqlExe"} HARDCODE_DB_PATTERNS = [r"\bDBNAME\s*=", r"dbname\s*=\s*['\"]", r"\bDB_NAME\s*="] # -------------------------------------------------------------------------- # 结果收集 # -------------------------------------------------------------------------- RESULTS = [] # [(status, group, name, detail)] def record(status, group, name, detail=""): RESULTS.append((status, group, name, detail)) mark = {"PASS": "[PASS]", "FAIL": "[FAIL]", "SKIP": "[SKIP]"}.get(status, "[%s]" % status) line = "%s %s | %s" % (mark, group, name) if detail: line += " -> %s" % detail print(line) def check(cond, group, name, detail_ok="", detail_bad=""): record("PASS" if cond else "FAIL", group, name, detail_ok if cond else detail_bad) return bool(cond) def load_json(path): with open(path, "r", encoding="utf-8") as f: return json.load(f) # -------------------------------------------------------------------------- # A. 目录结构 # -------------------------------------------------------------------------- def group_a(): g = "A.目录结构" required = [ (PKG_DIR, "包目录 pbl_blueprint/"), (os.path.join(PKG_DIR, "__init__.py"), "pbl_blueprint/__init__.py"), (os.path.join(PKG_DIR, "init.py"), "pbl_blueprint/init.py"), (os.path.join(PKG_DIR, "errors.py"), "pbl_blueprint/errors.py"), (os.path.join(REPO_ROOT, "pyproject.toml"), "pyproject.toml"), (os.path.join(HERE, "load_path.py"), "scripts/load_path.py"), (os.path.join(REPO_ROOT, "skill", "SKILL.md"), "skill/SKILL.md"), (MODELS_DIR, "pbl_blueprint/models/"), (JSON_DIR, "pbl_blueprint/json/"), (SQL_DIR, "pbl_blueprint/sql/"), ] for path, label in required: check(os.path.exists(path), g, label, "存在 %s" % path, "缺失 %s" % path) # 违禁替代文件 forbidden = ["module.json", "rp.json"] for name in forbidden: bad = os.path.exists(os.path.join(REPO_ROOT, name)) or os.path.exists(os.path.join(PKG_DIR, name)) check(not bad, g, "无违禁替代文件 %s" % name, "未发现", "发现违禁文件 %s" % name) # 模块不得有独立 app.py（模块非部署单元） has_app = os.path.exists(os.path.join(REPO_ROOT, "app.py")) or os.path.exists(os.path.join(PKG_DIR, "app.py")) check(not has_app, g, "模块无独立 app.py", "符合（模块非部署单元）", "发现 app.py，模块不应独立部署") # -------------------------------------------------------------------------- # B. 表定义（models/*.json 四段式） # -------------------------------------------------------------------------- def group_b(): g = "B.表定义" if not os.path.isdir(MODELS_DIR): record("FAIL", g, "models 目录", "缺失 %s" % MODELS_DIR) return files = sorted([f for f in os.listdir(MODELS_DIR) if f.endswith(".json")]) check(len(files) == len(EXPECTED_TABLES), g, "models/*.json 数量 = %d" % len(EXPECTED_TABLES), "实际 %d" % len(files), "实际 %d，期望 %d" % (len(files), len(EXPECTED_TABLES))) names = set(f[:-5] for f in files) missing = [t for t in EXPECTED_TABLES if t not in names] check(not missing, g, "表集合与设计定稿一致", "11 表齐备", "缺失表: %s" % ",".join(missing)) for fn in files: path = os.path.join(MODELS_DIR, fn) tbl = fn[:-5] try: d = load_json(path) except Exception as e: # noqa: BLE001 record("FAIL", g, "%s JSON 可解析" % tbl, "", "解析失败: %s" % e) continue # 四段式 segs = ["summary", "fields", "indexes", "codes"] lack = [s for s in segs if s not in d] check(not lack, g, "%s 四段式齐备" % tbl, "summary/fields/indexes/codes", "缺段: %s" % ",".join(lack)) if lack: continue fields = d.get("fields") or {} # primary = ["id"] primary = d.get("primary") if primary is None: primary = (d.get("summary") or {}).get("primary") check(primary == ["id"], g, "%s primary=[\"id\"]" % tbl, "primary=%s" % (primary,), "primary=%s，期望 ['id']" % (primary,)) # id 为 str32 idf = fields.get("id") or {} idt = str(idf.get("type", "")) check(idt.startswith("str") and "32" in idt, g, "%s id 类型 str32" % tbl, "id.type=%s" % idt, "id.type=%s，期望 str32" % idt) # tenant_id 为首业务字段（id 之后第一个） keys = list(fields.keys()) biz = [k for k in keys if k != "id"] check(bool(biz) and biz[0] == "tenant_id", g, "%s tenant_id 为首业务字段" % tbl, "字段序: id, %s" % (biz[0] if biz else "-"), "首业务字段=%s，期望 tenant_id" % (biz[0] if biz else "-")) # 金额字段 double(18,2) money_bad = [] for k, v in fields.items(): if not isinstance(v, dict): continue t = str(v.get("type", "")) is_money_name = ("amount" in k) or ("money" in k) or ("price" in k) or ("fee" in k) is_money_type = t.startswith("double") or t.startswith("decimal") or t.startswith("money") if is_money_name or is_money_type: if "18" not in t or "2" not in t: money_bad.append("%s:%s" % (k, t)) if money_bad: record("FAIL", g, "%s 金额字段 double(18,2)" % tbl, "", "不合规: %s" % ",".join(money_bad)) else: record("PASS", g, "%s 金额字段 double(18,2)" % tbl, "无金额字段或均合规") # indexes 中 tenant_id 打头（复合索引首列） idx = d.get("indexes") or {} bad_idx = [] if isinstance(idx, dict): for ik, iv in idx.items(): cols = iv.get("fields") if isinstance(iv, dict) else iv if isinstance(cols, str): cols = [c.strip() for c in cols.split(",")] if isinstance(cols, (list, tuple)) and cols and cols[0] != "tenant_id": bad_idx.append("%s%s" % (ik, list(cols))) if bad_idx: record("FAIL", g, "%s 索引 tenant_id 打头" % tbl, "", "首列非 tenant_id: %s" % ";".join(bad_idx)) else: record("PASS", g, "%s 索引 tenant_id 打头" % tbl, "全部复合索引首列为 tenant_id") # -------------------------------------------------------------------------- # C. CRUD 契约（json/*.json） # -------------------------------------------------------------------------- def group_c(): g = "C.CRUD契约" if not os.path.isdir(JSON_DIR): record("FAIL", g, "json 目录", "缺失 %s" % JSON_DIR) return files = sorted([f for f in os.listdir(JSON_DIR) if f.endswith(".json")]) check(len(files) == len(EXPECTED_TABLES), g, "json/*.json 数量 = %d" % len(EXPECTED_TABLES), "实际 %d" % len(files), "实际 %d，期望 %d" % (len(files), len(EXPECTED_TABLES))) for fn in files: path = os.path.join(JSON_DIR, fn) tbl = fn[:-5] try: d = load_json(path) except Exception as e: # noqa: BLE001 record("FAIL", g, "%s JSON 可解析" % tbl, "", "解析失败: %s" % e) continue # 根键仅 tblname + params，无自创 table/list 格式 root = set(d.keys()) check(root == {"tblname", "params"}, g, "%s 根键 = tblname+params" % tbl, "根键=%s" % sorted(root), "根键=%s，期望 ['params','tblname']" % sorted(root)) check("table" not in root and "list" not in root, g, "%s 无自创 table/list 格式" % tbl, "符合 crud-definition-spec", "发现自创键") params = d.get("params") or {} # editable 为三个 .dspy URL ed = params.get("editable") ed_list = ed if isinstance(ed, list) else ([ed] if ed else []) dspy = [u for u in ed_list if isinstance(u, str) and u.endswith(".dspy")] check(len(ed_list) == 3 and len(dspy) == 3, g, "%s editable = 3 个 .dspy URL" % tbl, "editable=%s" % (ed_list,), "editable=%s，期望 3 个 .dspy" % (ed_list,)) # browserfields 非空 bf = params.get("browserfields") bf_nonempty = bool(bf) and (len(bf) > 0 if isinstance(bf, (list, dict, str)) else True) check(bf_nonempty, g, "%s browserfields 非空" % tbl, "%d 项" % (len(bf) if hasattr(bf, "__len__") else 1), "browserfields 为空/缺失") # tblname 与文件名一致 check(d.get("tblname") == tbl, g, "%s tblname 与文件名一致" % tbl, "tblname=%s" % d.get("tblname"), "tblname=%s，文件名=%s" % (d.get("tblname"), tbl)) # -------------------------------------------------------------------------- # D. 代码质量 # -------------------------------------------------------------------------- def iter_py_files(): out = [] for root, dirs, fs in os.walk(REPO_ROOT): dirs[:] = [x for x in dirs if x not in (".git", "__pycache__", ".venv", "build", "dist")] for f in fs: if f.endswith(".py"): out.append(os.path.join(root, f)) return sorted(out) def group_d(): g = "D.代码质量" pys = iter_py_files() check(len(pys) > 0, g, "存在 .py 源文件", "%d 个" % len(pys), "未找到任何 .py") # py_compile 全量 try: r = subprocess.run([sys.executable, "-m", "py_compile"] + pys, stdout=subprocess.PIPE, stderr=subprocess.PIPE) out = (r.stdout or b"").decode("utf-8", "replace") + (r.stderr or b"").decode("utf-8", "replace") check(r.returncode == 0, g, "py_compile 全量编译（%d 文件）" % len(pys), "rc=0", "rc=%s\n%s" % (r.returncode, out.strip()[:2000])) except Exception as e: # noqa: BLE001 record("FAIL", g, "py_compile 全量编译", "", "执行异常: %s" % e) # 禁硬编码库名 hits = [] for p in pys: try: txt = open(p, "r", encoding="utf-8").read() except Exception: # noqa: BLE001 continue for i, line in enumerate(txt.splitlines(), 1): s = line.strip() if s.startswith("#"): continue for pat in HARDCODE_DB_PATTERNS: if re.search(pat, line): hits.append("%s:%d:%s" % (os.path.relpath(p, REPO_ROOT), i, s[:120])) check(not hits, g, "禁硬编码库名（DBNAME=/dbname='…'/DB_NAME=）", "无命中，统一走 ServerEnv().get_module_dbname('pbl_blueprint')", "命中 %d 处:\n %s" % (len(hits), "\n ".join(hits[:20]))) # sqlor API 白名单 used = set() for p in pys: try: txt = open(p, "r", encoding="utf-8").read() except Exception: # noqa: BLE001 continue for m in re.finditer(r"\bsor\.([A-Za-z_][A-Za-z0-9_]*)", txt): used.add(m.group(1)) illegal = sorted(used - SQLOR_WHITELIST) check(not illegal, g, "sqlor API 白名单（仅 C/U/D/R/I/sqlExe）", "实际使用: %s" % (sorted(used) or "无"), "非法 API: %s（禁编造 save/list/insert）" % ",".join(illegal)) # -------------------------------------------------------------------------- # E. 函数三处注册同步 # -------------------------------------------------------------------------- def group_e(): g = "E.注册同步" init_py = os.path.join(PKG_DIR, "__init__.py") loader = os.path.join(PKG_DIR, "init.py") if not (os.path.exists(init_py) and os.path.exists(loader)): record("FAIL", g, "__init__.py / init.py 存在", "", "缺失，无法核验三处注册") return init_txt = open(init_py, "r", encoding="utf-8").read() load_txt = open(loader, "r", encoding="utf-8").read() # errors.py 14 符号在 __init__.py 中导出 lack = [s for s in ERROR_SYMBOLS if s not in init_txt] check(not lack, g, "__init__.py 导出 errors 14 符号", "14 符号齐备", "缺导出: %s" % ",".join(lack)) # load_pbl_blueprint 入口存在 check("def load_pbl_blueprint" in load_txt, g, "init.py 定义 load_pbl_blueprint()", "入口存在", "未找到 load_pbl_blueprint 定义") # get_module_dbname 取库名（禁硬编码） check("get_module_dbname" in load_txt, g, "init.py 用 get_module_dbname 取库名", "符合规范", "未使用 ServerEnv().get_module_dbname") # env 注册（api 函数挂到 ServerEnv） has_reg = bool(re.search(r"(register|set_api|env\[|setattr\(|add_function|reg_func)", load_txt)) check(has_reg, g, "init.py 向 env 注册契约函数", "发现注册语句", "未发现任何注册语句") # load_path.py RBAC 路径显式枚举、无通配符 lp = os.path.join(HERE, "load_path.py") if os.path.exists(lp): txt = open(lp, "r", encoding="utf-8").read() wild = [] for i, line in enumerate(txt.splitlines(), 1): s = line.strip() if s.startswith("#"): continue for m in re.finditer(r"['\"]([^'\"]*)['\"]", line): v = m.group(1) if ("%" in v or "*" in v) and "/" in v: wild.append("%d:%s" % (i, v)) check(not wild, g, "load_path.py RBAC 路径无 %/* 通配符", "全部显式枚举", "发现通配路径: %s" % ";".join(wild[:10])) check(len(txt.strip()) > 0, g, "load_path.py 非空", "已注册路径", "文件为空") else: record("FAIL", g, "load_path.py 存在", "", "缺失 %s" % lp) # -------------------------------------------------------------------------- # F. 契约运行（import） # -------------------------------------------------------------------------- def group_f(): g = "F.契约运行" if REPO_ROOT not in sys.path: sys.path.insert(0, REPO_ROOT) # F1: errors.py 可独立 import 且 14 符号齐备 try: import importlib m = importlib.import_module("pbl_blueprint.errors") lack = [s for s in ERROR_SYMBOLS if not hasattr(m, s)] check(not lack, g, "pbl_blueprint.errors 导出 14 符号", "14 符号齐备", "缺符号: %s" % ",".join(lack)) # 语义抽检 r_ok = m.ok({"id": "1"}) check(m.is_ok(r_ok) and r_ok["code"] == m.ERR_OK, g, "ok() 返回 code=0", "%s" % (r_ok,), "异常: %s" % (r_ok,)) r_fail = m.fail(m.ERR_TENANT_MISSING) check(r_fail["code"] == m.ERR_TENANT_MISSING and not m.is_ok(r_fail), g, "fail(ERR_TENANT_MISSING) 非成功", "%s" % (r_fail,), "%s" % (r_fail,)) r_err = m.err() check(r_err["code"] == m.ERR_INTERNAL, g, "err() 缺省 ERR_INTERNAL", "%s" % (r_err,), "%s" % (r_err,)) try: raise m.PblBlueprintError(m.ERR_NOT_FOUND, "蓝图不存在") except m.PblBlueprintError as e: check(e.code == m.ERR_NOT_FOUND and e.to_dict()["code"] == m.ERR_NOT_FOUND, g, "PblBlueprintError 抛出与 to_dict()", "%s" % (e.to_dict(),), "%s" % (e.to_dict(),)) except Exception as e: # noqa: BLE001 record("FAIL", g, "pbl_blueprint.errors 可 import", "", "%s: %s" % (type(e).__name__, e)) # F2: 整包 import（依赖 sqlor/ahserver 运行时则 SKIP，不算 FAIL） try: import importlib importlib.import_module("pbl_blueprint") record("PASS", g, "import pbl_blueprint 整包", "成功") except ImportError as e: miss = str(e) if any(k in miss for k in ("sqlor", "ahserver", "sage", "sagelib", "DictObject", "DBPools")): record("SKIP", g, "import pbl_blueprint 整包", "运行时依赖缺失（CI 环境预期）: %s" % miss) else: record("FAIL", g, "import pbl_blueprint 整包", "", "ImportError: %s" % miss) except Exception as e: # noqa: BLE001 record("FAIL", g, "import pbl_blueprint 整包", "", "%s: %s" % (type(e).__name__, e)) # -------------------------------------------------------------------------- # G. 租户 fail-closed # -------------------------------------------------------------------------- def group_g(): g = "G.租户fail-closed" tp = os.path.join(PKG_DIR, "tenant.py") if not os.path.exists(tp): record("FAIL", g, "tenant.py 存在", "", "缺失 %s" % tp) return txt = open(tp, "r", encoding="utf-8").read() check("ERR_TENANT_MISSING" in txt, g, "tenant.py 使用 ERR_TENANT_MISSING 拒绝", "fail-closed 拒绝路径存在", "未见 ERR_TENANT_MISSING") # 不得有默认租户兜底 fallback = re.search(r"tenant_id\s*=\s*tenant_id\s+or\s+['\"]", txt) or \ re.search(r"DEFAULT_TENANT\s*=\s*['\"][^'\"]+['\"]", txt) check(fallback is None, g, "tenant.py 无默认租户兜底", "缺失即拒绝，不静默填充", "发现默认租户兜底: %s" % (fallback.group(0) if fallback else "")) # 运行时抽检（可 import 时） if REPO_ROOT not in sys.path: sys.path.insert(0, REPO_ROOT) try: import importlib t = importlib.import_module("pbl_blueprint.tenant") fn = None for cand in ("require_tenant", "require_tenant_id", "get_tenant_id", "current_tenant_id"): if hasattr(t, cand): fn = getattr(t, cand) break if fn is None: record("SKIP", g, "租户缺失运行时拒绝", "未找到租户取值函数，静态检查已覆盖") else: try: fn({}) record("FAIL", g, "租户缺失运行时拒绝", "", "缺 tenant_id 未抛错/未拒绝") except Exception as e: # noqa: BLE001 record("PASS", g, "租户缺失运行时拒绝", "%s: %s" % (type(e).__name__, e)) except ImportError as e: record("SKIP", g, "租户缺失运行时拒绝", "运行时依赖缺失: %s" % e) except Exception as e: # noqa: BLE001 record("SKIP", g, "租户缺失运行时拒绝", "%s: %s" % (type(e).__name__, e)) # -------------------------------------------------------------------------- # H. DDL 生成 # -------------------------------------------------------------------------- def group_h(): g = "H.DDL生成" if not os.path.isdir(SQL_DIR): record("FAIL", g, "sql 目录", "缺失 %s" % SQL_DIR) return sqls = sorted([f for f in os.listdir(SQL_DIR) if f.endswith(".sql")]) check(len(sqls) >= 2, g, "sql/*.sql 存在（core + subobjects）", "%s" % ",".join(sqls), "实际 %d 个: %s" % (len(sqls), sqls)) blob = "" for fn in sqls: try: blob += open(os.path.join(SQL_DIR, fn), "r", encoding="utf-8").read() + "\n" except Exception as e: # noqa: BLE001 record("FAIL", g, "%s 可读" % fn, "", "%s" % e) ddl_tables = set(m.group(1).lower() for m in re.finditer( r"CREATE\s+TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?[`\"]?([A-Za-z0-9_]+)[`\"]?", blob, re.I)) missing = [t for t in EXPECTED_TABLES if t.lower() not in ddl_tables] check(not missing, g, "DDL 覆盖 11 张表", "CREATE TABLE %d 张" % len(ddl_tables), "DDL 缺表: %s" % ",".join(missing)) # 每张表 DDL 含 tenant_id no_tenant = [] for t in EXPECTED_TABLES: seg = re.search(r"CREATE\s+TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?[`\"]?%s[`\"]?\s*\((.*?)\n\)" % re.escape(t), blob, re.I | re.S) body = seg.group(1) if seg else "" if seg and "tenant_id" not in body: no_tenant.append(t) if no_tenant: record("FAIL", g, "DDL 各表含 tenant_id", "", "缺 tenant_id: %s" % ",".join(no_tenant)) else: record("PASS", g, "DDL 各表含 tenant_id", "已核验（或 DDL 由 models 生成）") # models 与 DDL 表集合一致 if os.path.isdir(MODELS_DIR): mset = set(f[:-5] for f in os.listdir(MODELS_DIR) if f.endswith(".json")) diff1 = sorted(mset - set(EXPECTED_TABLES)) check(not diff1, g, "models 无多余表", "一致", "models 多出: %s" % ",".join(diff1)) # -------------------------------------------------------------------------- # main # -------------------------------------------------------------------------- def main(): print("=" * 78) print("pbl_blueprint M1a 自查门禁") print("仓库根 : %s" % REPO_ROOT) print("包目录 : %s" % PKG_DIR) print("python : %s" % sys.version.replace("\n", " ")) print("=" * 78) if not os.path.isdir(REPO_ROOT): print("FATAL: 仓库根不存在 %s" % REPO_ROOT) return 2 for fn in (group_a, group_b, group_c, group_d, group_e, group_f, group_g, group_h): print("-" * 78) try: fn() except Exception as e: # noqa: BLE001 record("FAIL", fn.__name__, "检查组执行异常", "%s: %s" % (type(e).__name__, e)) total = len(RESULTS) n_pass = sum(1 for r in RESULTS if r[0] == "PASS") n_fail = sum(1 for r in RESULTS if r[0] == "FAIL") n_skip = sum(1 for r in RESULTS if r[0] == "SKIP") print("-" * 78) print("统计: 总计 %d | PASS %d | FAIL %d | SKIP %d" % (total, n_pass, n_fail, n_skip)) if n_fail: print("失败项清单:") for st, gp, nm, dt in RESULTS: if st == "FAIL": print(" - %s | %s | %s" % (gp, nm, (dt or "").replace("\n", " ")[:300])) print("结论: HAS FAILURES") # ★ 本轮修复项 B：有失败项必须 exit 非 0，否则门禁形同虚设 return 1 print("结论: ALL PASS%s" % ("（含 %d 项 SKIP，运行时依赖缺失，不计失败）" % n_skip if n_skip else "")) return 0 if __name__ == "__main__": try: rc = main() except KeyboardInterrupt: rc = 2 except Exception as _e: # noqa: BLE001 print("FATAL: 自查脚本自身异常: %s: %s" % (type(_e).__name__, _e)) rc = 2 print("exit_code=%s" % rc) sys.exit(rc)