diff --git a/models/pbl_game_definition.json b/models/pbl_game_definition.json index b086a86..0cf7564 100644 --- a/models/pbl_game_definition.json +++ b/models/pbl_game_definition.json @@ -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": [] -} +} \ No newline at end of file diff --git a/models/pbl_script_rule.json b/models/pbl_script_rule.json new file mode 100644 index 0000000..3794d6d --- /dev/null +++ b/models/pbl_script_rule.json @@ -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": [] +} diff --git a/pbl_compiler/exporter.py b/pbl_compiler/exporter.py new file mode 100644 index 0000000..ee27868 --- /dev/null +++ b/pbl_compiler/exporter.py @@ -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) diff --git a/pbl_compiler/script_mapping.py b/pbl_compiler/script_mapping.py new file mode 100644 index 0000000..cb9625b --- /dev/null +++ b/pbl_compiler/script_mapping.py @@ -0,0 +1,1279 @@ +# -*- coding: utf-8 -*- +""" +pbl_compiler.script_mapping —— M3b 第12章：event → condition → response (ECR) 到 +script_engine `script_type=1` 规则 JSON 的确定性映射。 + +设计铁律（与 M3a canonical.py 一致）： + 1. **确定性**：同一蓝图版本快照 → 逐字节相同的规则 JSON。 + - `json.dumps(..., sort_keys=True, separators=(',', ':'), ensure_ascii=False)`（零空白） + - 剥离易变字段（created_at / uuid / timestamp ...） + - 顶层规则按 (-priority, rule_id) 稳定排序；rule_id 缺失时由内容哈希派生 + 2. **fail-closed**：tenant_id 缺失、未知 action（strict 模式）、操作数缺失 → 记 issue 并阻断导出， + 绝不静默产出半成品规则。 + 3. **纯函数层**：本文件不做任何 DB / IO，全部可离线单测（scripts/test_m3b_mapping.py）。 + 落库与产物导出在 exporter.py。 + +规则 JSON（script_type=1）契约 v1.0： +{ + "schema_version": "1.0", + "rule_id": "...", # 稳定主键，来源 id 或内容哈希派生 + "source_id": "...", # 蓝图子对象原始 id（可空） + "name": "...", + "description": "...", + "priority": 100, # 大者先执行 + "enabled": true, + "event": {"kind": "event", "type": "...", "source": "...", "match": {...}}, + "condition": {"kind": "all|any|not|compare|expr|always", ...}, + "responses": [{"kind": "action", "action": "...", "target": {...}, "params": {...}, + "delay_ms": 0, "repeat": 1}], + "limits": {"cooldown_ms": 0, "max_triggers": 0}, # 0 = 不限 + "tags": ["..."], + "origin": {"blueprint_id": "...", "blueprint_version": "...", + "sub_object_id": "...", "source_index": 0} +} +""" + +import hashlib +import json +import re + +__all__ = [ + 'RULE_SCHEMA_VERSION', + 'CANONICAL_OPS', + 'OP_ALIASES', + 'VOLATILE_KEYS', + 'CompileError', + 'Issue', + 'canonical_dumps', + 'sha256_text', + 'strip_volatile', + 'normalize_operand', + 'normalize_condition', + 'normalize_event', + 'normalize_responses', + 'normalize_ecr', + 'iter_ecr_sources', + 'build_rules', + 'validate_rule', + 'validate_rules', + 'sort_rules', + 'canonical_rules_json', + 'rules_hash', + 'build_rules_manifest', + 'diff_rules', +] + +RULE_SCHEMA_VERSION = '1.0' + +# --------------------------------------------------------------------------- +# 错误与 issue +# --------------------------------------------------------------------------- + + +class CompileError(Exception): + """编译期硬错误（fail-closed）。code 用于前端/日志定位。""" + + def __init__(self, code, message, detail=None): + super().__init__('[%s] %s' % (code, message)) + self.code = code + self.message = message + self.detail = detail or {} + + +def Issue(severity, code, message, rule_id='', path=''): + """统一 issue 结构（dict，便于 JSON 序列化）。severity: error|warning|info""" + return { + 'severity': severity, + 'code': code, + 'message': message, + 'rule_id': rule_id or '', + 'path': path or '', + } + + +# --------------------------------------------------------------------------- +# 确定性序列化 / 哈希 / 易变字段剥离 +# --------------------------------------------------------------------------- + +VOLATILE_KEYS = frozenset([ + 'created_at', 'updated_at', 'create_time', 'update_time', 'created_by', 'updated_by', + 'creator', 'modifier', 'timestamp', 'ts', 'uuid', '_id', 'trace_id', 'request_id', + 'generated_at', 'expire_at', 'deleted_at', 'revision', 'etag', +]) + + +def canonical_dumps(obj): + """确定性 JSON：sort_keys + 零空白 + 非 ASCII 原样。优先复用 M3a canonical 实现。""" + try: # pragma: no cover - 依赖同包 canonical.py（M3a 产物） + from . import canonical as _c + for fn in ('canonical_dumps', 'dumps_canonical', 'canonical_json'): + f = getattr(_c, fn, None) + if callable(f): + return f(obj) + except Exception: + pass + return json.dumps(obj, ensure_ascii=False, sort_keys=True, separators=(',', ':')) + + +def sha256_text(text): + if isinstance(text, str): + text = text.encode('utf-8') + return hashlib.sha256(text).hexdigest() + + +def strip_volatile(obj, depth=0): + """递归剥离易变字段；list 保序（顺序是语义的一部分），dict 只删键不改值语义。""" + if depth > 32: + return obj + if isinstance(obj, dict): + out = {} + for k, v in obj.items(): + ks = str(k) + if ks in VOLATILE_KEYS or ks.startswith('_volatile'): + continue + out[ks] = strip_volatile(v, depth + 1) + return out + if isinstance(obj, (list, tuple)): + return [strip_volatile(v, depth + 1) for v in obj] + return obj + + +# --------------------------------------------------------------------------- +# 标量工具 +# --------------------------------------------------------------------------- + +_SNAKE_RE_1 = re.compile(r'(.)([A-Z][a-z]+)') +_SNAKE_RE_2 = re.compile(r'([a-z0-9])([A-Z])') +_NON_WORD_RE = re.compile(r'[^0-9a-zA-Z\u4e00-\u9fa5]+') + + +def to_snake(name): + """`onEnter Scene` / `OnEnterScene` → `on_enter_scene`（确定性归一）。""" + if name is None: + return '' + s = str(name).strip() + if not s: + return '' + s = _SNAKE_RE_1.sub(r'\1_\2', s) + s = _SNAKE_RE_2.sub(r'\1_\2', s) + s = _NON_WORD_RE.sub('_', s).strip('_') + s = re.sub(r'_+', '_', s) + return s.lower() + + +def first_str(d, keys, default=''): + for k in keys: + if isinstance(d, dict) and d.get(k) not in (None, ''): + return str(d.get(k)).strip() + return default + + +def to_int(v, default=0): + try: + if v is None or v == '': + return int(default) + if isinstance(v, bool): + return int(v) + return int(float(v)) + except (TypeError, ValueError): + return int(default) + + +def to_bool(v, default=True): + if v is None: + return bool(default) + if isinstance(v, bool): + return v + if isinstance(v, (int, float)): + return v != 0 + s = str(v).strip().lower() + if s in ('1', 'true', 'yes', 'y', 'on', 'enabled', 'active', '是'): + return True + if s in ('0', 'false', 'no', 'n', 'off', 'disabled', 'inactive', '否'): + return False + return bool(default) + + +def to_number_or_raw(v): + """常量值归一：数字字符串转数字（保证 1 与 '1' 编译结果一致），其余原样。""" + if isinstance(v, bool) or v is None: + return v + if isinstance(v, (int, float)): + return v + if isinstance(v, str): + s = v.strip() + if re.fullmatch(r'-?\d+', s): + return int(s) + if re.fullmatch(r'-?\d+\.\d+', s): + return float(s) + if s.lower() in ('true', 'false'): + return s.lower() == 'true' + if s.lower() in ('null', 'none'): + return None + return v + return v + + +# --------------------------------------------------------------------------- +# 操作符白名单（script_engine script_type=1 求值器契约） +# --------------------------------------------------------------------------- + +OP_ALIASES = { + '==': 'eq', '=': 'eq', 'eq': 'eq', 'equal': 'eq', 'equals': 'eq', 'is': 'eq', '相同': 'eq', + '!=': 'ne', '<>': 'ne', 'ne': 'ne', 'noteq': 'ne', 'not_equal': 'ne', 'is_not': 'ne', + '>': 'gt', 'gt': 'gt', 'greater_than': 'gt', '大于': 'gt', + '>=': 'gte', '=>': 'gte', 'gte': 'gte', 'ge': 'gte', 'greater_or_equal': 'gte', '大于等于': 'gte', + '<': 'lt', 'lt': 'lt', 'less_than': 'lt', '小于': 'lt', + '<=': 'lte', '=<': 'lte', 'lte': 'lte', 'le': 'lte', 'less_or_equal': 'lte', '小于等于': 'lte', + 'in': 'in', 'contains': 'contains', 'has': 'contains', '包含': 'contains', + 'not_in': 'not_in', 'nin': 'not_in', 'notcontains': 'not_contains', + 'not_contains': 'not_contains', + 'startswith': 'startswith', 'endswith': 'endswith', + 'matches': 'matches', 'regex': 'matches', 'regexp': 'matches', + 'is_true': 'is_true', 'istrue': 'is_true', 'true': 'is_true', + 'is_false': 'is_false', 'isfalse': 'is_false', 'false': 'is_false', + 'exists': 'exists', 'not_exists': 'not_exists', 'notexists': 'not_exists', + 'between': 'between', +} + +# 一元 / 二元 / 三元操作符 +OP_ARITY = { + 'eq': 2, 'ne': 2, 'gt': 2, 'gte': 2, 'lt': 2, 'lte': 2, + 'in': 2, 'not_in': 2, 'contains': 2, 'not_contains': 2, + 'startswith': 2, 'endswith': 2, 'matches': 2, 'between': 3, + 'is_true': 1, 'is_false': 1, 'exists': 1, 'not_exists': 1, +} + +CANONICAL_OPS = tuple(sorted(OP_ARITY.keys())) + +# 符号操作符（长者优先，直接子串扫描） +_SYMBOL_OPS = sorted([a for a in OP_ALIASES if not a[0].isalnum()], key=len, reverse=True) +# 字母操作符（必须词边界匹配，避免 `remaining` 里的 `in` 误命中） +_WORD_OPS = sorted([a for a in OP_ALIASES if a[0].isalnum()], key=len, reverse=True) +_WORD_OP_RE = re.compile( + r'(? len(p): + return {'kind': 'var', 'name': s[len(p):].strip().strip('{}')} + if s.startswith('{{') and s.endswith('}}'): + return {'kind': 'var', 'name': s[2:-2].strip()} + for p in _VAR_NS_PREFIXES: + if s.startswith(p) and len(s) > len(p): + return {'kind': 'var', 'name': s} + return {'kind': 'const', 'value': to_number_or_raw(value)} + + return {'kind': 'const', 'value': to_number_or_raw(value)} + + +# --------------------------------------------------------------------------- +# condition 归一（递归布尔树） +# --------------------------------------------------------------------------- + +_ALWAYS = {'kind': 'always'} + + +_IDENT_RE = re.compile(r'^[A-Za-z_\u4e00-\u9fa5][0-9A-Za-z_\.\u4e00-\u9fa5\[\]\'\"]*$') + + +def _operand_from_expr_token(token, issues, path, rule_id): + """字符串表达式里的操作数：裸标识符 → var，其余走通用归一。""" + t = str(token).strip().strip('()') + if _IDENT_RE.match(t) and not re.fullmatch(r'-?\d+(\.\d+)?', t): + return {'kind': 'var', 'name': t} + return normalize_operand(t, issues, path, rule_id) + + +def _parse_string_expr(text, issues, path, rule_id): + """尽力解析 `score >= 60` / `state.done == true` 形式的字符串条件。""" + s = str(text).strip() + if not s: + return dict(_ALWAYS) + low = s.lower() + if low in ('true', 'always', '1', '是'): + return dict(_ALWAYS) + if low in ('false', 'never', '0', '否'): + return {'kind': 'not', 'item': dict(_ALWAYS)} + candidates = [] + for op in _SYMBOL_OPS: + idx = s.find(op) + while idx > 0: + candidates.append((idx, op)) + idx = s.find(op, idx + 1) + for m in _WORD_OP_RE.finditer(s): + if m.start() > 0: + candidates.append((m.start(), m.group(1))) + # 取最靠左的操作符；同位置取最长（`>=` 优于 `>`） + candidates.sort(key=lambda c: (c[0], -len(c[1]))) + for idx, op in candidates: + left = s[:idx].strip() + right = s[idx + len(op):].strip() + if not left: + continue + canon = OP_ALIASES.get(op.lower()) or OP_ALIASES.get(op) + if not canon: + continue + arity = OP_ARITY.get(canon, 2) + if arity == 1: + if right: + continue + return {'kind': 'compare', 'op': canon, + 'left': _operand_from_expr_token(left, issues, path + '.left', rule_id)} + if not right: + continue + return { + 'kind': 'compare', + 'op': canon, + 'left': _operand_from_expr_token(left, issues, path + '.left', rule_id), + 'right': _operand_from_expr_token(right, issues, path + '.right', rule_id), + } + issues.append(Issue('warning', 'W_COND_EXPR_UNPARSED', + '字符串条件未能解析为结构化比较，按 expr 原样保留: %s' % s, rule_id, path)) + return {'kind': 'expr', 'text': s} + + +def normalize_condition(raw, issues, path='condition', rule_id=''): + """归一条件树。返回 None 表示「无条件」（调用方替换为 always）。""" + if raw is None or raw == '' or raw == [] or raw == {}: + return None + + if isinstance(raw, str): + return _parse_string_expr(raw, issues, path, rule_id) + + if isinstance(raw, (list, tuple)): + items = [normalize_condition(x, issues, '%s[%d]' % (path, i), rule_id) + for i, x in enumerate(raw)] + items = [i for i in items if i is not None] + if not items: + return None + if len(items) == 1: + return items[0] + return {'kind': 'all', 'items': items} + + if not isinstance(raw, dict): + issues.append(Issue('warning', 'W_COND_SHAPE', '条件类型不支持，按常量真处理', rule_id, path)) + return dict(_ALWAYS) + + kind = str(raw.get('kind') or raw.get('type') or '').strip().lower() + + # —— GD（M3a gd_builder.build_events）形态：all_of / any_of / none_of 字符串表达式数组 —— + for key, comb in (('all_of', 'all'), ('allof', 'all'), ('and', 'all'), + ('any_of', 'any'), ('anyof', 'any'), ('or', 'any'), + ('none_of', 'not'), ('noneof', 'not'), ('nor', 'not')): + if key in raw and isinstance(raw.get(key), (list, tuple, str, dict)): + val = raw[key] + items_raw = val if isinstance(val, (list, tuple)) else [val] + items = [normalize_condition(x, issues, '%s.%s[%d]' % (path, key, i), rule_id) + for i, x in enumerate(items_raw)] + items = [i for i in items if i is not None] + if not items: + return dict(_ALWAYS) + node = items[0] if len(items) == 1 else {'kind': comb if comb != 'not' else 'all', + 'items': items} + if comb == 'not': + node = {'kind': 'not', + 'item': (items[0] if len(items) == 1 + else {'kind': 'any', 'items': items})} + # 与其它键并存时（如 all_of + min_count）合并为 all + rest = {k: v for k, v in raw.items() + if str(k).strip().lower().replace('_', '') != key.replace('_', '')} + if rest: + extra = normalize_condition(rest, issues, path + '.rest', rule_id) + if extra is not None and extra != dict(_ALWAYS): + node = {'kind': 'all', 'items': [node, extra]} + return node + + # —— 计数型条件（GD evidence: {'min_count': N}）—— + for key, op in (('min_count', 'gte'), ('max_count', 'lte'), ('count', 'eq'), + ('min', 'gte'), ('max', 'lte')): + if key in raw and raw.get(key) not in (None, ''): + subject = raw.get('subject') or raw.get('field') or raw.get('var') or key + node = {'kind': 'compare', 'op': op, + 'left': normalize_operand(subject, issues, path + '.left', rule_id), + 'right': normalize_operand(raw.get(key), issues, path + '.right', rule_id)} + rest = {k: v for k, v in raw.items() + if str(k).strip().lower() not in (key, 'subject', 'field', 'var')} + if rest: + extra = normalize_condition(rest, issues, path + '.rest', rule_id) + if extra is not None and extra != dict(_ALWAYS): + node = {'kind': 'all', 'items': [node, extra]} + return node + + # 布尔组合子 + if kind in ('all', 'and', 'every'): + children = raw.get('items') or raw.get('conditions') or raw.get('all') or raw.get('children') or [] + items = [normalize_condition(c, issues, '%s.all[%d]' % (path, i), rule_id) + for i, c in enumerate(children if isinstance(children, (list, tuple)) else [children])] + items = [i for i in items if i is not None] + if not items: + return dict(_ALWAYS) + return {'kind': 'all', 'items': items} if len(items) > 1 else items[0] + + if kind in ('any', 'or', 'some'): + children = raw.get('items') or raw.get('conditions') or raw.get('any') or raw.get('children') or [] + items = [normalize_condition(c, issues, '%s.any[%d]' % (path, i), rule_id) + for i, c in enumerate(children if isinstance(children, (list, tuple)) else [children])] + items = [i for i in items if i is not None] + if not items: + return dict(_ALWAYS) + return {'kind': 'any', 'items': items} if len(items) > 1 else items[0] + + if kind in ('not', 'negate'): + child = raw.get('item') or raw.get('condition') or raw.get('not') + if isinstance(child, (list, tuple)) and child: + child = child[0] + item = normalize_condition(child, issues, path + '.not', rule_id) + if item is None: + return {'kind': 'not', 'item': dict(_ALWAYS)} + return {'kind': 'not', 'item': item} + + if kind in ('always', 'true', 'unconditional'): + return dict(_ALWAYS) + + if kind == 'expr' or (raw.get('expr') and not raw.get('op') and not raw.get('operator')): + text = str(raw.get('text') or raw.get('expr') or '').strip() + if not text: + return dict(_ALWAYS) + issues.append(Issue('info', 'I_COND_EXPR', '保留表达式条件: %s' % text, rule_id, path)) + return {'kind': 'expr', 'text': text} + + # 比较节点 + op_raw = raw.get('op') or raw.get('operator') or raw.get('cmp') or raw.get('compare') or raw.get('rel') + if op_raw is not None: + canon = OP_ALIASES.get(str(op_raw).strip().lower()) + if not canon: + issues.append(Issue('error', 'E_COND_OP_UNKNOWN', + '未知条件操作符: %r（白名单: %s）' % (op_raw, ', '.join(CANONICAL_OPS)), + rule_id, path)) + canon = 'eq' + arity = OP_ARITY.get(canon, 2) + left_raw = raw.get('left') + if left_raw is None: + left_raw = raw.get('field') if raw.get('field') is not None else raw.get('var') + if left_raw is None: + left_raw = raw.get('name') + right_raw = raw.get('right') + if right_raw is None: + right_raw = raw.get('value') if raw.get('value') is not None else raw.get('v') + if right_raw is None: + right_raw = raw.get('to') + + if left_raw is None: + issues.append(Issue('error', 'E_COND_OPERAND', '比较条件缺少左操作数', rule_id, path)) + left_raw = '' + + node = {'kind': 'compare', 'op': canon, + 'left': normalize_operand(left_raw, issues, path + '.left', rule_id)} + + if arity >= 2: + if canon in ('in', 'not_in') and isinstance(right_raw, str): + right_raw = [x.strip() for x in right_raw.split(',') if x.strip()] + if canon == 'between': + pair = right_raw if isinstance(right_raw, (list, tuple)) else \ + [raw.get('from'), raw.get('to')] + pair = [p for p in pair if p is not None] + if len(pair) != 2: + issues.append(Issue('error', 'E_COND_BETWEEN', + 'between 需要 [from, to] 两个边界', rule_id, path)) + pair = (pair + [None, None])[:2] + node['right'] = {'kind': 'list', + 'items': [normalize_operand(p, issues, path + '.right', rule_id) + for p in pair]} + else: + if right_raw is None: + issues.append(Issue('error', 'E_COND_OPERAND', + '操作符 %s 缺少右操作数' % canon, rule_id, path)) + right_raw = '' + node['right'] = normalize_operand(right_raw, issues, path + '.right', rule_id) + return node + + # 隐式：{field, value} / {var, op, value} + if raw.get('field') or raw.get('var') or raw.get('name'): + return normalize_condition( + {'op': raw.get('op') or raw.get('operator') or 'eq', + 'left': raw.get('field') or raw.get('var') or raw.get('name'), + 'right': raw.get('value', raw.get('v'))}, + issues, path, rule_id) + + issues.append(Issue('warning', 'W_COND_SHAPE', + '无法识别的条件结构，按 expr 保留', rule_id, path)) + return {'kind': 'expr', 'text': canonical_dumps(strip_volatile(raw))} + + +# --------------------------------------------------------------------------- +# event 归一 +# --------------------------------------------------------------------------- + +EVENT_TYPE_ALIASES = { + 'onenter': 'enter', 'on_enter': 'enter', 'sceneenter': 'enter', 'scene_enter': 'enter', + 'onexit': 'exit', 'on_exit': 'exit', 'sceneexit': 'exit', 'scene_exit': 'exit', + 'onclick': 'click', 'on_click': 'click', 'click': 'click', 'tap': 'click', + 'ontimer': 'timer', 'on_timer': 'timer', 'tick': 'timer', 'interval': 'timer', + 'oncollision': 'collision', 'on_collision': 'collision', 'collide': 'collision', + 'onmessage': 'message', 'on_message': 'message', 'msg': 'message', + 'onchange': 'change', 'on_change': 'change', 'valuechange': 'change', 'value_change': 'change', + 'onstart': 'start', 'on_start': 'start', 'gamestart': 'start', 'game_start': 'start', + 'onend': 'end', 'on_end': 'end', 'gameend': 'end', 'game_end': 'end', 'finish': 'end', + 'ontrigger': 'trigger', 'on_trigger': 'trigger', 'custom': 'custom', + 'onsubmit': 'submit', 'on_submit': 'submit', 'submit': 'submit', + 'onanswer': 'answer', 'on_answer': 'answer', 'answer': 'answer', + 'onvaluechange': 'change', 'onvarchange': 'change', 'onvariablechange': 'change', + 'onkeypress': 'keypress', 'on_key_press': 'keypress', 'keypress': 'keypress', + 'oninput': 'input', 'on_input': 'input', 'input': 'input', + 'onwin': 'win', 'on_win': 'win', 'win': 'win', 'onlose': 'lose', 'on_lose': 'lose', + 'lose': 'lose', 'oncomplete': 'complete', 'on_complete': 'complete', + 'complete': 'complete', 'onpick': 'pick', 'on_pick': 'pick', 'pick': 'pick', + 'ondrop': 'drop', 'on_drop': 'drop', 'drop': 'drop', +} + + +def normalize_event(raw, issues, path='event', rule_id='', known_types=None): + """归一事件触发器。缺失 → type='unknown' + error issue（fail-closed 由调用方决定）。""" + if raw is None or raw == '' or raw == {} or raw == []: + issues.append(Issue('error', 'E_EVENT_MISSING', '规则缺少 event 触发器', rule_id, path)) + return {'kind': 'event', 'type': 'unknown', 'source': '', 'match': {}} + + if isinstance(raw, str): + raw = {'type': raw} + elif isinstance(raw, (list, tuple)): + raw = raw[0] if raw else {} + issues.append(Issue('warning', 'W_EVENT_MULTI', + '一个规则只支持单事件触发，已取第一个', rule_id, path)) + if not isinstance(raw, dict): + issues.append(Issue('error', 'E_EVENT_SHAPE', '事件结构不支持: %r' % type(raw).__name__, + rule_id, path)) + return {'kind': 'event', 'type': 'unknown', 'source': '', 'match': {}} + + etype_raw = first_str(raw, ('type', 'event', 'event_type', 'name', 'trigger', 'on')) + etype = to_snake(etype_raw) + if etype: + compact = etype.replace('_', '') + base = etype[3:] if etype.startswith('on_') else etype + base_compact = base.replace('_', '') + etype = (EVENT_TYPE_ALIASES.get(compact) + or EVENT_TYPE_ALIASES.get(base_compact) + or base or etype) + etype = etype or 'unknown' + if not etype_raw: + issues.append(Issue('error', 'E_EVENT_MISSING', '事件缺少 type', rule_id, path)) + elif known_types and etype not in known_types \ + and to_snake(etype_raw) not in {to_snake(t) for t in known_types}: + issues.append(Issue('warning', 'W_EVENT_TYPE_UNKNOWN', + '事件类型 %s 未在能力/枚举登记表中' % etype, rule_id, path)) + + source = first_str(raw, ('source', 'source_id', 'from', 'emitter', 'target', 'object')) + # GD 形态 source = 'pbl_mission:12' → 拆 kind / id，保留原串 + source_kind = '' + source_ref = source + if source and ':' in source: + source_kind, _, source_ref = source.partition(':') + source_kind = to_snake(source_kind) + + match_raw = raw.get('match') or raw.get('filter') or raw.get('args') or raw.get('payload') or {} + match = {} + if isinstance(match_raw, dict): + for k, v in strip_volatile(match_raw).items(): + ks = to_snake(k) + if not ks: + continue + if isinstance(v, (dict, list, tuple)): + match[ks] = strip_volatile(v) + else: + match[ks] = to_number_or_raw(v) + elif match_raw not in (None, '', [], {}): + issues.append(Issue('warning', 'W_EVENT_MATCH_SHAPE', + 'event.match 必须是对象，已忽略', rule_id, path)) + + # timer 事件的周期参数提升到 match，保证运行时可调度 + for key in ('interval_ms', 'interval', 'delay_ms', 'delay', 'period_ms', 'period'): + if raw.get(key) not in (None, ''): + match.setdefault('interval_ms', to_int(raw.get(key), 0)) + + ev = {'kind': 'event', 'type': etype, 'source': source, 'match': match} + if source_kind: + ev['source_kind'] = source_kind + ev['source_ref'] = source_ref + return ev + + +# --------------------------------------------------------------------------- +# response 归一 +# --------------------------------------------------------------------------- + +RESPONSE_ALIASES = { + 'then': 'responses', 'response': 'responses', 'actions': 'responses', + 'action': 'responses', 'do': 'responses', 'effects': 'responses', 'effect': 'responses', +} + + +def _normalize_one_response(raw, index, issues, path, rule_id, known_actions, strict): + if raw is None or raw == '': + return None + if isinstance(raw, str): + raw = {'action': raw} + if not isinstance(raw, dict): + issues.append(Issue('error', 'E_RESP_SHAPE', + '响应必须是对象或字符串，得到 %s' % type(raw).__name__, rule_id, path)) + return None + + action_raw = first_str(raw, ('action', 'type', 'kind', 'name', 'op', 'command', 'do')) + action = to_snake(action_raw) + if not action: + issues.append(Issue('error', 'E_RESP_ACTION_MISSING', '响应缺少 action', rule_id, path)) + return None + + if known_actions is not None and action not in known_actions: + sev = 'error' if strict else 'warning' + issues.append(Issue(sev, 'E_CAP_UNKNOWN' if strict else 'W_CAP_UNKNOWN', + '响应 action `%s` 未在 pbl_capability_registry 登记' + % action, rule_id, path)) + if strict: + return None + + target_raw = raw.get('target') + if target_raw is None: + target_raw = raw.get('to') if raw.get('to') is not None else raw.get('object') + if isinstance(target_raw, dict): + target = {'kind': str(target_raw.get('kind') or target_raw.get('type') or 'entity'), + 'id': first_str(target_raw, ('id', 'entity_id', 'name', 'value'))} + elif isinstance(target_raw, (list, tuple)): + target = {'kind': 'list', 'id': '', + 'ids': sorted({str(x) for x in target_raw if str(x)})} + elif target_raw in (None, ''): + target = {'kind': 'none', 'id': ''} + else: + s = str(target_raw).strip() + if s.lower() in ('self', 'this', 'trigger', 'source'): + target = {'kind': 'self', 'id': ''} + elif s.lower() in ('all', 'everyone', '*'): + target = {'kind': 'all', 'id': ''} + else: + target = {'kind': 'entity', 'id': s} + + params_raw = raw.get('params') + if params_raw is None: + params_raw = raw.get('args') or raw.get('payload') or raw.get('data') + if params_raw is None: + # GD 形态：response 项把业务字段平铺（{'op':'emit','event':'x','capability':'pbl.y'}） + _RESERVED = {'action', 'type', 'kind', 'name', 'op', 'command', 'do', + 'target', 'to', 'object', 'params', 'args', 'payload', 'data', + 'delay_ms', 'delay', 'repeat', 'times', 'condition', 'when', 'if'} + flat = {k: v for k, v in raw.items() if str(k).strip().lower() not in _RESERVED} + params_raw = strip_volatile(flat) + if not isinstance(params_raw, dict): + if params_raw in (None, '', [], {}): + params_raw = {} + else: + params_raw = {'value': to_number_or_raw(params_raw)} + params = {} + for k, v in strip_volatile(params_raw).items(): + ks = to_snake(k) + if not ks: + continue + params[ks] = _normalize_param_value(v) + + # 常见同义参数键归一 + for a, b in (('val', 'value'), ('amount', 'value'), ('delta', 'value'), + ('text', 'message'), ('msg', 'message'), ('scene', 'scene_id'), + ('next_scene', 'scene_id'), ('to_scene', 'scene_id')): + if a in params and b not in params: + params[b] = params.pop(a) + + out = { + 'kind': 'action', + 'action': action, + 'target': target, + 'params': params, + 'delay_ms': max(0, to_int(raw.get('delay_ms', raw.get('delay', 0)), 0)), + 'repeat': max(0, to_int(raw.get('repeat', raw.get('times', 1)), 1)), + } + cap = first_str(raw, ('capability', 'capability_key', 'cap')) or \ + first_str(params, ('capability', 'capability_key')) + if cap: + out['capability'] = cap + params.pop('capability', None) + params.pop('capability_key', None) + ev = first_str(raw, ('event', 'emit', 'emit_event')) + if ev and action in ('emit', 'emit_event', 'fire'): + params.setdefault('event', to_snake(ev) or ev) + if raw.get('condition') not in (None, '', [], {}): + sub = normalize_condition(raw.get('condition'), issues, path + '.condition', rule_id) + if sub is not None: + out['condition'] = sub + return out + + +def _normalize_param_value(v, depth=0): + if depth > 16: + return None + if isinstance(v, dict): + return {to_snake(k) or str(k): _normalize_param_value(x, depth + 1) + for k, x in strip_volatile(v).items()} + if isinstance(v, (list, tuple)): + return [_normalize_param_value(x, depth + 1) for x in v] + return to_number_or_raw(v) + + +def normalize_responses(raw, issues, path='responses', rule_id='', + known_actions=None, strict=True): + if raw is None or raw == '' or raw == {} or raw == []: + issues.append(Issue('error', 'E_RESP_MISSING', '规则缺少 response（至少一个动作）', + rule_id, path)) + return [] + if isinstance(raw, dict): + # {"responses": [...]} 嵌套 或 单个动作对象 + inner = None + for k, v in raw.items(): + if str(k).strip().lower() in RESPONSE_ALIASES and isinstance(v, (list, tuple, dict)): + inner = v + break + raw = inner if inner is not None else [raw] + if not isinstance(raw, (list, tuple)): + raw = [raw] + out = [] + for i, item in enumerate(raw): + r = _normalize_one_response(item, i, issues, '%s[%d]' % (path, i), + rule_id, known_actions, strict) + if r is not None: + out.append(r) + if not out: + issues.append(Issue('error', 'E_RESP_EMPTY', + '规则响应归一后为空（strict 模式下未登记 action 会被剔除）', + rule_id, path)) + return out + + +# --------------------------------------------------------------------------- +# ECR 源枚举（兼容蓝图快照的多种形态） +# --------------------------------------------------------------------------- + +_ECR_LIST_KEYS = ('ecr', 'ecrs', 'rules', 'rule', 'event_rules', 'logic', 'behaviors', + 'triggers', 'event_conditions', 'scripts') +_EVENT_LIST_KEYS = ('events', 'event', 'scene_events') + + +def iter_ecr_sources(snapshot): + """从蓝图版本快照里确定性地枚举 ECR 原始记录。 + + 支持形态： + A. snapshot['rules' | 'ecr' | 'logic' | ...] = [ {event, condition, responses}, ... ] + B. snapshot['events'] = [ {id, type, conditions, responses}, ... ] （事件内嵌条件/响应） + C. snapshot['scenes'] = [ {events: [...] } ] （场景内嵌事件） + D. snapshot['children' | 'sub_objects' | 'objects'] = [ {kind:'rule'|'event', ...} ] + + 返回 [(source_path, raw_dict), ...]，按 source_path 排序保证确定性。 + """ + found = [] + if not isinstance(snapshot, dict): + return found + + def _walk(node, prefix, depth=0): + if depth > 8 or not isinstance(node, dict): + return + for key in _ECR_LIST_KEYS: + val = node.get(key) + if isinstance(val, dict): + val = [val] + if isinstance(val, (list, tuple)): + for i, item in enumerate(val): + if isinstance(item, dict) and _looks_like_ecr(item): + found.append(('%s.%s[%d]' % (prefix, key, i), item)) + for key in _EVENT_LIST_KEYS: + val = node.get(key) + if isinstance(val, dict): + val = [val] + if isinstance(val, (list, tuple)): + for i, item in enumerate(val): + if isinstance(item, dict) and _looks_like_ecr(item): + found.append(('%s.%s[%d]' % (prefix, key, i), item)) + for key in ('scenes', 'scene', 'children', 'sub_objects', 'objects', 'nodes', 'items'): + val = node.get(key) + if isinstance(val, dict): + val = [val] + if isinstance(val, (list, tuple)): + for i, item in enumerate(val): + if isinstance(item, dict): + _walk(item, '%s.%s[%d]' % (prefix, key, i), depth + 1) + + _walk(snapshot, 'snapshot') + + # 去重（同一对象可能同时出现在 rules 与 events 下）：按内容哈希保留首个路径 + seen = {} + uniq = [] + for path, raw in sorted(found, key=lambda x: x[0]): + h = sha256_text(canonical_dumps(strip_volatile(raw))) + if h in seen: + continue + seen[h] = path + uniq.append((path, raw)) + return uniq + + +_TRIGGER_KEYS = {'event', 'trigger', 'on', 'type', 'event_type'} +_EFFECT_KEYS = {'responses', 'response', 'actions', 'action', 'then', 'do', 'effects', 'effect'} +_COND_KEYS = {'condition', 'conditions', 'when', 'if'} +# 容器 / 实体标记键：命中且无触发器时不视为 ECR（避免把 scene/entity 误当规则） +_CONTAINER_KEYS = {'scenes', 'scene', 'children', 'sub_objects', 'objects', 'nodes', + 'events', 'rules', 'items', 'entities', 'components', 'assets', + 'position', 'transform', 'mesh', 'variables', 'ui'} +_ECR_KINDS = {'rule', 'ecr', 'event_rule', 'logic', 'behavior', 'trigger', 'event'} + + +def _looks_like_ecr(d): + keys = {str(k).strip().lower() for k in d.keys()} + has_trigger = bool(keys & _TRIGGER_KEYS) + has_effect = bool(keys & _EFFECT_KEYS) + has_cond = bool(keys & _COND_KEYS) + kind = str(d.get('kind') or d.get('object_type') or d.get('sub_type') or '').lower() + if kind in _ECR_KINDS: + return True + score = sum([has_trigger, has_effect, has_cond]) + if score == 0: + return False + if (keys & _CONTAINER_KEYS) and not has_trigger: + return False + # 只有响应没有触发器：需带 id/name 标识才当规则（畸形规则要报错，不能静默丢） + if not has_trigger and not has_cond: + return bool(keys & {'id', 'name', 'rule_id', 'key', 'code'}) + return True + + +# --------------------------------------------------------------------------- +# 单条 ECR → 规则 +# --------------------------------------------------------------------------- + + +def _derive_rule_id(rule_core): + return 'rule_' + sha256_text(canonical_dumps(rule_core))[:16] + + +def normalize_ecr(raw, index, ctx=None, source_path=''): + """把一条原始 ECR 记录归一为 script_type=1 规则 JSON。返回 (rule, issues)。""" + ctx = ctx or {} + issues = [] + raw = strip_volatile(raw) if isinstance(raw, dict) else {} + + source_id = first_str(raw, ('id', 'rule_id', 'event_key', 'key', 'code', + 'object_id', 'sub_object_id')) + name = first_str(raw, ('name', 'title', 'label', 'rule_name')) or \ + 'rule_%03d' % (index + 1) + description = first_str(raw, ('description', 'desc', 'remark', 'memo')) + priority = to_int(raw.get('priority', raw.get('order', raw.get('weight', 100))), 100) + enabled = to_bool(raw.get('enabled', raw.get('active', raw.get('is_active', True))), True) + + rule_id_hint = source_id + tmp_id = rule_id_hint or ('pending_%d' % index) + + event = normalize_event( + raw.get('event') if raw.get('event') is not None else + (raw.get('trigger') if raw.get('trigger') is not None else raw.get('on')), + issues, 'event', tmp_id, ctx.get('event_types')) + + # 形态 B：事件对象自身即触发器（{id,type,conditions,responses}） + if event.get('type') == 'unknown' and (raw.get('type') or raw.get('event_type')): + event = normalize_event({'type': raw.get('type') or raw.get('event_type'), + 'source': raw.get('source') or raw.get('source_id'), + 'match': raw.get('match') or raw.get('args') or {}}, + issues, 'event', tmp_id, ctx.get('event_types')) + issues[:] = [i for i in issues if i['code'] != 'E_EVENT_MISSING'] + + cond_raw = raw.get('condition') + if cond_raw is None: + cond_raw = raw.get('conditions') + if cond_raw is None: + cond_raw = raw.get('when') + if cond_raw is None: + cond_raw = raw.get('if') + condition = normalize_condition(cond_raw, issues, 'condition', tmp_id) + if condition is None: + condition = dict(_ALWAYS) + + resp_raw = raw.get('responses') + for k in ('response', 'actions', 'action', 'then', 'do', 'effects'): + if resp_raw is None: + resp_raw = raw.get(k) + responses = normalize_responses(resp_raw, issues, 'responses', tmp_id, + ctx.get('known_actions'), + bool(ctx.get('strict', True))) + + limits = raw.get('limits') if isinstance(raw.get('limits'), dict) else {} + limits_out = { + 'cooldown_ms': max(0, to_int( + limits.get('cooldown_ms', raw.get('cooldown_ms', raw.get('cooldown', 0))), 0)), + 'max_triggers': max(0, to_int( + limits.get('max_triggers', raw.get('max_triggers', raw.get('max_times', 0))), 0)), + } + + tags_raw = raw.get('tags') or raw.get('tag') or raw.get('labels') or [] + if isinstance(tags_raw, str): + tags_raw = [t for t in re.split(r'[,;|\s]+', tags_raw) if t] + tags = sorted({to_snake(t) or str(t).strip() for t in tags_raw if str(t).strip()}) + + origin = { + 'blueprint_id': str(ctx.get('blueprint_id') or ''), + 'blueprint_version': str(ctx.get('blueprint_version') or ''), + 'sub_object_id': source_id, + 'source_index': int(index), + 'source_path': source_path or '', + } + + core = { + 'schema_version': RULE_SCHEMA_VERSION, + 'name': name, + 'description': description, + 'priority': priority, + 'enabled': bool(enabled), + 'event': event, + 'condition': condition, + 'responses': responses, + 'limits': limits_out, + 'tags': tags, + 'origin': origin, + } + + rule_id = source_id if source_id else _derive_rule_id(core) + rule = dict(core) + rule['rule_id'] = rule_id + rule['source_id'] = source_id + # 键序无关（canonical_dumps sort_keys），但保持可读顺序 + ordered = { + 'schema_version': rule['schema_version'], + 'rule_id': rule['rule_id'], + 'source_id': rule['source_id'], + 'name': rule['name'], + 'description': rule['description'], + 'priority': rule['priority'], + 'enabled': rule['enabled'], + 'event': rule['event'], + 'condition': rule['condition'], + 'responses': rule['responses'], + 'limits': rule['limits'], + 'tags': rule['tags'], + 'origin': rule['origin'], + } + for i in issues: + i['rule_id'] = rule_id + return ordered, issues + + +# --------------------------------------------------------------------------- +# 批量编译 +# --------------------------------------------------------------------------- + + +def sort_rules(rules): + """确定性排序：priority 降序 → rule_id 升序。""" + return sorted(rules, key=lambda r: (-to_int(r.get('priority'), 0), str(r.get('rule_id')))) + + +def build_rules(snapshot, ctx=None): + """蓝图版本快照 → (rules, issues)。 + + ctx: {tenant_id, blueprint_id, blueprint_version, known_actions(set|None), + event_types(set|None), strict(bool, 默认 True)} + """ + ctx = dict(ctx or {}) + ctx.setdefault('strict', True) + issues = [] + + tenant_id = str(ctx.get('tenant_id') or '').strip() + if not tenant_id: + raise CompileError('E_TENANT_REQUIRED', 'tenant_id 缺失，拒绝编译（多租户铁律）') + + if not isinstance(snapshot, dict) or not snapshot: + raise CompileError('E_SNAPSHOT_EMPTY', '蓝图版本快照为空，无法映射规则') + + sources = iter_ecr_sources(snapshot) + if not sources: + issues.append(Issue('warning', 'W_NO_ECR', '快照中未发现 event→condition→response 记录')) + + rules = [] + seen_ids = {} + dropped = 0 + for idx, (path, raw) in enumerate(sources): + rule, ri = normalize_ecr(raw, idx, ctx, path) + issues.extend(ri) + # fail-closed：触发器无效或响应为空的规则不进入产物（错误已记 issue，编译整体失败） + if not rule['responses'] or (rule['event'] or {}).get('type') in (None, '', 'unknown'): + dropped += 1 + issues.append(Issue('error', 'E_RULE_DROPPED', + '规则被剔除（event 无效或 responses 为空），不产出半成品: %s' + % (rule.get('source_id') or rule.get('name') or path), + rule['rule_id'], path)) + continue + rid = rule['rule_id'] + if rid in seen_ids: + issues.append(Issue('error', 'E_RULE_ID_DUP', + 'rule_id 重复: %s（首次出现 %s，本次 %s）' + % (rid, seen_ids[rid], path), rid, path)) + rule['rule_id'] = rid + '_' + sha256_text(path)[:6] + rid = rule['rule_id'] + seen_ids[rid] = path + rules.append(rule) + + rules = sort_rules(rules) + # origin.source_index 保留原始枚举序，确保可追溯；dropped 计数供上层告警 + if dropped: + issues.append(Issue('error', 'E_RULES_DROPPED_TOTAL', + '共剔除 %d 条无效规则' % dropped)) + return rules, issues + + +# --------------------------------------------------------------------------- +# 校验 +# --------------------------------------------------------------------------- + +_REQUIRED_RULE_KEYS = ('schema_version', 'rule_id', 'name', 'priority', 'enabled', + 'event', 'condition', 'responses', 'limits', 'tags', 'origin') + + +def validate_rule(rule): + """校验单条已归一规则，返回 issues（不修改入参）。""" + issues = [] + if not isinstance(rule, dict): + return [Issue('error', 'E_RULE_SHAPE', '规则必须是对象')] + rid = str(rule.get('rule_id') or '') + for k in _REQUIRED_RULE_KEYS: + if k not in rule: + issues.append(Issue('error', 'E_RULE_KEY_MISSING', '规则缺少字段 %s' % k, rid)) + if not rid: + issues.append(Issue('error', 'E_RULE_ID_MISSING', '规则缺少 rule_id')) + if str(rule.get('schema_version') or '') != RULE_SCHEMA_VERSION: + issues.append(Issue('error', 'E_RULE_SCHEMA', + 'schema_version 必须为 %s' % RULE_SCHEMA_VERSION, rid)) + ev = rule.get('event') or {} + if not isinstance(ev, dict) or not ev.get('type') or ev.get('type') == 'unknown': + issues.append(Issue('error', 'E_EVENT_MISSING', '规则 event.type 无效', rid, 'event')) + resp = rule.get('responses') + if not isinstance(resp, list) or not resp: + issues.append(Issue('error', 'E_RESP_EMPTY', '规则至少需要一个 response', rid, 'responses')) + else: + for i, r in enumerate(resp): + if not isinstance(r, dict) or not r.get('action'): + issues.append(Issue('error', 'E_RESP_ACTION_MISSING', + 'response[%d] 缺少 action' % i, rid, 'responses[%d]' % i)) + elif not re.fullmatch(r'[a-z][a-z0-9_]{0,63}', str(r.get('action'))): + issues.append(Issue('error', 'E_RESP_ACTION_FORMAT', + 'action 必须是 lower_snake（≤64字符）: %r' % r.get('action'), + rid, 'responses[%d]' % i)) + _validate_condition(rule.get('condition'), issues, rid, 'condition') + return issues + + +def _validate_condition(node, issues, rid, path, depth=0): + if depth > 24: + issues.append(Issue('error', 'E_COND_DEPTH', '条件嵌套过深（>24）', rid, path)) + return + if node is None: + issues.append(Issue('error', 'E_COND_MISSING', '条件为空', rid, path)) + return + if not isinstance(node, dict): + issues.append(Issue('error', 'E_COND_SHAPE', '条件必须是对象', rid, path)) + return + kind = node.get('kind') + if kind == 'always': + return + if kind in ('all', 'any'): + items = node.get('items') + if not isinstance(items, list) or not items: + issues.append(Issue('error', 'E_COND_ITEMS', '%s 需要非空 items' % kind, rid, path)) + return + for i, it in enumerate(items): + _validate_condition(it, issues, rid, '%s.%s[%d]' % (path, kind, i), depth + 1) + return + if kind == 'not': + _validate_condition(node.get('item'), issues, rid, path + '.not', depth + 1) + return + if kind == 'expr': + if not str(node.get('text') or '').strip(): + issues.append(Issue('error', 'E_COND_EXPR_EMPTY', 'expr 条件缺少 text', rid, path)) + return + if kind == 'compare': + op = node.get('op') + if op not in OP_ARITY: + issues.append(Issue('error', 'E_COND_OP_UNKNOWN', + '未知操作符 %r' % op, rid, path)) + return + arity = OP_ARITY[op] + left = node.get('left') + if not isinstance(left, dict) or not left.get('kind'): + issues.append(Issue('error', 'E_COND_OPERAND', '左操作数无效', rid, path + '.left')) + elif left.get('kind') == 'var' and not left.get('name'): + issues.append(Issue('error', 'E_COND_OPERAND', '变量操作数 name 为空', rid, path + '.left')) + if arity >= 2: + right = node.get('right') + if not isinstance(right, dict) or not right.get('kind'): + issues.append(Issue('error', 'E_COND_OPERAND', '右操作数无效', rid, path + '.right')) + elif op == 'between': + items = right.get('items') or [] + if len(items) != 2 or any(i.get('value') is None for i in items + if isinstance(i, dict)): + issues.append(Issue('error', 'E_COND_BETWEEN', + 'between 需要两个非空边界', rid, path + '.right')) + return + issues.append(Issue('error', 'E_COND_KIND', '未知条件 kind: %r' % kind, rid, path)) + + +def validate_rules(rules): + issues = [] + ids = set() + for r in rules or []: + rid = str((r or {}).get('rule_id') or '') + if rid and rid in ids: + issues.append(Issue('error', 'E_RULE_ID_DUP', 'rule_id 重复: %s' % rid, rid)) + ids.add(rid) + issues.extend(validate_rule(r)) + return issues + + +def has_errors(issues): + return any(i.get('severity') == 'error' for i in issues or []) + + +# --------------------------------------------------------------------------- +# 产物序列化 +# --------------------------------------------------------------------------- + + +def canonical_rules_json(rules): + """规则数组 → 确定性 JSON 文本（script_engine script_type=1 直接可加载）。""" + return canonical_dumps(sort_rules(rules or [])) + + +def rule_content_json(rule): + """单条规则 → 确定性 JSON 文本（写入 script 表 content 字段）。""" + return canonical_dumps(rule) + + +def rules_hash(rules): + return sha256_text(canonical_rules_json(rules)) + + +def count_conditions(node): + if not isinstance(node, dict): + return 0 + kind = node.get('kind') + if kind in ('all', 'any'): + return 1 + sum(count_conditions(i) for i in (node.get('items') or [])) + if kind == 'not': + return 1 + count_conditions(node.get('item')) + return 1 + + +def build_rules_manifest(rules, extra=None): + """规则清单（供版本比对 / 导出校验 / GD 引用）。确定性：不含时间戳。""" + rules = sort_rules(rules or []) + entries = [] + for r in rules: + entries.append({ + 'rule_id': str(r.get('rule_id') or ''), + 'name': str(r.get('name') or ''), + 'priority': to_int(r.get('priority'), 0), + 'enabled': bool(r.get('enabled')), + 'event_type': str((r.get('event') or {}).get('type') or ''), + 'condition_count': count_conditions(r.get('condition')), + 'response_count': len(r.get('responses') or []), + 'content_hash': sha256_text(canonical_dumps(r)), + }) + manifest = { + 'schema_version': RULE_SCHEMA_VERSION, + 'script_type': 1, + 'rule_count': len(entries), + 'enabled_count': sum(1 for e in entries if e['enabled']), + 'disabled_count': sum(1 for e in entries if not e['enabled']), + 'response_total': sum(e['response_count'] for e in entries), + 'condition_total': sum(e['condition_count'] for e in entries), + 'rules_hash': rules_hash(rules), + 'rules': entries, + } + if extra: + manifest['extra'] = strip_volatile(extra) + return manifest + + +def diff_rules(old_rules, new_rules): + """版本比对：added / removed / changed（按 content_hash）。确定性输出。""" + def _idx(rs): + out = {} + for r in rs or []: + rid = str((r or {}).get('rule_id') or '') + if rid: + out[rid] = sha256_text(canonical_dumps(r)) + return out + + o, n = _idx(old_rules), _idx(new_rules) + added = sorted(k for k in n if k not in o) + removed = sorted(k for k in o if k not in n) + changed = sorted(k for k in n if k in o and o[k] != n[k]) + return { + 'added': added, 'removed': removed, 'changed': changed, + 'unchanged_count': len([k for k in n if k in o and o[k] == n[k]]), + 'added_count': len(added), 'removed_count': len(removed), + 'changed_count': len(changed), + 'old_rules_hash': rules_hash(old_rules or []), + 'new_rules_hash': rules_hash(new_rules or []), + } diff --git a/scripts/test_m3b_mapping.py b/scripts/test_m3b_mapping.py new file mode 100644 index 0000000..f5ad711 --- /dev/null +++ b/scripts/test_m3b_mapping.py @@ -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())