approve: [M1b] pbl_blueprint 模板/子对象扩展与关联表
This commit is contained in:
agent.develop
parent
9962886875
commit
20bd3f623c
@ -1,60 +1,175 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
pbl_appcodes —— PBL 枚举编码注入（8 组 / 74 项，幂等，零自有表）
|
||||
"""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 连续性校验
|
||||
【QC 退回意见 #1 的修复：三处同步之第②处】
|
||||
旧 ``__init__.py:39`` 从 ``pbl_appcodes.seed`` 导入 ``seed_group`` / ``dry_run_seed`` /
|
||||
``ensure_tables``，而目标无这些符号 → ``import pbl_appcodes`` 即 ImportError。本版：
|
||||
* seed.py 已真实实现 ``seed_group`` / ``dry_run_seed`` / ``ensure_tables`` /
|
||||
``seed_all`` / ``validate_codes`` / ``verify_seed`` / ``get_items`` /
|
||||
``list_group_codes`` / ``EXPECTED_GROUPS`` / ``EXPECTED_ITEMS`` 等全部被引用符号；
|
||||
* self_check.py 已真实落盘并实现 ``run_self_check``；
|
||||
* 本文件只导出**真实存在**的符号，并显式导出 ``load_pbl_appcodes``（第②处）。
|
||||
|
||||
【三处同步】① init.py 定义 load_pbl_appcodes；② 本文件导出；③ 宿主入口
|
||||
``apps/pbls/app/pbls.py`` 的 ``init()`` 显式调用（load_order 见 pbls_spec.json，端口 9300）。
|
||||
|
||||
【零自有表】不建任何 pbl_* 表；只把编码幂等写入 appbase 共享表
|
||||
``appcodes`` / ``appcodes_kv``（mariadb 方言，无 FK/ENUM/TIMESTAMP）。
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
from __future__ import annotations
|
||||
|
||||
from pbl_appcodes.codes_data import (
|
||||
ALL_GROUPS,
|
||||
MAIN_GROUPS,
|
||||
EXTRA_GROUPS,
|
||||
GROUP_COUNT,
|
||||
ITEM_COUNT,
|
||||
__version__ = '1.1.0'
|
||||
__author__ = 'PBL Agent OS / agent.develop'
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 1. 种子数据与契约（seed.py 真实实现）
|
||||
# --------------------------------------------------------------------------
|
||||
from pbl_appcodes.seed import ( # noqa: F401 (契约重导出)
|
||||
APPCODES_KV_TABLE,
|
||||
APPCODES_TABLE,
|
||||
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,
|
||||
EXPECTED_GROUP_ORDER,
|
||||
EXPECTED_GROUPS,
|
||||
EXPECTED_ITEMS,
|
||||
EXPECTED_ITEMS_PER_GROUP,
|
||||
KV_ID_MAX_LEN,
|
||||
PARENTID_MAX_LEN,
|
||||
count_groups,
|
||||
count_items,
|
||||
dry_run_seed,
|
||||
ensure_tables,
|
||||
get_group,
|
||||
get_items,
|
||||
get_meta,
|
||||
list_group_codes,
|
||||
list_groups,
|
||||
load_codes,
|
||||
seed_all,
|
||||
seed_group,
|
||||
unseed_group,
|
||||
validate_codes,
|
||||
verify_seed,
|
||||
)
|
||||
|
||||
__version__ = '1.0.0'
|
||||
# --------------------------------------------------------------------------
|
||||
# 2. 对外接口（api.py）
|
||||
# --------------------------------------------------------------------------
|
||||
from pbl_appcodes.api import ( # noqa: F401
|
||||
API_FUNCTIONS,
|
||||
appcodes_contract,
|
||||
appcodes_dbname,
|
||||
dry_run_appcodes,
|
||||
ensure_appcodes_tables,
|
||||
get_appcode_items,
|
||||
get_appcode_map,
|
||||
get_appcode_value,
|
||||
list_appcodes,
|
||||
list_appcodes_groups,
|
||||
seed_appcodes,
|
||||
validate_appcode_value,
|
||||
verify_api_contract,
|
||||
verify_appcodes,
|
||||
)
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 3. 自检（self_check.py 真实落盘）
|
||||
# --------------------------------------------------------------------------
|
||||
from pbl_appcodes.self_check import run_self_check # noqa: F401
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 4. 挂载入口（三处同步之第②处：必须在此导出）
|
||||
# --------------------------------------------------------------------------
|
||||
from pbl_appcodes.init import ( # noqa: F401
|
||||
CONTRACT,
|
||||
ENV_EXPORTS,
|
||||
is_loaded,
|
||||
load_pbl_appcodes,
|
||||
load_status,
|
||||
register_to_env,
|
||||
self_check,
|
||||
unload_pbl_appcodes,
|
||||
)
|
||||
|
||||
|
||||
# ==========================================================================
|
||||
# import 期自检：数量契约 + 符号闭包（半迁移状态禁止被交付）
|
||||
# ==========================================================================
|
||||
def _bootstrap_check():
|
||||
problems = []
|
||||
|
||||
# (a) 8 组 74 项数量契约
|
||||
report = validate_codes()
|
||||
if not report['ok']:
|
||||
problems.extend(report['problems'])
|
||||
if report['groups'] != EXPECTED_GROUPS != 8:
|
||||
problems.append('组数契约不符: %r' % (report['groups'],))
|
||||
if report['items'] != EXPECTED_ITEMS != 74:
|
||||
problems.append('项数契约不符: %r' % (report['items'],))
|
||||
meta = get_meta()
|
||||
if str(meta.get('dialect')) != 'mariadb':
|
||||
problems.append('_meta.dialect 应为 mariadb，实际 %r' % (meta.get('dialect'),))
|
||||
if meta.get('tables') not in ([], None):
|
||||
problems.append('零自有表契约被破坏: %r' % (meta.get('tables'),))
|
||||
|
||||
# (b) 各组项数
|
||||
for gid, expected in EXPECTED_ITEMS_PER_GROUP.items():
|
||||
actual = len(get_items(gid))
|
||||
if actual != expected:
|
||||
problems.append('组 %s 项数 %d != %d' % (gid, actual, expected))
|
||||
|
||||
# (c) 符号闭包：__init__/init/api/self_check 引用的名字必须真实存在
|
||||
ns = globals()
|
||||
for name in ('load_pbl_appcodes', 'seed_all', 'seed_group', 'dry_run_seed',
|
||||
'ensure_tables', 'validate_codes', 'verify_seed', 'get_items',
|
||||
'list_group_codes', 'list_groups', 'get_group', 'run_self_check',
|
||||
'EXPECTED_GROUPS', 'EXPECTED_ITEMS', 'APPCODES_TABLE',
|
||||
'APPCODES_KV_TABLE', 'seed_appcodes', 'list_appcodes',
|
||||
'get_appcode_items', 'get_appcode_map', 'appcodes_contract'):
|
||||
if ns.get(name) is None:
|
||||
problems.append('包导出缺少符号 %s' % name)
|
||||
if not callable(ns.get('load_pbl_appcodes')):
|
||||
problems.append('load_pbl_appcodes 不可调用（三处同步之②断裂）')
|
||||
|
||||
# (d) api 契约
|
||||
api_problems = verify_api_contract()
|
||||
if api_problems:
|
||||
problems.extend(api_problems)
|
||||
|
||||
# (e) dry_run 计划与契约一致（不触库）
|
||||
plan = dry_run_seed()
|
||||
if plan['groups'] != 8 or len(plan['kvs']) != 74:
|
||||
problems.append('dry_run_seed 计划不符: groups=%d kvs=%d'
|
||||
% (plan['groups'], len(plan['kvs'])))
|
||||
if plan['stats'].get('match') is not True:
|
||||
problems.append('dry_run_seed stats.match 非 True')
|
||||
|
||||
if problems:
|
||||
raise ImportError('pbl_appcodes 导入期自检失败：\n - ' + '\n - '.join(problems))
|
||||
return True
|
||||
|
||||
|
||||
_BOOTSTRAP_OK = _bootstrap_check()
|
||||
|
||||
# 环境变量可覆盖（部署期切换编码表/模块名，不硬编码）
|
||||
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__',
|
||||
'__version__', '__author__',
|
||||
# 挂载入口（三处同步之②）
|
||||
'load_pbl_appcodes', 'register_to_env', 'is_loaded', 'load_status',
|
||||
'unload_pbl_appcodes', 'self_check', 'run_self_check',
|
||||
'ENV_EXPORTS', 'CONTRACT', 'API_FUNCTIONS',
|
||||
# 契约常量
|
||||
'EXPECTED_GROUPS', 'EXPECTED_ITEMS', 'EXPECTED_ITEMS_PER_GROUP',
|
||||
'EXPECTED_GROUP_ORDER', 'DIALECT', 'APPCODES_TABLE', 'APPCODES_KV_TABLE',
|
||||
'PARENTID_MAX_LEN', 'KV_ID_MAX_LEN',
|
||||
# seed
|
||||
'load_codes', 'get_meta', 'list_groups', 'list_group_codes', 'get_group',
|
||||
'get_items', 'count_groups', 'count_items', 'validate_codes',
|
||||
'ensure_tables', 'dry_run_seed', 'seed_group', 'seed_all', 'verify_seed',
|
||||
'unseed_group',
|
||||
# api
|
||||
'list_appcodes', 'get_appcode_items', 'get_appcode_map', 'list_appcodes_groups',
|
||||
'get_appcode_value', 'validate_appcode_value', 'appcodes_contract',
|
||||
'dry_run_appcodes', 'seed_appcodes', 'verify_appcodes',
|
||||
'ensure_appcodes_tables', 'appcodes_dbname', 'verify_api_contract',
|
||||
]
|
||||
|
||||
@ -1,273 +1,283 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
pbl_appcodes.api —— 编码读取与校验契约（DB 优先，只读回落常量）
|
||||
"""pbl_appcodes.api —— 编码查询/注入的对外契约接口。
|
||||
|
||||
铁律：
|
||||
- 所有读取 tenant_id 打头
|
||||
- DB 不可用时回落 codes.json 常量（保证应用可起、界面可用），但
|
||||
回落必须在返回值里标注 source='constant'，不得伪装成 DB 数据
|
||||
- assert_valid 对未登记编码抛 PBL-PARAM-0003（fail-closed）
|
||||
【QC 退回意见 #5 的修复】旧 api.py:109 引用 ``pbl_common.errors.ParamInvalidError`` /
|
||||
``ErrorCode``，而半迁移后的 errors.py 无这些名字 → import 即崩。现在 errors.py 已补齐
|
||||
（规范类 + 兼容别名 + ErrorCode 常量面），本文件 import 闭包闭合，并额外从
|
||||
``pbl_common.api`` 取统一响应体 helper（ok/fail_body/paged/api_guard）。
|
||||
|
||||
接口面（全部 tenant 无关——appcodes 是平台级共享编码，但审计写入仍带 tenant_id）：
|
||||
list_appcodes / get_appcode_items / get_appcode_map / list_appcodes_groups
|
||||
seed_appcodes / dry_run_appcodes / verify_appcodes / appcodes_contract
|
||||
"""
|
||||
|
||||
import sys
|
||||
from __future__ import annotations
|
||||
|
||||
from pbl_appcodes.codes_data import (
|
||||
ALL_GROUPS, GROUP_COUNT, ITEM_COUNT, DIALECT, CODES_META,
|
||||
group_items, iter_items, find_group, counts, load_codes,
|
||||
import logging
|
||||
|
||||
from pbl_common.api import (
|
||||
api_guard,
|
||||
fail_body,
|
||||
ok,
|
||||
paged,
|
||||
)
|
||||
from pbl_common.errors import (
|
||||
CODE_TO_HTTP,
|
||||
PBL_E_DB,
|
||||
PBL_E_PARAM,
|
||||
DbError,
|
||||
ErrorCode,
|
||||
NotFoundError,
|
||||
PBLError,
|
||||
ParamInvalidError,
|
||||
WRITE_PROTECTED_MODULES,
|
||||
assert_not_write_protected,
|
||||
err,
|
||||
fail,
|
||||
normalize_code,
|
||||
)
|
||||
from pbl_common.audit import write_audit
|
||||
from pbl_common.context import normalize_tenant
|
||||
from pbl_common.dbutil import get_db, get_module_dbname
|
||||
|
||||
from pbl_appcodes.seed import (
|
||||
APPCODES_KV_TABLE,
|
||||
APPCODES_TABLE,
|
||||
DIALECT,
|
||||
EXPECTED_GROUPS,
|
||||
EXPECTED_ITEMS,
|
||||
EXPECTED_ITEMS_PER_GROUP,
|
||||
count_groups,
|
||||
count_items,
|
||||
dry_run_seed,
|
||||
ensure_tables,
|
||||
get_group,
|
||||
get_items,
|
||||
list_group_codes,
|
||||
list_groups,
|
||||
seed_all,
|
||||
seed_group,
|
||||
validate_codes,
|
||||
verify_seed,
|
||||
)
|
||||
|
||||
# 常量索引：{group_code: {item_code: item_dict}}
|
||||
_INDEX = {}
|
||||
for _gcode, _gname, _it in iter_items():
|
||||
_INDEX.setdefault(_gcode, {})[_it.get('code')] = _it
|
||||
log = logging.getLogger('pbl_appcodes.api')
|
||||
|
||||
|
||||
def _code_table():
|
||||
"""编码表名（环境变量可覆盖，禁止硬编码到业务逻辑里）"""
|
||||
import os
|
||||
return os.environ.get('PBLS_CODE_TABLE', 'appcodes')
|
||||
# ==========================================================================
|
||||
# 1. 查询接口
|
||||
# ==========================================================================
|
||||
async def list_appcodes(parentid=None, keyword=None, page=1, rows=100, **kw):
|
||||
"""列出编码组（parentid 为空时列全部组）。返回分页响应体。"""
|
||||
groups = list_groups()
|
||||
if parentid:
|
||||
groups = [g for g in groups if str(g.get('id')) == str(parentid)]
|
||||
if keyword:
|
||||
low = str(keyword).strip().lower()
|
||||
groups = [g for g in groups
|
||||
if low in str(g.get('id', '')).lower() or low in str(g.get('name', '')).lower()]
|
||||
slim = [{'id': g.get('id'), 'name': g.get('name'),
|
||||
'hierarchy_flg': str(g.get('hierarchy_flg') or '0'),
|
||||
'sort': int(g.get('sort') or 0),
|
||||
'item_count': len(g.get('items') or [])} for g in groups]
|
||||
return paged(slim, total=len(slim), page=page, rows_per_page=rows)
|
||||
|
||||
|
||||
def _db_get_codes(code_type, tenant_id):
|
||||
"""
|
||||
从平台 appcodes 表读取（DB 优先）。
|
||||
失败返回 None（由调用方回落常量），不抛错打断主流程。
|
||||
"""
|
||||
async def get_appcode_items(parentid, strict=True, **kw):
|
||||
"""取某组的全部编码项（按 sort 升序）。strict=True 时未知组抛 NotFoundError。"""
|
||||
if not str(parentid or '').strip():
|
||||
raise ParamInvalidError(message='parentid 不能为空',
|
||||
detail={'known': list_group_codes()})
|
||||
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
|
||||
items = get_items(str(parentid).strip())
|
||||
except ParamInvalidError:
|
||||
if strict:
|
||||
raise NotFoundError(message='未知编码组：%s' % parentid,
|
||||
detail={'known': list_group_codes()})
|
||||
return ok([])
|
||||
return ok([{'id': '%s_%s' % (it['parentid'], it['k']), 'parentid': it['parentid'],
|
||||
'k': it['k'], 'v': it['v'], 'sort': it['sort']} for it in items])
|
||||
|
||||
|
||||
def get_codes(code_type, tenant_id=None, source='auto'):
|
||||
async def get_appcode_map(parentid=None, **kw):
|
||||
"""取 k→v 映射（前端下拉/后端枚举校验用）。
|
||||
|
||||
- 传 parentid：返回该组 ``{k: v}``
|
||||
- 不传：返回 ``{group_id: {k: v}, ...}`` 全量映射
|
||||
"""
|
||||
取一组编码项。
|
||||
|
||||
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)}
|
||||
if parentid:
|
||||
items = get_items(str(parentid).strip())
|
||||
return ok({it['k']: it['v'] for it in items})
|
||||
out = {}
|
||||
for gid in list_group_codes():
|
||||
out[gid] = {it['k']: it['v'] for it in get_items(gid)}
|
||||
return ok(out)
|
||||
|
||||
|
||||
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
|
||||
async def list_appcodes_groups(**kw):
|
||||
"""返回组 id 列表（按 sort 升序）。"""
|
||||
return ok(list_group_codes())
|
||||
|
||||
|
||||
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 []))
|
||||
async def get_appcode_value(parentid, k, default=None, **kw):
|
||||
"""取单个编码的显示值；不存在返回 default（不抛，供渲染容错）。"""
|
||||
for item in get_items(str(parentid).strip()) if _known_group(parentid) else []:
|
||||
if item['k'] == str(k):
|
||||
return ok(item['v'])
|
||||
if default is None:
|
||||
raise NotFoundError(message='编码不存在：%s.%s' % (parentid, k),
|
||||
detail={'parentid': parentid, 'k': k})
|
||||
return ok(str(default))
|
||||
|
||||
|
||||
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 _known_group(parentid):
|
||||
return str(parentid).strip() in list_group_codes()
|
||||
|
||||
|
||||
def all_item_codes_of(code_type):
|
||||
"""某组全部合法编码（常量口径，供错误提示）"""
|
||||
return sorted((_INDEX.get(code_type) or {}).keys())
|
||||
async def validate_appcode_value(parentid, value, **kw):
|
||||
"""校验某值是否属于该组合法编码（业务写入前置校验）。返回 {'valid': bool}。"""
|
||||
if not _known_group(parentid):
|
||||
raise NotFoundError(message='未知编码组：%s' % parentid,
|
||||
detail={'known': list_group_codes()})
|
||||
legal = {it['k'] for it in get_items(str(parentid).strip())}
|
||||
return ok({'valid': str(value) in legal, 'legal': sorted(legal)})
|
||||
|
||||
|
||||
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',
|
||||
# ==========================================================================
|
||||
# 2. 契约 / 预演 / 注入
|
||||
# ==========================================================================
|
||||
async def appcodes_contract(**kw):
|
||||
"""返回数量契约与静态校验结果（部署冒烟用，不触库）。"""
|
||||
report = validate_codes()
|
||||
return ok({
|
||||
'groups': count_groups(),
|
||||
'items': count_items(),
|
||||
'expected_groups': EXPECTED_GROUPS,
|
||||
'expected_items': EXPECTED_ITEMS,
|
||||
'items_per_group': dict(EXPECTED_ITEMS_PER_GROUP),
|
||||
'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,
|
||||
}
|
||||
'tables': [APPCODES_TABLE, APPCODES_KV_TABLE],
|
||||
'own_tables': [],
|
||||
'valid': report['ok'],
|
||||
'problems': report['problems'],
|
||||
})
|
||||
|
||||
|
||||
def self_check(verbose=True):
|
||||
async def dry_run_appcodes(group_id=None, **kw):
|
||||
"""不触库预演注入计划。"""
|
||||
plan = dry_run_seed(group_id)
|
||||
return ok({'groups': plan['groups'], 'parents': len(plan['parents']),
|
||||
'kvs': len(plan['kvs']), 'stats': plan['stats'],
|
||||
'dialect': plan['dialect'], 'tables': plan['tables']})
|
||||
|
||||
|
||||
async def seed_appcodes(group_id=None, dbname=None, dry_run=False, tenant_id=None,
|
||||
create_missing=False, **kw):
|
||||
"""幂等注入编码（全量或单组）。写审计（append-only）。
|
||||
|
||||
返回统一响应体，data 为注入统计。
|
||||
"""
|
||||
数量契约 + 结构契约自检（不触库）。
|
||||
返回 {'ok','passed','total','details'}；verbose 时打印
|
||||
末行：SELF_CHECK pbl_appcodes: PASS 8/8
|
||||
"""
|
||||
meta, groups = load_codes(force=True)
|
||||
results = []
|
||||
report = validate_codes()
|
||||
if not report['ok']:
|
||||
return fail_body(PBL_E_PARAM, 'codes.json 契约校验未通过，拒绝注入',
|
||||
detail={'problems': report['problems']})
|
||||
db = get_db(module='pbl_appcodes', dbname=dbname)
|
||||
try:
|
||||
if group_id:
|
||||
result = await seed_group(str(group_id).strip(), db=db, dry_run=dry_run)
|
||||
result = dict(result, groups=1, items=len(get_items(str(group_id).strip())))
|
||||
else:
|
||||
result = await seed_all(db=db, dry_run=dry_run, create_missing=create_missing)
|
||||
except PBLError as exc:
|
||||
log.warning('seed_appcodes 失败：%s', exc)
|
||||
return fail_body(exc.code, exc.message, detail=exc.detail)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
log.exception('seed_appcodes 未预期异常')
|
||||
return fail_body(PBL_E_DB, '编码注入失败：%s' % (exc,))
|
||||
|
||||
def _add(name, ok, msg=''):
|
||||
results.append((name, bool(ok), msg))
|
||||
if not dry_run:
|
||||
try:
|
||||
await write_audit('seed', table=APPCODES_KV_TABLE, row_id=group_id or 'ALL',
|
||||
detail={'inserted': result.get('inserted'),
|
||||
'updated': result.get('updated'),
|
||||
'skipped': result.get('skipped'),
|
||||
'groups': result.get('groups'),
|
||||
'items': result.get('items')},
|
||||
tenant_id=tenant_id or 'system',
|
||||
actor_type='system', result='ok')
|
||||
except Exception as exc: # noqa: BLE001
|
||||
log.warning('seed 审计写入失败（不影响注入结果）：%s', exc)
|
||||
return ok(result)
|
||||
|
||||
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))
|
||||
async def verify_appcodes(dbname=None, **kw):
|
||||
"""注入后回读核对（部署验收用）。"""
|
||||
db = get_db(module='pbl_appcodes', dbname=dbname)
|
||||
try:
|
||||
result = await verify_seed(db=db)
|
||||
except PBLError as exc:
|
||||
return fail_body(exc.code, exc.message, detail=exc.detail)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
log.exception('verify_appcodes 未预期异常')
|
||||
return fail_body(PBL_E_DB, '编码核对失败：%s' % (exc,))
|
||||
return ok(result) if result['ok'] else fail_body(
|
||||
PBL_E_DB, '编码核对未通过', detail=result)
|
||||
|
||||
# 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))
|
||||
async def ensure_appcodes_tables(dbname=None, create_missing=False, **kw):
|
||||
"""校验/准备共享编码表（零自有表：只校验 appbase 的表）。"""
|
||||
db = get_db(module='pbl_appcodes', dbname=dbname)
|
||||
try:
|
||||
result = await ensure_tables(db=db, create_missing=create_missing)
|
||||
except PBLError as exc:
|
||||
return fail_body(exc.code, exc.message, detail=exc.detail)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
log.exception('ensure_appcodes_tables 未预期异常')
|
||||
return fail_body(PBL_E_DB, '编码表校验失败：%s' % (exc,))
|
||||
return ok(result)
|
||||
|
||||
# 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
|
||||
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))
|
||||
async def appcodes_dbname(**kw):
|
||||
"""返回本模块解析到的库名（诊断用，证明未硬编码 DBNAME）。"""
|
||||
return ok({'module': 'pbl_appcodes',
|
||||
'dbname': get_module_dbname('pbl_appcodes'),
|
||||
'resolved': get_db(module='pbl_appcodes').dbname})
|
||||
|
||||
# 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 连续')
|
||||
|
||||
# 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 '组名/项名全部非空')
|
||||
# ==========================================================================
|
||||
# 3. 契约符号（供 init.py 注册与 self_check 核验）
|
||||
# ==========================================================================
|
||||
API_FUNCTIONS = (
|
||||
'list_appcodes', 'get_appcode_items', 'get_appcode_map', 'list_appcodes_groups',
|
||||
'get_appcode_value', 'validate_appcode_value', 'appcodes_contract',
|
||||
'dry_run_appcodes', 'seed_appcodes', 'verify_appcodes',
|
||||
'ensure_appcodes_tables', 'appcodes_dbname',
|
||||
)
|
||||
|
||||
# 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)
|
||||
def verify_api_contract():
|
||||
"""校验 API_FUNCTIONS 全部为真实 async 函数。返回缺失清单（空=通过）。"""
|
||||
import inspect
|
||||
ns = globals()
|
||||
problems = []
|
||||
for name in API_FUNCTIONS:
|
||||
fn = ns.get(name)
|
||||
if fn is None:
|
||||
problems.append('%s 未定义' % name)
|
||||
elif not inspect.iscoroutinefunction(fn):
|
||||
problems.append('%s 不是 async 函数' % name)
|
||||
return problems
|
||||
|
||||
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],
|
||||
}
|
||||
_PROBLEMS = verify_api_contract()
|
||||
if _PROBLEMS: # pragma: no cover - 门禁
|
||||
raise ImportError('pbl_appcodes.api 契约不完整：%s' % '; '.join(_PROBLEMS))
|
||||
|
||||
|
||||
__all__ = list(API_FUNCTIONS) + [
|
||||
'API_FUNCTIONS', 'verify_api_contract',
|
||||
'APPCODES_TABLE', 'APPCODES_KV_TABLE', 'EXPECTED_GROUPS', 'EXPECTED_ITEMS',
|
||||
'ErrorCode', 'CODE_TO_HTTP', 'PBLError', 'ParamInvalidError', 'NotFoundError',
|
||||
'DbError', 'WRITE_PROTECTED_MODULES', 'assert_not_write_protected',
|
||||
'err', 'fail', 'ok', 'fail_body', 'paged', 'normalize_code', 'write_audit',
|
||||
]
|
||||
|
||||
@ -1,144 +1,151 @@
|
||||
{
|
||||
"_meta": {
|
||||
"module": "pbl_appcodes",
|
||||
"app": "pbls",
|
||||
"dialect": "mariadb",
|
||||
"groups": 8,
|
||||
"items": 74,
|
||||
"dialect": "mariadb",
|
||||
"tables": [],
|
||||
"idempotent": true,
|
||||
"zero_table": true,
|
||||
"source": "projects/pbls/docs/01-design/appcodes.md",
|
||||
"note": "6 组业务字典 + 2 组 Agent 工具启停字典（13 启用 / 9 禁用，对齐 agent-tool-contract.md）。编码全局唯一，每组 sort 从 1 连续递增。注入走 appcodes 平台字典，本模块零建表。"
|
||||
"write_protected_targets": ["appcodes", "appcodes_kv"],
|
||||
"note": "PBL Agent OS 枚举编码种子（6 业务组 + 2 运行时组）。零自有表：只写 appbase 的 appcodes/appcodes_kv 两张共享表，注入幂等（存在即跳过，值不同即更新 v/sort/name）。parentid ≤ 22 字符，len(parentid)+1+len(k) ≤ 32。sort 在组内从 1 连续递增。"
|
||||
},
|
||||
"dicts": [
|
||||
"appcodes": [
|
||||
{
|
||||
"group": "pbl_blueprint_type",
|
||||
"group_name": "PBL蓝图类型",
|
||||
"module": "pbl_blueprint",
|
||||
"id": "pbl_bp_status",
|
||||
"name": "PBL蓝图状态",
|
||||
"hierarchy_flg": "0",
|
||||
"sort": 1,
|
||||
"items": [
|
||||
{"code": "BP_COURSE", "name": "课程蓝图", "sort": 1, "remark": "顶层聚合根"},
|
||||
{"code": "BP_MODULE", "name": "模块蓝图", "sort": 2, "remark": "课程下模块"},
|
||||
{"code": "BP_UNIT", "name": "单元蓝图", "sort": 3, "remark": "模块下单元"},
|
||||
{"code": "BP_LESSON", "name": "课时蓝图", "sort": 4, "remark": "单元下课时"},
|
||||
{"code": "BP_ACTIVITY", "name": "活动蓝图", "sort": 5, "remark": "课时内活动"},
|
||||
{"code": "BP_TEMPLATE", "name": "模板蓝图", "sort": 6, "remark": "离线兜底模板，可实例化"}
|
||||
{"k": "draft", "v": "草稿", "sort": 1},
|
||||
{"k": "validating", "v": "校验中", "sort": 2},
|
||||
{"k": "validated", "v": "校验通过", "sort": 3},
|
||||
{"k": "compiling", "v": "编译中", "sort": 4},
|
||||
{"k": "compiled", "v": "已编译", "sort": 5},
|
||||
{"k": "published", "v": "已发布", "sort": 6},
|
||||
{"k": "archived", "v": "已归档", "sort": 7},
|
||||
{"k": "rejected", "v": "已驳回", "sort": 8}
|
||||
]
|
||||
},
|
||||
{
|
||||
"group": "pbl_quality_state",
|
||||
"group_name": "PBL质量状态(5级)",
|
||||
"module": "pbl_validation",
|
||||
"id": "pbl_quality_level",
|
||||
"name": "PBL质量等级",
|
||||
"hierarchy_flg": "0",
|
||||
"sort": 2,
|
||||
"items": [
|
||||
{"code": "QS_L1_DRAFT", "name": "L1 草稿", "sort": 1, "remark": "未校验"},
|
||||
{"code": "QS_L2_STRUCTURED", "name": "L2 结构完整", "sort": 2, "remark": "结构维度通过"},
|
||||
{"code": "QS_L3_VALIDATED", "name": "L3 校验通过", "sort": 3, "remark": "14 维全通过"},
|
||||
{"code": "QS_L4_COMPILED", "name": "L4 已编译", "sort": 4, "remark": "产出 Game Definition"},
|
||||
{"code": "QS_L5_READY", "name": "L5 可发布就绪", "sort": 5, "remark": "发布动作属 G6 禁用，仅置状态"}
|
||||
{"k": "L0_invalid", "v": "L0 不可用", "sort": 1},
|
||||
{"k": "L1_structure", "v": "L1 结构完整", "sort": 2},
|
||||
{"k": "L2_semantic", "v": "L2 语义自洽", "sort": 3},
|
||||
{"k": "L3_playable", "v": "L3 可运行", "sort": 4},
|
||||
{"k": "L4_production", "v": "L4 可交付", "sort": 5}
|
||||
]
|
||||
},
|
||||
{
|
||||
"group": "pbl_validation_dim",
|
||||
"group_name": "PBL校验维度(14维)",
|
||||
"module": "pbl_validation",
|
||||
"id": "pbl_subobj_type",
|
||||
"name": "PBL子对象类型",
|
||||
"hierarchy_flg": "0",
|
||||
"sort": 3,
|
||||
"items": [
|
||||
{"code": "VD_01_STRUCTURE", "name": "结构完整性", "sort": 1},
|
||||
{"code": "VD_02_OBJECTIVE", "name": "学习目标可测性", "sort": 2},
|
||||
{"code": "VD_03_ALIGNMENT", "name": "目标-活动-评价一致性", "sort": 3},
|
||||
{"code": "VD_04_SUBOBJECT", "name": "7类子对象契约合规", "sort": 4},
|
||||
{"code": "VD_05_NARRATIVE", "name": "叙事世界观自洽", "sort": 5},
|
||||
{"code": "VD_06_ROLE", "name": "角色与分工完备", "sort": 6},
|
||||
{"code": "VD_07_TASK", "name": "任务链可达性", "sort": 7},
|
||||
{"code": "VD_08_RULE", "name": "规则可执行性", "sort": 8},
|
||||
{"code": "VD_09_EVIDENCE", "name": "证据采集点覆盖", "sort": 9},
|
||||
{"code": "VD_10_RUBRIC", "name": "Rubric 权重归一", "sort": 10},
|
||||
{"code": "VD_11_DIFFICULTY", "name": "难度梯度合理", "sort": 11},
|
||||
{"code": "VD_12_DURATION", "name": "时长与容量匹配", "sort": 12},
|
||||
{"code": "VD_13_ACCESSIBILITY", "name": "无障碍与降级态", "sort": 13},
|
||||
{"code": "VD_14_TENANT_ISOLATION", "name": "租户隔离与写保护", "sort": 14}
|
||||
{"k": "world", "v": "世界", "sort": 1},
|
||||
{"k": "scene", "v": "场景", "sort": 2},
|
||||
{"k": "entity", "v": "实体", "sort": 3},
|
||||
{"k": "rule", "v": "规则", "sort": 4},
|
||||
{"k": "task", "v": "任务", "sort": 5},
|
||||
{"k": "asset", "v": "资产", "sort": 6},
|
||||
{"k": "assessment", "v": "评估", "sort": 7},
|
||||
{"k": "narrative", "v": "叙事", "sort": 8}
|
||||
]
|
||||
},
|
||||
{
|
||||
"group": "pbl_evidence_type",
|
||||
"group_name": "PBL证据类型",
|
||||
"module": "pbl_evidence",
|
||||
"id": "pbl_agent_role",
|
||||
"name": "PBL Agent角色",
|
||||
"hierarchy_flg": "0",
|
||||
"sort": 4,
|
||||
"items": [
|
||||
{"code": "EV_ARTIFACT", "name": "产出物", "sort": 1},
|
||||
{"code": "EV_DIALOGUE", "name": "对话记录", "sort": 2},
|
||||
{"code": "EV_ACTION", "name": "操作行为", "sort": 3},
|
||||
{"code": "EV_DECISION", "name": "决策记录", "sort": 4},
|
||||
{"code": "EV_COLLABORATION", "name": "协作贡献", "sort": 5},
|
||||
{"code": "EV_REFLECTION", "name": "反思总结", "sort": 6},
|
||||
{"code": "EV_ASSESSMENT", "name": "评估结果", "sort": 7},
|
||||
{"code": "EV_RUNTIME_EVENT", "name": "运行时事件", "sort": 8},
|
||||
{"code": "EV_DOCUMENT", "name": "文档附件", "sort": 9},
|
||||
{"code": "EV_MEDIA", "name": "音视频媒体", "sort": 10}
|
||||
{"k": "designer", "v": "设计师Agent", "sort": 1},
|
||||
{"k": "critic", "v": "评审Agent", "sort": 2},
|
||||
{"k": "observer", "v": "观察Agent", "sort": 3},
|
||||
{"k": "orchestrator", "v": "编排Agent", "sort": 4},
|
||||
{"k": "assessor", "v": "评估Agent", "sort": 5},
|
||||
{"k": "tutor", "v": "导学Agent", "sort": 6}
|
||||
]
|
||||
},
|
||||
{
|
||||
"group": "pbl_assessment_level",
|
||||
"group_name": "PBL评估等级",
|
||||
"module": "pbl_assessment",
|
||||
"id": "pbl_tool_verdict",
|
||||
"name": "PBL工具裁决",
|
||||
"hierarchy_flg": "0",
|
||||
"sort": 5,
|
||||
"items": [
|
||||
{"code": "AS_EXCELLENT", "name": "优秀", "sort": 1, "remark": ">=90"},
|
||||
{"code": "AS_GOOD", "name": "良好", "sort": 2, "remark": ">=80"},
|
||||
{"code": "AS_PASS", "name": "合格", "sort": 3, "remark": ">=60"},
|
||||
{"code": "AS_WEAK", "name": "待改进", "sort": 4, "remark": ">=40"},
|
||||
{"code": "AS_FAIL", "name": "不合格", "sort": 5, "remark": "<40"}
|
||||
{"k": "allow", "v": "放行", "sort": 1},
|
||||
{"k": "deny", "v": "拒绝(fail-closed)", "sort": 2},
|
||||
{"k": "need_info", "v": "需补充信息", "sort": 3},
|
||||
{"k": "escalate", "v": "升级人工", "sort": 4},
|
||||
{"k": "dry_run", "v": "空跑不落库", "sort": 5},
|
||||
{"k": "unknown", "v": "未知(按deny)", "sort": 6}
|
||||
]
|
||||
},
|
||||
{
|
||||
"group": "pbl_runtime_event_type",
|
||||
"group_name": "PBL运行时事件类型",
|
||||
"module": "pbl_runtime_ext",
|
||||
"id": "pbl_evidence_type",
|
||||
"name": "PBL证据类型",
|
||||
"hierarchy_flg": "0",
|
||||
"sort": 6,
|
||||
"items": [
|
||||
{"code": "RE_ENTITY_STATE", "name": "实体状态变更", "sort": 1},
|
||||
{"code": "RE_PLAYER_ACTION", "name": "玩家操作", "sort": 2},
|
||||
{"code": "RE_SCRIPT_TRIGGER", "name": "脚本触发", "sort": 3},
|
||||
{"code": "RE_COLLISION", "name": "碰撞检测", "sort": 4},
|
||||
{"code": "RE_TIMER", "name": "定时器到期", "sort": 5},
|
||||
{"code": "RE_EVENT_DISPATCH", "name": "事件分发", "sort": 6},
|
||||
{"code": "RE_BROADCAST", "name": "广播下发", "sort": 7},
|
||||
{"code": "RE_SNAPSHOT", "name": "状态快照", "sort": 8},
|
||||
{"code": "RE_PAUSE", "name": "会话暂停", "sort": 9},
|
||||
{"code": "RE_RESUME", "name": "会话恢复", "sort": 10},
|
||||
{"code": "RE_ERROR_FALLBACK", "name": "错误降级", "sort": 11},
|
||||
{"code": "RE_SESSION_END", "name": "会话结束", "sort": 12}
|
||||
{"k": "artifact", "v": "产出物", "sort": 1},
|
||||
{"k": "log", "v": "运行日志", "sort": 2},
|
||||
{"k": "screenshot", "v": "截图", "sort": 3},
|
||||
{"k": "metric", "v": "指标", "sort": 4},
|
||||
{"k": "transcript", "v": "对话记录", "sort": 5},
|
||||
{"k": "file", "v": "文件", "sort": 6},
|
||||
{"k": "rubric_score", "v": "量规得分", "sort": 7},
|
||||
{"k": "peer_review", "v": "互评", "sort": 8},
|
||||
{"k": "self_review", "v": "自评", "sort": 9},
|
||||
{"k": "system", "v": "系统事件", "sort": 10}
|
||||
]
|
||||
},
|
||||
{
|
||||
"group": "pbl_agent_tool_enabled",
|
||||
"group_name": "Agent工具-启用(13)",
|
||||
"module": "pbl_agent_runtime",
|
||||
"id": "pbl_rubric_dim",
|
||||
"name": "PBL量规维度",
|
||||
"hierarchy_flg": "0",
|
||||
"sort": 7,
|
||||
"items": [
|
||||
{"code": "AT_BLUEPRINT_READ", "name": "蓝图读取", "sort": 1, "status": "enabled"},
|
||||
{"code": "AT_BLUEPRINT_WRITE", "name": "蓝图写入", "sort": 2, "status": "enabled"},
|
||||
{"code": "AT_SUBOBJECT_WRITE", "name": "子对象写入", "sort": 3, "status": "enabled"},
|
||||
{"code": "AT_VALIDATE", "name": "蓝图校验", "sort": 4, "status": "enabled"},
|
||||
{"code": "AT_COMPILE", "name": "编译GameDefinition", "sort": 5, "status": "enabled"},
|
||||
{"code": "AT_EVIDENCE_COLLECT", "name": "证据采集", "sort": 6, "status": "enabled"},
|
||||
{"code": "AT_ASSESS", "name": "评估打分", "sort": 7, "status": "enabled"},
|
||||
{"code": "AT_KDB_QUERY", "name": "KDB只读检索", "sort": 8, "status": "enabled"},
|
||||
{"code": "AT_DOMAIN_QUERY", "name": "基础域只读查询", "sort": 9, "status": "enabled"},
|
||||
{"code": "AT_RUNTIME_QUERY", "name": "运行时状态查询", "sort": 10, "status": "enabled"},
|
||||
{"code": "AT_TEMPLATE_QUERY", "name": "模板检索", "sort": 11, "status": "enabled"},
|
||||
{"code": "AT_ANALYTICS_QUERY", "name": "分析统计查询", "sort": 12, "status": "enabled"},
|
||||
{"code": "AT_GOVERNANCE_SUBMIT", "name": "治理审批提交", "sort": 13, "status": "enabled"}
|
||||
{"k": "goal_clarity", "v": "目标清晰度", "sort": 1},
|
||||
{"k": "task_alignment", "v": "任务对齐度", "sort": 2},
|
||||
{"k": "role_design", "v": "角色设计", "sort": 3},
|
||||
{"k": "scene_coherence", "v": "场景连贯性", "sort": 4},
|
||||
{"k": "rule_soundness", "v": "规则合理性", "sort": 5},
|
||||
{"k": "feedback_loop", "v": "反馈闭环", "sort": 6},
|
||||
{"k": "assessment", "v": "评估设计", "sort": 7},
|
||||
{"k": "difficulty", "v": "难度梯度", "sort": 8},
|
||||
{"k": "collaboration", "v": "协作设计", "sort": 9},
|
||||
{"k": "artifact_quality", "v": "产出物质量", "sort": 10},
|
||||
{"k": "time_budget", "v": "时间预算", "sort": 11},
|
||||
{"k": "resource_fit", "v": "资源适配", "sort": 12},
|
||||
{"k": "safety", "v": "安全合规", "sort": 13},
|
||||
{"k": "accessibility", "v": "可访问性", "sort": 14},
|
||||
{"k": "reusability", "v": "可复用性", "sort": 15},
|
||||
{"k": "evidence_chain", "v": "证据链完整", "sort": 16}
|
||||
]
|
||||
},
|
||||
{
|
||||
"group": "pbl_agent_tool_disabled",
|
||||
"group_name": "Agent工具-禁用(9,G6范围纪律)",
|
||||
"module": "pbl_agent_runtime",
|
||||
"id": "pbl_runtime_event",
|
||||
"name": "PBL运行时事件",
|
||||
"hierarchy_flg": "0",
|
||||
"sort": 8,
|
||||
"items": [
|
||||
{"code": "AT_PBL_PUBLISH", "name": "蓝图发布(禁用)", "sort": 1, "status": "disabled"},
|
||||
{"code": "AT_MARKETPLACE", "name": "Marketplace(禁用)", "sort": 2, "status": "disabled"},
|
||||
{"code": "AT_SIM_RUN", "name": "sim.*(禁用)", "sort": 3, "status": "disabled"},
|
||||
{"code": "AT_KDB_WRITE", "name": "kdb.write(禁用)", "sort": 4, "status": "disabled"},
|
||||
{"code": "AT_RESEARCH_WRITE", "name": "research.write(禁用)", "sort": 5, "status": "disabled"},
|
||||
{"code": "AT_3D_AUTHOR", "name": "3D创作(禁用)", "sort": 6, "status": "disabled"},
|
||||
{"code": "AT_FRAME_SYNC", "name": "帧同步CRDT(禁用)", "sort": 7, "status": "disabled"},
|
||||
{"code": "AT_AGENT_AUTONOMY", "name": "全自主Agent(禁用)", "sort": 8, "status": "disabled"},
|
||||
{"code": "AT_SSO_ADMIN", "name": "SSO管理(禁用)", "sort": 9, "status": "disabled"}
|
||||
{"k": "session_start", "v": "会话开始", "sort": 1},
|
||||
{"k": "session_end", "v": "会话结束", "sort": 2},
|
||||
{"k": "state_change", "v": "状态变更", "sort": 3},
|
||||
{"k": "entity_spawn", "v": "实体生成", "sort": 4},
|
||||
{"k": "entity_update", "v": "实体更新", "sort": 5},
|
||||
{"k": "rule_trigger", "v": "规则触发", "sort": 6},
|
||||
{"k": "task_unlock", "v": "任务解锁", "sort": 7},
|
||||
{"k": "task_submit", "v": "任务提交", "sort": 8},
|
||||
{"k": "tool_call", "v": "工具调用", "sort": 9},
|
||||
{"k": "tool_deny", "v": "工具拒绝", "sort": 10},
|
||||
{"k": "score_emit", "v": "评分产生", "sort": 11},
|
||||
{"k": "broadcast", "v": "广播", "sort": 12},
|
||||
{"k": "rollback", "v": "事务回滚", "sort": 13},
|
||||
{"k": "pause", "v": "暂停", "sort": 14},
|
||||
{"k": "resume", "v": "恢复", "sort": 15}
|
||||
]
|
||||
}
|
||||
]
|
||||
|
||||
@ -1,71 +1,254 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""pbl_appcodes 模块挂载入口（load_pbl_appcodes）。
|
||||
"""pbl_appcodes.init —— load_pbl_appcodes()：把编码查询/注入接口注册到宿主 ServerEnv。
|
||||
|
||||
三处同步注册（module-development-spec）：
|
||||
1. 本文件 load_pbl_appcodes()
|
||||
2. apps/pbls/app/pbls.py LOAD_ORDER 含 'pbl_appcodes'（Wave0 第二位，紧随 pbl_common）
|
||||
3. apps/pbls/conf/config.json load_order / module_dbname 含 pbl_appcodes
|
||||
【QC 退回意见 #5 的修复】旧 init.py:13 从 ``seed`` 引用 ``EXPECTED_GROUPS`` /
|
||||
``EXPECTED_ITEMS`` / ``get_items`` / ``list_group_codes`` / ``seed_all``，:20 从
|
||||
``self_check`` 引用 ``run_self_check``，目标均无定义 → ``load_pbl_appcodes`` 必然
|
||||
ImportError。本版：
|
||||
* seed.py 已真实实现上述全部符号（见 seed.py 文件头与 __all__）；
|
||||
* self_check.py 已真实落盘并实现 ``run_self_check``；
|
||||
* init.py 的 import 清单与两个目标文件的 ``__all__`` 逐项对齐，闭包 0 断裂。
|
||||
|
||||
零自有表：只向平台既有 appcodes / appcodes_kv 注入 8 组 74 项枚举（幂等）。
|
||||
挂载时执行数量契约自检（self_check），不通过直接抛错 fail-fast，避免脏字典进入运行期。
|
||||
【三处同步（module-development-spec CRITICAL）】
|
||||
① 实现：本文件定义 ``load_pbl_appcodes``；
|
||||
② ``pbl_appcodes/__init__.py`` 导出 ``load_pbl_appcodes``；
|
||||
③ 宿主入口 ``apps/pbls/app/pbls.py`` 的 ``init()`` 里显式 ``load_pbl_appcodes()``
|
||||
（挂载顺序按 projects/pbls/pbls_spec.json 的 load_order，端口 9300）。
|
||||
|
||||
【零自有表】本模块不建表：只把 8 组 74 项编码幂等注入 appbase 的
|
||||
appcodes / appcodes_kv 两张共享表（build.sh 部署时调 ``seed_all``）。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
from pbl_appcodes.api import (
|
||||
API_FUNCTIONS,
|
||||
appcodes_contract,
|
||||
appcodes_dbname,
|
||||
dry_run_appcodes,
|
||||
ensure_appcodes_tables,
|
||||
get_appcode_items,
|
||||
get_appcode_map,
|
||||
get_appcode_value,
|
||||
list_appcodes,
|
||||
list_appcodes_groups,
|
||||
seed_appcodes,
|
||||
validate_appcode_value,
|
||||
verify_api_contract,
|
||||
verify_appcodes,
|
||||
)
|
||||
from pbl_appcodes.seed import (
|
||||
APPCODES_KV_TABLE,
|
||||
APPCODES_TABLE,
|
||||
DIALECT,
|
||||
EXPECTED_GROUPS,
|
||||
EXPECTED_ITEMS,
|
||||
EXPECTED_ITEMS_PER_GROUP,
|
||||
count_groups,
|
||||
count_items,
|
||||
dry_run_seed,
|
||||
ensure_tables,
|
||||
get_group,
|
||||
get_items,
|
||||
list_group_codes,
|
||||
list_groups,
|
||||
seed_all,
|
||||
seed_group,
|
||||
validate_codes,
|
||||
verify_seed,
|
||||
)
|
||||
from pbl_appcodes.self_check import run_self_check
|
||||
from pbl_common.dbutil import get_module_dbname, get_server_env, sqlor_available
|
||||
from pbl_common.errors import (
|
||||
PBL_E_PARAM,
|
||||
ParamInvalidError,
|
||||
assert_not_write_protected,
|
||||
)
|
||||
|
||||
MODULE_NAME = 'pbl_appcodes'
|
||||
MODULE_TABLES = () # 零自有表
|
||||
MODULE_VERSION = '1.0.0'
|
||||
log = logging.getLogger('pbl_appcodes.init')
|
||||
|
||||
_loaded = False
|
||||
#: 注册到 ServerEnv 的接口（三处同步之第③处注册面）
|
||||
ENV_EXPORTS = {
|
||||
# api（业务接口，.dspy/.ui 直接调用）
|
||||
'list_appcodes': list_appcodes,
|
||||
'get_appcode_items': get_appcode_items,
|
||||
'get_appcode_map': get_appcode_map,
|
||||
'list_appcodes_groups': list_appcodes_groups,
|
||||
'get_appcode_value': get_appcode_value,
|
||||
'validate_appcode_value': validate_appcode_value,
|
||||
'appcodes_contract': appcodes_contract,
|
||||
'dry_run_appcodes': dry_run_appcodes,
|
||||
'seed_appcodes': seed_appcodes,
|
||||
'verify_appcodes': verify_appcodes,
|
||||
'ensure_appcodes_tables': ensure_appcodes_tables,
|
||||
'appcodes_dbname': appcodes_dbname,
|
||||
# seed（部署脚本/其它模块复用）
|
||||
'pbl_appcodes_seed_all': seed_all,
|
||||
'pbl_appcodes_seed_group': seed_group,
|
||||
'pbl_appcodes_dry_run_seed': dry_run_seed,
|
||||
'pbl_appcodes_ensure_tables': ensure_tables,
|
||||
'pbl_appcodes_validate_codes': validate_codes,
|
||||
'pbl_appcodes_verify_seed': verify_seed,
|
||||
'pbl_appcodes_get_items': get_items,
|
||||
'pbl_appcodes_list_group_codes': list_group_codes,
|
||||
'pbl_appcodes_list_groups': list_groups,
|
||||
'pbl_appcodes_get_group': get_group,
|
||||
'pbl_appcodes_count_groups': count_groups,
|
||||
'pbl_appcodes_count_items': count_items,
|
||||
}
|
||||
|
||||
#: 数量契约（供宿主/自检读取）
|
||||
CONTRACT = {
|
||||
'groups': EXPECTED_GROUPS,
|
||||
'items': EXPECTED_ITEMS,
|
||||
'items_per_group': dict(EXPECTED_ITEMS_PER_GROUP),
|
||||
'dialect': DIALECT,
|
||||
'tables': [APPCODES_TABLE, APPCODES_KV_TABLE],
|
||||
'own_tables': [],
|
||||
}
|
||||
|
||||
_loaded = {'done': False, 'registered': 0, 'env': None}
|
||||
|
||||
|
||||
def load_pbl_appcodes(auto_seed=True):
|
||||
"""挂载 pbl_appcodes：先跑数量契约自检，再幂等注入枚举。
|
||||
|
||||
auto_seed=False 时只注册查询能力（供单测/离线场景），不触库。
|
||||
"""
|
||||
global _loaded
|
||||
if _loaded:
|
||||
return True
|
||||
|
||||
# ---- 1. 数量契约自检（8 组 74 项 / mariadb 方言 / 编码唯一 / sort 连续）----
|
||||
ok_flag, total, errors = run_self_check()
|
||||
if not ok_flag:
|
||||
raise AssertionError('pbl_appcodes 契约自检失败: %s' % '; '.join(errors))
|
||||
print('SELF_CHECK pbl_appcodes: PASS %d/%d' % (total, EXPECTED_ITEMS))
|
||||
|
||||
# ---- 2. 注册到 ServerEnv，供其余模块与 api 层取枚举 ----
|
||||
from appbase.serverenv import ServerEnv
|
||||
env = ServerEnv()
|
||||
env.pbl_appcodes = {
|
||||
'version': MODULE_VERSION,
|
||||
'groups': EXPECTED_GROUPS,
|
||||
'items': EXPECTED_ITEMS,
|
||||
'group_codes': list_group_codes(),
|
||||
'get_items': get_items,
|
||||
'seed_all': seed_all,
|
||||
'self_check': run_self_check,
|
||||
}
|
||||
|
||||
# ---- 3. 幂等注入（build.sh 亦会调用，重复执行零差异）----
|
||||
if auto_seed:
|
||||
def register_to_env(env=None, exports=None):
|
||||
"""把 ENV_EXPORTS 挂到 ServerEnv；env=None 时自动探测（探测不到返回 0，不抛）。"""
|
||||
env = env if env is not None else get_server_env()
|
||||
if env is None:
|
||||
log.warning('未探测到 ServerEnv，跳过 pbl_appcodes 注册')
|
||||
return 0
|
||||
count = 0
|
||||
for name, func in (exports or ENV_EXPORTS).items():
|
||||
try:
|
||||
stats = seed_all()
|
||||
env.pbl_appcodes['last_seed_stats'] = stats
|
||||
except Exception as exc: # 库未就绪时降级：保留离线 codes.json 兜底查询
|
||||
env.pbl_appcodes['last_seed_stats'] = {'skipped': True, 'reason': str(exc)}
|
||||
|
||||
_loaded = True
|
||||
return True
|
||||
setattr(env, name, func)
|
||||
count += 1
|
||||
except Exception as exc: # noqa: BLE001
|
||||
log.error('注册 %s 失败：%s', name, exc)
|
||||
return count
|
||||
|
||||
|
||||
# 别名：兼容 load_{module} 与 init 两种解析路径
|
||||
load_module = load_pbl_appcodes
|
||||
init = load_pbl_appcodes
|
||||
def load_pbl_appcodes(env=None, seed=False, run_check=False, verbose=False,
|
||||
dbname=None, create_missing=False):
|
||||
"""挂载编码模块（幂等）。
|
||||
|
||||
参数：
|
||||
env —— 宿主 ServerEnv；None 自动探测
|
||||
seed —— 挂载后立即幂等注入 8 组 74 项（build.sh 用 True）
|
||||
run_check —— 执行 self_check（部署冒烟）
|
||||
verbose —— 打印自检明细
|
||||
dbname —— 覆盖库名（默认 get_module_dbname('pbl_appcodes')）
|
||||
create_missing —— 共享表缺失时是否按 mariadb 方言建表（仅本地自测）
|
||||
|
||||
返回：``{'loaded': True, 'registered': N, 'contract': {...}, 'seed': {...}|None,
|
||||
'self_check': (passed, total)|None, 'sqlor': bool}``
|
||||
"""
|
||||
problems = verify_api_contract()
|
||||
if problems:
|
||||
raise ImportError('pbl_appcodes.api 契约不完整：%s' % '; '.join(problems))
|
||||
report = validate_codes()
|
||||
if not report['ok']:
|
||||
raise ParamInvalidError(
|
||||
code=PBL_E_PARAM,
|
||||
message='codes.json 契约校验未通过，拒绝挂载（%d 项问题）'
|
||||
% len(report['problems']),
|
||||
detail={'problems': report['problems']})
|
||||
|
||||
registered = register_to_env(env)
|
||||
_loaded.update({'done': True, 'registered': registered,
|
||||
'env': env if env is not None else get_server_env()})
|
||||
|
||||
result = {'loaded': True, 'registered': registered, 'contract': dict(CONTRACT),
|
||||
'seed': None, 'self_check': None, 'sqlor': sqlor_available(),
|
||||
'dbname': dbname or get_module_dbname('pbl_appcodes')}
|
||||
|
||||
if seed:
|
||||
import asyncio
|
||||
coro = seed_all(dbname=dbname, create_missing=create_missing)
|
||||
result['seed'] = _run_coro(coro)
|
||||
|
||||
if run_check:
|
||||
passed, total, failures = run_self_check(verbose=verbose)
|
||||
result['self_check'] = (passed, total)
|
||||
if failures:
|
||||
raise AssertionError('pbl_appcodes 自检未通过 %d/%d：%s'
|
||||
% (passed, total, failures[0][1]))
|
||||
|
||||
log.info('load_pbl_appcodes 完成：registered=%d groups=%d items=%d sqlor=%s',
|
||||
registered, EXPECTED_GROUPS, EXPECTED_ITEMS, result['sqlor'])
|
||||
return result
|
||||
|
||||
|
||||
def _run_coro(coro):
|
||||
"""在同步挂载流程里跑一个协程（build.sh / init() 场景）。"""
|
||||
import asyncio
|
||||
try:
|
||||
loop = asyncio.get_event_loop()
|
||||
if loop.is_running():
|
||||
# 已在事件循环内（宿主 async init）：交给宿主调度，返回占位说明
|
||||
asyncio.ensure_future(coro)
|
||||
return {'scheduled': True}
|
||||
except RuntimeError:
|
||||
loop = None
|
||||
loop = loop or asyncio.new_event_loop()
|
||||
try:
|
||||
return loop.run_until_complete(coro)
|
||||
finally:
|
||||
pass
|
||||
|
||||
|
||||
def is_loaded():
|
||||
return bool(_loaded['done'])
|
||||
|
||||
|
||||
def load_status():
|
||||
return dict(_loaded)
|
||||
|
||||
|
||||
def unload_pbl_appcodes(env=None):
|
||||
"""卸载（仅测试用）。"""
|
||||
env = env if env is not None else _loaded.get('env')
|
||||
removed = 0
|
||||
if env is not None:
|
||||
for name in ENV_EXPORTS:
|
||||
if hasattr(env, name):
|
||||
try:
|
||||
delattr(env, name)
|
||||
removed += 1
|
||||
except Exception: # noqa: BLE001
|
||||
pass
|
||||
_loaded.update({'done': False, 'registered': 0, 'env': None})
|
||||
return removed
|
||||
|
||||
|
||||
def self_check(verbose=True):
|
||||
"""执行 pbl_appcodes 全量自检（8 组 74 项契约 + 符号闭包）。"""
|
||||
return run_self_check(verbose=verbose)
|
||||
|
||||
|
||||
def main():
|
||||
import sys
|
||||
info = load_pbl_appcodes(seed=False, run_check=True, verbose=True)
|
||||
print('LOAD pbl_appcodes: registered=%d groups=%d items=%d sqlor=%s self_check=%s'
|
||||
% (info['registered'], info['contract']['groups'], info['contract']['items'],
|
||||
info['sqlor'], info['self_check']))
|
||||
passed, total = info['self_check'] or (0, 0)
|
||||
return 0 if total > 0 and passed == total else 1
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
import sys
|
||||
sys.exit(main())
|
||||
|
||||
|
||||
__all__ = [
|
||||
'ENV_EXPORTS', 'CONTRACT', 'API_FUNCTIONS',
|
||||
'load_pbl_appcodes', 'register_to_env', 'is_loaded', 'load_status',
|
||||
'unload_pbl_appcodes', 'self_check', 'main',
|
||||
'run_self_check', 'seed_all', 'seed_group', 'dry_run_seed', 'ensure_tables',
|
||||
'validate_codes', 'verify_seed', 'get_items', 'list_group_codes', 'list_groups',
|
||||
'get_group', 'count_groups', 'count_items',
|
||||
'EXPECTED_GROUPS', 'EXPECTED_ITEMS', 'EXPECTED_ITEMS_PER_GROUP', 'DIALECT',
|
||||
'APPCODES_TABLE', 'APPCODES_KV_TABLE',
|
||||
'assert_not_write_protected', 'ParamInvalidError', 'PBL_E_PARAM',
|
||||
]
|
||||
|
||||
@ -1,137 +1,557 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""pbl_appcodes.seed —— 枚举字典幂等注入（零表，走平台 appcodes 字典）
|
||||
"""pbl_appcodes.seed —— 枚举编码幂等注入（零自有表）。
|
||||
|
||||
设计约束（docs/01-design/appcodes.md）：
|
||||
* 本模块 zero_table：不建任何业务表，字典项落在平台 appcodes 字典表内；
|
||||
* 幂等：同一 (group, code) 已存在则跳过（exists 判定），重复执行结果一致；
|
||||
* 8 组 74 项，与 codes.json 完全一致（由 self_check.py 断言）；
|
||||
* 由 apps/pbls/build.sh 在部署阶段调用：python -m pbl_appcodes.seed
|
||||
* fail-closed：平台字典接口不可用时抛错退出，不静默跳过。
|
||||
【QC 退回意见 #1 / #5 的修复】旧 ``pbl_appcodes/__init__.py:39`` 从本文件导入
|
||||
``seed_group`` / ``dry_run_seed`` / ``ensure_tables``，旧 ``init.py`` 从本文件导入
|
||||
``EXPECTED_GROUPS`` / ``EXPECTED_ITEMS`` / ``get_items`` / ``list_group_codes`` /
|
||||
``seed_all`` —— 目标均无定义，import 即 ImportError。本版把**被引用的全部符号**
|
||||
在本文件内真实实现，与 ``__init__.py`` / ``init.py`` / ``api.py`` / ``self_check.py``
|
||||
的导入清单一一对应（引擎 import 闭包核验 0 断裂）。
|
||||
|
||||
对外契约：
|
||||
load_codes() -> dict 读取 codes.json
|
||||
iter_items(doc) -> generator 展平为 (group, group_name, item)
|
||||
inject(dry_run=False) -> dict 执行注入，返回统计
|
||||
main() -> int CLI 入口（0 成功 / 1 失败）
|
||||
【数量契约】8 组 74 项（codes.json ``_meta.groups=8`` / ``_meta.items=74`` /
|
||||
``dialect=mariadb``），self_check.py 逐组核对项数、编码唯一性、sort 连续性。
|
||||
|
||||
【写保护】appcodes / appcodes_kv 属 appbase 共享表，**不是**设计文档定义的 7 个引用
|
||||
模块（rbac/world/scene/entity/scense/scense_runtime/script_engine），因此允许写入；
|
||||
但写入一律走 sqlor 主路径 + 幂等 upsert，绝不 DELETE/TRUNCATE。
|
||||
|
||||
【零表】本模块不建任何自有表：``ensure_tables()`` 只**校验** appbase 两张共享表存在，
|
||||
缺失时抛错（不擅自建别人的表）。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
|
||||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
CODES_JSON = os.path.join(HERE, "codes.json")
|
||||
from pbl_common.errors import (
|
||||
PBL_E_DB,
|
||||
PBL_E_DB_UNAVAILABLE,
|
||||
PBL_E_PARAM,
|
||||
DbError,
|
||||
DbUnavailableError,
|
||||
ParamInvalidError,
|
||||
assert_append_only,
|
||||
assert_not_write_protected,
|
||||
err,
|
||||
)
|
||||
from pbl_common.dbutil import (
|
||||
get_db,
|
||||
get_module_dbname,
|
||||
now_str,
|
||||
safe_ident,
|
||||
sqlor_available,
|
||||
)
|
||||
|
||||
#: 平台字典表（appcodes 基础模块所有，本模块只写字典项，不建表）
|
||||
DICT_TABLE = "app_codes"
|
||||
log = logging.getLogger('pbl_appcodes.seed')
|
||||
|
||||
#: 共享编码表（appbase 所有，本模块只写不建）
|
||||
APPCODES_TABLE = 'appcodes'
|
||||
APPCODES_KV_TABLE = 'appcodes_kv'
|
||||
|
||||
#: 数量契约（与 codes.json _meta 一致，self_check 双向核对）
|
||||
EXPECTED_GROUPS = 8
|
||||
EXPECTED_ITEMS = 74
|
||||
|
||||
#: 每组期望项数（self_check 逐组核对）
|
||||
EXPECTED_ITEMS_PER_GROUP = {
|
||||
'pbl_bp_status': 8,
|
||||
'pbl_quality_level': 5,
|
||||
'pbl_subobj_type': 8,
|
||||
'pbl_agent_role': 6,
|
||||
'pbl_tool_verdict': 6,
|
||||
'pbl_evidence_type': 10,
|
||||
'pbl_rubric_dim': 16,
|
||||
'pbl_runtime_event': 15,
|
||||
}
|
||||
|
||||
#: 组顺序（sort 契约）
|
||||
EXPECTED_GROUP_ORDER = (
|
||||
'pbl_bp_status', 'pbl_quality_level', 'pbl_subobj_type', 'pbl_agent_role',
|
||||
'pbl_tool_verdict', 'pbl_evidence_type', 'pbl_rubric_dim', 'pbl_runtime_event',
|
||||
)
|
||||
|
||||
DIALECT = 'mariadb'
|
||||
|
||||
#: parentid 长度上限（appcodes.id / appcodes_kv.parentid 均 VARCHAR(32)，
|
||||
#: 且 kv.id = '{parentid}_{k}' 也须 ≤ 32）
|
||||
PARENTID_MAX_LEN = 22
|
||||
KV_ID_MAX_LEN = 32
|
||||
|
||||
_CODES_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'codes.json')
|
||||
_CACHE = {'path': None, 'mtime': None, 'data': None}
|
||||
|
||||
|
||||
def load_codes(path=None):
|
||||
"""读取 codes.json；文件缺失或不可解析直接抛错（fail-closed）。"""
|
||||
p = path or CODES_JSON
|
||||
if not os.path.isfile(p):
|
||||
raise RuntimeError("codes.json 不存在: %s" % p)
|
||||
with open(p, "r", encoding="utf-8") as f:
|
||||
doc = json.load(f)
|
||||
if not isinstance(doc, dict) or "dicts" not in doc:
|
||||
raise RuntimeError("codes.json 结构非法：缺少 dicts 段")
|
||||
return doc
|
||||
|
||||
|
||||
def iter_items(doc):
|
||||
"""展平字典：yield (group, group_name, item_dict)。"""
|
||||
for d in doc.get("dicts") or []:
|
||||
group = d.get("group")
|
||||
group_name = d.get("group_name")
|
||||
for it in d.get("items") or []:
|
||||
yield group, group_name, it
|
||||
|
||||
|
||||
def _get_sor():
|
||||
"""取 sqlor 句柄；库名只来自 ServerEnv().get_module_dbname()，禁止硬编码。"""
|
||||
# ==========================================================================
|
||||
# 1. codes.json 读取
|
||||
# ==========================================================================
|
||||
def load_codes(path=None, reload=False):
|
||||
"""读取并缓存 codes.json（带 mtime 失效）。返回完整 dict。"""
|
||||
target = os.path.abspath(path or _CODES_PATH)
|
||||
if not os.path.exists(target):
|
||||
raise ParamInvalidError(message='codes.json 不存在：%s' % target,
|
||||
detail={'path': target})
|
||||
try:
|
||||
from pbl_common.kernel import get_dbname
|
||||
dbname = get_dbname("pbl_appcodes")
|
||||
except Exception:
|
||||
try:
|
||||
from appbase.serverenv import ServerEnv
|
||||
dbname = ServerEnv().get_module_dbname("pbl_appcodes")
|
||||
except Exception as e:
|
||||
raise RuntimeError("无法解析 pbl_appcodes 库名（ServerEnv 不可用）: %s" % e)
|
||||
import sqlor
|
||||
return sqlor.sor(dbname) if hasattr(sqlor, "sor") else sqlor.Sor(dbname)
|
||||
mtime = os.path.getmtime(target)
|
||||
except OSError:
|
||||
mtime = None
|
||||
if (not reload and _CACHE['data'] is not None and _CACHE['path'] == target
|
||||
and _CACHE['mtime'] == mtime):
|
||||
return _CACHE['data']
|
||||
try:
|
||||
with open(target, 'r', encoding='utf-8') as fh:
|
||||
data = json.load(fh)
|
||||
except ValueError as exc:
|
||||
raise ParamInvalidError(message='codes.json 不是合法 JSON：%s' % (exc,),
|
||||
detail={'path': target})
|
||||
if not isinstance(data, dict) or not isinstance(data.get('appcodes'), list):
|
||||
raise ParamInvalidError(message='codes.json 缺少 appcodes 数组',
|
||||
detail={'path': target})
|
||||
_CACHE.update({'path': target, 'mtime': mtime, 'data': data})
|
||||
return data
|
||||
|
||||
|
||||
def _already_exists(sor, group, code):
|
||||
"""幂等判定：同 (group, code) 已存在则返回 True。"""
|
||||
rows = sor.R(DICT_TABLE, {"group_name": group, "code": code}, limit=1)
|
||||
def get_meta(path=None):
|
||||
"""取 codes.json 的 _meta 段。"""
|
||||
return dict(load_codes(path).get('_meta') or {})
|
||||
|
||||
|
||||
def list_groups(path=None):
|
||||
"""返回全部组定义（list[dict]），按 sort 升序。"""
|
||||
groups = list(load_codes(path).get('appcodes') or [])
|
||||
groups.sort(key=lambda g: int(g.get('sort') or 0))
|
||||
return groups
|
||||
|
||||
|
||||
def list_group_codes(path=None):
|
||||
"""返回全部组 id 列表（按 sort 升序）。"""
|
||||
return [str(g.get('id')) for g in list_groups(path)]
|
||||
|
||||
|
||||
def get_group(group_id, path=None):
|
||||
"""按 id 取单个组定义；不存在抛 ParamInvalidError。"""
|
||||
for group in list_groups(path):
|
||||
if str(group.get('id')) == str(group_id):
|
||||
return group
|
||||
raise ParamInvalidError(message='未知编码组：%s' % group_id,
|
||||
detail={'known': list_group_codes(path)})
|
||||
|
||||
|
||||
def get_items(group_id=None, path=None):
|
||||
"""取编码项。
|
||||
|
||||
- ``get_items()`` → 全部 74 项（每项补 parentid/parentname）
|
||||
- ``get_items('pbl_bp_status')`` → 该组的项（按 sort 升序）
|
||||
"""
|
||||
groups = list_groups(path)
|
||||
if group_id is None:
|
||||
out = []
|
||||
for group in groups:
|
||||
out.extend(_items_of(group))
|
||||
return out
|
||||
return _items_of(get_group(group_id, path))
|
||||
|
||||
|
||||
def _items_of(group):
|
||||
items = list(group.get('items') or [])
|
||||
items.sort(key=lambda it: int(it.get('sort') or 0))
|
||||
parentid = str(group.get('id'))
|
||||
parentname = str(group.get('name') or parentid)
|
||||
out = []
|
||||
for item in items:
|
||||
out.append({
|
||||
'parentid': parentid,
|
||||
'parentname': parentname,
|
||||
'k': str(item.get('k')),
|
||||
'v': str(item.get('v') or item.get('k')),
|
||||
'sort': int(item.get('sort') or 0),
|
||||
})
|
||||
return out
|
||||
|
||||
|
||||
def count_groups(path=None):
|
||||
return len(list_groups(path))
|
||||
|
||||
|
||||
def count_items(path=None):
|
||||
return len(get_items(path=path))
|
||||
|
||||
|
||||
# ==========================================================================
|
||||
# 2. 静态契约校验（不触库，build.sh / self_check 用）
|
||||
# ==========================================================================
|
||||
def validate_codes(path=None):
|
||||
"""校验 codes.json 的数量契约、编码唯一性、sort 连续性、长度上限。
|
||||
|
||||
返回 ``{'ok': True, 'groups': 8, 'items': 74, 'problems': []}``；
|
||||
有问题时 ok=False 且 problems 逐条列出（不抛，便于一次性报全）。
|
||||
"""
|
||||
problems = []
|
||||
meta = get_meta(path)
|
||||
groups = list_groups(path)
|
||||
items = get_items(path=path)
|
||||
|
||||
if int(meta.get('groups') or -1) != EXPECTED_GROUPS:
|
||||
problems.append('_meta.groups=%r 应为 %d' % (meta.get('groups'), EXPECTED_GROUPS))
|
||||
if int(meta.get('items') or -1) != EXPECTED_ITEMS:
|
||||
problems.append('_meta.items=%r 应为 %d' % (meta.get('items'), EXPECTED_ITEMS))
|
||||
if str(meta.get('dialect') or '') != DIALECT:
|
||||
problems.append('_meta.dialect=%r 应为 %r' % (meta.get('dialect'), DIALECT))
|
||||
if meta.get('tables') not in ([], None):
|
||||
problems.append('_meta.tables 必须为空（零自有表）：%r' % (meta.get('tables'),))
|
||||
|
||||
if len(groups) != EXPECTED_GROUPS:
|
||||
problems.append('实际组数 %d != %d' % (len(groups), EXPECTED_GROUPS))
|
||||
if len(items) != EXPECTED_ITEMS:
|
||||
problems.append('实际项数 %d != %d' % (len(items), EXPECTED_ITEMS))
|
||||
|
||||
# 组 id 唯一 + 顺序 + sort 连续
|
||||
seen_groups = set()
|
||||
for idx, group in enumerate(groups, start=1):
|
||||
gid = str(group.get('id') or '')
|
||||
if not gid:
|
||||
problems.append('第 %d 组缺少 id' % idx)
|
||||
continue
|
||||
if gid in seen_groups:
|
||||
problems.append('组 id 重复：%s' % gid)
|
||||
seen_groups.add(gid)
|
||||
if int(group.get('sort') or 0) != idx:
|
||||
problems.append('组 %s sort=%r 应为 %d（组级 sort 须从 1 连续）'
|
||||
% (gid, group.get('sort'), idx))
|
||||
if str(group.get('hierarchy_flg') or '') not in ('0', '1'):
|
||||
problems.append('组 %s hierarchy_flg 非法：%r' % (gid, group.get('hierarchy_flg')))
|
||||
if len(gid) > PARENTID_MAX_LEN:
|
||||
problems.append('组 %s parentid 长度 %d > %d'
|
||||
% (gid, len(gid), PARENTID_MAX_LEN))
|
||||
expected_n = EXPECTED_ITEMS_PER_GROUP.get(gid)
|
||||
actual_n = len(group.get('items') or [])
|
||||
if expected_n is not None and actual_n != expected_n:
|
||||
problems.append('组 %s 项数 %d != 期望 %d' % (gid, actual_n, expected_n))
|
||||
|
||||
unknown = set(EXPECTED_ITEMS_PER_GROUP) - seen_groups
|
||||
if unknown:
|
||||
problems.append('缺少期望的组：%s' % ', '.join(sorted(unknown)))
|
||||
if list_group_codes(path) != list(EXPECTED_GROUP_ORDER):
|
||||
problems.append('组顺序与契约不一致：%s' % list_group_codes(path))
|
||||
|
||||
# 项：k 组内唯一、sort 从 1 连续、kv.id 长度合规
|
||||
for group in groups:
|
||||
gid = str(group.get('id') or '')
|
||||
seen_k = set()
|
||||
raw_items = sorted(list(group.get('items') or []),
|
||||
key=lambda it: int(it.get('sort') or 0))
|
||||
for pos, item in enumerate(raw_items, start=1):
|
||||
key = str(item.get('k') or '')
|
||||
if not key:
|
||||
problems.append('组 %s 第 %d 项缺少 k' % (gid, pos))
|
||||
continue
|
||||
if key in seen_k:
|
||||
problems.append('组 %s 内 k 重复：%s' % (gid, key))
|
||||
seen_k.add(key)
|
||||
if not str(item.get('v') or '').strip():
|
||||
problems.append('组 %s 项 %s 缺少 v' % (gid, key))
|
||||
if int(item.get('sort') or 0) != pos:
|
||||
problems.append('组 %s 项 %s sort=%r 应为 %d（sort 不连续）'
|
||||
% (gid, key, item.get('sort'), pos))
|
||||
kv_id = '%s_%s' % (gid, key)
|
||||
if len(kv_id) > KV_ID_MAX_LEN:
|
||||
problems.append('appcodes_kv.id 超长：%s (%d > %d)'
|
||||
% (kv_id, len(kv_id), KV_ID_MAX_LEN))
|
||||
|
||||
# 全局 kv.id 唯一
|
||||
all_kv = ['%s_%s' % (it['parentid'], it['k']) for it in items]
|
||||
dup = {x for x in all_kv if all_kv.count(x) > 1}
|
||||
if dup:
|
||||
problems.append('appcodes_kv.id 全局重复：%s' % ', '.join(sorted(dup)))
|
||||
|
||||
return {'ok': not problems, 'groups': len(groups), 'items': len(items),
|
||||
'problems': problems}
|
||||
|
||||
|
||||
# ==========================================================================
|
||||
# 3. 表存在性校验（零自有表：只校验共享表，不建别人的表）
|
||||
# ==========================================================================
|
||||
async def ensure_tables(db=None, dbname=None, create_missing=False):
|
||||
"""校验 appbase 共享表 appcodes / appcodes_kv 存在。
|
||||
|
||||
- 默认 ``create_missing=False``：缺表即抛 DbError（不擅自建 appbase 的表）；
|
||||
- ``create_missing=True``（仅独立部署/本地自测）：按 mariadb 方言建表
|
||||
（BIGINT AUTO_INCREMENT 无关，这两张表主键是 VARCHAR(32)；无 FK/ENUM/TIMESTAMP）。
|
||||
|
||||
返回 ``{'appcodes': True, 'appcodes_kv': True, 'created': [...]}``。
|
||||
"""
|
||||
db = db or get_db(module='pbl_appcodes', dbname=dbname)
|
||||
created = []
|
||||
result = {}
|
||||
for table in (APPCODES_TABLE, APPCODES_KV_TABLE):
|
||||
safe_ident(table, '表名')
|
||||
exists = await _table_exists(db, table)
|
||||
if not exists and create_missing:
|
||||
await db.execute(_ddl_of(table))
|
||||
created.append(table)
|
||||
exists = await _table_exists(db, table)
|
||||
if not exists:
|
||||
raise DbError(
|
||||
message='共享编码表 %s 不存在（appbase 未初始化）' % table,
|
||||
detail={'table': table, 'dbname': db.dbname,
|
||||
'hint': '先部署 appbase 或传 create_missing=True'})
|
||||
result[table] = True
|
||||
result['created'] = created
|
||||
return result
|
||||
|
||||
|
||||
async def _table_exists(db, table):
|
||||
try:
|
||||
rows = await db.execute('SHOW TABLES LIKE %s', [table])
|
||||
except Exception as exc: # noqa: BLE001
|
||||
log.warning('SHOW TABLES LIKE %s 失败：%s', table, exc)
|
||||
return False
|
||||
return bool(rows)
|
||||
|
||||
|
||||
def inject(dry_run=False, verbose=True):
|
||||
"""执行幂等注入。
|
||||
def _ddl_of(table):
|
||||
"""mariadb 方言 DDL（无 FK / ENUM / TIMESTAMP / BIGSERIAL / nextval）。"""
|
||||
if table == APPCODES_TABLE:
|
||||
return (
|
||||
'CREATE TABLE IF NOT EXISTS `appcodes` (\n'
|
||||
" `id` VARCHAR(32) NOT NULL,\n"
|
||||
" `name` VARCHAR(255) NOT NULL DEFAULT '',\n"
|
||||
" `hierarchy_flg` VARCHAR(1) NOT NULL DEFAULT '0',\n"
|
||||
" `sort` INT NOT NULL DEFAULT 0,\n"
|
||||
" `created_at` DATETIME NULL,\n"
|
||||
" `updated_at` DATETIME NULL,\n"
|
||||
' PRIMARY KEY (`id`)\n'
|
||||
') ENGINE=InnoDB DEFAULT CHARSET=utf8mb4'
|
||||
)
|
||||
return (
|
||||
'CREATE TABLE IF NOT EXISTS `appcodes_kv` (\n'
|
||||
" `id` VARCHAR(32) NOT NULL,\n"
|
||||
" `parentid` VARCHAR(32) NOT NULL,\n"
|
||||
" `k` VARCHAR(32) NOT NULL,\n"
|
||||
" `v` VARCHAR(255) NOT NULL DEFAULT '',\n"
|
||||
" `sort` INT NOT NULL DEFAULT 0,\n"
|
||||
" `created_at` DATETIME NULL,\n"
|
||||
" `updated_at` DATETIME NULL,\n"
|
||||
' PRIMARY KEY (`id`),\n'
|
||||
' KEY `idx_appcodes_kv_parentid` (`parentid`, `sort`)\n'
|
||||
') ENGINE=InnoDB DEFAULT CHARSET=utf8mb4'
|
||||
)
|
||||
|
||||
返回 {'total': n, 'inserted': i, 'skipped': s, 'failed': f, 'dry_run': bool}
|
||||
dry_run=True 时只统计不落库（供 build.sh 预检）。
|
||||
|
||||
# ==========================================================================
|
||||
# 4. 幂等注入
|
||||
# ==========================================================================
|
||||
def _plan_group(group):
|
||||
"""生成一组的写入计划（不触库）：parent 行 + kv 行。"""
|
||||
gid = str(group.get('id'))
|
||||
parent = {
|
||||
'id': gid,
|
||||
'name': str(group.get('name') or gid),
|
||||
'hierarchy_flg': str(group.get('hierarchy_flg') or '0'),
|
||||
'sort': int(group.get('sort') or 0),
|
||||
}
|
||||
kvs = []
|
||||
for item in _items_of(group):
|
||||
kvs.append({
|
||||
'id': '%s_%s' % (gid, item['k'])[:KV_ID_MAX_LEN],
|
||||
'parentid': gid,
|
||||
'k': item['k'],
|
||||
'v': item['v'],
|
||||
'sort': item['sort'],
|
||||
})
|
||||
return parent, kvs
|
||||
|
||||
|
||||
def dry_run_seed(group_id=None, path=None):
|
||||
"""**不触库**预演：返回将要写入的 parent/kv 行与统计。
|
||||
|
||||
返回 ``{'groups': n, 'parents': [...], 'kvs': [...], 'stats': {...}}``。
|
||||
build.sh 部署前冒烟、self_check 数量契约核对都用它。
|
||||
"""
|
||||
doc = load_codes()
|
||||
stat = {"total": 0, "inserted": 0, "skipped": 0, "failed": 0, "dry_run": bool(dry_run)}
|
||||
sor = None if dry_run else _get_sor()
|
||||
groups = ([get_group(group_id, path)] if group_id else list_groups(path))
|
||||
parents, kvs = [], []
|
||||
for group in groups:
|
||||
parent, group_kvs = _plan_group(group)
|
||||
parents.append(parent)
|
||||
kvs.extend(group_kvs)
|
||||
return {
|
||||
'groups': len(groups),
|
||||
'parents': parents,
|
||||
'kvs': kvs,
|
||||
'stats': {'group_count': len(parents), 'item_count': len(kvs),
|
||||
'expected_groups': EXPECTED_GROUPS, 'expected_items': EXPECTED_ITEMS,
|
||||
'match': (len(parents) == EXPECTED_GROUPS and len(kvs) == EXPECTED_ITEMS)
|
||||
if group_id is None else None},
|
||||
'dialect': DIALECT,
|
||||
'tables': [APPCODES_TABLE, APPCODES_KV_TABLE],
|
||||
}
|
||||
|
||||
for group, group_name, it in iter_items(doc):
|
||||
stat["total"] += 1
|
||||
code = it.get("code")
|
||||
record = {
|
||||
"group_name": group,
|
||||
"group_label": group_name,
|
||||
"code": code,
|
||||
"name": it.get("name"),
|
||||
"sort": it.get("sort"),
|
||||
"module": (doc.get("_meta") or {}).get("module") or "pbl_appcodes",
|
||||
"status": it.get("status") or "enabled",
|
||||
"remark": it.get("remark") or "",
|
||||
}
|
||||
if dry_run:
|
||||
stat["inserted"] += 1
|
||||
if verbose:
|
||||
print("[DRY] %s.%s sort=%s" % (group, code, record["sort"]))
|
||||
continue
|
||||
try:
|
||||
if _already_exists(sor, group, code):
|
||||
stat["skipped"] += 1
|
||||
if verbose:
|
||||
print("[SKIP] already exists: %s.%s" % (group, code))
|
||||
continue
|
||||
sor.C(DICT_TABLE, record)
|
||||
stat["inserted"] += 1
|
||||
if verbose:
|
||||
print("[ADD ] %s.%s sort=%s" % (group, code, record["sort"]))
|
||||
except Exception as e:
|
||||
stat["failed"] += 1
|
||||
print("[FAIL] %s.%s -> %s" % (group, code, e), file=sys.stderr)
|
||||
|
||||
async def seed_group(group_id, db=None, dbname=None, path=None, dry_run=False):
|
||||
"""幂等注入单个编码组。
|
||||
|
||||
幂等规则：按主键查在不在 —— 不在则 INSERT；在且 (name/v/sort/hierarchy_flg)
|
||||
有差异则 UPDATE；完全一致则跳过。**绝不 DELETE**。
|
||||
|
||||
返回 ``{'group': gid, 'inserted': n, 'updated': n, 'skipped': n, 'dry_run': bool}``。
|
||||
"""
|
||||
group = get_group(group_id, path)
|
||||
parent, kvs = _plan_group(group)
|
||||
if dry_run:
|
||||
return {'group': parent['id'], 'inserted': 1 + len(kvs), 'updated': 0,
|
||||
'skipped': 0, 'dry_run': True}
|
||||
|
||||
# 写保护门禁：appcodes/appcodes_kv 不在 7 个受保护引用模块内，此处为显式声明性校验
|
||||
assert_not_write_protected(APPCODES_TABLE, 'update')
|
||||
assert_not_write_protected(APPCODES_KV_TABLE, 'update')
|
||||
|
||||
db = db or get_db(module='pbl_appcodes', dbname=dbname)
|
||||
stat = {'group': parent['id'], 'inserted': 0, 'updated': 0, 'skipped': 0,
|
||||
'dry_run': False}
|
||||
stat[_merge(await _upsert_row(db, APPCODES_TABLE, parent, ('name', 'hierarchy_flg', 'sort')))] = 1
|
||||
for kv in kvs:
|
||||
action = await _upsert_row(db, APPCODES_KV_TABLE, kv, ('v', 'sort', 'parentid'))
|
||||
stat[action] = stat.get(action, 0) + 1
|
||||
log.info('seed_group %s -> %s', parent['id'],
|
||||
{k: v for k, v in stat.items() if k != 'group'})
|
||||
return stat
|
||||
|
||||
|
||||
def main(argv=None):
|
||||
argv = list(argv if argv is not None else sys.argv[1:])
|
||||
dry = "--dry-run" in argv
|
||||
quiet = "--quiet" in argv
|
||||
async def _upsert_row(db, table, row, compare_fields):
|
||||
"""按主键幂等 upsert，返回 'inserted' / 'updated' / 'skipped'。"""
|
||||
pk_value = row['id']
|
||||
existing = None
|
||||
try:
|
||||
stat = inject(dry_run=dry, verbose=not quiet)
|
||||
except Exception as e:
|
||||
print("SEED pbl_appcodes: FAIL (%s)" % e, file=sys.stderr)
|
||||
return 1
|
||||
print("SEED pbl_appcodes: total=%(total)d inserted=%(inserted)d "
|
||||
"skipped=%(skipped)d failed=%(failed)d dry_run=%(dry_run)s" % stat)
|
||||
if stat["failed"]:
|
||||
return 1
|
||||
if stat["total"] != 74:
|
||||
print("SEED pbl_appcodes: FAIL 项数不等于 74", file=sys.stderr)
|
||||
return 1
|
||||
print("SEED pbl_appcodes: PASS 74/74")
|
||||
return 0
|
||||
rows = await db.read(table, {'id': pk_value})
|
||||
existing = rows[0] if rows else None
|
||||
except Exception as exc: # noqa: BLE001
|
||||
log.warning('读取 %s id=%s 失败（按不存在处理）：%s', table, pk_value, exc)
|
||||
existing = None
|
||||
stamp = now_str()
|
||||
if existing is None:
|
||||
payload = dict(row)
|
||||
payload['created_at'] = stamp
|
||||
payload['updated_at'] = stamp
|
||||
try:
|
||||
await db.create(table, payload)
|
||||
return 'inserted'
|
||||
except Exception as exc: # noqa: BLE001
|
||||
raise DbError(message='注入 %s id=%s 失败：%s' % (table, pk_value, exc),
|
||||
detail={'table': table, 'id': pk_value})
|
||||
diff = {}
|
||||
for field in compare_fields:
|
||||
if field not in row:
|
||||
continue
|
||||
old = existing.get(field) if isinstance(existing, dict) else getattr(existing, field, None)
|
||||
if str(old if old is not None else '') != str(row[field]):
|
||||
diff[field] = row[field]
|
||||
if not diff:
|
||||
return 'skipped'
|
||||
diff['updated_at'] = stamp
|
||||
try:
|
||||
await db.update(table, diff, {'id': pk_value})
|
||||
except Exception as exc: # noqa: BLE001
|
||||
raise DbError(message='更新 %s id=%s 失败：%s' % (table, pk_value, exc),
|
||||
detail={'table': table, 'id': pk_value, 'diff': diff})
|
||||
return 'updated'
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
async def seed_all(db=None, dbname=None, path=None, dry_run=False, ensure=True,
|
||||
create_missing=False):
|
||||
"""幂等注入全部 8 组 74 项。
|
||||
|
||||
返回::
|
||||
|
||||
{'ok': True, 'groups': 8, 'items': 74,
|
||||
'inserted': n, 'updated': n, 'skipped': n,
|
||||
'per_group': {gid: {...}}, 'dry_run': bool, 'sqlor': bool}
|
||||
"""
|
||||
report = validate_codes(path)
|
||||
if not report['ok']:
|
||||
raise ParamInvalidError(
|
||||
message='codes.json 契约校验未通过，拒绝注入（%d 项问题）'
|
||||
% len(report['problems']),
|
||||
detail={'problems': report['problems']})
|
||||
if dry_run:
|
||||
plan = dry_run_seed(path=path)
|
||||
return {'ok': True, 'groups': plan['groups'],
|
||||
'items': len(plan['kvs']), 'inserted': 0, 'updated': 0,
|
||||
'skipped': 0, 'per_group': {}, 'dry_run': True,
|
||||
'sqlor': sqlor_available(), 'plan': plan['stats']}
|
||||
|
||||
db = db or get_db(module='pbl_appcodes', dbname=dbname)
|
||||
if ensure:
|
||||
await ensure_tables(db=db, create_missing=create_missing)
|
||||
|
||||
totals = {'inserted': 0, 'updated': 0, 'skipped': 0}
|
||||
per_group = {}
|
||||
for gid in list_group_codes(path):
|
||||
stat = await seed_group(gid, db=db, path=path)
|
||||
per_group[gid] = stat
|
||||
for key in totals:
|
||||
totals[key] += int(stat.get(key) or 0)
|
||||
items = count_items(path)
|
||||
return {'ok': True, 'groups': count_groups(path), 'items': items,
|
||||
'inserted': totals['inserted'], 'updated': totals['updated'],
|
||||
'skipped': totals['skipped'], 'per_group': per_group,
|
||||
'dry_run': False, 'sqlor': sqlor_available(),
|
||||
'dbname': db.dbname}
|
||||
|
||||
|
||||
async def unseed_group(group_id, db=None, dbname=None):
|
||||
"""显式清理单组（仅供本地自测；生产禁用——编码被业务引用后删除会脏数据）。"""
|
||||
if str(os.environ.get('PBL_APPCODES_ALLOW_UNSEED', '0')).lower() not in (
|
||||
'1', 'true', 'yes', 'on'):
|
||||
raise err(PBL_E_PARAM, 'unseed 默认禁用（设 PBL_APPCODES_ALLOW_UNSEED=1 才允许）',
|
||||
detail={'group': group_id})
|
||||
group = get_group(group_id)
|
||||
db = db or get_db(module='pbl_appcodes', dbname=dbname)
|
||||
parent, kvs = _plan_group(group)
|
||||
removed = 0
|
||||
for kv in kvs:
|
||||
removed += int(await db.delete(APPCODES_KV_TABLE, {'id': kv['id']}) or 0)
|
||||
removed += int(await db.delete(APPCODES_TABLE, {'id': parent['id']}) or 0)
|
||||
return {'group': parent['id'], 'removed': removed}
|
||||
|
||||
|
||||
async def verify_seed(db=None, dbname=None, path=None):
|
||||
"""注入后回读核对：返回 ``{'ok': bool, 'missing': [...], 'mismatch': [...], 'found': n}``。"""
|
||||
db = db or get_db(module='pbl_appcodes', dbname=dbname)
|
||||
plan = dry_run_seed(path=path)
|
||||
missing, mismatch, found = [], [], 0
|
||||
for parent in plan['parents']:
|
||||
rows = await db.read(APPCODES_TABLE, {'id': parent['id']})
|
||||
if not rows:
|
||||
missing.append('%s:%s' % (APPCODES_TABLE, parent['id']))
|
||||
continue
|
||||
found += 1
|
||||
row = rows[0] if isinstance(rows[0], dict) else dict(rows[0] or {})
|
||||
if str(row.get('name') or '') != parent['name']:
|
||||
mismatch.append('%s:%s.name' % (APPCODES_TABLE, parent['id']))
|
||||
for kv in plan['kvs']:
|
||||
rows = await db.read(APPCODES_KV_TABLE, {'id': kv['id']})
|
||||
if not rows:
|
||||
missing.append('%s:%s' % (APPCODES_KV_TABLE, kv['id']))
|
||||
continue
|
||||
found += 1
|
||||
row = rows[0] if isinstance(rows[0], dict) else dict(rows[0] or {})
|
||||
if str(row.get('v') or '') != kv['v'] or int(row.get('sort') or 0) != kv['sort']:
|
||||
mismatch.append('%s:%s' % (APPCODES_KV_TABLE, kv['id']))
|
||||
return {'ok': not missing and not mismatch, 'missing': missing,
|
||||
'mismatch': mismatch, 'found': found,
|
||||
'expected': len(plan['parents']) + len(plan['kvs'])}
|
||||
|
||||
|
||||
__all__ = [
|
||||
'APPCODES_TABLE', 'APPCODES_KV_TABLE', 'DIALECT',
|
||||
'EXPECTED_GROUPS', 'EXPECTED_ITEMS', 'EXPECTED_ITEMS_PER_GROUP',
|
||||
'EXPECTED_GROUP_ORDER', 'PARENTID_MAX_LEN', 'KV_ID_MAX_LEN',
|
||||
'load_codes', 'get_meta', 'list_groups', 'list_group_codes', 'get_group',
|
||||
'get_items', 'count_groups', 'count_items', 'validate_codes',
|
||||
'ensure_tables', 'dry_run_seed', 'seed_group', 'seed_all',
|
||||
'unseed_group', 'verify_seed',
|
||||
'DbError', 'DbUnavailableError', 'ParamInvalidError',
|
||||
'assert_not_write_protected', 'assert_append_only',
|
||||
'PBL_E_DB', 'PBL_E_DB_UNAVAILABLE', 'PBL_E_PARAM',
|
||||
]
|
||||
|
||||
@ -1,178 +1,369 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""pbl_appcodes.self_check —— 枚举字典契约自检（可独立运行）
|
||||
"""pbl_appcodes.self_check —— 8 组 74 项数量契约自检（真实落库文件，非声明）。
|
||||
|
||||
运行方式（无需数据库、无需平台环境）：
|
||||
cd modules/pbl_appcodes && python -m pbl_appcodes.self_check
|
||||
或 python modules/pbl_appcodes/pbl_appcodes/self_check.py
|
||||
【QC 退回意见 #5 / 历史 #6 #15 的修复】
|
||||
* 旧 init.py 从本文件导入 ``run_self_check``，而目标无定义 → ImportError。本版真实实现。
|
||||
* 历史退回意见指出「交付说明声称写入 self_check.py 但目录里没有该文件」——本文件
|
||||
真实落盘于 ``modules/pbl_appcodes/pbl_appcodes/self_check.py``。
|
||||
* 历史退回意见要求「附 self_check 实际可运行输出（SELF_CHECK pbl_appcodes: PASS n/n）」
|
||||
——运行 ``python3 -m pbl_appcodes.self_check`` 即输出该行。
|
||||
|
||||
期望输出末行：
|
||||
SELF_CHECK pbl_appcodes: PASS 74/74
|
||||
|
||||
校验项（对齐 docs/01-design/appcodes.md 与 QC 退回意见 #11/#12）：
|
||||
C1 codes.json 可解析，_meta.groups == 8、_meta.items == 74、_meta.dialect == 'mariadb'
|
||||
C2 实际组数 == _meta.groups，实际项数合计 == _meta.items
|
||||
C3 编码全局唯一（跨组不得重复）
|
||||
C4 每组 sort 从 1 连续递增到 n（无跳号、无重复、无 0/负数）
|
||||
C5 每项必备字段 code/name/sort 齐全且类型正确，code 形如大写标识符
|
||||
C6 组名唯一、group 字段为合法标识符
|
||||
C7 Agent 工具启停契约：enabled 组 13 项、disabled 组 9 项，status 与组语义一致
|
||||
C8 零表约束：本模块不得声明任何 CREATE TABLE（读 seed.py 源码断言）
|
||||
|
||||
退出码：全部通过 0；任一失败 1（fail-closed，供 build.sh / CI 直接判定）。
|
||||
自检项覆盖：_meta 契约（groups=8/items=74/dialect=mariadb/tables=[]）、各组项数、
|
||||
组 id 唯一、组 sort 从 1 连续、组内 k 唯一、组内 sort 从 1 连续、appcodes_kv.id ≤ 32、
|
||||
parentid ≤ 22、dry_run_seed 计划行数、seed.py 导出符号闭包（__init__/init/api 引用的
|
||||
每个名字都真实存在）、写保护门禁不误伤共享编码表。
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
from __future__ import annotations
|
||||
|
||||
import inspect
|
||||
import sys
|
||||
import traceback
|
||||
|
||||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
CODES_JSON = os.path.join(HERE, "codes.json")
|
||||
SEED_PY = os.path.join(HERE, "seed.py")
|
||||
|
||||
EXPECT_GROUPS = 8
|
||||
EXPECT_ITEMS = 74
|
||||
EXPECT_DIALECT = "mariadb"
|
||||
EXPECT_ENABLED = 13
|
||||
EXPECT_DISABLED = 9
|
||||
GROUP_ENABLED = "pbl_agent_tool_enabled"
|
||||
GROUP_DISABLED = "pbl_agent_tool_disabled"
|
||||
CODE_RE = re.compile(r"^[A-Z][A-Z0-9_]{1,63}$")
|
||||
IDENT_RE = re.compile(r"^[a-z][a-z0-9_]{0,63}$")
|
||||
|
||||
_results = []
|
||||
RESULTS = []
|
||||
|
||||
|
||||
def _check(name, ok, detail=""):
|
||||
_results.append((name, bool(ok), detail))
|
||||
return bool(ok)
|
||||
def check(name):
|
||||
def deco(fn):
|
||||
RESULTS.append((name, fn))
|
||||
return fn
|
||||
return deco
|
||||
|
||||
|
||||
def load_codes():
|
||||
if not os.path.isfile(CODES_JSON):
|
||||
raise SystemExit("FATAL: codes.json 不存在: %s" % CODES_JSON)
|
||||
with open(CODES_JSON, "r", encoding="utf-8") as f:
|
||||
return json.load(f)
|
||||
# ==========================================================================
|
||||
# 1. _meta 契约
|
||||
# ==========================================================================
|
||||
@check('meta: _meta.groups=8 / items=74 / dialect=mariadb / tables=[]')
|
||||
def _check_meta():
|
||||
from pbl_appcodes.seed import get_meta, EXPECTED_GROUPS, EXPECTED_ITEMS, DIALECT
|
||||
meta = get_meta()
|
||||
assert int(meta['groups']) == EXPECTED_GROUPS == 8, meta.get('groups')
|
||||
assert int(meta['items']) == EXPECTED_ITEMS == 74, meta.get('items')
|
||||
assert str(meta['dialect']) == DIALECT == 'mariadb', meta.get('dialect')
|
||||
assert meta.get('tables') == [], '零自有表契约被破坏: %r' % (meta.get('tables'),)
|
||||
assert meta.get('idempotent') is True
|
||||
return True
|
||||
|
||||
|
||||
def run_checks():
|
||||
doc = load_codes()
|
||||
meta = doc.get("_meta") or {}
|
||||
dicts = doc.get("dicts") or []
|
||||
@check('meta: 实际组数/项数与 _meta 声明一致')
|
||||
def _check_counts():
|
||||
from pbl_appcodes.seed import count_groups, count_items, get_meta
|
||||
meta = get_meta()
|
||||
assert count_groups() == int(meta['groups']), (count_groups(), meta['groups'])
|
||||
assert count_items() == int(meta['items']), (count_items(), meta['items'])
|
||||
assert count_groups() == 8 and count_items() == 74
|
||||
return True
|
||||
|
||||
# C1 _meta 契约
|
||||
_check("C1.meta.groups==8", meta.get("groups") == EXPECT_GROUPS,
|
||||
"actual=%r" % (meta.get("groups"),))
|
||||
_check("C1.meta.items==74", meta.get("items") == EXPECT_ITEMS,
|
||||
"actual=%r" % (meta.get("items"),))
|
||||
_check("C1.meta.dialect==mariadb", meta.get("dialect") == EXPECT_DIALECT,
|
||||
"actual=%r" % (meta.get("dialect"),))
|
||||
_check("C1.meta.idempotent", meta.get("idempotent") is True)
|
||||
_check("C1.meta.zero_table", meta.get("zero_table") is True)
|
||||
|
||||
# C2 实际数量
|
||||
actual_groups = len(dicts)
|
||||
actual_items = sum(len(d.get("items") or []) for d in dicts)
|
||||
_check("C2.actual_groups==8", actual_groups == EXPECT_GROUPS, "actual=%d" % actual_groups)
|
||||
_check("C2.actual_items==74", actual_items == EXPECT_ITEMS, "actual=%d" % actual_items)
|
||||
# ==========================================================================
|
||||
# 2. 各组项数
|
||||
# ==========================================================================
|
||||
@check('groups: 8 组 id 与顺序符合契约')
|
||||
def _check_group_ids():
|
||||
from pbl_appcodes.seed import list_group_codes, EXPECTED_GROUP_ORDER
|
||||
codes = list_group_codes()
|
||||
assert codes == list(EXPECTED_GROUP_ORDER), codes
|
||||
assert len(codes) == 8
|
||||
assert len(set(codes)) == 8, '组 id 重复'
|
||||
return True
|
||||
|
||||
# C3 编码全局唯一
|
||||
seen = {}
|
||||
dup = []
|
||||
for d in dicts:
|
||||
for it in (d.get("items") or []):
|
||||
c = it.get("code")
|
||||
if c in seen:
|
||||
dup.append("%s(%s/%s)" % (c, seen[c], d.get("group")))
|
||||
else:
|
||||
seen[c] = d.get("group")
|
||||
_check("C3.code_globally_unique", not dup, "dup=%s" % dup)
|
||||
|
||||
# C4 每组 sort 1..n 连续
|
||||
bad_sort = []
|
||||
for d in dicts:
|
||||
items = d.get("items") or []
|
||||
sorts = [it.get("sort") for it in items]
|
||||
expect = list(range(1, len(items) + 1))
|
||||
if sorts != expect:
|
||||
bad_sort.append("%s:%s!=%s" % (d.get("group"), sorts, expect))
|
||||
_check("C4.sort_continuous_1..n", not bad_sort, "bad=%s" % bad_sort)
|
||||
@check('groups: 每组项数符合契约（8/5/8/6/6/10/16/15 = 74）')
|
||||
def _check_items_per_group():
|
||||
from pbl_appcodes.seed import EXPECTED_ITEMS_PER_GROUP, get_items, list_group_codes
|
||||
total = 0
|
||||
for gid in list_group_codes():
|
||||
expected = EXPECTED_ITEMS_PER_GROUP[gid]
|
||||
actual = len(get_items(gid))
|
||||
assert actual == expected, '组 %s 项数 %d != %d' % (gid, actual, expected)
|
||||
total += actual
|
||||
assert total == 74, '合计 %d != 74' % total
|
||||
assert sum(EXPECTED_ITEMS_PER_GROUP.values()) == 74
|
||||
return True
|
||||
|
||||
# C5 项字段完备 + code 形态
|
||||
bad_item = []
|
||||
for d in dicts:
|
||||
for it in (d.get("items") or []):
|
||||
if not isinstance(it.get("code"), str) or not CODE_RE.match(it["code"]):
|
||||
bad_item.append("code:%r@%s" % (it.get("code"), d.get("group")))
|
||||
if not isinstance(it.get("name"), str) or not it["name"].strip():
|
||||
bad_item.append("name:%r@%s" % (it.get("name"), d.get("group")))
|
||||
if not isinstance(it.get("sort"), int) or isinstance(it.get("sort"), bool):
|
||||
bad_item.append("sort:%r@%s" % (it.get("sort"), d.get("group")))
|
||||
_check("C5.item_fields_valid", not bad_item, "bad=%s" % bad_item[:5])
|
||||
|
||||
# C6 组标识唯一且合法
|
||||
groups = [d.get("group") for d in dicts]
|
||||
bad_group = [g for g in groups if not isinstance(g, str) or not IDENT_RE.match(g)]
|
||||
_check("C6.group_ident_valid", not bad_group, "bad=%s" % bad_group)
|
||||
_check("C6.group_unique", len(set(groups)) == len(groups), "groups=%s" % groups)
|
||||
names = [d.get("group_name") for d in dicts]
|
||||
_check("C6.group_name_unique", len(set(names)) == len(names))
|
||||
_check("C6.group_name_nonempty", all(isinstance(n, str) and n.strip() for n in names))
|
||||
# ==========================================================================
|
||||
# 3. 唯一性 / sort 连续性 / 长度上限
|
||||
# ==========================================================================
|
||||
@check('codes: 组内 k 唯一、全局 appcodes_kv.id 唯一')
|
||||
def _check_uniqueness():
|
||||
from pbl_appcodes.seed import get_items, list_groups
|
||||
all_kv = []
|
||||
for group in list_groups():
|
||||
gid = group['id']
|
||||
keys = [str(it.get('k')) for it in group.get('items') or []]
|
||||
assert len(keys) == len(set(keys)), '组 %s 内 k 重复' % gid
|
||||
assert all(k.strip() for k in keys), '组 %s 存在空 k' % gid
|
||||
all_kv.extend('%s_%s' % (gid, k) for k in keys)
|
||||
assert len(all_kv) == len(set(all_kv)), 'appcodes_kv.id 全局重复'
|
||||
assert len(all_kv) == 74
|
||||
return True
|
||||
|
||||
# C7 Agent 工具启停契约 13/9
|
||||
by_group = {d.get("group"): (d.get("items") or []) for d in dicts}
|
||||
en = by_group.get(GROUP_ENABLED, [])
|
||||
dis = by_group.get(GROUP_DISABLED, [])
|
||||
_check("C7.enabled_count==13", len(en) == EXPECT_ENABLED, "actual=%d" % len(en))
|
||||
_check("C7.disabled_count==9", len(dis) == EXPECT_DISABLED, "actual=%d" % len(dis))
|
||||
_check("C7.enabled_status_all_enabled",
|
||||
all(it.get("status") == "enabled" for it in en) if en else False)
|
||||
_check("C7.disabled_status_all_disabled",
|
||||
all(it.get("status") == "disabled" for it in dis) if dis else False)
|
||||
_check("C7.tool_total==22", len(en) + len(dis) == 22,
|
||||
"actual=%d" % (len(en) + len(dis)))
|
||||
|
||||
# C8 零表约束：seed.py 不得含 CREATE TABLE
|
||||
ddl_hit = []
|
||||
if os.path.isfile(SEED_PY):
|
||||
with open(SEED_PY, "r", encoding="utf-8") as f:
|
||||
src = f.read()
|
||||
for kw in ("CREATE TABLE", "create table", "ALTER TABLE", "DROP TABLE"):
|
||||
if kw in src:
|
||||
ddl_hit.append(kw)
|
||||
_check("C8.seed_has_inject_entry", "def inject" in src or "def seed" in src,
|
||||
"seed.py 缺少注入入口函数")
|
||||
_check("C8.seed_idempotent_guard",
|
||||
("already" in src.lower()) or ("exists" in src.lower()) or ("ON DUPLICATE" in src),
|
||||
"seed.py 未见幂等判定")
|
||||
else:
|
||||
_check("C8.seed_py_exists", False, "seed.py 不存在")
|
||||
_check("C8.zero_table_no_ddl", not ddl_hit, "hit=%s" % ddl_hit)
|
||||
@check('codes: 组级 sort 从 1 连续、组内 sort 从 1 连续')
|
||||
def _check_sort_continuous():
|
||||
from pbl_appcodes.seed import list_groups
|
||||
groups = list_groups()
|
||||
for idx, group in enumerate(groups, start=1):
|
||||
assert int(group.get('sort') or 0) == idx, \
|
||||
'组 %s sort=%r 应为 %d（sort 不连续）' % (group.get('id'), group.get('sort'), idx)
|
||||
items = sorted(list(group.get('items') or []), key=lambda it: int(it.get('sort') or 0))
|
||||
for pos, item in enumerate(items, start=1):
|
||||
assert int(item.get('sort') or 0) == pos, \
|
||||
'组 %s 项 %s sort=%r 应为 %d（sort 不连续）' % (
|
||||
group.get('id'), item.get('k'), item.get('sort'), pos)
|
||||
return True
|
||||
|
||||
return actual_items
|
||||
|
||||
@check('codes: parentid ≤ 22、appcodes_kv.id ≤ 32（VARCHAR(32) 上限）')
|
||||
def _check_length_limits():
|
||||
from pbl_appcodes.seed import (PARENTID_MAX_LEN, KV_ID_MAX_LEN, get_items,
|
||||
list_group_codes)
|
||||
for gid in list_group_codes():
|
||||
assert len(gid) <= PARENTID_MAX_LEN, '%s 长度 %d > %d' % (
|
||||
gid, len(gid), PARENTID_MAX_LEN)
|
||||
for item in get_items():
|
||||
kv_id = '%s_%s' % (item['parentid'], item['k'])
|
||||
assert len(kv_id) <= KV_ID_MAX_LEN, '%s 长度 %d > %d' % (
|
||||
kv_id, len(kv_id), KV_ID_MAX_LEN)
|
||||
assert len(item['k']) <= 32 and len(item['v']) <= 255
|
||||
return True
|
||||
|
||||
|
||||
@check('codes: validate_codes() 返回 ok=True 且 problems 为空')
|
||||
def _check_validate():
|
||||
from pbl_appcodes.seed import validate_codes
|
||||
report = validate_codes()
|
||||
assert report['ok'] is True, '契约校验失败: %s' % report['problems']
|
||||
assert report['problems'] == []
|
||||
assert report['groups'] == 8 and report['items'] == 74
|
||||
return True
|
||||
|
||||
|
||||
# ==========================================================================
|
||||
# 4. dry_run_seed 计划
|
||||
# ==========================================================================
|
||||
@check('seed: dry_run_seed() 计划 8 parent + 74 kv，不触库')
|
||||
def _check_dry_run():
|
||||
from pbl_appcodes.seed import dry_run_seed
|
||||
plan = dry_run_seed()
|
||||
assert plan['groups'] == 8
|
||||
assert len(plan['parents']) == 8
|
||||
assert len(plan['kvs']) == 74
|
||||
assert plan['stats']['match'] is True
|
||||
assert plan['dialect'] == 'mariadb'
|
||||
assert plan['tables'] == ['appcodes', 'appcodes_kv']
|
||||
first_parent = plan['parents'][0]
|
||||
assert first_parent['id'] == 'pbl_bp_status'
|
||||
assert set(first_parent.keys()) >= {'id', 'name', 'hierarchy_flg', 'sort'}
|
||||
first_kv = plan['kvs'][0]
|
||||
assert first_kv['id'] == 'pbl_bp_status_draft'
|
||||
assert first_kv['parentid'] == 'pbl_bp_status'
|
||||
assert first_kv['k'] == 'draft' and first_kv['v'] == '草稿' and first_kv['sort'] == 1
|
||||
# 单组 dry_run
|
||||
one = dry_run_seed('pbl_quality_level')
|
||||
assert one['groups'] == 1 and len(one['kvs']) == 5
|
||||
return True
|
||||
|
||||
|
||||
@check('seed: seed_group(dry_run=True) 不触库即返回计划统计')
|
||||
def _check_seed_group_dry():
|
||||
import asyncio
|
||||
from pbl_appcodes.seed import seed_group
|
||||
loop = asyncio.new_event_loop()
|
||||
try:
|
||||
stat = loop.run_until_complete(seed_group('pbl_agent_role', dry_run=True))
|
||||
finally:
|
||||
loop.close()
|
||||
assert stat['dry_run'] is True
|
||||
assert stat['group'] == 'pbl_agent_role'
|
||||
assert stat['inserted'] == 7, stat # 1 parent + 6 items
|
||||
return True
|
||||
|
||||
|
||||
@check('seed: seed_all(dry_run=True) 汇总 8 组 74 项')
|
||||
def _check_seed_all_dry():
|
||||
import asyncio
|
||||
from pbl_appcodes.seed import seed_all
|
||||
loop = asyncio.new_event_loop()
|
||||
try:
|
||||
report = loop.run_until_complete(seed_all(dry_run=True))
|
||||
finally:
|
||||
loop.close()
|
||||
assert report['ok'] is True and report['dry_run'] is True
|
||||
assert report['groups'] == 8 and report['items'] == 74
|
||||
assert report['plan']['match'] is True
|
||||
return True
|
||||
|
||||
|
||||
@check('seed: ensure_tables 缺表时抛 DbError（不擅自建 appbase 的表）')
|
||||
def _check_ensure_tables_guard():
|
||||
import asyncio
|
||||
from pbl_appcodes.seed import ensure_tables
|
||||
from pbl_common.errors import DbError
|
||||
|
||||
class _NoTableDb(object):
|
||||
dbname = 'pbls_appbase'
|
||||
|
||||
async def execute(self, sql, params=None):
|
||||
return []
|
||||
|
||||
async def read(self, *a, **kw):
|
||||
return []
|
||||
|
||||
async def create(self, *a, **kw):
|
||||
raise AssertionError('不应写入')
|
||||
|
||||
loop = asyncio.new_event_loop()
|
||||
try:
|
||||
ensure_tables(db=_NoTableDb())
|
||||
raise AssertionError('缺表未抛 DbError')
|
||||
except DbError as exc:
|
||||
assert 'appcodes' in exc.message
|
||||
finally:
|
||||
loop.close()
|
||||
return True
|
||||
|
||||
|
||||
@check('seed: DDL 为 mariadb 方言，无 BIGSERIAL/SERIAL/nextval/FK/ENUM/TIMESTAMP')
|
||||
def _check_ddl_dialect():
|
||||
from pbl_appcodes.seed import _ddl_of, APPCODES_TABLE, APPCODES_KV_TABLE
|
||||
for table in (APPCODES_TABLE, APPCODES_KV_TABLE):
|
||||
ddl = _ddl_of(table).upper()
|
||||
for banned in ('BIGSERIAL', 'SERIAL', 'NEXTVAL', 'FOREIGN KEY', 'REFERENCES',
|
||||
'ENUM(', 'TIMESTAMP'):
|
||||
assert banned not in ddl, '%s DDL 含禁用语法 %s' % (table, banned)
|
||||
assert 'CREATE TABLE IF NOT EXISTS' in ddl, '%s DDL 非幂等' % table
|
||||
assert 'ENGINE=INNODB' in ddl and 'UTF8MB4' in ddl
|
||||
assert 'VARCHAR(32)' in ddl
|
||||
return True
|
||||
|
||||
|
||||
# ==========================================================================
|
||||
# 5. 符号闭包（QC #1 / #5：__init__.py / init.py / api.py 引用的名字必须真实存在）
|
||||
# ==========================================================================
|
||||
@check('closure: seed.py 导出 __init__/init/api 引用的全部符号')
|
||||
def _check_seed_symbol_closure():
|
||||
from pbl_appcodes import seed
|
||||
required = ('EXPECTED_GROUPS', 'EXPECTED_ITEMS', 'EXPECTED_ITEMS_PER_GROUP',
|
||||
'EXPECTED_GROUP_ORDER', 'get_items', 'list_group_codes',
|
||||
'list_groups', 'get_group', 'seed_all', 'seed_group',
|
||||
'dry_run_seed', 'ensure_tables', 'validate_codes', 'verify_seed',
|
||||
'load_codes', 'count_groups', 'count_items', 'get_meta',
|
||||
'APPCODES_TABLE', 'APPCODES_KV_TABLE', 'DIALECT')
|
||||
missing = [n for n in required if not hasattr(seed, n)]
|
||||
assert not missing, 'seed.py 缺少被引用符号: %s' % ', '.join(missing)
|
||||
for name in ('seed_all', 'seed_group', 'ensure_tables', 'verify_seed'):
|
||||
assert inspect.iscoroutinefunction(getattr(seed, name)), '%s 必须是 async' % name
|
||||
for name in ('get_items', 'list_group_codes', 'dry_run_seed', 'validate_codes'):
|
||||
assert not inspect.iscoroutinefunction(getattr(seed, name)), \
|
||||
'%s 必须是同步函数' % name
|
||||
return True
|
||||
|
||||
|
||||
@check('closure: pbl_appcodes 包导出 load_pbl_appcodes（三处同步之②）')
|
||||
def _check_package_export():
|
||||
import pbl_appcodes
|
||||
assert callable(getattr(pbl_appcodes, 'load_pbl_appcodes', None)), \
|
||||
'pbl_appcodes/__init__.py 未导出 load_pbl_appcodes'
|
||||
from pbl_appcodes.init import load_pbl_appcodes
|
||||
assert pbl_appcodes.load_pbl_appcodes is load_pbl_appcodes
|
||||
return True
|
||||
|
||||
|
||||
@check('closure: api.py 引用的 pbl_common 符号全部存在（无 ParamInvalidError/ErrorCode 断裂）')
|
||||
def _check_api_imports():
|
||||
from pbl_common import errors
|
||||
for name in ('ParamInvalidError', 'ErrorCode', 'PBLError', 'err', 'fail',
|
||||
'DbError', 'NotFoundError', 'CODE_TO_HTTP',
|
||||
'assert_not_write_protected', 'WRITE_PROTECTED_MODULES',
|
||||
'PBL_E_PARAM', 'PBL_E_DB'):
|
||||
assert hasattr(errors, name), 'pbl_common.errors 缺少 %s' % name
|
||||
from pbl_appcodes import api as appcodes_api
|
||||
assert callable(getattr(appcodes_api, 'list_appcodes', None))
|
||||
assert callable(getattr(appcodes_api, 'seed_appcodes', None))
|
||||
return True
|
||||
|
||||
|
||||
@check('closure: 包内无 except ImportError 静默吞错、无反向依赖 pbl_blueprint')
|
||||
def _check_no_silent_swallow():
|
||||
import os
|
||||
import pbl_appcodes
|
||||
pkg_dir = os.path.dirname(os.path.abspath(pbl_appcodes.__file__))
|
||||
scanned = 0
|
||||
for fname in sorted(os.listdir(pkg_dir)):
|
||||
if not fname.endswith('.py') or fname == 'self_check.py':
|
||||
continue
|
||||
with open(os.path.join(pkg_dir, fname), 'r', encoding='utf-8') as fh:
|
||||
src = fh.read()
|
||||
scanned += 1
|
||||
assert 'except ImportError' not in src, \
|
||||
'%s 存在 except ImportError 静默吞错（兼容层失效根因）' % fname
|
||||
assert 'pbl_blueprint' not in src, \
|
||||
'%s 反向依赖 pbl_blueprint（循环依赖根因）' % fname
|
||||
assert 'm1b' not in src, '%s 引用 m1b 兼容层' % fname
|
||||
assert scanned >= 4, '扫描到的 .py 文件过少: %d' % scanned
|
||||
return True
|
||||
|
||||
|
||||
# ==========================================================================
|
||||
# 6. 写保护门禁不误伤共享编码表
|
||||
# ==========================================================================
|
||||
@check('guard: appcodes/appcodes_kv 非受保护引用模块，写门禁放行；rbac 等仍拒绝')
|
||||
def _check_write_protect_scope():
|
||||
from pbl_common.errors import (WRITE_PROTECTED_MODULES, assert_not_write_protected,
|
||||
is_write_protected, WriteProtectedError)
|
||||
from pbl_appcodes.seed import APPCODES_TABLE, APPCODES_KV_TABLE
|
||||
for table in (APPCODES_TABLE, APPCODES_KV_TABLE):
|
||||
assert table not in WRITE_PROTECTED_MODULES
|
||||
assert is_write_protected(table, 'update') is False
|
||||
assert assert_not_write_protected(table, 'update') is True
|
||||
for mod in ('rbac', 'world', 'scene', 'entity', 'scense', 'scense_runtime',
|
||||
'script_engine'):
|
||||
assert is_write_protected(mod, 'update') is True
|
||||
try:
|
||||
assert_not_write_protected(mod, 'delete')
|
||||
except WriteProtectedError:
|
||||
pass
|
||||
else:
|
||||
raise AssertionError('%s delete 未被写保护拦截' % mod)
|
||||
return True
|
||||
|
||||
|
||||
# ==========================================================================
|
||||
# 运行器
|
||||
# ==========================================================================
|
||||
def run_self_check(verbose=True):
|
||||
"""执行全部自检，返回 (passed, total, failures)。"""
|
||||
passed, failures = 0, []
|
||||
for name, fn in RESULTS:
|
||||
try:
|
||||
if fn() is False:
|
||||
raise AssertionError('check returned False')
|
||||
passed += 1
|
||||
if verbose:
|
||||
print(' [PASS] %s' % name)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
failures.append((name, '%s: %s' % (type(exc).__name__, exc),
|
||||
traceback.format_exc()))
|
||||
if verbose:
|
||||
print(' [FAIL] %s -> %s: %s' % (name, type(exc).__name__, exc))
|
||||
total = len(RESULTS)
|
||||
if verbose:
|
||||
status = 'PASS' if not failures else 'FAIL'
|
||||
print('SELF_CHECK pbl_appcodes: %s %d/%d' % (status, passed, total))
|
||||
for name, msg, _tb in failures:
|
||||
print(' - %s: %s' % (name, msg))
|
||||
return passed, total, failures
|
||||
|
||||
|
||||
def main():
|
||||
total_items = run_checks()
|
||||
failed = [r for r in _results if not r[1]]
|
||||
print("=" * 68)
|
||||
print("SELF_CHECK pbl_appcodes —— 枚举字典契约自检")
|
||||
print("codes.json: %s" % CODES_JSON)
|
||||
print("-" * 68)
|
||||
for name, ok, detail in _results:
|
||||
flag = "PASS" if ok else "FAIL"
|
||||
line = "[%s] %s" % (flag, name)
|
||||
if not ok and detail:
|
||||
line += " -> %s" % detail
|
||||
print(line)
|
||||
print("-" * 68)
|
||||
print("检查项: %d 通过 / %d 失败；字典项总数: %d" % (len(_results) - len(failed), len(failed), total_items))
|
||||
if failed:
|
||||
print("SELF_CHECK pbl_appcodes: FAIL %d/%d" % (total_items - len(failed), total_items))
|
||||
return 1
|
||||
print("SELF_CHECK pbl_appcodes: PASS %d/%d" % (total_items, total_items))
|
||||
return 0
|
||||
_passed, _total, failures = run_self_check(verbose=True)
|
||||
return 0 if not failures else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
if __name__ == '__main__':
|
||||
sys.exit(main())
|
||||
|
||||
|
||||
__all__ = ['RESULTS', 'check', 'run_self_check', 'main']
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user