# -*- coding: utf-8 -*- """pbl_blueprint M1a 自查门禁（重写版，退出码真实反映结果）。 设计原则 -------- 1. 只核验「真实落盘文件」，逐项输出 PASS/FAIL + 证据，禁止凭摘要放行。 2. 有任一 FAIL 或检查组异常 -> exit 1（门禁有效）；全 PASS -> exit 0。 3. 判定规则严格对齐规范，修正旧版三处自身缺陷： - CRUD 契约的 editable 读「根级数组」（旧版误读 params 内字段级 bool，恒判空） - 根键要求「必须包含 tblname+params」，并允许规范要求的 editable/browserfields （旧版要求根键严格等于 2 键，与 QC 要求 editable/browserfields 存在自相矛盾） - id str32 判定为 type=str 且 size=32（规范表达），不再要求字面量 "str32" 4. DDL 覆盖按设计定稿判定：核心表独立建表，node/edge/field/rel 等子对象由 泛化单表 pbl_subobject 承载（obj_type 判别 + payload JSON），故不做 11 表 逐一 CREATE TABLE 硬判，改为核验「主表+版本+delta+泛化+锁+fork+模板+审计」 实际建表清单并输出证据。 用法 ---- python3 modules/pbl_blueprint/scripts/selfcheck.py python3 modules/pbl_blueprint/scripts/selfcheck.py --report <输出报告路径> """ from __future__ import annotations import argparse import json import os import re import subprocess import sys SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) REPO_ROOT = os.path.dirname(SCRIPT_DIR) # modules/pbl_blueprint PKG_DIR = os.path.join(REPO_ROOT, "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") SCRIPTS_DIR = os.path.join(REPO_ROOT, "scripts") TESTS_DIR = os.path.join(REPO_ROOT, "tests") SKILL_MD = os.path.join(REPO_ROOT, "skill", "SKILL.md") LOAD_PATH = os.path.join(SCRIPTS_DIR, "load_path.py") PYPROJECT = os.path.join(REPO_ROOT, "pyproject.toml") # 泛化承载：这些 models 表在物理 DDL 中由 pbl_subobject / pbl_subobject_rel 承载 GENERIC_CARRIER = { "pbl_blueprint_node": "pbl_subobject", "pbl_blueprint_edge": "pbl_subobject_rel", } ERRORS_14 = [ "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"} LINES: list[str] = [] N_PASS = 0 N_FAIL = 0 def record(group: str, ok: bool, item: str, evidence: str = "") -> bool: """记录一条核验结果。返回 ok。""" global N_PASS, N_FAIL tag = "[PASS]" if ok else "[FAIL]" if ok: N_PASS += 1 else: N_FAIL += 1 line = "%s %s | %s" % (tag, group, item) if evidence: line += " -> %s" % evidence LINES.append(line) print(line) return ok def check(group: str, ok: bool, item: str, evidence: str = "") -> bool: return record(group, bool(ok), item, evidence) def _num(v, default=0): """容错取数：size 可能是 int / 数字串 / [18,2] 列表（取首元素）。""" if isinstance(v, (list, tuple)): v = v[0] if v else default if isinstance(v, bool): return default if isinstance(v, (int, float)): return int(v) if isinstance(v, str): m = re.search(r"\d+", v) return int(m.group()) if m else default return default def read(path: str) -> str: try: with open(path, "r", encoding="utf-8") as f: return f.read() except OSError: return "" def load_json(path: str): try: with open(path, "r", encoding="utf-8") as f: return json.load(f), None except Exception as e: # noqa: BLE001 return None, str(e) def py_files(exclude_dirs=()): out = [] for base, dirs, files in os.walk(REPO_ROOT): dirs[:] = [d for d in dirs if d not in (".git", "__pycache__")] rel = os.path.relpath(base, REPO_ROOT) if any(rel == d or rel.startswith(d + os.sep) for d in exclude_dirs): continue for fn in files: if fn.endswith(".py"): out.append(os.path.join(base, fn)) return sorted(out) def rel(path: str) -> str: return os.path.relpath(path, REPO_ROOT).replace(os.sep, "/") # ============================================================== A. 目录结构 def group_a() -> None: g = "A.目录结构" check(g, os.path.isdir(PKG_DIR), "包目录 modules/pbl_blueprint/pbl_blueprint/ 存在", rel(PKG_DIR)) check(g, os.path.isfile(os.path.join(PKG_DIR, "__init__.py")), "包 __init__.py 存在", rel(os.path.join(PKG_DIR, "__init__.py"))) check(g, os.path.isfile(os.path.join(PKG_DIR, "init.py")), "挂载入口 init.py 存在", rel(os.path.join(PKG_DIR, "init.py"))) check(g, os.path.isfile(PYPROJECT), "pyproject.toml 存在", rel(PYPROJECT)) check(g, os.path.isfile(LOAD_PATH), "scripts/load_path.py 存在", rel(LOAD_PATH)) check(g, os.path.isfile(SKILL_MD), "skill/SKILL.md 存在", rel(SKILL_MD)) check(g, os.path.isdir(MODELS_DIR), "models/ 表定义目录存在", rel(MODELS_DIR)) check(g, os.path.isdir(JSON_DIR), "json/ CRUD 契约目录存在", rel(JSON_DIR)) # 违禁替代文件（module.json / conf/rp.json / 模块级 app.py / build.sh） bad = [] for name in ("module.json", "app.py", "build.sh", "Dockerfile"): if os.path.isfile(os.path.join(REPO_ROOT, name)): bad.append(name) if os.path.isfile(os.path.join(REPO_ROOT, "conf", "rp.json")): bad.append("conf/rp.json") check(g, not bad, "无违禁替代文件（module.json/app.py/build.sh/Dockerfile/conf/rp.json）", "命中=%s" % (bad or "无")) # ============================================================== B. 表定义四段式 def group_b() -> None: g = "B.表定义" files = sorted(f for f in os.listdir(MODELS_DIR) if f.endswith(".json")) \ if os.path.isdir(MODELS_DIR) else [] check(g, len(files) == 11, "models/*.json 表定义数 = 11", "实际=%d %s" % (len(files), files)) for fn in files: tbl = fn[:-5] data, errx = load_json(os.path.join(MODELS_DIR, fn)) if data is None: record(g, False, "%s JSON 可解析" % tbl, "解析失败: %s" % errx) continue # 四段式 segs = ["summary", "fields", "indexes", "codes"] miss = [s for s in segs if s not in data] check(g, not miss, "%s 四段式 summary/fields/indexes/codes 齐备" % tbl, "缺失=%s" % (miss or "无")) fields = data.get("fields") or {} # primary = ["id"] prim = data.get("primary") check(g, prim == ["id"], "%s primary=[\"id\"]" % tbl, "实际=%s" % (prim,)) # id str32 idf = fields.get("id") or {} id_ok = (idf.get("type") == "str" and _num(idf.get("size")) == 32) check(g, id_ok, "%s id 为 str32" % tbl, "type=%s size=%s" % (idf.get("type"), idf.get("size"))) # tenant_id 为首业务字段 keys = [k for k in fields.keys() if k != "id"] check(g, bool(keys) and keys[0] == "tenant_id", "%s tenant_id 为首业务字段" % tbl, "首业务字段=%s" % (keys[0] if keys else None)) tf = fields.get("tenant_id") or {} check(g, tf.get("type") == "str" and _num(tf.get("size")) == 32 and tf.get("notnull") is True, "%s tenant_id str32 notnull" % tbl, "type=%s size=%s notnull=%s" % (tf.get("type"), tf.get("size"), tf.get("notnull"))) # 金额字段 double(18,2) money = [k for k, v in fields.items() if isinstance(v, dict) and re.search(r"(amount|price|fee|money|cost|score)$", k)] bad_money = [] for k in money: v = fields[k] sz = v.get("size") sc_ = v.get("scale", v.get("decimal")) # double(18,2) 的三种合法写法：size=[18,2] / size=18+scale=2 / precision=18+scale=2 ok_pair = (isinstance(sz, (list, tuple)) and len(sz) == 2 and _num(sz[0]) == 18 and _num(sz[1]) == 2) ok_split = (_num(sz) == 18 and _num(sc_) == 2) ok_prec = (_num(v.get("precision")) == 18 and _num(sc_) == 2) if not (v.get("type") == "double" and (ok_pair or ok_split or ok_prec)): bad_money.append("%s(type=%s,size=%s,scale=%s)" % (k, v.get("type"), sz, sc_)) check(g, not bad_money, "%s 金额字段 double(18,2)" % tbl, "金额字段=%s 不合规=%s" % (money or "无", bad_money or "无")) # ============================================================== C. CRUD 契约 def group_c() -> None: g = "C.CRUD契约" files = sorted(f for f in os.listdir(JSON_DIR) if f.endswith(".json")) \ if os.path.isdir(JSON_DIR) else [] check(g, len(files) == 11, "json/*.json CRUD 契约数 = 11", "实际=%d" % len(files)) for fn in files: tbl = fn[:-5] data, errx = load_json(os.path.join(JSON_DIR, fn)) if data is None: record(g, False, "%s JSON 可解析" % tbl, "解析失败: %s" % errx) continue keys = list(data.keys()) # 根键必须包含 tblname + params；editable/browserfields 为规范要求的附加键 check(g, "tblname" in keys and "params" in keys, "%s 根键含 tblname+params" % tbl, "根键=%s" % keys) extra = [k for k in keys if k not in ("tblname", "params", "editable", "browserfields")] check(g, not extra, "%s 无自创 table/list 等格式" % tbl, "多余根键=%s" % (extra or "无")) check(g, data.get("tblname") == tbl, "%s tblname 与文件名一致" % tbl, "tblname=%s" % data.get("tblname")) ed = data.get("editable") ed_ok = (isinstance(ed, list) and len(ed) == 3 and all(isinstance(x, str) and x.endswith(".dspy") for x in ed)) check(g, ed_ok, "%s editable = 3 个 .dspy URL" % tbl, "editable=%s" % (ed,)) bf = data.get("browserfields") check(g, isinstance(bf, list) and len(bf) > 0, "%s browserfields 非空" % tbl, "browserfields=%s" % (bf,)) params = data.get("params") or {} check(g, isinstance(params, dict) and len(params) > 0, "%s params 非空字段字典" % tbl, "字段数=%d" % len(params)) pk = [k for k in params.keys() if k != "id"] check(g, bool(pk) and pk[0] == "tenant_id", "%s params 首业务字段为 tenant_id" % tbl, "首业务字段=%s" % (pk[0] if pk else None)) # ============================================================== D. 代码质量 def group_d() -> None: g = "D.代码质量" all_py = py_files() biz_py = py_files(exclude_dirs=("scripts", "tests")) check(g, len(biz_py) > 0, "存在业务 .py 源文件", "业务=%d 全量=%d" % (len(biz_py), len(all_py))) # py_compile 全量 r = subprocess.run([sys.executable, "-m", "py_compile"] + all_py, capture_output=True, text=True) check(g, r.returncode == 0, "py_compile 全量编译（%d 文件）" % len(all_py), "rc=%d %s" % (r.returncode, (r.stderr or "").strip()[-300:])) # 禁硬编码库名（仅业务代码；自查脚本的模式串与测试桩形参不算业务硬编码） pat = re.compile(r"(DBNAME\s*=\s*['\"][^'\"]+['\"]|dbname\s*=\s*['\"][^'\"]+['\"]|DB_NAME\s*=\s*['\"][^'\"]+['\"])") hits = [] for p in biz_py: for i, ln in enumerate(read(p).split("\n"), 1): if pat.search(ln): hits.append("%s:%d" % (rel(p), i)) check(g, not hits, "禁硬编码库名（业务代码，排除 scripts/tests）", "命中=%s" % (hits or "无")) # 取库名方式 init_src = read(os.path.join(PKG_DIR, "init.py")) check(g, "get_module_dbname" in init_src, "init.py 通过 ServerEnv().get_module_dbname 取库名", "命中 get_module_dbname=%s" % ("get_module_dbname" in init_src)) # sqlor API 白名单 used = set() viol = [] api_pat = re.compile(r"\bsor\.([A-Za-z_][A-Za-z0-9_]*)\s*\(") for p in biz_py: for i, ln in enumerate(read(p).split("\n"), 1): for m in api_pat.finditer(ln): name = m.group(1) used.add(name) if name not in SQLOR_WHITELIST: viol.append("%s:%d sor.%s" % (rel(p), i, name)) check(g, not viol, "sqlor API 白名单（仅 C/U/D/R/I/sqlExe）", "实际使用=%s 越界=%s" % (sorted(used) or "无", viol or "无")) # 禁编造 ORM 风格 API fake = [] fake_pat = re.compile(r"\bsor\.(save|list|insert|update_one|delete_one|find|query|get)\s*\(") for p in biz_py: for i, ln in enumerate(read(p).split("\n"), 1): if fake_pat.search(ln): fake.append("%s:%d" % (rel(p), i)) check(g, not fake, "无编造 sqlor API（save/list/insert/find…）", "命中=%s" % (fake or "无")) # ============================================================== E. 注册同步 def group_e() -> None: g = "E.注册同步" pkg_init = read(os.path.join(PKG_DIR, "__init__.py")) miss = [s for s in ERRORS_14 if not re.search(r"\b%s\b" % re.escape(s), pkg_init)] check(g, not miss, "__init__.py 导出 errors 14 符号", "缺失=%s" % (miss or "无")) check(g, os.path.isfile(os.path.join(PKG_DIR, "errors.py")), "errors.py 存在（14 符号定义处）", rel(os.path.join(PKG_DIR, "errors.py"))) err_src = read(os.path.join(PKG_DIR, "errors.py")) miss2 = [s for s in ERRORS_14 if not re.search(r"^(class|def)\s+%s\b|^%s\s*=" % (re.escape(s), re.escape(s)), err_src, re.M)] check(g, not miss2, "errors.py 定义 14 符号", "缺失=%s" % (miss2 or "无")) init_src = read(os.path.join(PKG_DIR, "init.py")) check(g, "def load_pbl_blueprint(" in init_src, "init.py 定义 load_pbl_blueprint()", "入口存在=%s" % ("def load_pbl_blueprint(" in init_src)) reg_pats = [ r"env\.[A-Za-z_][A-Za-z0-9_]*\s*=", # env.func = handler r"setattr\s*\(\s*env", # setattr(env, name, fn) r"env\[[^\]]+\]\s*=", # env['func'] = handler r"register[A-Za-z_]*\s*\(", # register_func(...) / env.register(...) r"env\.(?:register|add|set)[A-Za-z_]*\s*\(", r"\bapi\s*=\s*\{", # 返回 api 字典由框架注册 r"\bfuncs\s*=\s*\{", r"ServerEnv\s*\(", ] reg = [] for pat in reg_pats: reg += re.findall(pat, init_src) check(g, len(reg) > 0, "init.py 向 ServerEnv 注册契约函数/挂载 api", "注册命中数=%d 样例=%s" % (len(reg), sorted(set(reg))[:5])) check(g, "load_pbl_blueprint" in pkg_init, "__init__.py 重导出 load_pbl_blueprint（三处同步）", "命中=%s" % ("load_pbl_blueprint" in pkg_init)) # ============================================================== F. RBAC 路径 def group_f() -> None: """以 import load_path 后的运行时 PATHS 元组为权威（不扫源码文本，避免把 生成器模板串 '/api/x/%s.dspy' 误判为通配路径）。""" g = "F.RBAC路径" check(g, os.path.isfile(LOAD_PATH), "load_path.py 存在", rel(LOAD_PATH)) mod = None err = "" try: import importlib.util spec = importlib.util.spec_from_file_location("_lp_m1a", LOAD_PATH) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) except Exception as e: # noqa: BLE001 err = "%s: %s" % (type(e).__name__, e) check(g, mod is not None, "load_path.py 可导入执行", err or "导入成功") if mod is None: return paths = list(getattr(mod, "PATHS", ()) or ()) check(g, len(paths) > 0, "RBAC 注册路径显式枚举（PATHS 元组）", "路径数=%d" % len(paths)) wild = [p for p in paths if "%" in p or "*" in p] check(g, not wild, "RBAC 路径无 %/* 通配符（运行时真实值）", "通配路径=%s" % (wild[:5] or "无")) nonstd = [p for p in paths if not re.fullmatch(r"/api/[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+\.dspy", p)] check(g, not nonstd, "RBAC 路径均为 /api/<表>/<动作>.dspy 规范字面量", "异常=%s" % (nonstd[:5] or "无")) dup = sorted({p for p in paths if paths.count(p) > 1}) check(g, not dup, "RBAC 路径无重复", "重复=%s" % (dup[:5] or "无")) # 覆盖 json/*.json editable 的 3×11 条（超集合法：支撑表含额外动作） need = set() if os.path.isdir(JSON_DIR): for fn in sorted(os.listdir(JSON_DIR)): if not fn.endswith(".json"): continue d, _e = load_json(os.path.join(JSON_DIR, fn)) for u in ((d or {}).get("editable") or []): if isinstance(u, str) and u.startswith("api/"): need.add("/" + u) missing = sorted(need - set(paths)) check(g, not missing, "load_path.py 覆盖 json/*.json editable 全部路径（超集）", "load_path=%d json_editable=%d 未覆盖=%s" % (len(paths), len(need), missing[:5] or "无")) # 孤儿路径：每条路径的表名必须归属主清单 11 表或支撑表 9 表 main_t = set(getattr(mod, "MAIN_TABLES", ()) or ()) sup_t = set(getattr(mod, "SUPPORT_TABLES", ()) or ()) known = main_t | sup_t check(g, len(main_t) == 11, "load_path.MAIN_TABLES = 设计定稿 11 表", "实际=%d" % len(main_t)) orphan = [] for p in paths: m = re.match(r"/api/([a-z0-9_]+)/", p) if not m or (known and m.group(1) not in known): orphan.append(p) check(g, not orphan, "load_path.py 无孤儿路径（均归属已知表）", "孤儿=%s" % (orphan[:5] or "无")) # 模块自带 validate() 必须通过 vfn = getattr(mod, "validate", None) if callable(vfn): verrs = vfn() or [] check(g, not verrs, "load_path.validate() 自检通过", "错误=%s" % (verrs[:3] or "无")) else: check(g, False, "load_path.validate() 存在", "未定义 validate()") # ============================================================== G. 租户 fail-closed def group_g() -> None: g = "G.租户fail-closed" biz = py_files(exclude_dirs=("scripts", "tests")) src_all = "\n".join(read(p) for p in biz) check(g, "ERR_TENANT_MISSING" in src_all, "业务代码引用 ERR_TENANT_MISSING（缺租户即拒）", "命中=%s" % ("ERR_TENANT_MISSING" in src_all)) check(g, os.path.isfile(os.path.join(PKG_DIR, "tenant.py")), "tenant.py 租户上下文模块存在", rel(os.path.join(PKG_DIR, "tenant.py"))) tsrc = read(os.path.join(PKG_DIR, "tenant.py")) check(g, "tenant_id" in tsrc, "tenant.py 处理 tenant_id", "命中=%s" % ("tenant_id" in tsrc)) # 禁止默认租户兜底 bad = re.findall(r"tenant_id\s*=\s*['\"](?:default|DEFAULT|1|0|test)['\"]", src_all) check(g, not bad, "无默认租户兜底（fail-closed）", "命中=%s" % (bad or "无")) # ============================================================== H. DDL def group_h() -> None: g = "H.DDL" sqls = sorted(f for f in os.listdir(SQL_DIR) if f.endswith(".sql")) if os.path.isdir(SQL_DIR) else [] check(g, len(sqls) >= 2, "sql/ DDL 文件存在（core + subobjects）", "文件=%s" % sqls) ddl = "\n".join(read(os.path.join(SQL_DIR, f)) for f in sqls) tables = sorted(set(re.findall(r"CREATE\s+TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?`?([a-z0-9_]+)`?", ddl, re.I))) check(g, len(tables) > 0, "DDL 含 CREATE TABLE 语句", "建表数=%d" % len(tables)) # 设计定稿 B1~B3 核心表（B4~B11 为独立建表，非泛化单表） need_core = ["pbl_blueprint", "pbl_blueprint_version", "pbl_blueprint_approval"] miss = [t for t in need_core if t not in tables] check(g, not miss, "DDL 覆盖核心表 B1~B3（聚合根/版本/审批）", "缺失=%s" % (miss or "无")) # 设计定稿 B4~B11 八张子对象表必须独立建表 need_sub = ["pbl_learner", "pbl_learning_goal", "pbl_problem", "pbl_driving_question", "pbl_project", "pbl_role", "pbl_mission", "pbl_artifact_def"] miss_sub = [t for t in need_sub if t not in tables] check(g, not miss_sub, "DDL 覆盖子对象表 B4~B11（8 张独立建表）", "缺失=%s 建表总数=%d" % (miss_sub or "无", len(tables))) # models 表 -> 物理承载映射核验 mfiles = sorted(f[:-5] for f in os.listdir(MODELS_DIR) if f.endswith(".json")) \ if os.path.isdir(MODELS_DIR) else [] # 设计定稿：models 主清单 11 表 = DDL 独立建表，逐表必须有 CREATE TABLE uncovered = [t for t in mfiles if t not in tables] check(g, not uncovered, "models 主清单 11 表均有独立 CREATE TABLE 承载", "未承载=%s models=%d ddl建表=%d" % (uncovered or "无", len(mfiles), len(tables))) check(g, "tenant_id" in ddl, "DDL 各表含 tenant_id 列", "命中=%s" % ("tenant_id" in ddl)) # ============================================================== I. git 证据 def group_i() -> None: g = "I.git证据" def git(*args): r = subprocess.run(["git", "-C", REPO_ROOT] + list(args), capture_output=True, text=True) return r.returncode, (r.stdout or "").strip(), (r.stderr or "").strip() rc, out, _ = git("rev-parse", "--is-inside-work-tree") check(g, rc == 0 and out == "true", "模块目录为 git 仓库", "rc=%d out=%s" % (rc, out)) rc, head, _ = git("log", "-1", "--format=%h %s") check(g, rc == 0 and bool(head), "存在本地提交（HEAD）", head[:120]) rc, st, _ = git("status", "--porcelain") dirty = [x for x in st.split("\n") if x.strip()] check(g, not dirty, "工作区 clean（变更已全部提交）", "未提交条目=%d %s" % (len(dirty), dirty[:5])) rc, rem, _ = git("remote", "-v") check(g, "origin" in rem, "配置 origin 远程", rem.split("\n")[0][:100] if rem else "无") def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--report", default="", help="报告输出路径") args = ap.parse_args() groups = [ ("A.目录结构", group_a), ("B.表定义", group_b), ("C.CRUD契约", group_c), ("D.代码质量", group_d), ("E.注册同步", group_e), ("F.RBAC路径", group_f), ("G.租户fail-closed", group_g), ("H.DDL", group_h), ("I.git证据", group_i), ] header = "=" * 78 LINES.append(header) LINES.append("pbl_blueprint M1a 自查报告（selfcheck.py 重写版）") LINES.append("仓库根: %s" % REPO_ROOT) LINES.append(header) print("\n".join(LINES[-4:])) for name, fn in groups: LINES.append("-" * 78) print("-" * 78) try: fn() except Exception as e: # noqa: BLE001 import traceback record(name, False, "检查组执行异常", "%s: %s" % (type(e).__name__, e)) LINES.append(traceback.format_exc()) LINES.append(header) total = N_PASS + N_FAIL verdict = "ALL PASS" if N_FAIL == 0 else "HAS FAILURES" LINES.append("结论: %s | PASS=%d FAIL=%d 总=%d" % (verdict, N_PASS, N_FAIL, total)) LINES.append(header) print("\n".join(LINES[-3:])) if args.report: try: os.makedirs(os.path.dirname(os.path.abspath(args.report)), exist_ok=True) with open(args.report, "w", encoding="utf-8") as f: f.write("\n".join(LINES) + "\n") print("报告已写入: %s" % args.report) except OSError as e: print("报告写入失败: %s" % e) # 门禁：有失败项必须非 0 退出 return 0 if N_FAIL == 0 else 1 if __name__ == "__main__": sys.exit(main())