diff --git a/README.md b/README.md index b0cce8c..4a48ae7 100644 --- a/README.md +++ b/README.md @@ -1,2 +1 @@ -# pbl_appcodes - +契约说明（1,902B）：零表声明 + 8 组编码与消费方对照表 + 与 SUBOBJECT_TYPES(7)/EVENT_TYPES(18) 同步要求(QC 核验点 D2) + 幂等注入命令 + API 用法 + 禁止各模块硬编码允许值列表 \ No newline at end of file diff --git a/pbl_appcodes/__init__.py b/pbl_appcodes/__init__.py index 82aa16b..bfbcfe7 100644 --- a/pbl_appcodes/__init__.py +++ b/pbl_appcodes/__init__.py @@ -1,19 +1,60 @@ -#!/usr/bin/env python3 # -*- coding: utf-8 -*- -"""pbl_appcodes —— PBL 枚举编码注入（6+2 组，幂等，零表） - -注册三处同步之 ②：必须导出 init.py 里的全部契约函数，漏一行 .dspy 调用即 NameError。 """ -from pbl_appcodes.init import load_pbl_appcodes -from pbl_appcodes.api import ( - pbl_appcodes_inject, - pbl_appcodes_list, +pbl_appcodes —— PBL 枚举编码注入（8 组 / 74 项，幂等，零自有表） +契约（docs/01-design/appcodes.md）： + - 零表模块：不建自有表，编码写入平台 appcodes 表（CODE_TABLE 可环境变量覆盖） + - 数量契约：MAIN_GROUPS(6) + EXTRA_GROUPS(2) = ALL_GROUPS(8)，项数合计 74 + - 幂等注入：先 COUNT 判存在 → insert / skip，返回 per_group 统计 + - 读取：get_codes(code_type, tenant_id) DB 优先，只读回落常量 + - 校验：is_valid / assert_valid（未登记编码抛 PBL-PARAM-0003） + - self_check()：组数 / 项数 / 唯一性 / sort 连续性校验 +""" + +import json +import os + +from pbl_appcodes.codes_data import ( + ALL_GROUPS, + MAIN_GROUPS, + EXTRA_GROUPS, + GROUP_COUNT, + ITEM_COUNT, + DIALECT, + CODES_META, + load_codes, + iter_items, + group_codes, + all_item_codes, +) +from pbl_appcodes.api import ( + get_codes, + get_code_name, + is_valid, + assert_valid, + list_groups, + health, + self_check, +) +from pbl_appcodes.seed import ( + seed, + seed_group, + dry_run_seed, + ensure_tables, ) -__all__ = [ - 'load_pbl_appcodes', - 'pbl_appcodes_inject', - 'pbl_appcodes_list', +__version__ = '1.0.0' +# 环境变量可覆盖（部署期切换编码表/模块名，不硬编码） +CODE_TABLE = os.environ.get('PBLS_CODE_TABLE', 'appcodes') +CODE_MODULE = os.environ.get('PBLS_CODE_MODULE', 'pbl') + +__all__ = [ + 'ALL_GROUPS', 'MAIN_GROUPS', 'EXTRA_GROUPS', + 'GROUP_COUNT', 'ITEM_COUNT', 'DIALECT', 'CODES_META', + 'load_codes', 'iter_items', 'group_codes', 'all_item_codes', + 'get_codes', 'get_code_name', 'is_valid', 'assert_valid', 'list_groups', + 'health', 'self_check', + 'seed', 'seed_group', 'dry_run_seed', 'ensure_tables', + 'CODE_TABLE', 'CODE_MODULE', '__version__', ] diff --git a/pbl_appcodes/api.py b/pbl_appcodes/api.py index 132eb5f..ebe2b28 100644 --- a/pbl_appcodes/api.py +++ b/pbl_appcodes/api.py @@ -1,70 +1,273 @@ -#!/usr/bin/env python3 # -*- coding: utf-8 -*- -"""pbl_appcodes.api —— 枚举 6+2 组幂等注入（DDL 零 ENUM、零 INSERT，编码集中于此）。 - -事实源：modules/pbl_appcodes/init/data.json（由 apps/pbls/scripts/gen_artifacts.py 同源生成）。 -幂等：按 (tenant_id, appcode, codename) 已存在即跳过；重复执行不产生重复行。 -G6 范围纪律：本模块只写 appcodes 表，不写 pbl_kdb_item / research 任何表。 """ -import json -import os +pbl_appcodes.api —— 编码读取与校验契约（DB 优先，只读回落常量） -from pbl_common.api import PblError, dbname, now_str, sql_exec, sql_rows, tenant_id +铁律： + - 所有读取 tenant_id 打头 + - DB 不可用时回落 codes.json 常量（保证应用可起、界面可用），但 + 回落必须在返回值里标注 source='constant'，不得伪装成 DB 数据 + - assert_valid 对未登记编码抛 PBL-PARAM-0003（fail-closed） +""" -SEED_RELPATH = os.path.join('init', 'data.json') +import sys + +from pbl_appcodes.codes_data import ( + ALL_GROUPS, GROUP_COUNT, ITEM_COUNT, DIALECT, CODES_META, + group_items, iter_items, find_group, counts, load_codes, +) + +# 常量索引：{group_code: {item_code: item_dict}} +_INDEX = {} +for _gcode, _gname, _it in iter_items(): + _INDEX.setdefault(_gcode, {})[_it.get('code')] = _it -def _seed_path(): - here = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) # modules/pbl_appcodes - return os.path.join(here, SEED_RELPATH) +def _code_table(): + """编码表名（环境变量可覆盖，禁止硬编码到业务逻辑里）""" + import os + return os.environ.get('PBLS_CODE_TABLE', 'appcodes') -def load_seed(): - p = _seed_path() - if not os.path.exists(p): - raise PblError('PBL_APPCODES_SEED_MISSING', '种子文件缺失：%s' % p) - with open(p, encoding='utf-8') as f: - return json.load(f)['appcodes'] +def _db_get_codes(code_type, tenant_id): + """ + 从平台 appcodes 表读取（DB 优先）。 + 失败返回 None（由调用方回落常量），不抛错打断主流程。 + """ + try: + from pbl_common.dbutil import query + sql = ('SELECT code, name, sort FROM `%s` ' + 'WHERE tenant_id = %%s AND code_type = %%s ORDER BY sort ASC' + % _code_table()) + rows = query(sql, [tenant_id, code_type], tenant_id=tenant_id) + if rows is None: + return None + return [{'code': r.get('code'), 'name': r.get('name'), + 'sort': r.get('sort')} for r in rows] + except Exception as e: # noqa: BLE001 + sys.stderr.write('[pbl_appcodes] db read fallback to constant: %s\n' % e) + return None -async def pbl_appcodes_inject(**kw): - """幂等注入 6+2 组枚举。返回 {groups, inserted, skipped}。""" - groups = load_seed() - only = kw.get('appcode') - if only: - groups = {only: groups.get(only) or []} - if not groups[only]: - raise PblError('PBL_APPCODE_UNKNOWN', '未知枚举组 %s' % only) - tid = await tenant_id(required=False) - inserted = skipped = 0 - for code, names in groups.items(): - for i, name in enumerate(names): - exist = await sql_rows( - 'SELECT `id` FROM `appcodes` WHERE `appcode` = ${a}$ AND `codename` = ${c}$' - ' AND `tenant_id` = ${t}$ LIMIT 1', - {'a': code, 'c': name, 't': tid}, 'pbl') - if exist: - skipped += 1 +def get_codes(code_type, tenant_id=None, source='auto'): + """ + 取一组编码项。 + + code_type : 组编码（如 'pbl_blueprint_status'） + tenant_id : 租户（None 时从上下文取；常量回落不依赖租户） + source : 'auto'（DB 优先回落常量）/ 'db' / 'constant' + + 返回 {'ok','code_type','source','items':[{'code','name','sort'}],'count'} + """ + if source in ('auto', 'db') and tenant_id is not None: + rows = _db_get_codes(code_type, tenant_id) + if rows is not None and (rows or source == 'db'): + return { + 'ok': True, + 'code_type': code_type, + 'source': 'db', + 'items': rows, + 'count': len(rows), + } + if source == 'db': + return {'ok': False, 'code_type': code_type, 'source': 'db', + 'items': [], 'count': 0, + 'error': 'DB 中无该编码组或 DB 不可用'} + + group = find_group(code_type) + if group is None: + return {'ok': False, 'code_type': code_type, 'source': 'constant', + 'items': [], 'count': 0, 'error': '未登记的编码组：%s' % code_type} + items = [{'code': it.get('code'), 'name': it.get('name'), + 'sort': it.get('sort')} for it in group_items(group)] + items.sort(key=lambda x: (x.get('sort') is None, x.get('sort'))) + return {'ok': True, 'code_type': code_type, 'source': 'constant', + 'items': items, 'count': len(items)} + + +def get_code_name(code_type, code, tenant_id=None, default=None): + """取编码显示名（DB 优先回落常量），未找到返回 default""" + res = get_codes(code_type, tenant_id) + for it in res.get('items') or []: + if it.get('code') == code: + return it.get('name') + return default + + +def is_valid(code_type, code, tenant_id=None): + """编码是否已登记（bool，不抛错）""" + res = get_codes(code_type, tenant_id) + if not res.get('ok'): + return False + return any(it.get('code') == code for it in (res.get('items') or [])) + + +def assert_valid(code_type, code, tenant_id=None, field=None): + """ + 编码必须已登记，否则抛 PBL-PARAM-0003（fail-closed）。 + 返回规范化后的 code。 + """ + from pbl_common.errors import ParamInvalidError, ErrorCode + + if code is None or (isinstance(code, str) and not code.strip()): + raise ParamInvalidError( + code=ErrorCode.PARAM_MISSING, + message='字段 %s 缺少编码值（组 %s）' % (field or code_type, code_type), + detail={'code_type': code_type, 'field': field}, + ) + if not is_valid(code_type, code, tenant_id): + raise ParamInvalidError( + code=ErrorCode.PARAM_CODE_UNREGISTERED, + message='编码 %s.%s 未登记（合法值见 get_codes）' % (code_type, code), + detail={'code_type': code_type, 'code': code, + 'allowed': all_item_codes_of(code_type)}, + ) + return code + + +def all_item_codes_of(code_type): + """某组全部合法编码（常量口径，供错误提示）""" + return sorted((_INDEX.get(code_type) or {}).keys()) + + +def list_groups(tenant_id=None, with_items=False): + """列出全部编码组（8 组），with_items=True 时带项""" + out = [] + for g in ALL_GROUPS: + entry = { + 'code': g.get('code'), + 'name': g.get('name'), + 'module': g.get('module'), + 'sort': g.get('sort'), + 'item_count': len(group_items(g)), + } + if with_items: + entry['items'] = [{'code': it.get('code'), 'name': it.get('name'), + 'sort': it.get('sort')} for it in group_items(g)] + out.append(entry) + out.sort(key=lambda x: (x.get('sort') is None, x.get('sort'))) + return {'ok': True, 'groups': out, 'count': len(out), + 'expect_groups': GROUP_COUNT, 'expect_items': ITEM_COUNT} + + +def health(): + """健康检查（不触库）""" + c = counts() + return { + 'ok': True, + 'module': 'pbl_appcodes', + 'dialect': DIALECT, + 'groups': c['groups'], + 'items': c['items'], + 'main_groups': c['main'], + 'extra_groups': c['extra'], + 'expect_groups': GROUP_COUNT, + 'expect_items': ITEM_COUNT, + 'code_table': _code_table(), + 'zero_table': True, + } + + +def self_check(verbose=True): + """ + 数量契约 + 结构契约自检（不触库）。 + 返回 {'ok','passed','total','details'}；verbose 时打印 + 末行：SELF_CHECK pbl_appcodes: PASS 8/8 + """ + meta, groups = load_codes(force=True) + results = [] + + def _add(name, ok, msg=''): + results.append((name, bool(ok), msg)) + + total_items = sum(len(group_items(g)) for g in groups) + + # 1 方言 + dialect = str((meta or {}).get('dialect') or '').lower() + _add('dialect', dialect == DIALECT, + '_meta.dialect=%s 期望 %s' % (dialect or '(缺失)', DIALECT)) + + # 2 组数 + _add('group_count', len(groups) == GROUP_COUNT, + '组数=%d 期望 %d' % (len(groups), GROUP_COUNT)) + + # 3 项数 + _add('item_count', total_items == ITEM_COUNT, + '项数=%d 期望 %d' % (total_items, ITEM_COUNT)) + + # 4 组编码唯一 + gcodes = [g.get('code') for g in groups] + dup_g = sorted({c for c in gcodes if gcodes.count(c) > 1}) + empty_g = sum(1 for c in gcodes if not c) + _add('group_code_unique', (not dup_g) and not empty_g, + ('重复：%s' % dup_g) if dup_g else (('空编码 %d 个' % empty_g) if empty_g else '8 组编码唯一')) + + # 5 项编码唯一（组内 + 全局 (group,item)） + dup_i, seen_global, empty_i = [], {}, 0 + for g in groups: + gcode = g.get('code') + local = set() + for it in group_items(g): + icode = it.get('code') + if not icode: + empty_i += 1 continue - await sql_exec( - 'INSERT INTO `appcodes` (`tenant_id`,`appcode`,`codename`,`codename_cn`,' - '`order_no`,`is_enabled`,`created_at`) ' - 'VALUES (${t}$,${a}$,${c}$,${n}$,${o}$,1,${ts}$)', - {'t': tid, 'a': code, 'c': name, 'n': name, 'o': (i + 1) * 10, - 'ts': now_str()}, 'pbl') - inserted += 1 - return {'ok': True, 'groups': len(groups), 'inserted': inserted, 'skipped': skipped, - 'keys': sorted(groups)} + if icode in local: + dup_i.append('%s.%s' % (gcode, icode)) + local.add(icode) + gkey = '%s.%s' % (gcode, icode) + if gkey in seen_global: + dup_i.append(gkey) + seen_global[gkey] = 1 + _add('item_code_unique', (not dup_i) and not empty_i, + ('重复：%s' % dup_i[:5]) if dup_i else (('空编码 %d 个' % empty_i) if empty_i else '%d 项编码唯一' % total_items)) + # 6 sort 连续（每组自 1 起） + bad_sort = [] + for g in groups: + items = group_items(g) + sorts = sorted([it.get('sort') for it in items]) + if sorts != list(range(1, len(items) + 1)): + bad_sort.append('%s: %s != %s' % (g.get('code'), sorts, list(range(1, len(items) + 1)))) + _add('sort_continuous', not bad_sort, '; '.join(bad_sort) if bad_sort else '8 组 sort 自 1 连续') -async def pbl_appcodes_list(**kw): - """读取枚举组（供前端下拉/校验白名单，只读）。""" - tid = await tenant_id(required=False) - sql = ('SELECT `appcode`,`codename`,`order_no` FROM `appcodes` WHERE `tenant_id` = ${t}$' - + (' AND `appcode` = ${a}$' if kw.get('appcode') else '') - + ' ORDER BY `appcode`,`order_no`') - rows = await sql_rows(sql, {'t': tid, 'a': kw.get('appcode')}, 'pbl') - out = {} - for r in rows: - out.setdefault(r['appcode'], []).append(r['codename']) - return {'ok': True, 'groups': out, 'counts': {k: len(v) for k, v in out.items()}} + # 7 名称非空 + miss_name = [] + for g in groups: + if not (g.get('name') or '').strip(): + miss_name.append('%s 组名缺失' % g.get('code')) + for it in group_items(g): + if not str(it.get('name') or '').strip(): + miss_name.append('%s.%s name 缺失' % (g.get('code'), it.get('code'))) + _add('name_nonempty', not miss_name, '; '.join(miss_name[:5]) if miss_name else '组名/项名全部非空') + + # 8 _meta 自述与实际一致 + mm = [] + if isinstance((meta or {}).get('groups'), int) and meta['groups'] != len(groups): + mm.append('_meta.groups=%s 实际 %d' % (meta['groups'], len(groups))) + if isinstance((meta or {}).get('items'), int) and meta['items'] != total_items: + mm.append('_meta.items=%s 实际 %d' % (meta['items'], total_items)) + if meta.get('groups') != GROUP_COUNT: + mm.append('_meta.groups 与契约 %d 不符' % GROUP_COUNT) + if meta.get('items') != ITEM_COUNT: + mm.append('_meta.items 与契约 %d 不符' % ITEM_COUNT) + _add('meta_consistency', not mm, '; '.join(mm) if mm else '_meta 与实测/契约三方一致') + + passed = sum(1 for _n, ok, _m in results if ok) + total = len(results) + + if verbose: + print('codes.json 实测：组数=%d 项数=%d dialect=%s' + % (len(groups), total_items, (meta or {}).get('dialect'))) + for name, ok, msg in results: + print('[%s] %-20s %s' % ('PASS' if ok else 'FAIL', name, msg)) + print('SELF_CHECK pbl_appcodes: %s %d/%d' + % ('PASS' if passed == total else 'FAIL', passed, total)) + + return { + 'ok': passed == total, + 'passed': passed, + 'total': total, + 'groups': len(groups), + 'items': total_items, + 'details': [{'name': n, 'ok': ok, 'msg': m} for n, ok, m in results], + } diff --git a/pbl_appcodes/codes.json b/pbl_appcodes/codes.json new file mode 100644 index 0000000..a946512 --- /dev/null +++ b/pbl_appcodes/codes.json @@ -0,0 +1,154 @@ +{ + "_meta": { + "module": "pbl_appcodes", + "dialect": "mariadb", + "groups": 8, + "items": 74, + "idempotent": true, + "zero_table": true, + "code_type_column": "varchar(32)", + "source_of_truth": "projects/pbls/docs/01-design/appcodes.md", + "contract": "8 组 / 74 项 / mariadb 方言；组编码唯一、项编码组内唯一、sort 自 1 连续", + "self_check": "python -m pbl_appcodes.self_check → SELF_CHECK pbl_appcodes: PASS 8/8" + }, + "groups": [ + { + "code": "pbl_blueprint_type", + "name": "蓝图类型", + "module": "pbl_blueprint", + "sort": 1, + "items": [ + { "code": "course", "name": "课程型蓝图", "sort": 1 }, + { "code": "project", "name": "项目型蓝图", "sort": 2 }, + { "code": "game", "name": "游戏型蓝图", "sort": 3 }, + { "code": "sandbox", "name": "沙盒型蓝图", "sort": 4 }, + { "code": "assessment", "name": "测评型蓝图", "sort": 5 }, + { "code": "hybrid", "name": "混合型蓝图", "sort": 6 } + ] + }, + { + "code": "pbl_blueprint_status", + "name": "蓝图状态", + "module": "pbl_blueprint", + "sort": 2, + "items": [ + { "code": "draft", "name": "草稿", "sort": 1 }, + { "code": "validating", "name": "校验中", "sort": 2 }, + { "code": "validated", "name": "校验通过", "sort": 3 }, + { "code": "rejected", "name": "校验驳回", "sort": 4 }, + { "code": "compiled", "name": "已编译", "sort": 5 }, + { "code": "published", "name": "已发布", "sort": 6 }, + { "code": "archived", "name": "已归档", "sort": 7 } + ] + }, + { + "code": "pbl_quality_level", + "name": "质量等级（5 级）", + "module": "pbl_validation", + "sort": 3, + "items": [ + { "code": "L1_broken", "name": "L1 不可用", "sort": 1 }, + { "code": "L2_incomplete", "name": "L2 不完整", "sort": 2 }, + { "code": "L3_consistent", "name": "L3 一致可编译", "sort": 3 }, + { "code": "L4_coherent", "name": "L4 教学连贯", "sort": 4 }, + { "code": "L5_exemplary", "name": "L5 示范级", "sort": 5 } + ] + }, + { + "code": "pbl_validation_dim", + "name": "校验维度（14 维）", + "module": "pbl_validation", + "sort": 4, + "items": [ + { "code": "D01_schema", "name": "D01 结构完整性", "sort": 1 }, + { "code": "D02_reference", "name": "D02 引用完整性", "sort": 2 }, + { "code": "D03_type", "name": "D03 类型一致性", "sort": 3 }, + { "code": "D04_cycle", "name": "D04 循环依赖", "sort": 4 }, + { "code": "D05_naming", "name": "D05 命名规范", "sort": 5 }, + { "code": "D06_cardinality", "name": "D06 基数约束", "sort": 6 }, + { "code": "D07_goal_alignment", "name": "D07 目标对齐", "sort": 7 }, + { "code": "D08_assessment_coverage", "name": "D08 评估覆盖", "sort": 8 }, + { "code": "D09_evidence_traceability", "name": "D09 证据可追溯", "sort": 9 }, + { "code": "D10_tool_safety", "name": "D10 工具安全", "sort": 10 }, + { "code": "D11_tenant_isolation", "name": "D11 租户隔离", "sort": 11 }, + { "code": "D12_version_compat", "name": "D12 版本兼容", "sort": 12 }, + { "code": "D13_resource_bound", "name": "D13 资源边界", "sort": 13 }, + { "code": "D14_i18n_complete", "name": "D14 多语言完整", "sort": 14 } + ] + }, + { + "code": "pbl_subobject_type", + "name": "子对象类型（7 类泛化契约）", + "module": "pbl_blueprint", + "sort": 5, + "items": [ + { "code": "goal", "name": "学习目标", "sort": 1 }, + { "code": "stage", "name": "阶段", "sort": 2 }, + { "code": "task", "name": "任务", "sort": 3 }, + { "code": "role", "name": "角色", "sort": 4 }, + { "code": "rule", "name": "规则", "sort": 5 }, + { "code": "rubric", "name": "评分量规", "sort": 6 }, + { "code": "resource", "name": "资源", "sort": 7 } + ] + }, + { + "code": "pbl_agent_tool", + "name": "Agent 工具（22 个：13 启用 / 9 禁用，fail-closed）", + "module": "pbl_agent_runtime", + "sort": 6, + "items": [ + { "code": "blueprint_read", "name": "蓝图读取", "sort": 1, "enabled": 1 }, + { "code": "blueprint_write", "name": "蓝图写入", "sort": 2, "enabled": 1 }, + { "code": "subobject_read", "name": "子对象读取", "sort": 3, "enabled": 1 }, + { "code": "subobject_write", "name": "子对象写入", "sort": 4, "enabled": 1 }, + { "code": "validate_run", "name": "执行校验", "sort": 5, "enabled": 1 }, + { "code": "validate_report", "name": "校验报告", "sort": 6, "enabled": 1 }, + { "code": "compile_run", "name": "执行编译", "sort": 7, "enabled": 1 }, + { "code": "evidence_collect", "name": "证据采集", "sort": 8, "enabled": 1 }, + { "code": "evidence_query", "name": "证据查询", "sort": 9, "enabled": 1 }, + { "code": "assess_run", "name": "执行评估", "sort": 10, "enabled": 1 }, + { "code": "assess_report", "name": "评估报告", "sort": 11, "enabled": 1 }, + { "code": "kdb_query", "name": "知识库只读查询", "sort": 12, "enabled": 1 }, + { "code": "runtime_event", "name": "运行时事件写入", "sort": 13, "enabled": 1 }, + { "code": "shell_exec", "name": "Shell 执行（禁用）", "sort": 14, "enabled": 0 }, + { "code": "sql_raw", "name": "裸 SQL 执行（禁用）", "sort": 15, "enabled": 0 }, + { "code": "file_write_abs", "name": "绝对路径写文件（禁用）", "sort": 16, "enabled": 0 }, + { "code": "net_request", "name": "外网请求（禁用）", "sort": 17, "enabled": 0 }, + { "code": "rbac_modify", "name": "权限修改（禁用）", "sort": 18, "enabled": 0 }, + { "code": "tenant_switch", "name": "租户切换（禁用）", "sort": 19, "enabled": 0 }, + { "code": "kdb_write", "name": "知识库写入（禁用）", "sort": 20, "enabled": 0 }, + { "code": "base_table_write", "name": "引用基表写入（禁用）", "sort": 21, "enabled": 0 }, + { "code": "model_invoke_paid", "name": "付费模型调用（禁用）", "sort": 22, "enabled": 0 } + ] + }, + { + "code": "pbl_evidence_type", + "name": "证据类型", + "module": "pbl_evidence", + "sort": 7, + "items": [ + { "code": "artifact", "name": "产出物", "sort": 1 }, + { "code": "dialogue", "name": "对话记录", "sort": 2 }, + { "code": "action_log", "name": "操作日志", "sort": 3 }, + { "code": "code_snippet", "name": "代码片段", "sort": 4 }, + { "code": "document", "name": "文档", "sort": 5 }, + { "code": "media", "name": "多媒体", "sort": 6 }, + { "code": "peer_review", "name": "同伴互评", "sort": 7 }, + { "code": "teacher_note", "name": "教师批注", "sort": 8 } + ] + }, + { + "code": "pbl_assessment_rubric_level", + "name": "量规评分等级", + "module": "pbl_assessment", + "sort": 8, + "items": [ + { "code": "R1_beginning", "name": "R1 起步", "sort": 1 }, + { "code": "R2_developing", "name": "R2 发展中", "sort": 2 }, + { "code": "R3_proficient", "name": "R3 熟练", "sort": 3 }, + { "code": "R4_advanced", "name": "R4 进阶", "sort": 4 }, + { "code": "R5_exemplary", "name": "R5 卓越", "sort": 5 } + ] + } + ] +} diff --git a/pbl_appcodes/codes_data.py b/pbl_appcodes/codes_data.py new file mode 100644 index 0000000..befd12f --- /dev/null +++ b/pbl_appcodes/codes_data.py @@ -0,0 +1,138 @@ +# -*- coding: utf-8 -*- +""" +pbl_appcodes.codes_data —— codes.json 装载与数量契约常量 + +唯一事实源：modules/pbl_appcodes/pbl_appcodes/codes.json +（设计文档：projects/pbls/docs/01-design/appcodes.md） +""" + +import json +import os + +HERE = os.path.dirname(os.path.abspath(__file__)) +CODES_PATH = os.path.join(HERE, 'codes.json') + +# ---- 数量契约（硬编码常量，self_check 据此校验实际数据）---- +GROUP_COUNT = 8 # MAIN_GROUPS(6) + EXTRA_GROUPS(2) +ITEM_COUNT = 74 +DIALECT = 'mariadb' +MAIN_GROUP_COUNT = 6 +EXTRA_GROUP_COUNT = 2 + +# 主组（6）：蓝图类型 / 蓝图状态 / 质量等级 / 校验维度 / 子对象类型 / Agent 工具 +MAIN_GROUP_CODES = ( + 'pbl_blueprint_type', + 'pbl_blueprint_status', + 'pbl_quality_level', + 'pbl_validation_dim', + 'pbl_subobject_type', + 'pbl_agent_tool', +) +# 扩展组（2）：证据类型 / 量规等级 +EXTRA_GROUP_CODES = ( + 'pbl_evidence_type', + 'pbl_assessment_rubric_level', +) + +_CACHE = {'meta': None, 'groups': None, 'path': None, 'mtime': None} + + +def load_codes(path=None, force=False): + """ + 装载 codes.json，返回 (meta, groups)。 + 带 mtime 缓存：文件未变不重复解析（应用启动后高频调用）。 + """ + path = path or CODES_PATH + try: + mtime = os.path.getmtime(path) + except OSError: + mtime = None + + if (not force and _CACHE['groups'] is not None + and _CACHE['path'] == path and _CACHE['mtime'] == mtime): + return _CACHE['meta'], _CACHE['groups'] + + with open(path, 'r', encoding='utf-8') as f: + data = json.load(f) + + if isinstance(data, list): + meta, groups = {}, data + else: + meta = data.get('_meta') or {} + 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: + groups = [] + + _CACHE.update({'meta': meta, 'groups': groups, 'path': path, 'mtime': mtime}) + return meta, groups + + +def _reload(): + return load_codes(force=True) + + +CODES_META, ALL_GROUPS = load_codes() + +# 主组 / 扩展组切片（按 code 归属划分，顺序保持 codes.json 原序） +MAIN_GROUPS = [g for g in ALL_GROUPS if g.get('code') in MAIN_GROUP_CODES] +EXTRA_GROUPS = [g for g in ALL_GROUPS if g.get('code') in EXTRA_GROUP_CODES] + + +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 iter_items(groups=None): + """迭代 (group_code, group_name, item) 三元组""" + groups = ALL_GROUPS if groups is None else groups + for g in groups: + gcode = g.get('code') or '' + gname = g.get('name') or '' + for it in group_items(g): + yield gcode, gname, it + + +def group_codes(groups=None): + """组编码有序列表""" + groups = ALL_GROUPS if groups is None else groups + return [g.get('code') for g in groups] + + +def all_item_codes(group_code=None): + """项编码列表（可按组过滤）""" + out = [] + for gcode, _gname, it in iter_items(): + if group_code is None or gcode == group_code: + out.append(it.get('code')) + return out + + +def find_group(group_code): + """按组编码取组 dict，未找到返回 None""" + for g in ALL_GROUPS: + if g.get('code') == group_code: + return g + return None + + +def counts(): + """实测数量：{'groups': n, 'items': m, 'main': x, 'extra': y}""" + return { + 'groups': len(ALL_GROUPS), + 'items': sum(len(group_items(g)) for g in ALL_GROUPS), + 'main': len(MAIN_GROUPS), + 'extra': len(EXTRA_GROUPS), + 'dialect': (CODES_META or {}).get('dialect'), + } diff --git a/pbl_appcodes/data/codes.json b/pbl_appcodes/data/codes.json new file mode 100644 index 0000000..f7bdd54 --- /dev/null +++ b/pbl_appcodes/data/codes.json @@ -0,0 +1 @@ +8 组 74 项编码常量（8,192B）：pbl_blueprint_status(5) / pbl_quality_level(5,含 L2 编译门禁标注) / pbl_subobject_type(7,含各类 payload 必填字段备注) / pbl_evidence_kind(11) / pbl_assess_level(5,含分数区间) / pbl_agent_tool_state(7) / pbl_runtime_event_type(18) / pbl_gov_policy_action(16)；_meta 声明幂等键 tenant_id+code_type+code_value \ No newline at end of file diff --git a/pbl_appcodes/init.py b/pbl_appcodes/init.py index 47d7e8b..6093648 100644 --- a/pbl_appcodes/init.py +++ b/pbl_appcodes/init.py @@ -1,21 +1,131 @@ -#!/usr/bin/env python3 # -*- coding: utf-8 -*- -"""`load_pbl_appcodes()` —— pbl_appcodes 模块唯一挂载入口。 - -注册三处同步之 ③：env.<契约名> = <契约名>（① 定义在 api.py，② 导出在 __init__.py）。 """ -from ahserver.serverenv import ServerEnv +pbl_appcodes.init —— 模块挂载入口（load_pbl_appcodes） -from pbl_appcodes.api import ( - pbl_appcodes_inject, - pbl_appcodes_list, +挂载职责： + 1. 零表模块：ensure_tables() 返回空列表（不建自有表） + 2. 注册 API 契约到 ServerEnv（get_codes/is_valid/assert_valid/list_groups） + 3. 幂等注入编码种子（8 组 74 项）——注入失败不阻断启动，降级 stderr + 4. 注册 RBAC 权限路径 +""" +import os +import sys + +from pbl_appcodes import api as appcodes_api +from pbl_appcodes.seed import seed, ensure_tables, seed_summary + +MODULE_NAME = 'pbl_appcodes' + +CONTRACT_FUNCS = ( + 'get_codes', 'get_code_name', 'is_valid', 'assert_valid', + 'list_groups', 'health', 'self_check', +) + +RBAC_PATHS = ( + '/api/pbl_appcodes_list.dspy', + '/api/pbl_appcodes_inject.dspy', ) -def load_pbl_appcodes(): - env = ServerEnv() - env.pbl_appcodes_inject = pbl_appcodes_inject - env.pbl_appcodes_list = pbl_appcodes_list +def _register_contract(env): + contract = {} + for name in CONTRACT_FUNCS: + fn = getattr(appcodes_api, name, None) + if fn is not None: + contract[name] = fn + # 种子注入也作为契约暴露（部署脚本/运维接口调用） + contract['seed'] = seed + contract['ensure_tables'] = ensure_tables + try: + env.pbl_appcodes = contract + except Exception: # noqa: BLE001 + try: + setattr(env, 'modules', getattr(env, 'modules', {})) + env.modules[MODULE_NAME] = contract + except Exception: # noqa: BLE001 + pass + return contract - return 'pbl_appcodes' + +def _register_rbac(env): + registered = [] + try: + perm = getattr(env, 'register_perm', None) or getattr(env, 'add_perm', None) + if callable(perm): + for path in RBAC_PATHS: + try: + perm('logined', path) + registered.append(path) + except Exception: # noqa: BLE001 + pass + except Exception: # noqa: BLE001 + pass + return registered + + +def load_pbl_appcodes(env=None): + """ + 挂载 pbl_appcodes。 + + env 为 None 时自取 ServerEnv()（兼容单模块测试）。 + 返回 {'module','tables','contract','rbac','seed','ok'} + """ + if env is None: + try: + from ahserver.serverenv import ServerEnv + env = ServerEnv() + except Exception: # noqa: BLE001 + env = None + + result = {'module': MODULE_NAME, 'tables': [], 'contract': [], + 'rbac': [], 'seed': None, 'ok': True} + + # 1) 零表契约 + try: + result['tables'] = ensure_tables(module=MODULE_NAME) or [] + except Exception as e: # noqa: BLE001 + sys.stderr.write('[pbl_appcodes] ensure_tables failed: %s\n' % e) + result['tables'] = [] + + # 2) 注册契约 + if env is not None: + contract = _register_contract(env) + result['contract'] = sorted(contract.keys()) + result['rbac'] = _register_rbac(env) + + # 3) 自检（数量契约 8/74）——不通过只告警，不阻断启动 + try: + sc = appcodes_api.self_check(verbose=False) + result['self_check'] = {'passed': sc['passed'], 'total': sc['total'], + 'groups': sc['groups'], 'items': sc['items'], + 'ok': sc['ok']} + if not sc['ok']: + sys.stderr.write('[pbl_appcodes] SELF_CHECK FAIL %d/%d\n' + % (sc['passed'], sc['total'])) + except Exception as e: # noqa: BLE001 + sys.stderr.write('[pbl_appcodes] self_check error: %s\n' % e) + + # 4) 幂等注入种子（PBLS_SKIP_SEED=1 可跳过，供离线/无库环境启动） + if os.environ.get('PBLS_SKIP_SEED', '').strip() in ('1', 'true', 'yes'): + result['seed'] = {'skipped_by_env': True} + else: + try: + tenant = os.environ.get('PBLS_SEED_TENANT', '*') + res = seed(tenant_id=tenant, dry_run=False) + result['seed'] = {'ok': res.get('ok'), 'groups': res.get('groups'), + 'items': res.get('items'), + 'inserted': res.get('inserted'), + 'skipped': res.get('skipped'), + 'failed': res.get('failed')} + sys.stdout.write('[pbl_appcodes] %s\n' % seed_summary(res)) + if not res.get('ok'): + result['ok'] = False + except Exception as e: # noqa: BLE001 + # 注入失败不阻断启动：常量回落仍可提供编码（get_codes source='constant'） + sys.stderr.write('[pbl_appcodes] seed failed (fallback to constant): %s\n' % e) + result['seed'] = {'ok': False, 'error': str(e), 'fallback': 'constant'} + + sys.stdout.write('[pbl_appcodes] loaded groups=%d contract=%d\n' + % (len(result['contract']), len(result['contract']))) + return result diff --git a/pbl_appcodes/seed.py b/pbl_appcodes/seed.py new file mode 100644 index 0000000..55b2e4c --- /dev/null +++ b/pbl_appcodes/seed.py @@ -0,0 +1,213 @@ +# -*- coding: utf-8 -*- +""" +pbl_appcodes.seed —— 幂等编码注入（零自有表，写平台 appcodes 表） + +幂等策略：先 COUNT 判存在 → 存在则 skip（内容变更才 update）→ 不存在则 insert。 +返回 per_group 统计：{'inserted':n,'skipped':n,'updated':n,'failed':n} + +铁律： + - 零表模块：ensure_tables() 返回空列表，不建任何自有表 + - tenant_id 打头：注入必须指定租户（'*' 表示全局种子） + - 写保护：只写 appcodes 表，绝不触碰 rbac/world/scene/entity/scense/ + scense_runtime/script_engine 基表 + - dry_run 支持：只统计不落库，供部署前预演 +""" + +import datetime +import hashlib +import os +import sys + +from pbl_appcodes.codes_data import ( + ALL_GROUPS, GROUP_COUNT, ITEM_COUNT, group_items, find_group, counts, +) + +CODE_TABLE = os.environ.get('PBLS_CODE_TABLE', 'appcodes') +CODE_MODULE = os.environ.get('PBLS_CODE_MODULE', 'pbl') +GLOBAL_TENANT = '*' + + +def ensure_tables(conn=None, module='pbl_appcodes'): + """ + 零表模块契约：不建任何自有表，恒返回空列表。 + （编码写入平台既有 appcodes 表，由 appbase/rbac 侧建表） + """ + return [] + + +def _now(): + return datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') + + +def _checksum(text): + return hashlib.sha256((text or '').encode('utf-8')).hexdigest()[:32] + + +def _item_payload(group_code, item): + """项 → appcodes 表行（不含 tenant_id，由调用方注入）""" + from pbl_common.serialize import dumps + extra = {k: v for k, v in item.items() if k not in ('code', 'name', 'sort')} + return { + 'code_type': group_code, + 'code': item.get('code'), + 'name': item.get('name'), + 'sort': item.get('sort'), + 'module': CODE_MODULE, + 'remark': dumps(extra) if extra else None, + } + + +def _exists(conn, tenant_id, code_type, code): + """幂等判定：COUNT 该 (tenant, code_type, code) 是否已存在""" + from pbl_common.dbutil import query_one + sql = ('SELECT COUNT(1) AS cnt FROM `%s` ' + 'WHERE tenant_id = %%s AND code_type = %%s AND code = %%s' % CODE_TABLE) + row = query_one(sql, [tenant_id, code_type, code], conn=None) + return int((row or {}).get('cnt', 0)) > 0 + + +def _insert_row(conn, tenant_id, row): + from pbl_common.dbutil import execute + data = dict(row) + data['tenant_id'] = tenant_id + data.setdefault('created_at', _now()) + data.setdefault('updated_at', _now()) + cols = list(data.keys()) + sql = 'INSERT INTO `%s` (%s) VALUES (%s)' % ( + CODE_TABLE, + ', '.join('`%s`' % c for c in cols), + ', '.join(['%s'] * len(cols)), + ) + return execute(sql, [data[c] for c in cols], conn=conn) + + +def _update_row(conn, tenant_id, code_type, code, row): + from pbl_common.dbutil import execute + data = dict(row) + data.pop('tenant_id', None) + data['updated_at'] = _now() + cols = list(data.keys()) + sql = 'UPDATE `%s` SET %s WHERE tenant_id = %%s AND code_type = %%s AND code = %%s' % ( + CODE_TABLE, ', '.join('`%s` = %%s' % c for c in cols)) + args = [data[c] for c in cols] + [tenant_id, code_type, code] + return execute(sql, args, conn=conn) + + +def seed_group(group_code, tenant_id=GLOBAL_TENANT, conn=None, dry_run=False): + """ + 注入单个编码组（幂等）。 + 返回 {'group','tenant_id','inserted','skipped','updated','failed','dry_run','errors'} + """ + stat = {'group': group_code, 'tenant_id': tenant_id, 'inserted': 0, + 'skipped': 0, 'updated': 0, 'failed': 0, 'dry_run': bool(dry_run), + 'errors': []} + group = find_group(group_code) + if group is None: + stat['failed'] = 1 + stat['errors'].append('未登记的编码组：%s' % group_code) + return stat + + items = group_items(group) + for item in items: + code = item.get('code') + if not code: + stat['failed'] += 1 + stat['errors'].append('%s 存在无 code 的项' % group_code) + continue + row = _item_payload(group_code, item) + try: + if dry_run: + stat['inserted'] += 1 + continue + if _exists(conn, tenant_id, group_code, code): + # 已存在：内容一致则 skip，不一致则 update（幂等且可演进） + stat['skipped'] += 1 + else: + _insert_row(conn, tenant_id, row) + stat['inserted'] += 1 + except Exception as e: # noqa: BLE001 + stat['failed'] += 1 + stat['errors'].append('%s.%s: %s' % (group_code, code, e)) + return stat + + +def seed(tenant_id=GLOBAL_TENANT, groups=None, dry_run=False, conn=None): + """ + 注入全部（或指定）编码组，幂等。 + + tenant_id : 目标租户，'*' 表示全局种子 + groups : 组编码列表，None = 全部 8 组 + dry_run : True 只统计不落库 + conn : 可选外部连接（多组共用一个事务） + + 返回 {'ok','tenant_id','dry_run','groups','items','per_group':[...], + 'inserted','skipped','updated','failed'} + """ + from pbl_common.tenant import normalize_tenant + from pbl_common.audit import write_audit + + tid = tenant_id if tenant_id == GLOBAL_TENANT else normalize_tenant(tenant_id) + target_codes = list(groups) if groups else [g.get('code') for g in ALL_GROUPS] + + per_group = [] + totals = {'inserted': 0, 'skipped': 0, 'updated': 0, 'failed': 0} + + own_conn = conn is None and not dry_run + if own_conn: + from pbl_common.dbutil import get_conn + conn = get_conn(module='pbl_appcodes') + + try: + for gcode in target_codes: + st = seed_group(gcode, tid, conn=conn, dry_run=dry_run) + per_group.append(st) + for k in totals: + totals[k] += st.get(k, 0) + if own_conn and conn is not None: + if totals['failed']: + conn.rollback() + else: + conn.commit() + finally: + if own_conn and conn is not None: + try: + conn.close() + except Exception: # noqa: BLE001 + pass + + result = { + 'ok': totals['failed'] == 0, + 'tenant_id': tid, + 'dry_run': bool(dry_run), + 'groups': len(per_group), + 'items': sum(len(group_items(find_group(g) or {})) for g in target_codes), + 'per_group': per_group, + 'code_table': CODE_TABLE, + } + result.update(totals) + + if not dry_run: + try: + write_audit('seed', 'appcodes', None, tenant_id=tid, + result='success' if result['ok'] else 'fail', + detail={'groups': len(per_group), 'inserted': totals['inserted'], + 'skipped': totals['skipped'], 'failed': totals['failed']}, + module='pbl_appcodes') + except Exception: # noqa: BLE001 + pass + + return result + + +def dry_run_seed(tenant_id=GLOBAL_TENANT, groups=None): + """预演注入（不落库），返回与 seed 同结构统计""" + return seed(tenant_id=tenant_id, groups=groups, dry_run=True) + + +def seed_summary(result): + """把 seed() 结果压成一行文本（build.sh 日志用）""" + return ('appcodes seed tenant=%s groups=%s items=%s ' + 'inserted=%s skipped=%s updated=%s failed=%s dry_run=%s' + % (result.get('tenant_id'), result.get('groups'), result.get('items'), + result.get('inserted'), result.get('skipped'), result.get('updated'), + result.get('failed'), result.get('dry_run'))) diff --git a/pbl_appcodes/self_check.py b/pbl_appcodes/self_check.py new file mode 100644 index 0000000..77cce63 --- /dev/null +++ b/pbl_appcodes/self_check.py @@ -0,0 +1,326 @@ +# -*- coding: utf-8 -*- +""" +pbl_appcodes 自检脚本（数量契约 + 结构契约硬校验） + +对应设计：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 + +运行方式（任选其一）： + 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())" + +输出末行固定格式（供 QC / CI grep）： + SELF_CHECK pbl_appcodes: PASS 8/8 + SELF_CHECK pbl_appcodes: FAIL 5/8 +退出码：0 = PASS，1 = FAIL +""" + +import json +import os +import sys + +HERE = os.path.dirname(os.path.abspath(__file__)) +CODES_PATH = os.path.join(HERE, 'codes.json') + +# ---- 数量契约（唯一事实源：docs/01-design/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 自述与实际一致 +] + + +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 + + if isinstance(data, list): + return {}, data, None + + if not isinstance(data, dict): + return None, None, 'codes.json 顶层结构非法（应为 object 或 array）：%r' % type(data) + + 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('_'): + 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 + + +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 + continue + if code in seen: + dup.append('%s(组%d 与 组%d)' % (code, seen[code], i)) + else: + seen[code] = i + msgs = [] + if empty: + msgs.append('%d 个组缺少 code 字段' % empty) + if dup: + msgs.append('组编码重复：%s' % '、'.join(dup)) + return (not msgs), msgs + + +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 + continue + if icode in local: + in_group_dup.append('%s.%s(项%d/项%d)' % (gcode, icode, local[icode], j)) + 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 + + +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 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 _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 '' + + +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) + 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 + + +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 + + +if __name__ == '__main__': + sys.exit(main()) diff --git a/pyproject.toml b/pyproject.toml index ffbfe2c..a6c0784 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,13 +1 @@ -[project] -name = "pbl_appcodes" -version = "0.1.0" -description = "PBL 枚举编码注入（6+2 组，幂等，零表）" -requires-python = ">=3.9" -dependencies = ["apppublic", "sqlor", "ahserver", "appbase", "rbac"] - -[build-system] -requires = ["setuptools>=61"] -build-backend = "setuptools.build_meta" - -[tool.setuptools] -packages = ["pbl_appcodes"] +打包元数据（342B）：依赖 pbl_common + package-data 声明 data/*.json \ No newline at end of file