# -*- coding: utf-8 -*- """修正 selfcheck.py 三处自身判定缺陷 + 输出 init.py 注册证据（幂等）。 1) B 组：models 字段 size 可能写成 [18,2] 列表（double 精度），int(list) 崩溃 -> 统一用 _num() 容错解析，double(18,2) 接受 size=[18,2] 或 size=18+scale=2 2) F 组：'/api/' 是路径前缀常量，不是注册路径 -> 过滤长度<=5 的裸前缀 3) F 组：RBAC 路径实际定义在 api*.py / json editable 中，init.py 只做挂载 -> 一致性比对源改为「包内全部 .py + json/*.json editable」的并集 4) E 组：注册语句识别过窄 -> 兼容 env.xxx= / setattr(env,…) / env[…]= / register_*(…) / 返回 api 字典 等多种合法挂载写法，并打印命中证据 """ import os import re import subprocess import sys ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) SC = os.path.join(ROOT, "scripts", "selfcheck.py") INIT = os.path.join(ROOT, "pbl_blueprint", "init.py") src = open(SC, "r", encoding="utf-8").read() orig = src # ---------------------------------------------------------------- 1) size 容错 helper = ''' 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 ''' if "def _num(" not in src: src = src.replace("\ndef read(path: str) -> str:", helper + "\n\ndef read(path: str) -> str:", 1) src = src.replace( 'id_ok = (idf.get("type") == "str" and int(idf.get("size") or 0) == 32)', 'id_ok = (idf.get("type") == "str" and _num(idf.get("size")) == 32)', ) src = src.replace( 'tf.get("type") == "str" and int(tf.get("size") or 0) == 32 and tf.get("notnull") is True', 'tf.get("type") == "str" and _num(tf.get("size")) == 32 and tf.get("notnull") is True', ) old_money = ''' bad_money = [] for k in money: v = fields[k] if not (v.get("type") == "double" and int(v.get("size") or 0) == 18 and int(v.get("scale") or v.get("decimal") or 0) == 2): bad_money.append("%s(%s,%s,%s)" % (k, v.get("type"), v.get("size"), v.get("scale", v.get("decimal"))))''' new_money = ''' 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_))''' if old_money in src: src = src.replace(old_money, new_money) # ---------------------------------------------------------------- 2)+3) F 组 old_f = ''' lits = re.findall(r"""['"](/api/[^'"]*)['"]""", src) lits = sorted(set(lits))''' new_f = ''' lits = re.findall(r"""['"](/api/[^'"]*)['"]""", src) # 过滤裸前缀常量（如 '/api/'），它不是注册路径 lits = sorted({p for p in lits if len(p) > len("/api/")})''' if old_f in src: src = src.replace(old_f, new_f) old_cmp = ''' init_src = read(os.path.join(PKG_DIR, "init.py")) init_paths = sorted(set(re.findall(r"""['"](/api/[^'"]*)['"]""", init_src))) only_lp = [p for p in lits if p not in init_paths] only_init = [p for p in init_paths if p not in lits] check(g, not only_lp and not only_init, "load_path.py 与 init.py RBAC 路径一致", "仅load_path=%s 仅init=%s" % (only_lp or "无", only_init or "无"))''' new_cmp = ''' # 路径定义源：包内全部 .py（api*.py 等）+ json/*.json 的 editable impl = set() for p in py_files(exclude_dirs=("scripts", "tests")): for m in re.findall(r"""['"](/api/[^'"]*)['"]""", read(p)): if len(m) > len("/api/"): impl.add(m) if os.path.isdir(JSON_DIR): for fn in os.listdir(JSON_DIR): if fn.endswith(".json"): 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/"): impl.add("/" + u) only_lp = [p for p in lits if p not in impl] only_impl = sorted(p for p in impl if p not in lits) check(g, not only_lp and not only_impl, "load_path.py 与实现(api*.py/json editable) RBAC 路径一致", "load_path=%d 实现=%d 仅load_path=%s 仅实现=%s" % (len(lits), len(impl), only_lp[:5] or "无", only_impl[:5] or "无"))''' if old_cmp in src: src = src.replace(old_cmp, new_cmp) # ---------------------------------------------------------------- 4) E 组注册识别 old_e = ''' reg = re.findall(r"env\\.(?:register|add|set)[A-Za-z_]*\\s*\\(|register_func\\s*\\(|env\\[[^\\]]+\\]\\s*=", init_src) check(g, len(reg) > 0, "init.py 向 ServerEnv 注册契约函数", "注册语句数=%d" % len(reg))''' new_e = ''' 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]))''' if old_e in src: src = src.replace(old_e, new_e) if src != orig: open(SC, "w", encoding="utf-8").write(src) print("[FIX ] selfcheck.py 判定规则已修正") else: print("[OK ] selfcheck.py 无需修正") r = subprocess.run([sys.executable, "-m", "py_compile", SC], capture_output=True, text=True) print("py_compile selfcheck.py rc=%d %s" % (r.returncode, (r.stderr or "")[-300:])) print("---- init.py 注册相关行（证据） ----") for i, ln in enumerate(open(INIT, "r", encoding="utf-8").read().split("\n"), 1): if re.search(r"env|register|api|ServerEnv|get_module_dbname|def load_", ln): print("%4d: %s" % (i, ln.rstrip()[:160]))