feat(mcfold): P7 实例层(传入文件夹→拉起实例→读mod注册方块表→连服,含端到端验证)+ 统一测试入口

This commit is contained in:
pi
2026-09-25 14:36:21 +08:00
parent b0182de8c7
commit 5038c7f679
9 changed files with 517 additions and 3 deletions

View File

@ -60,6 +60,10 @@
**关键语义**:`unloaded`(没收到)与 `air`(真是空气)**必须可区分**
mod 方块计入 `unknown_state_ids`,世界摘要带告警
- [x] **P7** 实例层(`mcfold/instance.py`,8 项回归全绿)
**传入文件夹路径 → 拉起实例 → 读 mod 注册方块表 → 连服**
含:加载器版本归一化、裸名展开、歧义处理、越界 state id 归因到 mod
## 累计测试
| 套件 | 项数 |
@ -69,5 +73,9 @@
| 区块解析 | 11 |
| mod jar | 8 |
| 世界模型 | 7 |
| mod jar | 8 |
| 实例层 | 8 |
| 跨实现互操作 | 2 |
| **合计** | **40** |
| **合计** | **48** |
统一入口:`./run_tests.sh`

Binary file not shown.

View File

@ -37,6 +37,7 @@ class Client:
self.proto: dict = {}
self.proto_num: int = 0
self.compression: int = -1
# 允许直接指定协议数据文件(优先),否则按 data_root 找
self._path = Path(protocol_path) if protocol_path else None
# ---------- 协议数据 ----------
@ -67,6 +68,12 @@ class Client:
def _data_root(self) -> Path:
if self._path:
# 传进来的可能是 data 根,也可能是 data/pc —— 都容忍
for c in (self._path, self._path / "pc", self._path.parent):
if (c / "pc").is_dir():
return c
if c.name == "pc" and c.is_dir():
return c.parent
return self._path
cands = [
Path("/home/newqqagent/plugin_data/mc/runtime/node_modules/"

291
mcfold/instance.py Normal file
View File

@ -0,0 +1,291 @@
"""实例层:传入文件夹路径 → 拉起一个完整实例。
这是用户口径的落点(2026-09-25):
"支持传入文件夹路径,按照版本文件夹内容,拉起实例,
并读取 mod jar 注册方块表。然后链接服务端,获取到数据。"
实例文件夹 == 一个完整的 MC 客户端安装(与官方启动器一致):
instances/<name>/ ← 传进来的就是它(== .minecraft)
versions/<verId>/<verId>.json ← 决定协议版本(含加载器继承)
versions/<verId>/<verId>.jar ← 客户端本体(渲染器)
mods/*.jar ← mod(决定方块名清单)
config/ options.txt saves/ ← 实例私有
本模块把这些组合起来,并对**每一项能力标注可信度**:
* 协议版本 → 可信(来自 version.json + inheritsFrom)
* 方块名 → 原版可信;mod 方块名只有名字、没有 state id(附录 D)
* 世界数据 → 连接后才有;未加载区域显式可辨
"""
from __future__ import annotations
import os
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
from .chunk import BlockRegistry, MIN_Y, WORLD_HEIGHT, NUM_SECTIONS
from .client import Client
from .modjar import ModInfo, scan_mods, summarize
from .version import VersionInfo, list_versions, resolve_version
from .world import World
# 默认的内容根(libraries/assets 可共享,可只读)
DEFAULT_CONTENT_ROOTS = (
"/home/newqqagent/mcview/store",
"/home/newqqagent/plugindev/mcfold/store",
)
# 内置的方块数据(minecraft-data 的 blocks.json,按 majorVersion 组织)
DEFAULT_DATA_ROOT = (
"/home/newqqagent/plugin_data/mc/runtime/node_modules/"
"minecraft-data/minecraft-data/data/pc"
)
@dataclass
class InstanceStatus:
"""给 agent 看的实例现状(含每一项的可信度)。"""
path: str
version_id: str = ""
mc_version: str = ""
loader: str = ""
protocol: int = 0
mods_loaded: int = 0
mods_failed: int = 0
mod_blocks: int = 0
connected: bool = False
loaded_chunks: int = 0
unknown_states: int = 0
notes: list[str] = field(default_factory=list)
def to_dict(self) -> dict[str, Any]:
return {
"path": self.path,
"version": {
"id": self.version_id, "mc": self.mc_version,
"loader": self.loader, "protocol": self.protocol,
"trusted": "来自 version.json(含 inheritsFrom 归一化)",
},
"mods": {
"loaded": self.mods_loaded, "failed": self.mods_failed,
"block_names": self.mod_blocks,
"state_ids": None,
"note": "只有方块名,无 state id(附录 D)",
},
"world": {
"connected": self.connected,
"loaded_chunks": self.loaded_chunks,
"unknown_states": self.unknown_states,
},
"notes": self.notes,
}
class Instance:
"""一个 MC 实例。"""
def __init__(
self,
path: str | os.PathLike[str],
prefer_version: str | None = None,
content_root: str | os.PathLike[str] | None = None,
data_root: str | os.PathLike[str] | None = None,
):
self.path = Path(path)
self.prefer_version = prefer_version
self.content_root = Path(content_root) if content_root else self._find_content_root()
self.data_root = Path(data_root) if data_root else Path(DEFAULT_DATA_ROOT)
self.version: VersionInfo | None = None
self.mods: list[ModInfo] = []
self.registry: BlockRegistry | None = None
self.world = World()
self.client: Client | None = None
self.notes: list[str] = []
# ---------- 内容根 ----------
def _find_content_root(self) -> Path | None:
"""找可共享的内容根(含 libraries/ 的那个)。
实例文件夹自己带 libraries/ 时优先用它(自包含安装)。
"""
if (self.path / "libraries").is_dir():
return self.path
for c in DEFAULT_CONTENT_ROOTS:
cp = Path(c)
if cp.is_dir():
# 内容根下按 loader-version 组织
for sub in cp.iterdir():
if sub.is_dir() and (sub / "libraries").is_dir():
return sub
return None
# ---------- 装载 ----------
def load(self) -> "Instance":
"""读实例文件夹:版本 + mod + 方块注册表。"""
self.version = resolve_version(self.path, self.prefer_version)
if self.version is None:
raise FileNotFoundError(
f"{self.path} 下没有可用版本(需要 versions/<id>/<id>.json)")
mods_dir = self.path / "mods"
self.mods = scan_mods(mods_dir)
self._load_registry()
self._build_world()
if self.mods:
bad = [m for m in self.mods if not m.ok]
self.notes.append(
f"本实例含 {len(self.mods)} 个 mod"
+ (f"({len(bad)} 个解析失败)" if bad else "")
+ ";mod 方块仅有名字、无 state id,"
"世界查询遇到未注册 state id 会显示 unknown(id)。"
)
return self
def _load_registry(self) -> None:
"""载入方块注册表。
注意 minecraft-data 的组织方式:1.20.1 目录里只有 version.json
(majorVersion=1.20),真正的数据在 1.20/ 下 —— 所以要按 majorVersion 找。
"""
if self.version is None:
return
mc = self.version.mc_version
cands = [self.data_root / mc / "blocks.json"]
# 按 majorVersion 回退(1.20.1 -> 1.20)
parts = mc.split(".")
for take in range(len(parts) - 1, 0, -1):
cands.append(self.data_root / ".".join(parts[:take]) / "blocks.json")
for c in cands:
if c.is_file():
self.registry = BlockRegistry.from_json_file(str(c))
return
self.notes.append(f"未找到 {mc} 的方块数据 ⇒ 世界查询将只返回 state id")
def _build_world(self) -> None:
self.world = World(self.registry)
if self.version is not None:
base = self.version.raw or {}
# 1.18+ 的世界范围可由 version.json 提供;缺省用 1.20 的实测值
self.world.min_y = base.get("minY", MIN_Y)
# worldHeight 常见写法:height 或 直接用默认 384
self.world.num_sections = base.get("worldHeight", WORLD_HEIGHT) >> 4
# ---------- 连接 ----------
def connect(self, host: str, port: int, username: str) -> Client:
"""连服务端(协议版本由实例文件夹决定,不硬编码)。"""
if self.version is None:
self.load()
assert self.version is not None
self.client = Client(host, port, username, self.version.mc_version,
protocol_path=self.data_root)
self.client.load_protocol()
return self.client
# ---------- 能力查询 ----------
def mod_block_names(self) -> list[str]:
"""本实例 mod 提供的方块名(带命名空间)。"""
out: list[str] = []
for m in self.mods:
if m.ok:
out.extend(m.qualified_blocks())
return sorted(set(out))
def knows_block_name(self, name: str) -> bool:
"""该方块名是否可用(原版注册表里有,或来自已装 mod)。
用途:在 agent 下指令前校验,避免"你说的这个方块这服没有"。
命名空间处理(实测需求):agent 常常只说 `cask` 而不是
`expandeddelight:cask`。所以无命名空间时也去 mod 清单里找:
* 唯一命中 ⇒ 认识
* 多个 mod 都有同名方块 ⇒ **不**当成认识(歧义,应当让 agent 说全名)
"""
if self.registry:
bare = name.split(":", 1)[1] if ":" in name else name
if bare in self.registry.by_name:
return True
if ":" in name:
return name in set(self.mod_block_names())
# 无命名空间:在 mod 方块名的「裸名」里查,但要歧义敏感
return len(self._mod_hits_bare(name)) == 1
def _mod_hits_bare(self, bare: str) -> list[str]:
"""裸名 → 命中的完整限定名列表(用于歧义判定)。"""
hits: list[str] = []
for m in self.mods:
if m.ok and bare in m.block_names:
hits.extend(m.qualified_blocks_for(bare))
return hits
def expand_block_name(self, name: str) -> str | None:
"""把裸名展开为完整限定名;歧义或无命中返回 None。"""
if ":" in name:
return name if name in set(self.mod_block_names()) else None
hits = self._mod_hits_bare(name)
return hits[0] if len(hits) == 1 else None
def resolve_block_name(self, state_id: int) -> str:
"""state id → 名字,带 mod 归因(拿不到名字时明说是可能来自 mod)。"""
if self.registry is None:
return f"unknown({state_id})"
n = self.registry.name_of(state_id)
if n is not None:
return n
if self.mods:
return (f"unknown({state_id}) —— 本机注册表无此 state id;"
f"本实例有 {len([m for m in self.mods if m.ok])} 个 mod,"
f"极可能是 mod 方块")
return f"unknown({state_id})"
def _protocol_of(self, mc_version: str) -> int:
"""从 minecraft-data 的 version.json 读协议号(无需连接)。
注意:传入的必须是**归一化后的** mc 版本(version.py 已保证)。
加载器版本的 id(如 1.20.1-forge-47.4.0)在这里查不到,
所以调用方先用 VersionInfo.mc_version。
"""
import json as _json
parts = mc_version.split(".")
cands = [self.data_root / mc_version / "version.json"]
# 按 majorVersion 回退(1.20.1 -> 1.20)
for take in range(len(parts) - 1, 0, -1):
cands.append(self.data_root / ".".join(parts[:take]) / "version.json")
for cand in cands:
if cand.is_file():
try:
return int(_json.loads(cand.read_text()).get("version", 0))
except Exception:
return 0
return 0
# ---------- 摘要 ----------
def status(self) -> InstanceStatus:
st = InstanceStatus(path=str(self.path))
if self.version:
st.version_id = self.version.version_id
st.mc_version = self.version.mc_version
st.loader = self.version.loader
if self.client:
st.protocol = self.client.proto_num
elif self.version:
st.protocol = self._protocol_of(self.version.mc_version)
st.mods_loaded = len([m for m in self.mods if m.ok])
st.mods_failed = len([m for m in self.mods if not m.ok])
st.mod_blocks = len(self.mod_block_names())
st.loaded_chunks = self.world.loaded_count
st.unknown_states = len(self.world.unknown_state_ids)
st.notes = list(self.notes)
return st
def mods_summary(self) -> dict[str, Any]:
return summarize(self.mods)
def world_summary(self) -> dict[str, Any]:
return self.world.summary()

View File

@ -54,8 +54,18 @@ class ModInfo:
def qualified_blocks(self) -> list[str]:
"""带命名空间的方块名,形如 expandeddelight:cask。"""
ns = self.mod_id or (self.namespaces[0] if self.namespaces else "")
return [f"{ns}:{n}" for n in self.block_names] if ns else list(self.block_names)
return [f"{self.namespace}:{n}" for n in self.block_names]
def qualified_blocks_for(self, block_name: str) -> list[str]:
"""指定裸方块名在本 mod 下的完整限定名(不在则空)。"""
if block_name not in self.block_names:
return []
return [f"{self.namespace}:{block_name}"]
@property
def namespace(self) -> str:
"""该 mod 的命名空间(通常 == mod_id)。"""
return self.mod_id or (self.namespaces[0] if self.namespaces else "")
# ---------------- 元数据解析 ----------------

23
run_tests.sh Executable file
View File

@ -0,0 +1,23 @@
#!/usr/bin/env bash
# 统一测试入口。用法: ./run_tests.sh
cd "$(dirname "$0")"
total=0; fail=0
for t in tests/test_*.py; do
name=$(basename "$t")
out=$(python3 "$t" 2>&1)
if echo "$out" | grep -qE "全部通过|通过$"; then
n=$(echo "$out" | grep -c "^ ✓")
total=$((total+n))
printf ' %-24s ✓ %2d 项\n' "$name" "$n"
else
fail=$((fail+1))
printf ' %-24s ✗ 失败\n' "$name"
echo "$out" | tail -5 | sed 's/^/ /'
fi
done
echo
if [ $fail -eq 0 ]; then
echo "全部通过:$total 项,0 失败"
else
echo "失败 $fail 个套件(共 $total 项通过)"; exit 1
fi

175
tests/test_instance.py Normal file
View File

@ -0,0 +1,175 @@
"""实例层回归:传入文件夹 → 拉起实例(版本 + mod + 方块表)。"""
import sys, os, json, zipfile, tempfile, shutil
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from mcfold.instance import Instance
from mcfold.modjar import scan_mods
BLOCKS_MOD = "/home/minecraft/mods/[扩充乐事] expandeddelight-0.1.4.jar"
FORGE_MOD = "/home/mc-test/mods/curtain-mc1.20.1-1.3.2.jar"
REAL_INST = "/home/newqqagent/mcview/store/vanilla-1.20.1"
def build_instance(with_mods=True, version_id="1.20.1", mc=None):
"""造一个临时实例文件夹(完整 MC 客户端安装形状)。
version_id 是目录名(可能是加载器 id);mc 是它继承的原版版本。
"""
d = tempfile.mkdtemp(prefix="mcfold-inst-")
ver = version_id
vdir = os.path.join(d, "versions", ver)
os.makedirs(vdir, exist_ok=True)
obj = {"id": ver, "mainClass": "net.minecraft.client.main.Main", "assets": "5"}
if mc and mc != ver:
obj["inheritsFrom"] = mc
obj["mainClass"] = ("io.github.zekerzhayard.forgewrapper.installer.Main"
if "forge" in ver else
"net.fabricmc.loader.impl.launch.knot.KnotClient")
with open(os.path.join(vdir, f"{ver}.json"), "w") as f:
json.dump(obj, f)
os.makedirs(os.path.join(d, "mods"), exist_ok=True)
if with_mods:
for src in (BLOCKS_MOD, FORGE_MOD):
if os.path.isfile(src):
shutil.copy(src, os.path.join(d, "mods"))
return d
def test_load_from_folder():
"""★ 用户口径:传入文件夹路径 → 拉起实例。"""
d = build_instance()
try:
inst = Instance(d).load()
st = inst.status()
assert st.mc_version == "1.20.1", st.mc_version
assert st.protocol == 763, st.protocol
assert st.loader == "vanilla", st.loader
print(f" ✓ 传入文件夹拉起实例: {st.version_id} mc={st.mc_version} 协议={st.protocol}")
finally:
shutil.rmtree(d, ignore_errors=True)
def test_mods_registered():
"""★ 读取 mod jar 注册方块表。"""
d = build_instance()
try:
inst = Instance(d).load()
ms = inst.mods_summary()
assert ms["loaded"] == 2, ms
assert ms["block_name_count"] > 20, ms
names = inst.mod_block_names()
assert "expandeddelight:cask" in names, names[:5]
# 诚实边界必须体现在摘要里
assert "state id" in ms["note"]
print(f" ✓ mod 注册: {ms['loaded']} 个 mod, {ms['block_name_count']} 个方块名")
finally:
shutil.rmtree(d, ignore_errors=True)
def test_loader_version_normalization():
"""forge 实例:mc_version 必须归一化(否则协议选错)。"""
d = build_instance(version_id="1.20.1-forge-47.4.0", mc="1.20.1")
try:
inst = Instance(d, prefer_version="1.20.1-forge-47.4.0").load()
st = inst.status()
assert st.mc_version == "1.20.1", st.mc_version
assert st.loader == "forge", st.loader
assert st.protocol == 763, st.protocol
print(f" ✓ forge 归一化: id={st.version_id} → mc={st.mc_version} 协议={st.protocol}")
finally:
shutil.rmtree(d, ignore_errors=True)
def test_block_name_validation():
"""方块名校验:裸名可展开;歧义不误判。"""
d = build_instance()
try:
inst = Instance(d).load()
assert inst.knows_block_name("cask")
assert inst.expand_block_name("cask") == "expandeddelight:cask"
assert inst.knows_block_name("stone")
assert not inst.knows_block_name("definitely_not_a_block")
print(" ✓ 方块名校验: 裸名展开正确, 不存在的方块判否")
finally:
shutil.rmtree(d, ignore_errors=True)
def test_ambiguous_bare_name():
"""两个 mod 有同名方块时,裸名必须判为**不可用**(歧义)。"""
d = build_instance(with_mods=False)
try:
for mid in ("moda", "modb"):
with zipfile.ZipFile(os.path.join(d, "mods", f"{mid}.jar"), "w") as zf:
zf.writestr("fabric.mod.json",
json.dumps({"id": mid, "name": mid, "version": "1"}))
zf.writestr(f"assets/{mid}/blockstates/shared_block.json", "{}")
inst = Instance(d).load()
assert not inst.knows_block_name("shared_block"), "有歧义却判为认识"
assert inst.knows_block_name("moda:shared_block"), "限定名应可用"
print(" ✓ 歧义处理: 裸名(两 mod 同名)判否, 限定名可判是")
finally:
shutil.rmtree(d, ignore_errors=True)
def test_real_install_folder():
"""对真实安装目录验证(无 mod 的 vanilla)。"""
if not os.path.isdir(REAL_INST):
print(" - 真实安装目录不存在,跳过"); return
inst = Instance(REAL_INST).load()
st = inst.status()
assert st.mc_version == "1.20.1" and st.protocol == 763, (st.mc_version, st.protocol)
assert inst.registry is not None and len(inst.registry.blocks) > 900
print(f" ✓ 真实安装: {st.version_id} 协议={st.protocol} 原版方块={len(inst.registry.blocks)}")
def test_resolve_block_name_mod_attribution():
"""越界 state id 的报错必须**归因到 mod**(让降级可信)。"""
d = build_instance()
try:
inst = Instance(d).load()
msg = inst.resolve_block_name(99999)
assert msg.startswith("unknown("), msg
assert "mod" in msg, msg
print(f" ✓ 越界归因: {msg[:46]}…")
finally:
shutil.rmtree(d, ignore_errors=True)
def test_e2e_connect_from_instance():
"""★ 端到端:实例文件夹 → 连服 → 登录(协议由文件夹决定,不硬编码)。"""
import socket, subprocess, time
HOST, PORT = "127.0.0.1", 25599
runtime = "/home/newqqagent/plugin_data/mc/runtime"
def reachable():
try:
socket.create_connection((HOST, PORT), timeout=2).close(); return True
except OSError:
return False
proc = None
if not reachable():
src = os.path.join(os.path.dirname(os.path.abspath(__file__)), "support_testserver.js")
if not os.path.isfile(src):
print(" - 无 support_testserver.js,跳过"); return
shutil.copy(src, runtime)
proc = subprocess.Popen(["node", "support_testserver.js", str(PORT)], cwd=runtime,
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
for _ in range(50):
if reachable(): break
time.sleep(0.2)
if not reachable():
print(" - 服务端不可用,跳过")
if proc: proc.kill()
return
d = build_instance()
try:
inst = Instance(d).load()
c = inst.connect(HOST, PORT, "McfoldBot")
r = c.run_smoke()
assert r["protocol"] == 763 and r["login_ok"], r
print(f" ✓ 端到端: 文件夹→协议{r['protocol']}→登录成功 (mod {len(inst.mods)} 个)")
c.close()
finally:
shutil.rmtree(d, ignore_errors=True)
if proc:
proc.terminate()
try: proc.wait(timeout=5)
except Exception: proc.kill()
try: os.remove(os.path.join(runtime, "support_testserver.js"))
except OSError: pass
if __name__ == "__main__":
for fn in [test_load_from_folder, test_mods_registered, test_loader_version_normalization,
test_block_name_validation, test_ambiguous_bare_name, test_real_install_folder,
test_resolve_block_name_mod_attribution, test_e2e_connect_from_instance]:
fn()
print("\n实例层:全部通过")