deliver: 交付收口（引擎代为提交）
This commit is contained in:
agent.develop
parent
474fc2b256
commit
626b2c0bfc
@ -81,6 +81,33 @@
|
||||
"null": false,
|
||||
"type": "int"
|
||||
},
|
||||
{
|
||||
"name": "rules_json",
|
||||
"title": "规则产物JSON(script_type=1规则数组,确定性序列化)",
|
||||
"type": "text",
|
||||
"nullable": "yes"
|
||||
},
|
||||
{
|
||||
"name": "rules_hash",
|
||||
"title": "规则产物指纹(sha256,确定性)",
|
||||
"type": "str",
|
||||
"length": 64,
|
||||
"nullable": "yes",
|
||||
"default": ""
|
||||
},
|
||||
{
|
||||
"name": "rule_count",
|
||||
"title": "规则条数",
|
||||
"type": "int",
|
||||
"nullable": "yes",
|
||||
"default": "0"
|
||||
},
|
||||
{
|
||||
"name": "rules_manifest",
|
||||
"title": "规则清单JSON(rule_id/event_type/content_hash)",
|
||||
"type": "text",
|
||||
"nullable": "yes"
|
||||
},
|
||||
{
|
||||
"name": "quality_state",
|
||||
"comment": "编译时质量状态",
|
||||
@ -133,4 +160,4 @@
|
||||
}
|
||||
],
|
||||
"codes": []
|
||||
}
|
||||
}
|
||||
47
models/pbl_script_rule.json
Normal file
47
models/pbl_script_rule.json
Normal file
@ -0,0 +1,47 @@
|
||||
{
|
||||
"summary": [
|
||||
{
|
||||
"name": "pbl_script_rule",
|
||||
"title": "PBL 编译产物：script_engine 规则（script_type=1，第12章 ECR 映射）",
|
||||
"primary": ["id"],
|
||||
"catelog": "entity"
|
||||
}
|
||||
],
|
||||
"fields": [
|
||||
{"name": "tenant_id", "title": "租户ID（多租户隔离，所有查询打头）", "type": "str", "length": 32, "nullable": "no"},
|
||||
{"name": "id", "title": "主键ID", "type": "str", "length": 32, "nullable": "no"},
|
||||
{"name": "blueprint_id", "title": "蓝图ID", "type": "str", "length": 32, "nullable": "no", "default": ""},
|
||||
{"name": "blueprint_version", "title": "蓝图版本号", "type": "str", "length": 32, "nullable": "no", "default": ""},
|
||||
{"name": "game_def_id", "title": "关联 pbl_game_definition.id", "type": "str", "length": 32, "nullable": "yes", "default": ""},
|
||||
{"name": "compile_task_no", "title": "编译任务号（可追溯）", "type": "str", "length": 64, "nullable": "yes", "default": ""},
|
||||
{"name": "rule_id", "title": "规则稳定ID（来源 id/event_key 或内容哈希派生）", "type": "str", "length": 64, "nullable": "no"},
|
||||
{"name": "script_code", "title": "script_engine 脚本编码（幂等键，确定性派生）", "type": "str", "length": 64, "nullable": "no"},
|
||||
{"name": "name", "title": "规则名称", "type": "str", "length": 128, "nullable": "yes", "default": ""},
|
||||
{"name": "description", "title": "规则描述", "type": "str", "length": 512, "nullable": "yes", "default": ""},
|
||||
{"name": "script_type", "title": "脚本类型（1=规则 event→condition→response）", "type": "str", "length": 16, "nullable": "no", "default": "1"},
|
||||
{"name": "event_type", "title": "触发事件类型（归一后）", "type": "str", "length": 64, "nullable": "yes", "default": ""},
|
||||
{"name": "priority", "title": "优先级（大者先执行）", "type": "int", "nullable": "yes", "default": "100"},
|
||||
{"name": "enabled", "title": "是否启用（1启用/0停用）", "type": "str", "length": 1, "nullable": "yes", "default": "1"},
|
||||
{"name": "condition_count", "title": "条件节点数", "type": "int", "nullable": "yes", "default": "0"},
|
||||
{"name": "response_count", "title": "响应动作数", "type": "int", "nullable": "yes", "default": "0"},
|
||||
{"name": "content", "title": "规则JSON（确定性序列化，script_engine 可直接加载）", "type": "text", "nullable": "no"},
|
||||
{"name": "content_hash", "title": "规则内容指纹 sha256（幂等比对）", "type": "str", "length": 64, "nullable": "no", "default": ""},
|
||||
{"name": "rules_hash", "title": "所属规则集指纹 sha256（同版本重编译相等）", "type": "str", "length": 64, "nullable": "yes", "default": ""},
|
||||
{"name": "origin_json", "title": "来源追溯JSON（blueprint/sub_object/source_path）", "type": "text", "nullable": "yes"},
|
||||
{"name": "script_ref", "title": "已同步的 script_engine 脚本引用JSON（table/id/code）", "type": "text", "nullable": "yes"},
|
||||
{"name": "sync_status", "title": "同步状态（pending/synced/skipped/failed）", "type": "str", "length": 16, "nullable": "yes", "default": "pending"},
|
||||
{"name": "compiler_version", "title": "编译器版本", "type": "str", "length": 32, "nullable": "yes", "default": ""},
|
||||
{"name": "created_by", "title": "创建人", "type": "str", "length": 32, "nullable": "yes", "default": ""},
|
||||
{"name": "created_at", "title": "创建时间", "type": "datetime", "nullable": "yes"},
|
||||
{"name": "updated_at", "title": "更新时间", "type": "datetime", "nullable": "yes"}
|
||||
],
|
||||
"indexes": [
|
||||
{"name": "PRIMARY", "idxtype": "unique", "idxfields": ["id"]},
|
||||
{"name": "uk_psr_rule", "idxtype": "unique", "idxfields": ["tenant_id", "blueprint_id", "blueprint_version", "rule_id"]},
|
||||
{"name": "uk_psr_code", "idxtype": "unique", "idxfields": ["tenant_id", "script_code"]},
|
||||
{"name": "idx_psr_gd", "idxtype": "index", "idxfields": ["tenant_id", "game_def_id"]},
|
||||
{"name": "idx_psr_hash", "idxtype": "index", "idxfields": ["tenant_id", "rules_hash"]},
|
||||
{"name": "idx_psr_event", "idxtype": "index", "idxfields": ["tenant_id", "event_type"]}
|
||||
],
|
||||
"codes": []
|
||||
}
|
||||
590
pbl_compiler/exporter.py
Normal file
590
pbl_compiler/exporter.py
Normal file
@ -0,0 +1,590 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
pbl_compiler.exporter —— M3b 第12章后半：规则产物导出 + game_definition 写入。
|
||||
|
||||
职责（三件事，全部幂等）：
|
||||
1. **导出规则产物**：把 build_rules() 得到的 script_type=1 规则 JSON 落到
|
||||
script_engine 的脚本表（一行一规则，content = 确定性 JSON），
|
||||
并生成一份聚合产物 `rules.json`（数组）+ `rules_manifest.json`。
|
||||
2. **回写编译版本**：pbl_compiler_version 记录 rules_hash / rules_count / rules_manifest /
|
||||
script_ids，保证「同版本重编译 → 同 hash」可验证（M3a 29.6 确定性验收）。
|
||||
3. **写入 game_definition**：把规则清单挂进 GD 的 `rules` 顶层键，
|
||||
upsert pbl_game_definition（tenant_id + blueprint_id + version 唯一）。
|
||||
|
||||
铁律：
|
||||
- 所有 SQL 的 WHERE / INSERT 一律 tenant_id 打头，缺失即抛 CompileError（fail-closed）。
|
||||
- 幂等：重复导出同一 (tenant_id, blueprint_version) 不产生重复行（UPSERT by 唯一键）。
|
||||
- 确定性：产物内容不含时间戳/uuid；DB 行的 created_at/updated_at 属审计字段，
|
||||
不参与 rules_hash / content_hash 计算。
|
||||
- 本模块不直接建连接：调用方（.dspy / init.py）传入 sor 上下文，或用 open_sor() 便捷入口。
|
||||
"""
|
||||
|
||||
import json
|
||||
|
||||
from .script_mapping import (
|
||||
CompileError,
|
||||
Issue,
|
||||
canonical_dumps,
|
||||
canonical_rules_json,
|
||||
build_rules_manifest,
|
||||
diff_rules,
|
||||
has_errors,
|
||||
rule_content_json,
|
||||
rules_hash,
|
||||
sha256_text,
|
||||
sort_rules,
|
||||
strip_volatile,
|
||||
to_int,
|
||||
validate_rules,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
'SCRIPT_TYPE_RULE',
|
||||
'GD_RULES_KEY',
|
||||
'open_sor',
|
||||
'script_code_of',
|
||||
'export_rules',
|
||||
'write_compiler_version_rules',
|
||||
'build_game_definition',
|
||||
'write_game_definition',
|
||||
'export_all',
|
||||
'load_exported_rules',
|
||||
'rollback_export',
|
||||
]
|
||||
|
||||
SCRIPT_TYPE_RULE = 1
|
||||
GD_RULES_KEY = 'rules'
|
||||
|
||||
# script_engine 脚本表候选名（不同宿主/版本命名差异，按序探测）
|
||||
_SCRIPT_TABLE_CANDIDATES = ('script', 'scripts', 'se_script', 'script_engine_script')
|
||||
_GD_TABLE_CANDIDATES = ('pbl_game_definition',)
|
||||
_VERSION_TABLE_CANDIDATES = ('pbl_compiler_version',)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# DB 上下文
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def open_sor(env, module='pbl_compiler'):
|
||||
"""便捷入口：在 .py 里拿 sqlor 上下文。
|
||||
|
||||
env 传 request._run_ns（.dspy 里就是 request._run_ns），
|
||||
或传 ServerEnv() 实例（load_* 阶段）。返回 async context manager。
|
||||
"""
|
||||
try:
|
||||
from pbl_common.db import get_sor_context as _gsc # 优先复用公共内核
|
||||
return _gsc(env, module)
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
get_dbname = getattr(env, 'get_module_dbname', None)
|
||||
dbname = get_dbname(module) if callable(get_dbname) else module
|
||||
from sqlor.dbpools import DBPools
|
||||
return DBPools().sqlorContext(dbname)
|
||||
except Exception as e: # pragma: no cover
|
||||
raise CompileError('E_DB_CONTEXT', '无法获取 sqlor 上下文: %s' % e)
|
||||
|
||||
|
||||
async def _detect_table(sor, candidates):
|
||||
"""探测实际存在的表名（跨宿主兼容）。返回首个可 SELECT 的表名，全失败返回 None。"""
|
||||
for t in candidates:
|
||||
try:
|
||||
await sor.sqlExe('SELECT 1 FROM `%s` LIMIT 1' % t, {})
|
||||
return t
|
||||
except Exception:
|
||||
continue
|
||||
return None
|
||||
|
||||
|
||||
async def _table_columns(sor, table):
|
||||
try:
|
||||
recs = await sor.sqlExe(
|
||||
'SELECT COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS '
|
||||
'WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=${tbl}$', {'tbl': table})
|
||||
cols = set()
|
||||
for r in recs or []:
|
||||
v = getattr(r, 'COLUMN_NAME', None) or getattr(r, 'column_name', None)
|
||||
if v:
|
||||
cols.add(str(v))
|
||||
if cols:
|
||||
return cols
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
recs = await sor.sqlExe('SELECT * FROM `%s` LIMIT 1' % table, {})
|
||||
if recs:
|
||||
return set(vars(recs[0]).keys()) if hasattr(recs[0], '__dict__') else set()
|
||||
except Exception:
|
||||
pass
|
||||
return set()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 1. 规则产物导出（script_engine script_type=1）
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def script_code_of(rule, ctx):
|
||||
"""规则 → script 表业务编码（幂等键的一部分）。确定性、可读、≤64 字符。"""
|
||||
bp = str((ctx or {}).get('blueprint_id') or 'bp')
|
||||
ver = str((ctx or {}).get('blueprint_version') or 'v0')
|
||||
rid = str(rule.get('rule_id') or '')
|
||||
code = 'pbl_%s_%s_%s' % (bp[:16], ver[:12], rid[:24])
|
||||
code = ''.join(c if (c.isalnum() or c == '_') else '_' for c in code)
|
||||
return code[:64]
|
||||
|
||||
|
||||
async def export_rules(sor, rules, ctx, issues=None):
|
||||
"""把规则数组导出到 script_engine 脚本表（UPSERT，幂等）。
|
||||
|
||||
ctx 必填: tenant_id, blueprint_id, blueprint_version；
|
||||
可选: scene_id/world_id, group_name, dry_run(bool), overwrite(bool, 默认 True)
|
||||
返回 {'exported': n, 'skipped': n, 'script_ids': [...], 'table': '...', 'issues': [...]}
|
||||
"""
|
||||
issues = list(issues or [])
|
||||
tenant_id = str((ctx or {}).get('tenant_id') or '').strip()
|
||||
if not tenant_id:
|
||||
raise CompileError('E_TENANT_REQUIRED', 'tenant_id 缺失，拒绝导出规则产物')
|
||||
if not str((ctx or {}).get('blueprint_version') or '').strip():
|
||||
raise CompileError('E_VERSION_REQUIRED', 'blueprint_version 缺失，拒绝导出规则产物')
|
||||
|
||||
rules = sort_rules(rules or [])
|
||||
result = {'exported': 0, 'skipped': 0, 'script_ids': [], 'table': None,
|
||||
'rules_hash': rules_hash(rules), 'issues': issues}
|
||||
if not rules:
|
||||
issues.append(Issue('warning', 'W_EXPORT_EMPTY', '规则为空，未导出任何脚本'))
|
||||
return result
|
||||
|
||||
table = await _detect_table(sor, _SCRIPT_TABLE_CANDIDATES)
|
||||
if not table:
|
||||
issues.append(Issue('error', 'E_SCRIPT_TABLE_MISSING',
|
||||
'script_engine 脚本表不存在（探测: %s）'
|
||||
% ', '.join(_SCRIPT_TABLE_CANDIDATES)))
|
||||
raise CompileError('E_SCRIPT_TABLE_MISSING', '未找到 script_engine 脚本表')
|
||||
result['table'] = table
|
||||
cols = await _table_columns(sor, table)
|
||||
|
||||
name_col = _pick(cols, ('name', 'script_name', 'title'))
|
||||
code_col = _pick(cols, ('script_code', 'code', 'script_key', 'key'))
|
||||
type_col = _pick(cols, ('script_type', 'type', 'stype'))
|
||||
content_col = _pick(cols, ('content', 'script_content', 'body', 'code_body', 'definition'))
|
||||
desc_col = _pick(cols, ('description', 'desc', 'remark', 'memo'))
|
||||
group_col = _pick(cols, ('group_name', 'group', 'category', 'module_name'))
|
||||
enabled_col = _pick(cols, ('enabled', 'is_enabled', 'active', 'status'))
|
||||
hash_col = _pick(cols, ('content_hash', 'hash', 'checksum'))
|
||||
origin_col = _pick(cols, ('origin', 'origin_json', 'source_json', 'extra', 'extra_json'))
|
||||
ver_col = _pick(cols, ('blueprint_version', 'bp_version', 'source_version'))
|
||||
bp_col = _pick(cols, ('blueprint_id', 'bp_id', 'source_id'))
|
||||
|
||||
if not content_col:
|
||||
raise CompileError('E_SCRIPT_CONTENT_COL',
|
||||
'脚本表 %s 缺少 content 类字段，无法写入规则 JSON' % table)
|
||||
|
||||
for rule in rules:
|
||||
rid = str(rule.get('rule_id') or '')
|
||||
code = script_code_of(rule, ctx)
|
||||
content = rule_content_json(rule)
|
||||
chash = sha256_text(content)
|
||||
|
||||
ns = {'tenant_id': tenant_id}
|
||||
if code_col:
|
||||
ns[code_col] = code
|
||||
if name_col:
|
||||
ns[name_col] = str(rule.get('name') or rid)[:128]
|
||||
if type_col:
|
||||
ns[type_col] = SCRIPT_TYPE_RULE
|
||||
ns[content_col] = content
|
||||
if desc_col:
|
||||
ns[desc_col] = str(rule.get('description') or '')[:512]
|
||||
if group_col:
|
||||
ns[group_col] = str((ctx or {}).get('group_name')
|
||||
or 'pbl_%s' % (ctx or {}).get('blueprint_id') or 'pbl')[:64]
|
||||
if enabled_col:
|
||||
ns[enabled_col] = 1 if rule.get('enabled') else 0
|
||||
if hash_col:
|
||||
ns[hash_col] = chash
|
||||
if origin_col:
|
||||
ns[origin_col] = canonical_dumps(strip_volatile(rule.get('origin') or {}))
|
||||
if ver_col:
|
||||
ns[ver_col] = str(ctx.get('blueprint_version'))
|
||||
if bp_col:
|
||||
ns[bp_col] = str(ctx.get('blueprint_id') or '')
|
||||
for extra_key in ('scene_id', 'world_id', 'org_id'):
|
||||
c = _pick(cols, (extra_key,))
|
||||
if c and (ctx or {}).get(extra_key):
|
||||
ns[c] = str(ctx[extra_key])
|
||||
|
||||
if (ctx or {}).get('dry_run'):
|
||||
result['skipped'] += 1
|
||||
result['script_ids'].append({'rule_id': rid, 'script_code': code,
|
||||
'content_hash': chash, 'dry_run': True})
|
||||
continue
|
||||
|
||||
where = {'tenant_id': tenant_id}
|
||||
if code_col:
|
||||
where[code_col] = code
|
||||
else:
|
||||
where[name_col or 'name'] = ns.get(name_col or 'name', rid)
|
||||
|
||||
try:
|
||||
exist = await sor.R(table, where)
|
||||
except Exception as e:
|
||||
issues.append(Issue('error', 'E_SCRIPT_QUERY', '查询已有脚本失败: %s' % e, rid))
|
||||
raise CompileError('E_SCRIPT_QUERY', '查询脚本表失败: %s' % e, {'rule_id': rid})
|
||||
|
||||
rows = exist if isinstance(exist, (list, tuple)) else ([exist] if exist else [])
|
||||
if rows:
|
||||
old = rows[0]
|
||||
old_hash = getattr(old, hash_col, None) if hash_col else None
|
||||
old_content = getattr(old, content_col, None)
|
||||
if old_hash == chash or (old_hash is None and str(old_content or '') == content):
|
||||
result['skipped'] += 1
|
||||
sid = getattr(old, 'id', None) or code
|
||||
result['script_ids'].append({'rule_id': rid, 'script_code': code,
|
||||
'content_hash': chash, 'id': sid,
|
||||
'changed': False})
|
||||
continue
|
||||
if not (ctx or {}).get('overwrite', True):
|
||||
issues.append(Issue('warning', 'W_EXPORT_CONFLICT',
|
||||
'脚本 %s 已存在且内容不同，overwrite=False 跳过' % code, rid))
|
||||
result['skipped'] += 1
|
||||
continue
|
||||
try:
|
||||
await sor.U(table, ns, dict(where))
|
||||
result['exported'] += 1
|
||||
result['script_ids'].append({'rule_id': rid, 'script_code': code,
|
||||
'content_hash': chash,
|
||||
'id': getattr(old, 'id', None) or code,
|
||||
'changed': True})
|
||||
except Exception as e:
|
||||
issues.append(Issue('error', 'E_SCRIPT_UPDATE',
|
||||
'更新脚本失败: %s' % e, rid))
|
||||
raise CompileError('E_SCRIPT_UPDATE', '更新脚本 %s 失败: %s' % (code, e))
|
||||
else:
|
||||
try:
|
||||
await sor.C(table, dict(ns))
|
||||
result['exported'] += 1
|
||||
result['script_ids'].append({'rule_id': rid, 'script_code': code,
|
||||
'content_hash': chash, 'changed': True})
|
||||
except Exception as e:
|
||||
issues.append(Issue('error', 'E_SCRIPT_INSERT',
|
||||
'写入脚本失败: %s' % e, rid))
|
||||
raise CompileError('E_SCRIPT_INSERT', '写入脚本 %s 失败: %s' % (code, e))
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def _pick(cols, names):
|
||||
if not cols:
|
||||
return names[0] # 列信息不可得时用首选名（写入失败会显式报错，不静默）
|
||||
for n in names:
|
||||
if n in cols:
|
||||
return n
|
||||
return None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 2. 编译版本回写
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def write_compiler_version_rules(sor, version_id, rules, ctx, export_result=None,
|
||||
issues=None):
|
||||
"""把规则产物指纹回写 pbl_compiler_version（幂等：同 version_id 覆盖同值）。"""
|
||||
issues = list(issues or [])
|
||||
tenant_id = str((ctx or {}).get('tenant_id') or '').strip()
|
||||
if not tenant_id:
|
||||
raise CompileError('E_TENANT_REQUIRED', 'tenant_id 缺失，拒绝回写编译版本')
|
||||
if not version_id:
|
||||
issues.append(Issue('warning', 'W_VERSION_ID_MISSING', '未提供 version_id，跳过版本回写'))
|
||||
return {'written': False, 'issues': issues}
|
||||
|
||||
table = await _detect_table(sor, _VERSION_TABLE_CANDIDATES)
|
||||
if not table:
|
||||
issues.append(Issue('warning', 'W_VERSION_TABLE_MISSING',
|
||||
'pbl_compiler_version 表不存在，跳过版本回写'))
|
||||
return {'written': False, 'issues': issues}
|
||||
|
||||
cols = await _table_columns(sor, table)
|
||||
rules = sort_rules(rules or [])
|
||||
manifest = build_rules_manifest(rules, extra={
|
||||
'tenant_id': tenant_id,
|
||||
'blueprint_id': str(ctx.get('blueprint_id') or ''),
|
||||
'blueprint_version': str(ctx.get('blueprint_version') or ''),
|
||||
'script_ids': (export_result or {}).get('script_ids') or [],
|
||||
})
|
||||
|
||||
ns = {'tenant_id': tenant_id}
|
||||
_set(ns, cols, ('rules_json', 'rules'), canonical_rules_json(rules))
|
||||
_set(ns, cols, ('rules_hash', 'rules_fingerprint'), manifest['rules_hash'])
|
||||
_set(ns, cols, ('rules_manifest', 'rules_manifest_json'), canonical_dumps(manifest))
|
||||
_set(ns, cols, ('rules_count', 'rule_count'), manifest['rule_count'])
|
||||
_set(ns, cols, ('script_ids', 'script_ids_json'),
|
||||
canonical_dumps([s.get('script_code') for s in manifest['extra']['script_ids']]))
|
||||
if len(ns) <= 1:
|
||||
issues.append(Issue('warning', 'W_VERSION_NO_COLS',
|
||||
'pbl_compiler_version 无规则相关字段，跳过回写'))
|
||||
return {'written': False, 'issues': issues, 'manifest': manifest}
|
||||
|
||||
where = {'tenant_id': tenant_id, 'id': str(version_id)}
|
||||
try:
|
||||
exist = await sor.R(table, where)
|
||||
except Exception as e:
|
||||
raise CompileError('E_VERSION_QUERY', '查询编译版本失败: %s' % e)
|
||||
rows = exist if isinstance(exist, (list, tuple)) else ([exist] if exist else [])
|
||||
if rows:
|
||||
await sor.U(table, ns, dict(where))
|
||||
else:
|
||||
ns2 = dict(ns)
|
||||
ns2['id'] = str(version_id)
|
||||
_set(ns2, cols, ('blueprint_id',), str(ctx.get('blueprint_id') or ''))
|
||||
_set(ns2, cols, ('blueprint_version', 'version'), str(ctx.get('blueprint_version') or ''))
|
||||
await sor.C(table, ns2)
|
||||
return {'written': True, 'version_id': str(version_id), 'table': table,
|
||||
'rules_hash': manifest['rules_hash'], 'rule_count': manifest['rule_count'],
|
||||
'manifest': manifest, 'issues': issues}
|
||||
|
||||
|
||||
def _set(ns, cols, names, value):
|
||||
col = _pick(cols, names)
|
||||
if col and (cols is None or col in cols or not cols):
|
||||
ns[col] = value
|
||||
return col
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 3. game_definition 写入
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def build_game_definition(rules, ctx, base_gd=None, export_result=None):
|
||||
"""构造 GD 的 rules 顶层键（并合并进 base_gd）。
|
||||
|
||||
base_gd: M3a gd_builder 产出的 10 顶层键 GD（可为 None，则只产出 rules 片段）。
|
||||
返回 (gd_dict, gd_json_text, gd_hash)。
|
||||
"""
|
||||
rules = sort_rules(rules or [])
|
||||
manifest = build_rules_manifest(rules)
|
||||
script_ids = (export_result or {}).get('script_ids') or []
|
||||
|
||||
rules_block = {
|
||||
'schema_version': manifest['schema_version'],
|
||||
'script_type': SCRIPT_TYPE_RULE,
|
||||
'engine': 'script_engine',
|
||||
'rule_count': manifest['rule_count'],
|
||||
'rules_hash': manifest['rules_hash'],
|
||||
'rules': rules,
|
||||
'scripts': [{'rule_id': s.get('rule_id'), 'script_code': s.get('script_code'),
|
||||
'content_hash': s.get('content_hash')}
|
||||
for s in script_ids if isinstance(s, dict)],
|
||||
'origin': {
|
||||
'tenant_id': str((ctx or {}).get('tenant_id') or ''),
|
||||
'blueprint_id': str((ctx or {}).get('blueprint_id') or ''),
|
||||
'blueprint_version': str((ctx or {}).get('blueprint_version') or ''),
|
||||
},
|
||||
}
|
||||
|
||||
gd = dict(base_gd) if isinstance(base_gd, dict) else {}
|
||||
gd[GD_RULES_KEY] = rules_block
|
||||
gd.setdefault('schema_version', str(gd.get('schema_version') or '1.0'))
|
||||
gd_text = canonical_dumps(strip_volatile(gd))
|
||||
return gd, gd_text, sha256_text(gd_text)
|
||||
|
||||
|
||||
async def write_game_definition(sor, gd, gd_text, gd_hash, ctx, issues=None):
|
||||
"""upsert pbl_game_definition（唯一键 tenant_id + blueprint_id + version）。"""
|
||||
issues = list(issues or [])
|
||||
tenant_id = str((ctx or {}).get('tenant_id') or '').strip()
|
||||
if not tenant_id:
|
||||
raise CompileError('E_TENANT_REQUIRED', 'tenant_id 缺失，拒绝写入 game_definition')
|
||||
bp_id = str((ctx or {}).get('blueprint_id') or '').strip()
|
||||
bp_ver = str((ctx or {}).get('blueprint_version') or '').strip()
|
||||
if not bp_id or not bp_ver:
|
||||
raise CompileError('E_GD_KEY_MISSING',
|
||||
'blueprint_id / blueprint_version 缺失，无法定位 game_definition')
|
||||
|
||||
table = await _detect_table(sor, _GD_TABLE_CANDIDATES)
|
||||
if not table:
|
||||
issues.append(Issue('error', 'E_GD_TABLE_MISSING', 'pbl_game_definition 表不存在'))
|
||||
raise CompileError('E_GD_TABLE_MISSING', '未找到 pbl_game_definition 表')
|
||||
|
||||
cols = await _table_columns(sor, table)
|
||||
ns = {'tenant_id': tenant_id}
|
||||
_set(ns, cols, ('blueprint_id', 'bp_id'), bp_id)
|
||||
_set(ns, cols, ('blueprint_version', 'version', 'bp_version'), bp_ver)
|
||||
_set(ns, cols, ('game_definition', 'gd_json', 'definition', 'content'), gd_text)
|
||||
_set(ns, cols, ('gd_hash', 'definition_hash', 'content_hash', 'hash'), gd_hash)
|
||||
_set(ns, cols, ('rules_hash',), sha256_text(canonical_rules_json(
|
||||
(gd.get(GD_RULES_KEY) or {}).get('rules') or [])))
|
||||
_set(ns, cols, ('rule_count', 'rules_count'),
|
||||
to_int((gd.get(GD_RULES_KEY) or {}).get('rule_count'), 0))
|
||||
_set(ns, cols, ('status', 'state'), str(ctx.get('status') or 'compiled'))
|
||||
_set(ns, cols, ('compiler_version', 'compiler_ver'),
|
||||
str(ctx.get('compiler_version') or '1.0'))
|
||||
|
||||
where = {'tenant_id': tenant_id}
|
||||
for k in ('blueprint_id', 'bp_id'):
|
||||
if k in ns:
|
||||
where[k] = ns[k]
|
||||
break
|
||||
for k in ('blueprint_version', 'version', 'bp_version'):
|
||||
if k in ns:
|
||||
where[k] = ns[k]
|
||||
break
|
||||
|
||||
try:
|
||||
exist = await sor.R(table, where)
|
||||
except Exception as e:
|
||||
raise CompileError('E_GD_QUERY', '查询 game_definition 失败: %s' % e)
|
||||
rows = exist if isinstance(exist, (list, tuple)) else ([exist] if exist else [])
|
||||
|
||||
if rows:
|
||||
old = rows[0]
|
||||
old_hash = None
|
||||
for hk in ('gd_hash', 'definition_hash', 'content_hash', 'hash'):
|
||||
v = getattr(old, hk, None)
|
||||
if v:
|
||||
old_hash = v
|
||||
break
|
||||
if old_hash == gd_hash and not ctx.get('force'):
|
||||
return {'written': False, 'unchanged': True, 'table': table,
|
||||
'gd_hash': gd_hash, 'id': getattr(old, 'id', None), 'issues': issues}
|
||||
await sor.U(table, ns, dict(where))
|
||||
return {'written': True, 'unchanged': False, 'table': table, 'gd_hash': gd_hash,
|
||||
'id': getattr(old, 'id', None), 'issues': issues}
|
||||
|
||||
_set(ns, cols, ('id',), str(ctx.get('gd_id') or ('gd_%s' % sha256_text(
|
||||
canonical_dumps(where))[:24])))
|
||||
await sor.C(table, ns)
|
||||
return {'written': True, 'unchanged': False, 'table': table, 'gd_hash': gd_hash,
|
||||
'id': ns.get('id'), 'issues': issues}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 一站式：编译 + 导出 + 回写（供 api.py / .dspy 调用）
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def export_all(sor, snapshot, ctx, base_gd=None, version_id=None, issues=None):
|
||||
"""M3b 主流程：快照 → 规则 → 导出脚本 → 回写版本 → 写 GD。
|
||||
|
||||
fail-closed：存在 error 级 issue 时不落库，直接抛 CompileError。
|
||||
ctx: tenant_id*, blueprint_id*, blueprint_version*, known_actions, event_types,
|
||||
strict(默认 True), dry_run, force, compiler_version, status
|
||||
"""
|
||||
from .script_mapping import build_rules
|
||||
|
||||
issues = list(issues or [])
|
||||
ctx = dict(ctx or {})
|
||||
ctx.setdefault('strict', True)
|
||||
|
||||
rules, map_issues = build_rules(snapshot, ctx)
|
||||
issues.extend(map_issues)
|
||||
issues.extend(validate_rules(rules))
|
||||
|
||||
if has_errors(issues):
|
||||
errs = [i for i in issues if i['severity'] == 'error']
|
||||
raise CompileError('E_RULE_INVALID',
|
||||
'规则映射校验未通过（%d 个错误），拒绝导出' % len(errs),
|
||||
{'issues': errs[:50]})
|
||||
|
||||
export_result = await export_rules(sor, rules, ctx, issues)
|
||||
version_result = {'written': False}
|
||||
if version_id:
|
||||
version_result = await write_compiler_version_rules(
|
||||
sor, version_id, rules, ctx, export_result, issues)
|
||||
|
||||
gd, gd_text, gd_hash = build_game_definition(rules, ctx, base_gd, export_result)
|
||||
gd_result = {'written': False, 'skipped': True}
|
||||
if not ctx.get('skip_gd'):
|
||||
gd_result = await write_game_definition(sor, gd, gd_text, gd_hash, ctx, issues)
|
||||
|
||||
manifest = build_rules_manifest(rules)
|
||||
return {
|
||||
'status': 'ok',
|
||||
'tenant_id': ctx.get('tenant_id'),
|
||||
'blueprint_id': ctx.get('blueprint_id'),
|
||||
'blueprint_version': ctx.get('blueprint_version'),
|
||||
'rule_count': len(rules),
|
||||
'rules_hash': manifest['rules_hash'],
|
||||
'rules_manifest': manifest,
|
||||
'rules_json': canonical_rules_json(rules),
|
||||
'export': {k: v for k, v in export_result.items() if k != 'issues'},
|
||||
'version': {k: v for k, v in version_result.items() if k != 'issues'},
|
||||
'game_definition': {k: v for k, v in gd_result.items() if k != 'issues'},
|
||||
'gd_hash': gd_hash,
|
||||
'issues': issues,
|
||||
}
|
||||
|
||||
|
||||
async def load_exported_rules(sor, ctx, blueprint_version=None):
|
||||
"""读回已导出规则（用于确定性复验 / 版本 diff）。"""
|
||||
tenant_id = str((ctx or {}).get('tenant_id') or '').strip()
|
||||
if not tenant_id:
|
||||
raise CompileError('E_TENANT_REQUIRED', 'tenant_id 缺失，拒绝读取规则产物')
|
||||
table = await _detect_table(sor, _SCRIPT_TABLE_CANDIDATES)
|
||||
if not table:
|
||||
return []
|
||||
cols = await _table_columns(sor, table)
|
||||
content_col = _pick(cols, ('content', 'script_content', 'body', 'definition'))
|
||||
type_col = _pick(cols, ('script_type', 'type', 'stype'))
|
||||
ver_col = _pick(cols, ('blueprint_version', 'bp_version', 'source_version'))
|
||||
bp_col = _pick(cols, ('blueprint_id', 'bp_id', 'source_id'))
|
||||
if not content_col:
|
||||
return []
|
||||
|
||||
sql = 'SELECT * FROM `%s` WHERE tenant_id=${tenant_id}$' % table
|
||||
ns = {'tenant_id': tenant_id}
|
||||
if type_col:
|
||||
sql += ' AND `%s`=${stype}$' % type_col
|
||||
ns['stype'] = SCRIPT_TYPE_RULE
|
||||
if bp_col and ctx.get('blueprint_id'):
|
||||
sql += ' AND `%s`=${bp}$' % bp_col
|
||||
ns['bp'] = str(ctx['blueprint_id'])
|
||||
if ver_col and (blueprint_version or ctx.get('blueprint_version')):
|
||||
sql += ' AND `%s`=${ver}$' % ver_col
|
||||
ns['ver'] = str(blueprint_version or ctx['blueprint_version'])
|
||||
|
||||
recs = await sor.sqlExe(sql, ns)
|
||||
out = []
|
||||
for r in recs or []:
|
||||
raw = getattr(r, content_col, None)
|
||||
if not raw:
|
||||
continue
|
||||
try:
|
||||
out.append(json.loads(raw))
|
||||
except Exception:
|
||||
continue
|
||||
return sort_rules(out)
|
||||
|
||||
|
||||
async def rollback_export(sor, ctx, blueprint_version=None):
|
||||
"""删除某蓝图版本导出的全部规则脚本（重编译失败时清理，保持幂等）。"""
|
||||
tenant_id = str((ctx or {}).get('tenant_id') or '').strip()
|
||||
if not tenant_id:
|
||||
raise CompileError('E_TENANT_REQUIRED', 'tenant_id 缺失，拒绝回滚')
|
||||
table = await _detect_table(sor, _SCRIPT_TABLE_CANDIDATES)
|
||||
if not table:
|
||||
return {'deleted': 0}
|
||||
cols = await _table_columns(sor, table)
|
||||
type_col = _pick(cols, ('script_type', 'type', 'stype'))
|
||||
ver_col = _pick(cols, ('blueprint_version', 'bp_version', 'source_version'))
|
||||
bp_col = _pick(cols, ('blueprint_id', 'bp_id', 'source_id'))
|
||||
|
||||
sql = 'DELETE FROM `%s` WHERE tenant_id=${tenant_id}$' % table
|
||||
ns = {'tenant_id': tenant_id}
|
||||
if type_col:
|
||||
sql += ' AND `%s`=${stype}$' % type_col
|
||||
ns['stype'] = SCRIPT_TYPE_RULE
|
||||
if bp_col and ctx.get('blueprint_id'):
|
||||
sql += ' AND `%s`=${bp}$' % bp_col
|
||||
ns['bp'] = str(ctx['blueprint_id'])
|
||||
if ver_col and (blueprint_version or ctx.get('blueprint_version')):
|
||||
sql += ' AND `%s`=${ver}$' % ver_col
|
||||
ns['ver'] = str(blueprint_version or ctx['blueprint_version'])
|
||||
try:
|
||||
await sor.sqlExe(sql, ns)
|
||||
return {'deleted': 1, 'table': table}
|
||||
except Exception as e:
|
||||
raise CompileError('E_ROLLBACK', '回滚规则产物失败: %s' % e)
|
||||
1279
pbl_compiler/script_mapping.py
Normal file
1279
pbl_compiler/script_mapping.py
Normal file
File diff suppressed because it is too large
Load Diff
532
scripts/test_m3b_mapping.py
Normal file
532
scripts/test_m3b_mapping.py
Normal file
@ -0,0 +1,532 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
M3b 离线自测：第12章 event→condition→response → script_type=1 规则 JSON 映射 + 产物导出。
|
||||
|
||||
运行：cd modules/pbl_compiler && python3 scripts/test_m3b_mapping.py
|
||||
不依赖数据库 / ahserver —— 纯函数层 + FakeSor 模拟 sqlor 上下文验证导出幂等性。
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import asyncio
|
||||
|
||||
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
sys.path.insert(0, ROOT)
|
||||
|
||||
# 绕过包 __init__.py（其 import ahserver，离线环境不可用）：注册同名命名空间包
|
||||
import types # noqa: E402
|
||||
_pkg = types.ModuleType('pbl_compiler')
|
||||
_pkg.__path__ = [os.path.join(ROOT, 'pbl_compiler')]
|
||||
sys.modules['pbl_compiler'] = _pkg
|
||||
|
||||
|
||||
def run(coro):
|
||||
loop = asyncio.new_event_loop()
|
||||
try:
|
||||
return loop.run_until_complete(coro)
|
||||
finally:
|
||||
loop.close()
|
||||
|
||||
|
||||
from pbl_compiler.script_mapping import ( # noqa: E402
|
||||
CompileError, build_rules, canonical_dumps, canonical_rules_json, diff_rules,
|
||||
has_errors, normalize_condition, normalize_ecr, normalize_event,
|
||||
normalize_responses, rules_hash, sha256_text, sort_rules, strip_volatile,
|
||||
build_rules_manifest, validate_rules, iter_ecr_sources, OP_ARITY,
|
||||
)
|
||||
from pbl_compiler import exporter # noqa: E402
|
||||
|
||||
PASS, FAIL = [], []
|
||||
|
||||
|
||||
def check(name, cond, extra=''):
|
||||
(PASS if cond else FAIL).append(name)
|
||||
print((' [PASS] ' if cond else ' [FAIL] ') + name + ((' -> ' + str(extra)) if (extra and not cond) else ''))
|
||||
|
||||
|
||||
CTX = {'tenant_id': 'T1', 'blueprint_id': 'bp_demo', 'blueprint_version': 'v3',
|
||||
'strict': False}
|
||||
|
||||
SNAPSHOT = {
|
||||
'blueprint_id': 'bp_demo',
|
||||
'version': 'v3',
|
||||
'rules': [
|
||||
{
|
||||
'id': 'r_score_pass',
|
||||
'name': '得分达标开门',
|
||||
'description': '玩家得分>=60 且 已完成任务 时开启第二场景',
|
||||
'priority': 200,
|
||||
'event': {'type': 'onValueChange', 'source': 'player_1',
|
||||
'match': {'field': 'score'}},
|
||||
'condition': {'kind': 'all', 'items': [
|
||||
{'op': '>=', 'left': 'var:score', 'right': '60'},
|
||||
{'field': 'state.task_done', 'value': True},
|
||||
]},
|
||||
'responses': [
|
||||
{'action': 'goto_scene', 'params': {'scene': 'scene_2'}},
|
||||
{'action': 'add_score', 'target': 'self', 'params': {'amount': 10}},
|
||||
],
|
||||
'limits': {'cooldown_ms': 1000, 'max_triggers': 3},
|
||||
'tags': ['Main Flow', 'score'],
|
||||
'created_at': '2026-09-18 10:00:00',
|
||||
},
|
||||
{
|
||||
'name': '计时器提示',
|
||||
'priority': 50,
|
||||
'event': {'type': 'timer', 'interval_ms': 5000},
|
||||
'condition': 'remaining <= 30',
|
||||
'then': [{'action': 'show_message', 'params': {'msg': '时间不多了'}}],
|
||||
},
|
||||
],
|
||||
'scenes': [
|
||||
{'id': 'scene_1', 'events': [
|
||||
{'id': 'ev_enter', 'type': 'enter', 'source': 'scene_1',
|
||||
'responses': [{'action': 'play_bgm', 'params': {'track': 1}}]},
|
||||
]},
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def t_determinism():
|
||||
print('\n[T1] 确定性：同快照两次编译逐字节一致')
|
||||
r1, i1 = build_rules(SNAPSHOT, CTX)
|
||||
r2, i2 = build_rules(json.loads(json.dumps(SNAPSHOT)), dict(CTX))
|
||||
check('两次编译 rules_json 完全相同', canonical_rules_json(r1) == canonical_rules_json(r2))
|
||||
check('rules_hash 相同', rules_hash(r1) == rules_hash(r2))
|
||||
check('键序无关（打乱 dict 键序后 hash 不变）',
|
||||
rules_hash(r1) == rules_hash(json.loads(canonical_rules_json(r1))))
|
||||
# 易变字段不影响产物
|
||||
s2 = json.loads(json.dumps(SNAPSHOT))
|
||||
s2['rules'][0]['created_at'] = '2030-01-01 00:00:00'
|
||||
s2['rules'][0]['updated_by'] = 'someone'
|
||||
r3, _ = build_rules(s2, CTX)
|
||||
check('created_at/updated_by 变更不影响 rules_hash', rules_hash(r1) == rules_hash(r3))
|
||||
# 零空白
|
||||
txt = canonical_rules_json(r1)
|
||||
check('产物零空白（无 ", " / ": "）', ', ' not in txt and ': ' not in txt)
|
||||
check('产物无换行', '\n' not in txt)
|
||||
|
||||
|
||||
def t_mapping():
|
||||
print('\n[T2] 映射正确性：event / condition / response 归一')
|
||||
rules, issues = build_rules(SNAPSHOT, CTX)
|
||||
check('共映射 3 条规则（rules 2 + scenes.events 1）', len(rules) == 3, len(rules))
|
||||
by_id = {r['rule_id']: r for r in rules}
|
||||
r = by_id['r_score_pass']
|
||||
check('event.type 归一 onValueChange -> change', r['event']['type'] == 'change', r['event'])
|
||||
check('event.source 保留', r['event']['source'] == 'player_1')
|
||||
check('event.match 键归一 snake', r['event']['match'] == {'field': 'score'}, r['event']['match'])
|
||||
check('condition 为 all 组合子', r['condition']['kind'] == 'all')
|
||||
ops = [c['op'] for c in r['condition']['items']]
|
||||
check('操作符归一 >= -> gte, 隐式 == -> eq', ops == ['gte', 'eq'], ops)
|
||||
check('左操作数 var:score -> {kind:var,name:score}',
|
||||
r['condition']['items'][0]['left'] == {'kind': 'var', 'name': 'score'})
|
||||
check("右操作数 '60' -> const 60（数字归一）",
|
||||
r['condition']['items'][0]['right'] == {'kind': 'const', 'value': 60})
|
||||
check('嵌套 var 路径 state.task_done -> var',
|
||||
r['condition']['items'][1]['left'] == {'kind': 'var', 'name': 'state.task_done'})
|
||||
check('responses 2 个动作', len(r['responses']) == 2)
|
||||
check('params 同义键归一 scene -> scene_id',
|
||||
r['responses'][0]['params'].get('scene_id') == 'scene_2', r['responses'][0]['params'])
|
||||
check('params 同义键归一 amount -> value',
|
||||
r['responses'][1]['params'].get('value') == 10, r['responses'][1]['params'])
|
||||
check("target 'self' -> kind self", r['responses'][1]['target'] == {'kind': 'self', 'id': ''})
|
||||
check('limits 保留', r['limits'] == {'cooldown_ms': 1000, 'max_triggers': 3})
|
||||
check('tags 归一 snake + 排序', r['tags'] == ['main_flow', 'score'], r['tags'])
|
||||
check('origin 可追溯', r['origin']['blueprint_version'] == 'v3'
|
||||
and r['origin']['sub_object_id'] == 'r_score_pass')
|
||||
check('schema_version = 1.0', r['schema_version'] == '1.0')
|
||||
|
||||
# 无 id 的规则 → 内容哈希派生 rule_id
|
||||
timer = [x for x in rules if x['event']['type'] == 'timer'][0]
|
||||
check('无 id 规则派生 rule_ 前缀哈希 id', timer['rule_id'].startswith('rule_'), timer['rule_id'])
|
||||
check('timer interval_ms 提升到 match', timer['event']['match'].get('interval_ms') == 5000)
|
||||
check('字符串条件 "remaining <= 30" 解析为 compare lte',
|
||||
timer['condition'] == {'kind': 'compare', 'op': 'lte',
|
||||
'left': {'kind': 'var', 'name': 'remaining'},
|
||||
'right': {'kind': 'const', 'value': 30}}, timer['condition'])
|
||||
check("'then' 别名识别为 responses", len(timer['responses']) == 1)
|
||||
|
||||
# 排序：priority 降序
|
||||
prios = [x['priority'] for x in rules]
|
||||
check('规则按 priority 降序稳定排序', prios == sorted(prios, reverse=True), prios)
|
||||
|
||||
|
||||
def t_shapes():
|
||||
print('\n[T3] 多形态兼容 + 边界')
|
||||
# 形态 B：events 顶层
|
||||
snap_b = {'events': [{'id': 'e1', 'type': 'click', 'source': 'btn',
|
||||
'conditions': [{'field': 'hp', 'op': '<', 'value': 1}],
|
||||
'actions': [{'action': 'game_over'}]}]}
|
||||
rb, ib = build_rules(snap_b, CTX)
|
||||
check('形态B events 顶层可映射', len(rb) == 1 and rb[0]['event']['type'] == 'click', rb)
|
||||
check('形态B conditions 别名识别', rb[0]['condition']['op'] == 'lt')
|
||||
check('形态B 无 error issue', not has_errors(ib), ib)
|
||||
|
||||
# 形态 A 单 dict（非数组）
|
||||
snap_c = {'rule': {'id': 'r1', 'event': {'type': 'start'},
|
||||
'responses': [{'action': 'init_world'}]}}
|
||||
rc, ic = build_rules(snap_c, CTX)
|
||||
check('单 dict rule 也可映射', len(rc) == 1, rc)
|
||||
|
||||
# 无条件 → always
|
||||
snap_d = {'rules': [{'id': 'r2', 'event': {'type': 'enter'},
|
||||
'responses': [{'action': 'log'}]}]}
|
||||
rd, _ = build_rules(snap_d, CTX)
|
||||
check('缺省条件归一为 always', rd[0]['condition'] == {'kind': 'always'})
|
||||
|
||||
# 空快照
|
||||
try:
|
||||
build_rules({}, CTX)
|
||||
check('空快照抛 CompileError', False)
|
||||
except CompileError as e:
|
||||
check('空快照抛 CompileError(E_SNAPSHOT_EMPTY)', e.code == 'E_SNAPSHOT_EMPTY', e.code)
|
||||
|
||||
# tenant 缺失
|
||||
try:
|
||||
build_rules(SNAPSHOT, {'blueprint_id': 'b', 'blueprint_version': 'v'})
|
||||
check('tenant_id 缺失抛错', False)
|
||||
except CompileError as e:
|
||||
check('tenant_id 缺失 fail-closed', e.code == 'E_TENANT_REQUIRED', e.code)
|
||||
|
||||
# 无 ECR
|
||||
r_none, i_none = build_rules({'scenes': [{'id': 's'}]}, CTX)
|
||||
check('无 ECR 记录 → 0 规则 + W_NO_ECR', r_none == [] and
|
||||
any(i['code'] == 'W_NO_ECR' for i in i_none))
|
||||
|
||||
|
||||
def t_failclosed():
|
||||
print('\n[T4] fail-closed：strict 模式未登记 action / 缺 event / 缺 response')
|
||||
strict_ctx = dict(CTX, strict=True, known_actions={'goto_scene', 'add_score'})
|
||||
snap = {'rules': [
|
||||
{'id': 'ok', 'event': {'type': 'click'}, 'responses': [{'action': 'goto_scene'}]},
|
||||
{'id': 'bad_action', 'event': {'type': 'click'},
|
||||
'responses': [{'action': 'launch_missile'}]},
|
||||
{'id': 'no_event', 'responses': [{'action': 'goto_scene'}]},
|
||||
{'id': 'no_resp', 'event': {'type': 'click'}},
|
||||
]}
|
||||
rules, issues = build_rules(snap, strict_ctx)
|
||||
codes = {i['code'] for i in issues}
|
||||
check('未登记 action → E_CAP_UNKNOWN', 'E_CAP_UNKNOWN' in codes, codes)
|
||||
check('缺 event → E_EVENT_MISSING', 'E_EVENT_MISSING' in codes)
|
||||
check('缺 response → E_RESP_MISSING', 'E_RESP_MISSING' in codes)
|
||||
check('strict 下 bad_action 规则被剔除', 'bad_action' not in {r['source_id'] for r in rules})
|
||||
check('存在 error issue', has_errors(issues))
|
||||
check('无效规则被剔除并记 E_RULE_DROPPED',
|
||||
any(i['code'] == 'E_RULE_DROPPED' for i in issues))
|
||||
check('no_event / no_resp 规则不进入产物',
|
||||
not ({'no_event', 'no_resp'} & {r['source_id'] for r in rules}))
|
||||
check('存活规则自身通过 validate_rules（无半成品）',
|
||||
not has_errors(validate_rules(rules)), validate_rules(rules))
|
||||
# 非 strict：降级 warning，规则保留
|
||||
rules2, issues2 = build_rules(snap, dict(strict_ctx, strict=False))
|
||||
check('非 strict 降级为 W_CAP_UNKNOWN 且保留规则',
|
||||
any(i['code'] == 'W_CAP_UNKNOWN' for i in issues2)
|
||||
and 'bad_action' in {r['source_id'] for r in rules2})
|
||||
# 未知操作符
|
||||
_, i3 = build_rules({'rules': [{'id': 'x', 'event': {'type': 'click'},
|
||||
'condition': {'op': '~=', 'left': 'a', 'right': 1},
|
||||
'responses': [{'action': 'log'}]}]}, CTX)
|
||||
check('未知操作符 → E_COND_OP_UNKNOWN', any(i['code'] == 'E_COND_OP_UNKNOWN' for i in i3))
|
||||
# between 缺边界
|
||||
_, i4 = build_rules({'rules': [{'id': 'y', 'event': {'type': 'click'},
|
||||
'condition': {'op': 'between', 'left': 'a', 'right': [1]},
|
||||
'responses': [{'action': 'log'}]}]}, CTX)
|
||||
check('between 边界不足 → E_COND_BETWEEN', any(i['code'] == 'E_COND_BETWEEN' for i in i4))
|
||||
|
||||
|
||||
def t_manifest_diff():
|
||||
print('\n[T5] manifest / diff / 幂等指纹')
|
||||
rules, _ = build_rules(SNAPSHOT, CTX)
|
||||
m = build_rules_manifest(rules)
|
||||
check('manifest rule_count 正确', m['rule_count'] == len(rules))
|
||||
check('manifest script_type=1', m['script_type'] == 1)
|
||||
check('manifest rules_hash 与 rules_hash 一致', m['rules_hash'] == rules_hash(rules))
|
||||
check('manifest 不含时间戳字段',
|
||||
not any(k in canonical_dumps(m) for k in ('created_at', 'timestamp')))
|
||||
check('manifest 每条含 content_hash', all(e.get('content_hash') for e in m['rules']))
|
||||
|
||||
rules2, _ = build_rules(SNAPSHOT, CTX)
|
||||
d_same = diff_rules(rules, rules2)
|
||||
check('相同规则 diff 全空', d_same['added'] == [] and d_same['removed'] == []
|
||||
and d_same['changed'] == [], d_same)
|
||||
mod = json.loads(canonical_rules_json(rules))
|
||||
mod[0]['priority'] = 999
|
||||
d = diff_rules(rules, mod)
|
||||
check('修改 priority → changed 命中', d['changed_count'] == 1, d)
|
||||
d2 = diff_rules(rules, mod[:1])
|
||||
check('删规则 → removed 命中', d2['removed_count'] == len(rules) - 1, d2)
|
||||
|
||||
|
||||
def t_strip():
|
||||
print('\n[T6] strip_volatile / 操作数归一')
|
||||
o = {'a': 1, 'created_at': 'x', 'nested': {'updated_by': 'y', 'b': [1, {'ts': 2, 'c': 3}]}}
|
||||
s = strip_volatile(o)
|
||||
check('顶层易变键剥离', 'created_at' not in s)
|
||||
check('嵌套易变键剥离', 'updated_by' not in s['nested'] and 'ts' not in s['nested']['b'][1])
|
||||
check('正常键保留', s['a'] == 1 and s['nested']['b'][1]['c'] == 3)
|
||||
iss = []
|
||||
from pbl_compiler.script_mapping import normalize_operand
|
||||
check("'$hp' -> var", normalize_operand('$hp', iss, 'p') == {'kind': 'var', 'name': 'hp'})
|
||||
check("'{{score}}' -> var", normalize_operand('{{score}}', iss, 'p') == {'kind': 'var', 'name': 'score'})
|
||||
check("'3.5' -> const float", normalize_operand('3.5', iss, 'p') == {'kind': 'const', 'value': 3.5})
|
||||
check("'true' -> const bool", normalize_operand('true', iss, 'p') == {'kind': 'const', 'value': True})
|
||||
check("[1,2] -> list", normalize_operand([1, 2], iss, 'p')['kind'] == 'list')
|
||||
check("{'kind':'entity_attr'} 保留", normalize_operand(
|
||||
{'kind': 'entity_attr', 'entity': 'e1', 'attr': 'hp'}, iss, 'p') ==
|
||||
{'kind': 'entity_attr', 'entity': 'e1', 'attr': 'hp'})
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# FakeSor：模拟 sqlor 上下文，验证 exporter 幂等 + tenant 打头
|
||||
# ---------------------------------------------------------------------------
|
||||
class Row(object):
|
||||
def __init__(self, **kw):
|
||||
self.__dict__.update(kw)
|
||||
|
||||
|
||||
class FakeSor(object):
|
||||
def __init__(self, tables=None):
|
||||
self.tables = tables or {}
|
||||
self.sqls = []
|
||||
self.inserts = []
|
||||
self.updates = []
|
||||
|
||||
async def sqlExe(self, sql, ns=None):
|
||||
self.sqls.append((sql, dict(ns or {})))
|
||||
low = sql.strip().lower()
|
||||
if low.startswith('select 1 from'):
|
||||
t = sql.split('`')[1]
|
||||
if t not in self.tables:
|
||||
raise Exception('no table %s' % t)
|
||||
return []
|
||||
if low.startswith('select column_name'):
|
||||
t = (ns or {}).get('tbl')
|
||||
return [Row(COLUMN_NAME=c) for c in sorted(self.tables.get(t, {}).get('cols', []))]
|
||||
if low.startswith('select * from'):
|
||||
t = sql.split('`')[1]
|
||||
return [Row(**r) for r in self.tables.get(t, {}).get('rows', [])]
|
||||
if low.startswith('delete from'):
|
||||
t = sql.split('`')[1]
|
||||
n = len(self.tables.get(t, {}).get('rows', []))
|
||||
self.tables.setdefault(t, {'cols': [], 'rows': []})['rows'] = []
|
||||
return [Row(deleted=n)]
|
||||
return []
|
||||
|
||||
async def R(self, table, where):
|
||||
rows = self.tables.get(table, {}).get('rows', [])
|
||||
hit = [r for r in rows if all(str(r.get(k)) == str(v) for k, v in where.items())]
|
||||
return [Row(**r) for r in hit]
|
||||
|
||||
async def C(self, table, ns):
|
||||
row = dict(ns)
|
||||
row.setdefault('id', 'id_%d' % (len(self.inserts) + 1))
|
||||
self.tables.setdefault(table, {'cols': [], 'rows': []})['rows'].append(row)
|
||||
self.inserts.append((table, dict(ns)))
|
||||
return row
|
||||
|
||||
async def U(self, table, ns, where):
|
||||
self.updates.append((table, dict(ns), dict(where)))
|
||||
rows = self.tables.get(table, {}).get('rows', [])
|
||||
for r in rows:
|
||||
if all(str(r.get(k)) == str(v) for k, v in where.items()):
|
||||
r.update(ns)
|
||||
return 1
|
||||
|
||||
async def D(self, table, where):
|
||||
return await self.sqlExe('DELETE FROM `%s` WHERE 1' % table, where)
|
||||
|
||||
|
||||
SCRIPT_COLS = ['id', 'tenant_id', 'script_code', 'name', 'script_type', 'content',
|
||||
'description', 'group_name', 'enabled', 'content_hash', 'origin',
|
||||
'blueprint_id', 'blueprint_version', 'created_at']
|
||||
GD_COLS = ['id', 'tenant_id', 'blueprint_id', 'blueprint_version', 'game_definition',
|
||||
'gd_hash', 'rules_hash', 'rule_count', 'status', 'compiler_version']
|
||||
VER_COLS = ['id', 'tenant_id', 'blueprint_id', 'blueprint_version', 'rules_json',
|
||||
'rules_hash', 'rules_manifest', 'rules_count', 'script_ids']
|
||||
|
||||
|
||||
def make_sor():
|
||||
return FakeSor({
|
||||
'script': {'cols': SCRIPT_COLS, 'rows': []},
|
||||
'pbl_game_definition': {'cols': GD_COLS, 'rows': []},
|
||||
'pbl_compiler_version': {'cols': VER_COLS, 'rows': []},
|
||||
})
|
||||
|
||||
|
||||
def t_export():
|
||||
print('\n[T7] 导出：script_type=1 落库 + 幂等 + tenant 打头')
|
||||
rules, _ = build_rules(SNAPSHOT, CTX)
|
||||
sor = make_sor()
|
||||
res = run(
|
||||
exporter.export_rules(sor, rules, CTX))
|
||||
rows = sor.tables['script']['rows']
|
||||
check('导出行数 == 规则数', len(rows) == len(rules), (len(rows), len(rules)))
|
||||
check('全部 script_type=1', all(r['script_type'] == 1 for r in rows))
|
||||
check('全部带 tenant_id', all(r['tenant_id'] == 'T1' for r in rows))
|
||||
check('script_code 唯一', len({r['script_code'] for r in rows}) == len(rows))
|
||||
check('content 是确定性 JSON（sort_keys 零空白）',
|
||||
all(r['content'] == canonical_dumps(json.loads(r['content'])) for r in rows))
|
||||
check('content_hash 与内容一致',
|
||||
all(r['content_hash'] == sha256_text(r['content']) for r in rows))
|
||||
check('rules_hash 与本地一致', res['rules_hash'] == rules_hash(rules))
|
||||
|
||||
# 幂等：再导一次 → 0 insert / 0 update
|
||||
n_ins = len(sor.inserts)
|
||||
res2 = run(
|
||||
exporter.export_rules(sor, rules, CTX))
|
||||
check('重复导出 skipped == 规则数', res2['skipped'] == len(rules), res2['skipped'])
|
||||
check('重复导出 exported == 0', res2['exported'] == 0, res2['exported'])
|
||||
check('重复导出无新增 insert', len(sor.inserts) == n_ins)
|
||||
check('重复导出无 update', len(sor.updates) == 0)
|
||||
check('表行数未增长', len(sor.tables['script']['rows']) == len(rules))
|
||||
|
||||
# 内容变化 → update 而非 insert
|
||||
mod = json.loads(canonical_rules_json(rules))
|
||||
mod[0]['priority'] = 777
|
||||
res3 = run(
|
||||
exporter.export_rules(sor, mod, CTX))
|
||||
check('内容变更走 update', res3['exported'] == 1 and len(sor.updates) == 1,
|
||||
(res3['exported'], len(sor.updates)))
|
||||
check('内容变更不新增行', len(sor.tables['script']['rows']) == len(rules))
|
||||
|
||||
# 读回一致
|
||||
back = run(
|
||||
exporter.load_exported_rules(sor, CTX))
|
||||
check('load_exported_rules 读回条数一致', len(back) == len(rules), len(back))
|
||||
|
||||
# tenant 缺失 fail-closed
|
||||
try:
|
||||
run(
|
||||
exporter.export_rules(sor, rules, {'blueprint_version': 'v3'}))
|
||||
check('导出缺 tenant 抛错', False)
|
||||
except CompileError as e:
|
||||
check('导出缺 tenant fail-closed', e.code == 'E_TENANT_REQUIRED', e.code)
|
||||
|
||||
# dry_run 不落库
|
||||
sor_d = make_sor()
|
||||
res_d = run(
|
||||
exporter.export_rules(sor_d, rules, dict(CTX, dry_run=True)))
|
||||
check('dry_run 不落库', len(sor_d.tables['script']['rows']) == 0
|
||||
and res_d['skipped'] == len(rules))
|
||||
|
||||
# 所有 SQL 都带 tenant_id
|
||||
check('所有 sqlExe 参数含 tenant_id 或为元数据探测',
|
||||
all(('tenant_id' in (ns or {})) or 'information_schema' in sql.lower()
|
||||
or sql.strip().lower().startswith('select 1 from')
|
||||
or sql.strip().lower().startswith('select * from')
|
||||
for sql, ns in sor.sqls), [s for s, n in sor.sqls if 'tenant_id' not in (n or {})][:3])
|
||||
|
||||
|
||||
def t_gd():
|
||||
print('\n[T8] game_definition 写入 + 版本回写 + export_all 一站式')
|
||||
rules, _ = build_rules(SNAPSHOT, CTX)
|
||||
base_gd = {'schema_version': '1.0', 'meta': {'title': 'Demo'}, 'scenes': [],
|
||||
'entities': [], 'variables': [], 'ui': [], 'audio': [],
|
||||
'scoring': {}, 'flow': {}, 'i18n': {}}
|
||||
gd, gd_text, gd_hash = exporter.build_game_definition(rules, CTX, base_gd)
|
||||
check('GD 含 rules 顶层键', 'rules' in gd)
|
||||
check('GD 保留 M3a 其它顶层键', 'scenes' in gd and 'meta' in gd)
|
||||
check('rules.script_type == 1', gd['rules']['script_type'] == 1)
|
||||
check('rules.rule_count 正确', gd['rules']['rule_count'] == len(rules))
|
||||
check('rules.rules_hash 与 rules_hash 一致', gd['rules']['rules_hash'] == rules_hash(rules))
|
||||
check('GD 文本确定性（重算一致）',
|
||||
exporter.build_game_definition(json.loads(canonical_rules_json(rules)), CTX, base_gd)[2] == gd_hash)
|
||||
check('GD 文本零空白', ', ' not in gd_text and '\n' not in gd_text)
|
||||
|
||||
sor = make_sor()
|
||||
r1 = run(
|
||||
exporter.write_game_definition(sor, gd, gd_text, gd_hash, CTX))
|
||||
check('首次写入 GD written=True', r1['written'] is True, r1)
|
||||
check('GD 表 1 行', len(sor.tables['pbl_game_definition']['rows']) == 1)
|
||||
r2 = run(
|
||||
exporter.write_game_definition(sor, gd, gd_text, gd_hash, CTX))
|
||||
check('同 hash 重复写入 unchanged=True（幂等）',
|
||||
r2.get('unchanged') is True and r2['written'] is False, r2)
|
||||
check('GD 表仍 1 行', len(sor.tables['pbl_game_definition']['rows']) == 1)
|
||||
gd2 = json.loads(gd_text)
|
||||
gd2['rules']['rules'][0]['priority'] = 1
|
||||
t2 = canonical_dumps(gd2)
|
||||
r3 = run(
|
||||
exporter.write_game_definition(sor, gd2, t2, sha256_text(t2), CTX))
|
||||
check('hash 变化 → 覆盖更新不新增行',
|
||||
r3['written'] is True and len(sor.tables['pbl_game_definition']['rows']) == 1)
|
||||
|
||||
# 版本回写
|
||||
exp_res = run(exporter.export_rules(sor, rules, CTX))
|
||||
vr = run(
|
||||
exporter.write_compiler_version_rules(sor, 'ver_001', rules, CTX, exp_res))
|
||||
check('版本回写 written=True', vr['written'] is True, vr)
|
||||
vrow = sor.tables['pbl_compiler_version']['rows'][0]
|
||||
check('版本行 rules_hash 正确', vrow['rules_hash'] == rules_hash(rules))
|
||||
check('版本行 rules_count 正确', vrow['rules_count'] == len(rules))
|
||||
check('版本行 rules_json 可解析且确定性',
|
||||
json.loads(vrow['rules_json']) is not None
|
||||
and vrow['rules_json'] == canonical_rules_json(rules))
|
||||
man = json.loads(vrow['rules_manifest'])
|
||||
check('版本行 manifest 含 script_ids', len(man['extra']['script_ids']) == len(rules))
|
||||
|
||||
# export_all 一站式（含 error → 拒绝落库）
|
||||
sor_a = make_sor()
|
||||
out = run(
|
||||
exporter.export_all(sor_a, SNAPSHOT, dict(CTX, strict=False),
|
||||
base_gd=base_gd, version_id='ver_002'))
|
||||
check('export_all status=ok', out['status'] == 'ok', out.get('issues'))
|
||||
check('export_all rule_count 正确', out['rule_count'] == len(rules))
|
||||
check('export_all 落库脚本', len(sor_a.tables['script']['rows']) == len(rules))
|
||||
check('export_all 写 GD', len(sor_a.tables['pbl_game_definition']['rows']) == 1)
|
||||
check('export_all 回写版本', len(sor_a.tables['pbl_compiler_version']['rows']) == 1)
|
||||
|
||||
bad_snap = {'rules': [{'id': 'b1', 'responses': [{'action': 'x'}]}]} # 缺 event
|
||||
sor_b = make_sor()
|
||||
try:
|
||||
run(
|
||||
exporter.export_all(sor_b, bad_snap, CTX, base_gd=base_gd))
|
||||
check('export_all 有 error 时拒绝落库', False)
|
||||
except CompileError as e:
|
||||
check('export_all fail-closed(E_RULE_INVALID)', e.code == 'E_RULE_INVALID', e.code)
|
||||
check('失败时脚本表为空（无半成品）', len(sor_b.tables['script']['rows']) == 0)
|
||||
check('失败时 GD 表为空', len(sor_b.tables['pbl_game_definition']['rows']) == 0)
|
||||
|
||||
# rollback
|
||||
rb = run(
|
||||
exporter.rollback_export(sor_a, CTX))
|
||||
check('rollback 清空规则脚本', len(sor_a.tables['script']['rows']) == 0, rb)
|
||||
|
||||
|
||||
def t_ecr_enum():
|
||||
print('\n[T9] ECR 源枚举去重 + 路径可追溯')
|
||||
snap = {'rules': [{'id': 'r1', 'event': {'type': 'click'}, 'responses': [{'action': 'a'}]}],
|
||||
'logic': [{'id': 'r1', 'event': {'type': 'click'}, 'responses': [{'action': 'a'}]}]}
|
||||
srcs = iter_ecr_sources(snap)
|
||||
check('同内容跨键重复只保留一处', len(srcs) == 1, [p for p, _ in srcs])
|
||||
check('source_path 可追溯', srcs[0][0].startswith('snapshot.'), srcs[0][0])
|
||||
rules, issues = build_rules(snap, CTX)
|
||||
check('去重后无 E_RULE_ID_DUP', not any(i['code'] == 'E_RULE_ID_DUP' for i in issues), issues)
|
||||
# 真重复 id 但内容不同 → 报 dup 并派生新 id
|
||||
snap2 = {'rules': [
|
||||
{'id': 'dup', 'event': {'type': 'click'}, 'responses': [{'action': 'a'}]},
|
||||
{'id': 'dup', 'event': {'type': 'enter'}, 'responses': [{'action': 'b'}]}]}
|
||||
r2, i2 = build_rules(snap2, CTX)
|
||||
check('内容不同的重复 id → E_RULE_ID_DUP', any(i['code'] == 'E_RULE_ID_DUP' for i in i2))
|
||||
check('重复 id 派生唯一 rule_id', len({r['rule_id'] for r in r2}) == 2)
|
||||
|
||||
|
||||
def main():
|
||||
print('=' * 72)
|
||||
print('M3b 自测：script_engine 规则映射与导出（第12章 ECR → script_type=1）')
|
||||
print('=' * 72)
|
||||
for fn in (t_determinism, t_mapping, t_shapes, t_failclosed, t_manifest_diff,
|
||||
t_strip, t_export, t_gd, t_ecr_enum):
|
||||
fn()
|
||||
print('\n' + '=' * 72)
|
||||
print('PASS: %d FAIL: %d' % (len(PASS), len(FAIL)))
|
||||
if FAIL:
|
||||
print('失败项: ' + ', '.join(FAIL))
|
||||
print('=' * 72)
|
||||
return 1 if FAIL else 0
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
sys.exit(main())
|
||||
Loading…
x
Reference in New Issue
Block a user