270 lines
11 KiB
Python
270 lines
11 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""pbl_compiler.gd_rules_writer —— M3b 第12章收口：Game Definition 规则产物落库。
|
||
|
||
背景（本轮 M3b 整改的真实缺口）：
|
||
exporter.write_game_definition() 用「候选列名探测」写 GD，候选集是
|
||
('game_definition','gd_json','definition','content')，而 models/pbl_game_definition.json
|
||
的真实列名是 `definition_json` / `content_fingerprint` / `rules_json` / `rules_hash` /
|
||
`rule_count` / `rules_manifest` / `compile_status` / `quality_state` / `world_id` /
|
||
`scene_id` / `duration_ms` / `error_message`。探测不命中 → M3b 的规则映射产物
|
||
（rules_json / rules_hash / rule_count / rules_manifest）根本没写进 game_definition，
|
||
「规则导出与 game_definition 写入」这条需求等于没落地。
|
||
|
||
本文件是 M3b 的**权威 GD 写入器**：列名直接对齐 models/pbl_game_definition.json，
|
||
显式列名 INSERT/UPDATE（不靠位置、不靠猜），并把规则四件套一并落库。
|
||
|
||
铁律：
|
||
- 所有 SELECT/INSERT/UPDATE 一律 tenant_id 打头，缺失即抛 CompileError（fail-closed）。
|
||
- 幂等：同 (tenant_id, blueprint_id, blueprint_version_no) 重复写只更新不新增；
|
||
同输入必同 content_fingerprint / rules_hash（29.6 确定性，无时间戳无 uuid 进正文）。
|
||
- created_at/updated_at 属审计字段，只写审计列，绝不参与指纹计算。
|
||
"""
|
||
|
||
import json
|
||
|
||
from .script_mapping import (
|
||
CompileError,
|
||
Issue,
|
||
canonical_rules_json,
|
||
build_rules_manifest,
|
||
canonical_dumps,
|
||
sha256_text,
|
||
strip_volatile,
|
||
to_int,
|
||
)
|
||
|
||
__all__ = [
|
||
'GD_TABLE',
|
||
'GD_COLUMNS',
|
||
'gd_primary_id',
|
||
'build_gd_row',
|
||
'write_gd_rules',
|
||
'read_gd_rules',
|
||
'read_gd_rules_by_id',
|
||
]
|
||
|
||
#: 表名与列集合，唯一事实源 = models/pbl_game_definition.json（勿在此表外造列）
|
||
GD_TABLE = 'pbl_game_definition'
|
||
|
||
GD_COLUMNS = (
|
||
'id', 'tenant_id', 'blueprint_id', 'blueprint_version_no', 'compiler_version_id',
|
||
'content_fingerprint', 'definition_json',
|
||
'rules_json', 'rules_hash', 'rule_count', 'rules_manifest',
|
||
'world_id', 'scene_id', 'compile_status', 'quality_state',
|
||
'duration_ms', 'error_message',
|
||
'created_by', 'created_at', 'updated_by', 'updated_at', 'is_deleted',
|
||
)
|
||
|
||
#: 参与内容指纹的列（29.6：排除审计/耗时/状态等易变列）
|
||
_FINGERPRINT_COLUMNS = ('blueprint_id', 'blueprint_version_no', 'definition_json',
|
||
'rules_json')
|
||
|
||
|
||
def _text(value):
|
||
"""任意对象 → 稳定文本（None → 空串，dict/list → canonical JSON）。"""
|
||
if value is None:
|
||
return ''
|
||
if isinstance(value, str):
|
||
return value
|
||
if isinstance(value, (dict, list)):
|
||
return canonical_dumps(strip_volatile(value))
|
||
return str(value)
|
||
|
||
|
||
def gd_primary_id(tenant_id, blueprint_id, blueprint_version_no, content_fingerprint):
|
||
"""确定性主键：sha256(tenant|bp|ver|fp) 前 32 位十六进制。
|
||
|
||
不用 uuid/不自增 —— 保证「同输入重编译 → 同一行」，幂等 upsert 才成立。
|
||
"""
|
||
key = '|'.join([str(tenant_id or ''), str(blueprint_id or ''),
|
||
str(blueprint_version_no or ''), str(content_fingerprint or '')])
|
||
return sha256_text(key)[:32]
|
||
|
||
|
||
def content_fingerprint_of(row):
|
||
"""按参与指纹的列计算 GD 内容指纹（列序固定，与 sort_keys 无关）。"""
|
||
parts = []
|
||
for col in _FINGERPRINT_COLUMNS:
|
||
parts.append('%s=%s' % (col, _text(row.get(col))))
|
||
return sha256_text('\n'.join(parts))
|
||
|
||
|
||
def build_gd_row(gd, ctx, rules=None, manifest=None, base_definition_json=None,
|
||
issues=None):
|
||
"""组装一行 pbl_game_definition 记录（列名对齐 models，不猜列）。
|
||
|
||
gd: M3a gd_builder + M3b rules 顶层键合成后的完整 GD dict（10 顶层键 + rules）
|
||
ctx: tenant_id* / blueprint_id* / blueprint_version* / compiler_version_id /
|
||
world_id / scene_id / quality_state / compile_status / duration_ms /
|
||
created_by / error_message
|
||
rules: build_rules() 产出的 script_type=1 规则列表
|
||
返回 (row_dict, issues)
|
||
"""
|
||
issues = list(issues or [])
|
||
ctx = dict(ctx or {})
|
||
tenant_id = str(ctx.get('tenant_id') or '').strip()
|
||
if not tenant_id:
|
||
raise CompileError('E_TENANT_REQUIRED', 'tenant_id 缺失，拒绝写入 game_definition')
|
||
blueprint_id = str(ctx.get('blueprint_id') or ctx.get('bp_id') or '').strip()
|
||
blueprint_version_no = to_int(ctx.get('blueprint_version')
|
||
or ctx.get('blueprint_version_no'), 0)
|
||
if not blueprint_id or blueprint_version_no <= 0:
|
||
raise CompileError('E_GD_KEY_MISSING',
|
||
'blueprint_id / blueprint_version_no 缺失或非法，无法定位 game_definition')
|
||
|
||
rules = list(rules or [])
|
||
manifest = manifest or build_rules_manifest(rules)
|
||
gd = dict(gd or {})
|
||
gd_text = canonical_dumps(strip_volatile(gd))
|
||
rules_text = canonical_rules_json(rules)
|
||
|
||
row = {
|
||
'tenant_id': tenant_id,
|
||
'blueprint_id': blueprint_id,
|
||
'blueprint_version_no': blueprint_version_no,
|
||
'compiler_version_id': _text(ctx.get('compiler_version_id')
|
||
or ctx.get('compiler_version') or '')[:32],
|
||
'definition_json': base_definition_json if base_definition_json is not None else gd_text,
|
||
'rules_json': rules_text,
|
||
'rules_hash': _text(manifest.get('rules_hash'))[:64],
|
||
'rule_count': to_int(manifest.get('rule_count'), len(rules)),
|
||
'rules_manifest': canonical_dumps(strip_volatile(manifest)),
|
||
'world_id': _text(ctx.get('world_id'))[:32],
|
||
'scene_id': _text(ctx.get('scene_id'))[:32],
|
||
'compile_status': _text(ctx.get('compile_status') or ctx.get('status')
|
||
or 'compiled')[:32],
|
||
'quality_state': _text(ctx.get('quality_state'))[:64],
|
||
'duration_ms': to_int(ctx.get('duration_ms'), 0),
|
||
'error_message': _text(ctx.get('error_message'))[:60000],
|
||
'created_by': _text(ctx.get('created_by') or ctx.get('actor_id'))[:32],
|
||
'is_deleted': '0',
|
||
}
|
||
row['content_fingerprint'] = content_fingerprint_of(row)
|
||
row['id'] = gd_primary_id(row['tenant_id'], row['blueprint_id'],
|
||
row['blueprint_version_no'], row['content_fingerprint'])
|
||
|
||
if not row['rules_hash']:
|
||
issues.append(Issue('error', 'E_RULES_HASH_EMPTY',
|
||
'rules_manifest 未给出 rules_hash，规则产物不可信',
|
||
'', 'rules_hash'))
|
||
if row['rule_count'] != len(rules):
|
||
issues.append(Issue('warning', 'W_RULE_COUNT_MISMATCH',
|
||
'rule_count=%s 与实际规则数=%s 不一致'
|
||
% (row['rule_count'], len(rules)), '', 'rule_count'))
|
||
return row, issues
|
||
|
||
|
||
async def _available_columns(sor, table):
|
||
"""探测实际库中该表已有列（跨宿主/未跑 DDL 时降级，不猜列名写 SQL）。"""
|
||
try:
|
||
recs = await sor.sqlExe(
|
||
'SELECT COLUMN_NAME AS c FROM INFORMATION_SCHEMA.COLUMNS '
|
||
'WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=${tbl}$', {'tbl': table})
|
||
cols = set()
|
||
for r in recs or []:
|
||
v = getattr(r, 'c', None) or getattr(r, 'COLUMN_NAME', None) \
|
||
or getattr(r, 'column_name', None)
|
||
if v:
|
||
cols.add(str(v))
|
||
if cols:
|
||
return cols
|
||
except Exception:
|
||
pass
|
||
return set(GD_COLUMNS)
|
||
|
||
|
||
async def write_gd_rules(sor, row, ctx=None):
|
||
"""幂等 upsert 一行 game_definition（含规则四件套）。
|
||
|
||
唯一键：tenant_id + blueprint_id + blueprint_version_no（idx_gd_bp）。
|
||
返回 {'written','mode','gd_id','content_fingerprint','rules_hash','rule_count'}
|
||
"""
|
||
ctx = dict(ctx or {})
|
||
tenant_id = str(row.get('tenant_id') or '').strip()
|
||
if not tenant_id:
|
||
raise CompileError('E_TENANT_REQUIRED', 'tenant_id 缺失，拒绝写入 game_definition')
|
||
|
||
cols = await _available_columns(sor, GD_TABLE)
|
||
payload = {k: v for k, v in row.items() if k in cols and k != 'id'}
|
||
missing = [c for c in ('definition_json', 'rules_json', 'rules_hash', 'rule_count')
|
||
if c not in cols]
|
||
if missing:
|
||
raise CompileError('E_GD_COLUMN_MISSING',
|
||
'pbl_game_definition 缺少 M3b 规则产物列 %s，请先执行 models/'
|
||
'pbl_game_definition.json 生成的 DDL（json2ddl）' % missing)
|
||
|
||
where = {'tenant_id': tenant_id,
|
||
'blueprint_id': str(row.get('blueprint_id') or ''),
|
||
'blueprint_version_no': int(row.get('blueprint_version_no') or 0)}
|
||
try:
|
||
exist = await sor.R(GD_TABLE, dict(where, is_deleted='0'))
|
||
except Exception as e:
|
||
raise CompileError('E_GD_QUERY', '查询 game_definition 失败: %s' % e)
|
||
|
||
now = ctx.get('now_str') or ''
|
||
try:
|
||
if exist:
|
||
old_id = str(getattr(exist[0], 'id', '') or '')
|
||
upd = dict(payload)
|
||
if 'updated_at' in cols and now:
|
||
upd['updated_at'] = now
|
||
await sor.U(GD_TABLE, dict(upd, **{'id': old_id}))
|
||
mode, gd_id = 'update', old_id
|
||
else:
|
||
ins = dict(payload)
|
||
if 'id' in cols:
|
||
ins['id'] = row['id']
|
||
if 'created_at' in cols and now:
|
||
ins['created_at'] = now
|
||
await sor.C(GD_TABLE, ins)
|
||
mode, gd_id = 'insert', str(row.get('id') or '')
|
||
except Exception as e:
|
||
raise CompileError('E_GD_WRITE', '写入 game_definition 失败: %s' % e)
|
||
|
||
return {
|
||
'written': True,
|
||
'mode': mode,
|
||
'gd_id': gd_id,
|
||
'content_fingerprint': row.get('content_fingerprint'),
|
||
'rules_hash': row.get('rules_hash'),
|
||
'rule_count': row.get('rule_count'),
|
||
}
|
||
|
||
|
||
async def read_gd_rules(sor, tenant_id, blueprint_id, blueprint_version_no=None):
|
||
"""读回 GD（含规则四件套），rules_json 解析成 list。"""
|
||
if not tenant_id:
|
||
raise CompileError('E_TENANT_REQUIRED', 'tenant_id 缺失')
|
||
where = {'tenant_id': str(tenant_id), 'blueprint_id': str(blueprint_id or '')}
|
||
if blueprint_version_no not in (None, ''):
|
||
where['blueprint_version_no'] = int(blueprint_version_no)
|
||
recs = await sor.R(GD_TABLE, where)
|
||
out = []
|
||
for r in recs or []:
|
||
d = {c: getattr(r, c, None) for c in GD_COLUMNS}
|
||
if d.get('rules_json'):
|
||
try:
|
||
d['rules'] = json.loads(d['rules_json'])
|
||
except Exception:
|
||
d['rules'] = []
|
||
out.append(d)
|
||
return out
|
||
|
||
|
||
async def read_gd_rules_by_id(sor, tenant_id, gd_id):
|
||
"""按主键读单行（租户打头，跨租户查不到即 None）。"""
|
||
if not tenant_id or not gd_id:
|
||
raise CompileError('E_PARAM_MISSING', 'tenant_id / id 均为必填')
|
||
recs = await sor.R(GD_TABLE, {'tenant_id': str(tenant_id), 'id': str(gd_id)})
|
||
if not recs:
|
||
return None
|
||
r = recs[0]
|
||
d = {c: getattr(r, c, None) for c in GD_COLUMNS}
|
||
for key in ('rules_json', 'rules_manifest', 'definition_json'):
|
||
if d.get(key):
|
||
try:
|
||
d[key + '_parsed'] = json.loads(d[key])
|
||
except Exception:
|
||
d[key + '_parsed'] = None
|
||
return d
|