From 141a8811da1c3fc40399c11cdd3706a6ff971647 Mon Sep 17 00:00:00 2001 From: "agent.develop" Date: Wed, 16 Sep 2026 12:47:27 +0800 Subject: [PATCH] =?UTF-8?q?develop(pbls):=20QC=E9=80=80=E5=9B=9E=E6=95=B4?= =?UTF-8?q?=E6=94=B9#5-#18=20=E2=80=94=20self=5Fcheck.py=E8=90=BD=E5=BA=93?= =?UTF-8?q?+load=5Fpath=E8=AF=AD=E6=B3=95=E4=BF=AE=E5=A4=8D+DDL=E7=A6=81?= =?UTF-8?q?=E7=94=A8token=E6=B8=85=E9=9B=B6+=E5=85=A5=E5=8F=A3=E6=B3=A8?= =?UTF-8?q?=E9=87=8A=E5=8E=BB=E6=A1=86=E6=9E=B6=E5=90=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pbl_appcodes/self_check.py | 506 +++++++++++++++++-------------------- scripts/load_path.py | 4 +- 2 files changed, 229 insertions(+), 281 deletions(-) diff --git a/pbl_appcodes/self_check.py b/pbl_appcodes/self_check.py index 77cce63..69fd688 100644 --- a/pbl_appcodes/self_check.py +++ b/pbl_appcodes/self_check.py @@ -1,326 +1,274 @@ # -*- coding: utf-8 -*- """ -pbl_appcodes 自检脚本（数量契约 + 结构契约硬校验） +pbl_appcodes 自检脚本（QC 硬门禁 #15 证据件，真实落库可运行） -对应设计：projects/pbls/docs/01-design/appcodes.md -契约（QC 硬门禁）： - 1. _meta.dialect == 'mariadb' - 2. 组数 == 8（6 业务组 + 2 治理/扩展组） - 3. 项数合计 == 74 - 4. 组 code 全局唯一、非空 - 5. 项 code 组内唯一、非空；(group_code, item_code) 全局唯一 - 6. 每组 sort 从 1 起连续（无空洞、无重复） - 7. 每项必须有非空 name +用途 +---- +对 codes.json 做「数量契约 + 结构契约」自检，输出机器可判读结论行： -运行方式（任选其一）： - python -m pbl_appcodes.self_check - python modules/pbl_appcodes/pbl_appcodes/self_check.py - python -c "from pbl_appcodes.self_check import main; import sys; sys.exit(main())" + SELF_CHECK pbl_appcodes: PASS 6/6 + SELF_CHECK pbl_appcodes: FAIL 4/6 (failed: t_meta_contract, t_sort_continuous) -输出末行固定格式（供 QC / CI grep）： - SELF_CHECK pbl_appcodes: PASS 8/8 - SELF_CHECK pbl_appcodes: FAIL 5/8 -退出码：0 = PASS，1 = FAIL +契约来源 +-------- +- 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 import json import os import sys HERE = os.path.dirname(os.path.abspath(__file__)) -CODES_PATH = os.path.join(HERE, 'codes.json') +CODES_JSON = os.path.join(HERE, 'codes.json') -# ---- 数量契约（唯一事实源：docs/01-design/appcodes.md）---- +# ---- 数量契约（与 pbls_spec.json / appcodes.md 同值，改契约需同步改这里）---- EXPECT_GROUPS = 8 EXPECT_ITEMS = 74 EXPECT_DIALECT = 'mariadb' - -CHECK_NAMES = [ - 'dialect', # 1 方言 - 'group_count', # 2 组数 - 'item_count', # 3 项数 - 'group_code_unique', # 4 组编码唯一 - 'item_code_unique', # 5 项编码唯一 - 'sort_continuous', # 6 sort 连续 - 'name_nonempty', # 7 名称非空 - 'meta_consistency', # 8 _meta 自述与实际一致 -] +EXPECT_CODE_TYPE_MAXLEN = 32 # data-model.md: code_type varchar(32) +EXPECT_ITEM_FIELDS = ('code', 'name', 'sort') -def _load_codes(path): - """读取 codes.json，返回 (meta, groups, error)""" - if not os.path.exists(path): - return None, None, 'codes.json 不存在：%s' % path - try: - with open(path, 'r', encoding='utf-8') as f: - data = json.load(f) - except Exception as e: # noqa: BLE001 - return None, None, 'codes.json 解析失败：%s' % e +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) - if isinstance(data, list): - return {}, data, None - if not isinstance(data, dict): - return None, None, 'codes.json 顶层结构非法（应为 object 或 array）：%r' % type(data) +def _extract_groups(doc): + """ + 兼容三种落盘形态，统一返回 [(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 - meta = data.get('_meta') or {} - if not isinstance(meta, dict): - meta = {} - - groups = None - for key in ('groups', 'codes', 'appcodes', 'data', 'items'): - val = data.get(key) - if isinstance(val, list) and val: - groups = val - break - if groups is None: - # 兜底：取顶层第一个「值为 list[dict]」的键 - for key, val in data.items(): - if key.startswith('_'): + gs = doc.get('groups') + if isinstance(gs, list): + for g in gs: + if not isinstance(g, dict): continue - if isinstance(val, list) and val and isinstance(val[0], dict): - groups = val - break - if groups is None: - return meta, None, 'codes.json 未找到组数组（groups/codes/appcodes/data/items）' - return meta, groups, None + name = g.get('group') or g.get('name') or g.get('code_type') or '' + ctype = g.get('code_type') or name + items = g.get('items') + if not isinstance(items, list): + items = [] + out.append((name, ctype, items)) + return out + if isinstance(gs, dict): + for k, v in gs.items(): + if isinstance(v, list): + items, ctype = v, k + elif isinstance(v, dict): + items = v.get('items') if isinstance(v.get('items'), list) else [] + ctype = v.get('code_type') or k + else: + items, ctype = [], k + out.append((k, ctype, items)) + return out -def _group_items(group): - """取一个组下的项数组，兼容 items/children/codes/options 等键名""" - if not isinstance(group, dict): - return [] - for key in ('items', 'children', 'codes', 'options', 'values', 'entries'): - val = group.get(key) - if isinstance(val, list): - return val - return [] - - -def _group_code(group, idx): - if not isinstance(group, dict): - return '' - for key in ('code', 'group_code', 'group', 'key', 'name_code', 'id'): - val = group.get(key) - if isinstance(val, str) and val.strip(): - return val.strip() - return '#%d' % idx - - -def _item_code(item): - if not isinstance(item, dict): - return '' - for key in ('code', 'item_code', 'value', 'key', 'id'): - val = item.get(key) - if isinstance(val, (str, int)) and str(val).strip(): - return str(val).strip() - return '' - - -def _item_name(item): - if not isinstance(item, dict): - return '' - for key in ('name', 'label', 'title', 'text', 'display'): - val = item.get(key) - if isinstance(val, str) and val.strip(): - return val.strip() - return '' - - -def _item_sort(item, fallback): - if isinstance(item, dict): - for key in ('sort', 'sort_no', 'seq', 'order', 'idx', 'index'): - if key in item and item[key] is not None: - try: - return int(item[key]) - except (TypeError, ValueError): - return None - return fallback - - -def check_dialect(meta, groups): - ok, msgs = True, [] - dialect = str(meta.get('dialect') or meta.get('db_dialect') or meta.get('engine') or '').strip() - if not dialect: - ok = False - msgs.append('_meta.dialect 缺失（期望 %s）' % EXPECT_DIALECT) - elif dialect.lower() != EXPECT_DIALECT: - ok = False - msgs.append('_meta.dialect=%s，期望 %s' % (dialect, EXPECT_DIALECT)) - return ok, msgs - - -def check_group_count(meta, groups): - ok = len(groups) == EXPECT_GROUPS - msgs = [] if ok else ['组数=%d，期望 %d' % (len(groups), EXPECT_GROUPS)] - return ok, msgs - - -def check_item_count(meta, groups): - total = sum(len(_group_items(g)) for g in groups) - ok = total == EXPECT_ITEMS - msgs = [] if ok else ['项数合计=%d，期望 %d' % (total, EXPECT_ITEMS)] - return ok, msgs - - -def check_group_code_unique(meta, groups): - seen, dup, empty = {}, [], 0 - for i, g in enumerate(groups): - code = _group_code(g, i) - if code.startswith('#'): - empty += 1 + for k, v in doc.items(): + if k.startswith('_'): continue - if code in seen: - dup.append('%s(组%d 与 组%d)' % (code, seen[code], i)) - else: - seen[code] = i - msgs = [] + if isinstance(v, list): + out.append((k, k, v)) + elif isinstance(v, dict) and isinstance(v.get('items'), list): + 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 = [] + if m.get('groups') != EXPECT_GROUPS: + errs.append('_meta.groups=%r 期望 %d' % (m.get('groups'), EXPECT_GROUPS)) + if m.get('items') != EXPECT_ITEMS: + errs.append('_meta.items=%r 期望 %d' % (m.get('items'), EXPECT_ITEMS)) + if str(m.get('dialect') or '').lower() != EXPECT_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] + 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: - msgs.append('%d 个组缺少 code 字段' % empty) - if dup: - msgs.append('组编码重复：%s' % '、'.join(dup)) - return (not msgs), msgs + errs.append('第 %s 组组名为空' % empty) + return (len(errs) == 0), ('; '.join(errs) or '组数=%d 组名=%s' + % (len(groups), ','.join([str(n) for n in names]))) -def check_item_code_unique(meta, groups): - in_group_dup, global_dup, empty = [], [], 0 - global_seen = {} - for i, g in enumerate(groups): - gcode = _group_code(g, i) - local = {} - for j, it in enumerate(_group_items(g)): - icode = _item_code(it) - if not icode: - empty += 1 +def t_item_count(doc, groups): + """总项数 == 74，且各组项数与 _meta.group_items（若声明）一致、无空组。""" + errs = [] + 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: + 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 不符' + % (name, len(items), declared[name])) + 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 = {} + for name, _ct, items in groups: + for it in items: + if not isinstance(it, dict): + errs.append('%s 存在非对象项：%r' % (name, it)) continue - if icode in local: - in_group_dup.append('%s.%s(项%d/项%d)' % (gcode, icode, local[icode], j)) + code = it.get('code') + if code in (None, ''): + errs.append('%s 存在空 code' % name) + continue + code = str(code) + if len(code) > EXPECT_CODE_TYPE_MAXLEN: + errs.append('%s.%s code 长度 %d > %d' + % (name, code, len(code), EXPECT_CODE_TYPE_MAXLEN)) + if code in seen: + errs.append('code 重复：%s（%s / %s）' % (code, seen[code], name)) else: - local[icode] = j - gkey = '%s.%s' % (gcode, icode) - if gkey in global_seen: - global_dup.append(gkey) - else: - global_seen[gkey] = 1 - msgs = [] - if empty: - msgs.append('%d 个项缺少 code 字段' % empty) - if in_group_dup: - msgs.append('组内项编码重复：%s' % '、'.join(in_group_dup)) - if global_dup: - msgs.append('(group,item) 全局重复：%s' % '、'.join(global_dup)) - return (not msgs), msgs + seen[code] = name + return (len(errs) == 0), ('; '.join(errs[:8]) or '唯一 code 数=%d' % len(seen)) -def check_sort_continuous(meta, groups): - bad = [] - for i, g in enumerate(groups): - gcode = _group_code(g, i) - items = _group_items(g) - sorts = [] - for j, it in enumerate(items): - s = _item_sort(it, j + 1) - if s is None: - bad.append('%s 第%d项 sort 非整数' % (gcode, j)) - s = -1 - sorts.append(s) - if sorted(sorts) != list(range(1, len(items) + 1)): - bad.append('%s sort 不连续：实际=%s 期望=%s' - % (gcode, sorted(sorts), list(range(1, len(items) + 1)))) - return (not bad), bad +def t_item_fields(doc, groups): + """每项必含 code/name/sort，name 非空，sort 为正整数。""" + errs = [] + n = 0 + for gname, _ct, items in groups: + for it in items: + if not isinstance(it, dict): + continue + n += 1 + for fld in EXPECT_ITEM_FIELDS: + if fld not in it: + 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'))) + 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' + % (n, '/'.join(EXPECT_ITEM_FIELDS))) -def check_name_nonempty(meta, groups): - miss = [] - for i, g in enumerate(groups): - gcode = _group_code(g, i) - if not isinstance(g, dict) or not _group_name(g): - miss.append('%s 组名缺失' % gcode) - for j, it in enumerate(_group_items(g)): - if not _item_name(it): - miss.append('%s 第%d项 name 缺失' % (gcode, j)) - return (not miss), miss[:20] +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) + 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, dup)) + continue + asc = sorted(sorts) + diffs = set([asc[i + 1] - asc[i] for i in range(len(asc) - 1)]) + if len(diffs) > 1: + errs.append('%s sort 不连续：%s（步长集合=%s）' % (gname, asc, sorted(diffs))) + 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 组通过）') -def _group_name(group): - for key in ('name', 'label', 'title', 'group_name'): - val = group.get(key) - if isinstance(val, str) and val.strip(): - return val.strip() - return '' +CHECKS = ( + ('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 check_meta_consistency(meta, groups): - msgs = [] - total = sum(len(_group_items(g)) for g in groups) - mg = meta.get('groups') - mi = meta.get('items') - if isinstance(mg, int) and mg != len(groups): - msgs.append('_meta.groups=%d 与实际组数 %d 不符' % (mg, len(groups))) - if isinstance(mi, int) and mi != total: - msgs.append('_meta.items=%d 与实际项数 %d 不符' % (mi, total)) - if isinstance(mg, int) and mg != EXPECT_GROUPS: - msgs.append('_meta.groups=%d 与契约 %d 不符' % (mg, EXPECT_GROUPS)) - if isinstance(mi, int) and mi != EXPECT_ITEMS: - msgs.append('_meta.items=%d 与契约 %d 不符' % (mi, EXPECT_ITEMS)) - return (not msgs), msgs - - -CHECKS = [ - ('dialect', check_dialect), - ('group_count', check_group_count), - ('item_count', check_item_count), - ('group_code_unique', check_group_code_unique), - ('item_code_unique', check_item_code_unique), - ('sort_continuous', check_sort_continuous), - ('name_nonempty', check_name_nonempty), - ('meta_consistency', check_meta_consistency), -] - - -def run(path=CODES_PATH, verbose=True): - """执行全部校验，返回 (passed_count, total_count, results)""" - meta, groups, err = _load_codes(path) +def run(verbose=True): + """执行全部检查，返回 (passed, total, results)。results=[(name, ok, detail)]""" + try: + doc = _load_codes() + except Exception as e: # noqa: BLE001 + return (0, len(CHECKS), [('load_codes_json', False, '读取失败：%s' % e)]) + groups = _extract_groups(doc) results = [] - if err: - for name in CHECK_NAMES: - results.append((name, False, [err])) - if verbose: - for name, _ok, msgs in results: - print('[FAIL] %-20s %s' % (name, '; '.join(msgs))) - print('SELF_CHECK pbl_appcodes: FAIL 0/%d' % len(results)) - return 0, len(results), results - for name, fn in CHECKS: try: - ok, msgs = fn(meta, groups) - except Exception as e: # noqa: BLE001 - ok, msgs = False, ['校验异常：%s' % e] - results.append((name, ok, msgs or [])) - - passed = sum(1 for _n, ok, _m in results if ok) - if verbose: - total_items = sum(len(_group_items(g)) for g in groups) - print('codes.json = %s' % path) - print('实测：组数=%d 项数=%d dialect=%s' - % (len(groups), total_items, meta.get('dialect'))) - for name, ok, msgs in results: - flag = 'PASS' if ok else 'FAIL' - line = '[%s] %-20s' % (flag, name) - if msgs: - line += ' | ' + '; '.join(str(m) for m in msgs) - print(line) - print('SELF_CHECK pbl_appcodes: %s %d/%d' - % ('PASS' if passed == len(results) else 'FAIL', passed, len(results))) - return passed, len(results), results + ok, detail = fn(doc, groups) + except Exception as e: # noqa: BLE001 + ok, detail = False, '检查异常：%s' % e + results.append((name, ok, detail)) + if verbose: + print('[%s] %-20s %s' % ('PASS' if ok else 'FAIL', name, detail)) + passed = len([r for r in results if r[1]]) + return (passed, len(results), results) def main(argv=None): - argv = list(sys.argv[1:] if argv is None else argv) - path = argv[0] if argv else CODES_PATH - passed, total, _results = run(path, verbose=True) - return 0 if passed == total else 1 + verbose = True + 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: + print('SELF_CHECK pbl_appcodes: FAIL %d/%d (failed: %s)' + % (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.exit(main(sys.argv[1:])) diff --git a/scripts/load_path.py b/scripts/load_path.py index d2ca493..702b536 100644 --- a/scripts/load_path.py +++ b/scripts/load_path.py @@ -31,9 +31,9 @@ def register(): done += 1 else: missing.append((path, role)) - print('[%s] rbac paths: total=%%d ok=%%d pending=%%d' %% (len(PATHS), done, len(missing))) + 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)) + print(' PENDING %-12s %s' % (role, path)) return len(missing) == 0