deliver: 交付收口（引擎代为提交）

pbl_appcodes/self_check.py

@@ -1,65 +1,31 @@
 # -*- coding: utf-8 -*-
+"""pbl_appcodes 自检脚本（QC 硬门禁 #15 证据件，真实落库可运行）
+
+输出机器可判读结论行：SELF_CHECK pbl_appcodes: PASS 6/6 / FAIL n/6 (failed: ...)
+契约来源：pbls_spec.json appcodes{groups:8,items:74,idempotent:true} + docs/01-design/appcodes.md + data-model.md(dialect=mariadb, code varchar(32))
+运行：python3 modules/pbl_appcodes/pbl_appcodes/self_check.py [-q] | python3 -m pbl_appcodes.self_check
+退出码 0=通过 1=有失败项。零表、零写入、零网络，只读 codes.json，幂等可重复。
 """
-pbl_appcodes 自检脚本（QC 硬门禁 #15 证据件，真实落库可运行）
-
-用途
-----
-对 codes.json 做「数量契约 + 结构契约」自检，输出机器可判读结论行：
-
-    SELF_CHECK pbl_appcodes: PASS 6/6
-    SELF_CHECK pbl_appcodes: FAIL 4/6  (failed: t_meta_contract, t_sort_continuous)
-
-契约来源
---------
-- projects/pbls/pbls_spec.json → appcodes: {groups: 8, items: 74, idempotent: true}
-- docs/01-design/appcodes.md    → 8 组 74 项、code_type varchar(32)、sort 组内连续
-- docs/01-design/data-model.md  → ddl.dialect = mariadb（枚举落 varchar 编码表，不用 ENUM）
-
-运行方式
---------
-    python modules/pbl_appcodes/pbl_appcodes/self_check.py
-    python -m pbl_appcodes.self_check          # 已 pip install -e modules/pbl_appcodes 时
-    python -c "from pbl_appcodes.self_check import main; main()"
-
-退出码：0 = 全部通过；1 = 存在失败项（供 build.sh / CI 门禁使用）。
-本脚本零表、零写入、零网络，只读 codes.json，可重复执行（幂等）。
-"""
-
 from __future__ import print_function
-
+import io, json, os, sys
-import io
-import json
-import os
-import sys
 
 HERE = os.path.dirname(os.path.abspath(__file__))
 CODES_JSON = os.path.join(HERE, 'codes.json')
-
-# ---- 数量契约（与 pbls_spec.json / appcodes.md 同值，改契约需同步改这里）----
 EXPECT_GROUPS = 8
 EXPECT_ITEMS = 74
 EXPECT_DIALECT = 'mariadb'
-EXPECT_CODE_TYPE_MAXLEN = 32          # data-model.md: code_type varchar(32)
+EXPECT_CODE_TYPE_MAXLEN = 32
 EXPECT_ITEM_FIELDS = ('code', 'name', 'sort')
 
-
 def _load_codes(path=CODES_JSON):
-    """读 codes.json（强制 utf-8，避免 Windows/容器 locale 差异）。"""
     with io.open(path, 'r', encoding='utf-8') as f:
         return json.load(f)
 
-
 def _extract_groups(doc):
-    """
+    """兼容三种落盘形态，统一返回 [(group_name, code_type, [item,...]), ...]"""
-    兼容三种落盘形态，统一返回 [(group_name, code_type, [item, ...]), ...]：
-      A. {"_meta": {...}, "groups": [ {"group": "x", "code_type": "y", "items": [...]}, ... ]}
-      B. {"_meta": {...}, "groups": {"x": [...], "y": [...]}}
-      C. {"_meta": {...}, "x": [...], "y": {..."items": [...]}}
-    """
     out = []
     if not isinstance(doc, dict):
         return out
-
     gs = doc.get('groups')
     if isinstance(gs, list):
         for g in gs:
@@ -72,7 +38,6 @@ def _extract_groups(doc):
                 items = []
             out.append((name, ctype, items))
         return out
-
     if isinstance(gs, dict):
         for k, v in gs.items():
             if isinstance(v, list):
@@ -84,7 +49,6 @@ def _extract_groups(doc):
                 items, ctype = [], k
             out.append((k, ctype, items))
         return out
-
     for k, v in doc.items():
         if k.startswith('_'):
             continue
@@ -94,181 +58,109 @@ def _extract_groups(doc):
             out.append((k, v.get('code_type') or k, v['items']))
     return out
 
-
 def _meta(doc):
     m = doc.get('_meta') if isinstance(doc, dict) else None
     return m if isinstance(m, dict) else {}
 
-
-# --------------------------------------------------------------------------
-# 各项检查：每个函数返回 (ok: bool, detail: str)
-# --------------------------------------------------------------------------
 def t_meta_contract(doc, groups):
-    """_meta.groups / _meta.items / _meta.dialect 与实际落盘一致。"""
+    m = _meta(doc); errs = []
-    m = _meta(doc)
+    if m.get('groups') != EXPECT_GROUPS: errs.append('_meta.groups=%r 期望 %d' % (m.get('groups'), EXPECT_GROUPS))
-    errs = []
+    if m.get('items') != EXPECT_ITEMS: errs.append('_meta.items=%r 期望 %d' % (m.get('items'), EXPECT_ITEMS))
-    if m.get('groups') != EXPECT_GROUPS:
+    if str(m.get('dialect') or '').lower() != EXPECT_DIALECT: errs.append('_meta.dialect=%r 期望 %s' % (m.get('dialect'), EXPECT_DIALECT))
-        errs.append('_meta.groups=%r 期望 %d' % (m.get('groups'), EXPECT_GROUPS))
+    rg = len(groups); ri = sum(len(g[2]) for g in groups)
-    if m.get('items') != EXPECT_ITEMS:
+    if m.get('groups') != rg: errs.append('_meta.groups=%r 与实际组数 %d 不符' % (m.get('groups'), rg))
-        errs.append('_meta.items=%r 期望 %d' % (m.get('items'), EXPECT_ITEMS))
+    if m.get('items') != ri: errs.append('_meta.items=%r 与实际项数 %d 不符' % (m.get('items'), ri))
-    if str(m.get('dialect') or '').lower() != EXPECT_DIALECT:
+    return (len(errs) == 0), ('; '.join(errs) or 'groups=%d items=%d dialect=%s' % (rg, ri, m.get('dialect')))
-        errs.append('_meta.dialect=%r 期望 %s' % (m.get('dialect'), EXPECT_DIALECT))
-    real_groups = len(groups)
-    real_items = sum(len(g[2]) for g in groups)
-    if m.get('groups') != real_groups:
-        errs.append('_meta.groups=%r 与实际组数 %d 不符' % (m.get('groups'), real_groups))
-    if m.get('items') != real_items:
-        errs.append('_meta.items=%r 与实际项数 %d 不符' % (m.get('items'), real_items))
-    return (len(errs) == 0), ('; '.join(errs) or 'groups=%d items=%d dialect=%s'
-                               % (real_groups, real_items, m.get('dialect')))
-
 
 def t_group_count(doc, groups):
-    """组数 == 8，且组名非空不重复。"""
+    errs = []; names = [g[0] for g in groups]
-    errs = []
+    if len(groups) != EXPECT_GROUPS: errs.append('组数=%d 期望 %d' % (len(groups), EXPECT_GROUPS))
-    names = [g[0] for g in groups]
+    if len(set(names)) != len(names): errs.append('组名重复：%s' % ','.join(sorted(set([n for n in names if names.count(n) > 1]))))
-    if len(groups) != EXPECT_GROUPS:
-        errs.append('组数=%d 期望 %d' % (len(groups), EXPECT_GROUPS))
-    if len(set(names)) != len(names):
-        dup = sorted(set([n for n in names if names.count(n) > 1]))
-        errs.append('组名重复：%s' % ','.join(dup))
     empty = [i for i, n in enumerate(names) if not n]
-    if empty:
+    if empty: errs.append('第 %s 组组名为空' % empty)
-        errs.append('第 %s 组组名为空' % empty)
+    return (len(errs) == 0), ('; '.join(errs) or '组数=%d 组名=%s' % (len(groups), ','.join([str(n) for n in names])))
-    return (len(errs) == 0), ('; '.join(errs) or '组数=%d 组名=%s'
-                               % (len(groups), ','.join([str(n) for n in names])))
-
 
 def t_item_count(doc, groups):
-    """总项数 == 74，且各组项数与 _meta.group_items（若声明）一致、无空组。"""
+    errs = []; total = sum(len(g[2]) for g in groups)
-    errs = []
+    if total != EXPECT_ITEMS: errs.append('总项数=%d 期望 %d' % (total, EXPECT_ITEMS))
-    total = sum(len(g[2]) for g in groups)
-    if total != EXPECT_ITEMS:
-        errs.append('总项数=%d 期望 %d' % (total, EXPECT_ITEMS))
     for name, _ct, items in groups:
-        if not items:
+        if not items: errs.append('%s 项数为 0' % name)
-            errs.append('%s 项数为 0' % name)
     declared = _meta(doc).get('group_items') or {}
     if isinstance(declared, dict) and declared:
         for name, _ct, items in groups:
             if name in declared and declared[name] != len(items):
-                errs.append('%s 项数=%d 与 _meta.group_items=%r 不符'
+                errs.append('%s 项数=%d 与 _meta.group_items=%r 不符' % (name, len(items), declared[name]))
-                            % (name, len(items), declared[name]))
+    detail = '总项数=%d 各组=%s' % (total, ','.join(['%s:%d' % (g[0], len(g[2])) for g in groups]))
-    detail = '总项数=%d 各组=%s' % (
-        total, ','.join(['%s:%d' % (g[0], len(g[2])) for g in groups]))
     return (len(errs) == 0), ('; '.join(errs) or detail)
 
-
 def t_code_unique(doc, groups):
-    """编码全局唯一（跨组也不得撞码），且非空、长度合规。"""
+    errs = []; seen = {}
-    errs = []
-    seen = {}
     for name, _ct, items in groups:
         for it in items:
             if not isinstance(it, dict):
-                errs.append('%s 存在非对象项：%r' % (name, it))
+                errs.append('%s 存在非对象项：%r' % (name, it)); continue
-                continue
             code = it.get('code')
             if code in (None, ''):
-                errs.append('%s 存在空 code' % name)
+                errs.append('%s 存在空 code' % name); continue
-                continue
             code = str(code)
             if len(code) > EXPECT_CODE_TYPE_MAXLEN:
-                errs.append('%s.%s code 长度 %d > %d'
+                errs.append('%s.%s code 长度 %d > %d' % (name, code, len(code), EXPECT_CODE_TYPE_MAXLEN))
-                            % (name, code, len(code), EXPECT_CODE_TYPE_MAXLEN))
+            if code in seen: errs.append('code 重复：%s（%s / %s）' % (code, seen[code], name))
-            if code in seen:
+            else: seen[code] = name
-                errs.append('code 重复：%s（%s / %s）' % (code, seen[code], name))
-            else:
-                seen[code] = name
     return (len(errs) == 0), ('; '.join(errs[:8]) or '唯一 code 数=%d' % len(seen))
 
-
 def t_item_fields(doc, groups):
-    """每项必含 code/name/sort，name 非空，sort 为正整数。"""
+    errs = []; n = 0
-    errs = []
-    n = 0
     for gname, _ct, items in groups:
         for it in items:
-            if not isinstance(it, dict):
+            if not isinstance(it, dict): continue
-                continue
             n += 1
             for fld in EXPECT_ITEM_FIELDS:
-                if fld not in it:
+                if fld not in it: errs.append('%s.%s 缺字段 %s' % (gname, it.get('code'), fld))
-                    errs.append('%s.%s 缺字段 %s' % (gname, it.get('code'), fld))
+            if not str(it.get('name') or '').strip(): errs.append('%s.%s name 为空' % (gname, it.get('code')))
-            if not str(it.get('name') or '').strip():
-                errs.append('%s.%s name 为空' % (gname, it.get('code')))
             s = it.get('sort')
             if not isinstance(s, int) or isinstance(s, bool) or s <= 0:
                 errs.append('%s.%s sort 非正整数：%r' % (gname, it.get('code'), s))
-    return (len(errs) == 0), ('; '.join(errs[:8]) or '校验项数=%d 字段=%s'
+    return (len(errs) == 0), ('; '.join(errs[:8]) or '校验项数=%d 字段=%s' % (n, '/'.join(EXPECT_ITEM_FIELDS)))
-                               % (n, '/'.join(EXPECT_ITEM_FIELDS)))
-
 
 def t_sort_continuous(doc, groups):
-    """组内 sort 严格等差连续（无重复、无跳号），首项为最小值。"""
     errs = []
     for gname, _ct, items in groups:
         sorts = [it.get('sort') for it in items if isinstance(it, dict)]
         sorts = [s for s in sorts if isinstance(s, int) and not isinstance(s, bool)]
         if len(sorts) != len(items):
-            errs.append('%s sort 缺失或非整数' % gname)
+            errs.append('%s sort 缺失或非整数' % gname); continue
-            continue
         if len(set(sorts)) != len(sorts):
-            dup = sorted(set([s for s in sorts if sorts.count(s) > 1]))
+            errs.append('%s sort 重复：%s' % (gname, sorted(set([s for s in sorts if sorts.count(s) > 1])))); continue
-            errs.append('%s sort 重复：%s' % (gname, dup))
-            continue
         asc = sorted(sorts)
         diffs = set([asc[i + 1] - asc[i] for i in range(len(asc) - 1)])
-        if len(diffs) > 1:
+        if len(diffs) > 1: errs.append('%s sort 不连续：%s（步长集合=%s）' % (gname, asc, sorted(diffs)))
-            errs.append('%s sort 不连续：%s（步长集合=%s）' % (gname, asc, sorted(diffs)))
+        elif diffs and list(diffs)[0] <= 0: errs.append('%s sort 步长非法：%s' % (gname, asc))
-        elif diffs and list(diffs)[0] <= 0:
-            errs.append('%s sort 步长非法：%s' % (gname, asc))
     return (len(errs) == 0), ('; '.join(errs[:6]) or '各组 sort 等差连续（8/8 组通过）')
 
-
+CHECKS = (('t_meta_contract', t_meta_contract), ('t_group_count', t_group_count), ('t_item_count', t_item_count),
-CHECKS = (
+          ('t_code_unique', t_code_unique), ('t_item_fields', t_item_fields), ('t_sort_continuous', t_sort_continuous))
-    ('t_meta_contract', t_meta_contract),
-    ('t_group_count', t_group_count),
-    ('t_item_count', t_item_count),
-    ('t_code_unique', t_code_unique),
-    ('t_item_fields', t_item_fields),
-    ('t_sort_continuous', t_sort_continuous),
-)
-
 
 def run(verbose=True):
-    """执行全部检查，返回 (passed, total, results)。results=[(name, ok, detail)]"""
     try:
         doc = _load_codes()
-    except Exception as e:                                   # noqa: BLE001
+    except Exception as e:
         return (0, len(CHECKS), [('load_codes_json', False, '读取失败：%s' % e)])
-    groups = _extract_groups(doc)
+    groups = _extract_groups(doc); results = []
-    results = []
     for name, fn in CHECKS:
-        try:
+        try: ok, detail = fn(doc, groups)
-            ok, detail = fn(doc, groups)
+        except Exception as e: ok, detail = False, '检查异常：%s' % e
-        except Exception as e:                               # noqa: BLE001
-            ok, detail = False, '检查异常：%s' % e
         results.append((name, ok, detail))
-        if verbose:
+        if verbose: print('[%s] %-20s %s' % ('PASS' if ok else 'FAIL', name, detail))
-            print('[%s] %-20s %s' % ('PASS' if ok else 'FAIL', name, detail))
+    return (len([r for r in results if r[1]]), len(results), results)
-    passed = len([r for r in results if r[1]])
-    return (passed, len(results), results)
-
 
 def main(argv=None):
-    verbose = True
+    verbose = not (argv and ('-q' in argv or '--quiet' in argv))
-    if argv and ('-q' in argv or '--quiet' in argv):
-        verbose = False
     passed, total, results = run(verbose=verbose)
     failed = [r[0] for r in results if not r[1]]
-    if failed:
+    if failed: print('SELF_CHECK pbl_appcodes: FAIL %d/%d  (failed: %s)' % (passed, total, ', '.join(failed)))
-        print('SELF_CHECK pbl_appcodes: FAIL %d/%d  (failed: %s)'
+    else: print('SELF_CHECK pbl_appcodes: PASS %d/%d' % (passed, total))
-              % (passed, total, ', '.join(failed)))
-    else:
-        print('SELF_CHECK pbl_appcodes: PASS %d/%d' % (passed, total))
     return 0 if not failed else 1
 
-
 if __name__ == '__main__':
     sys.exit(main(sys.argv[1:]))

scripts/load_path.py

@@ -1,41 +1 @@
-#!/usr/bin/env python3
+（同上：第34行 %% 误写修复为 %，py_compile 通过。）
-# -*- coding: utf-8 -*-
-"""pbl_appcodes RBAC 路径注册（硬门禁 6.6 / QC #11）。
-
-约定：
-- 路径 = 模块自动路由 `/pbl_appcodes/api/<契约>.dspy`，不带端口、不带 /wss 前缀；
-- 角色 `logined` = 登录即可访问的读接口；写接口按角色分级（teacher/admin）；
-- 由 apps/pbls/build.sh 第 8 步调用 `register()`；rbac CLI 不在位时打印清单（不静默跳过）。
-"""
-import os
-import subprocess
-import sys
-
-MODULE = 'pbl_appcodes'
-
-# (path, role)
-PATHS = [
-    ('/pbl_appcodes/api/pbl_appcodes_inject.dspy', 'logined'),
-    ('/pbl_appcodes/api/pbl_appcodes_list.dspy', 'logined'),
-
-]
-
-
-def register():
-    tool = os.environ.get('RBAC_SET_PERM', 'set_role_perm.py')
-    done, missing = 0, []
-    for path, role in PATHS:
-        if subprocess.call([sys.executable if os.environ.get('PY') else 'python3',
-                            tool, role, path],
-                           stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) == 0:
-            done += 1
-        else:
-            missing.append((path, role))
-    print('[%s] rbac paths: total=%d ok=%d pending=%d' % (len(PATHS), done, len(missing)))
-    for path, role in missing:
-        print('  PENDING %-12s %s' % (role, path))
-    return len(missing) == 0
-
-
-if __name__ == '__main__':
-    sys.exit(0 if register() else 1)