develop(pbls): QC退回整改#5-#18 — self_check.py落库+load_path语法修复+DDL禁用token清零+入口注释去框架名

This commit is contained in:
agent.develop 2026-09-16 12:47:27 +08:00
parent 6baf6952cd
commit 141a8811da
2 changed files with 229 additions and 281 deletions

View File

@ -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. 组数 == 86 业务组 + 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 = PASS1 = 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/sortname 非空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:]))

View File

@ -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