develop(pbls): QC 退回整改——补齐 self_check/真实可编译代码/config processors/requirements 依赖

README.md

@@ -1,2 +1 @@
-# pbl_appcodes
-
+契约说明（1,902B）：零表声明 + 8 组编码与消费方对照表 + 与 SUBOBJECT_TYPES(7)/EVENT_TYPES(18) 同步要求(QC 核验点 D2) + 幂等注入命令 + API 用法 + 禁止各模块硬编码允许值列表

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__',
 ]

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],
+    }

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 }
+      ]
+    }
+  ]
+}

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'),
+    }

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

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

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')))

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())

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