chuanqi/apilot/bridge.py

"""Lupa 桥: VM 构建 + stub 装配 + 游戏 Lua 只读加载。

Task 6: bootstrap_runtime 建 VM+装 stub+配 package.path+注 networkUtil 公共面薄复刻
        +加载 netMessageDef;load_game_module 按 require 语义加载游戏原版模块。
        Task 7 (host.py) 依赖本模块接口: LUA_SRC / bootstrap_runtime / load_game_module。
"""
from __future__ import annotations

import json
import os
from typing import Any

import lupa.luajit21 as lj  # type: ignore

from apilot.stubs import install_all
from apilot.stubs.env import EnvData
from apilot.stubs.netclient import NetClientStub
from apilot.stubs.networkctl import NetworkCtlStub
from apilot.stubs.scheduler import Scheduler

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
LUA_SRC = os.path.join(ROOT, "lua脚本解密源码", "mod_fgcq", "stab", "scripts")


# 协议定义文件名(brief 指定第一个加载的游戏原文件)
_NETMSG_DEF = "network/netMessageDef.lua"


def bootstrap_runtime(nc: NetClientStub, ctl: NetworkCtlStub, sched: Scheduler,
                      env: EnvData, state_dir: str) -> Any:
    """建 Lupa VM,装 stub,设 package.path,注 class()/cjson/LuaSendMsg/... 公共面,加载 netMessageDef。"""
    lua = lj.LuaRuntime(unpack_returned_tuples=True)
    install_all(lua, nc, ctl, sched, env, state_dir)

    # package.path 指向游戏 scripts(只读),优先于系统默认
    lua.globals()["_LUA_SRC"] = LUA_SRC
    lua.execute("package.path = _LUA_SRC .. '/?.lua;' .. package.path")

    # PureMVC 极简 class()(单继承),够 framework.init 链: Facade/Mediator/Proxy/Notifier/...
    lua.execute(r"""
class = function(name, super)
    local cls = {__cname = name}
    if super ~= nil then
        setmetatable(cls, {__index = super})
        cls.super = super
    else
        setmetatable(cls, {__index = cls})
    end
    function cls.new(...)
        local inst = setmetatable({}, cls)
        if cls.ctor then cls.ctor(inst, ...) end
        return inst
    end
    return cls
end
""")

    # cjson 桥: Lua table<->JSON 字符串 走 Python json
    g = lua.globals()

    def _to_jsonable(obj: Any) -> Any:
        if isinstance(obj, dict):
            return {k: _to_jsonable(v) for k, v in obj.items()}
        if hasattr(obj, "items"):
            items = list(obj.items())
            if items and all(isinstance(k, int) and k > 0 for k, _ in items):
                keys = sorted(k for k, _ in items)
                if keys == list(range(1, len(items) + 1)):
                    return [_to_jsonable(v) for k, v in sorted(items)]
            return {k: _to_jsonable(v) for k, v in items}
        return obj

    def _enc(t: Any) -> str:
        return json.dumps(_to_jsonable(t), ensure_ascii=False, separators=(",", ":"))

    def _dec(s: str) -> Any:
        d = json.loads(s)
        return d if isinstance(d, dict) else {}

    g["_PY_cjson_encode"] = _enc
    g["_PY_cjson_decode"] = _dec

    # 薄复刻 networkUtil.lua 公共面(签名对齐;不真加载以免引入更多 C++ 面)
    lua.execute(r"""
cjson = cjson or {encode = function(t) return _PY_cjson_encode(t) end,
                  decode = function(s) return _PY_cjson_decode(s) end,
                  safe = nil}
slen = slen or function(s) return #s end

function LuaSendMsg(msgId, recog, p1, p2, p3, dataStr, dataLen)
    networkCtl:SendMsg(msgId, recog, p1, p2, p3, dataStr, dataLen)
end
function LuaRegisterMsgHandler(msgId, fn)
    networkCtl:RegisterLuaHandler(msgId, fn)
end
function SendTableToServer(msgId, t)
    local s = cjson.encode(t)
    LuaSendMsg(msgId, 0, 0, 0, 0, s, #s)
end
""")

    # 加载第一个游戏原文件(协议定义)
    nmd_path = os.path.join(LUA_SRC, _NETMSG_DEF)
    with open(nmd_path, encoding="utf-8") as f:
        lua.execute(f.read())
    return lua


def load_game_module(lua: Any, modname: str) -> None:
    """按 require 语义加载游戏原版模块(modname 用点号或斜杠均可)。"""
    lua.execute(f"require('{modname}')")