agentmc/mcfold/servers.py

"""M10b 加载器服务端：让 server.py 能起 **带 mod 的** 服务器。

## 为什么必须做

`server.py` 原先是硬编码 vanilla 直起（`java -jar server.jar`），
**装不了 mod** ⇒ 选项 A（测试服装 mod）根本走不通。
而用户 2026-09-26 已定：加载器都要做（vanilla / fabric / forge / neoforge），
因为框架本身定位是「给 agent 用的 MC 启动器」。

## 隔离沿用既有三层（`tests/test_version_isolation.py` 守同一口径）

    store/<loader>-<version>/server/   内容根：jar / libraries（只读共享）
    instances/<name>/                 实例：mods / config（各装各的）
    runtime/<name>/                   运行时：world / logs / pid（实例独占）

★ 关键：**mods 放实例，不放内容根**。
  一个 vanilla 内容根和它旁边若干 mod 实例共用，内容根保持干净；
  同一个内容根可以起多个「装了不同 mod 的服」。

## 一条纪律

装加载器是**下载 + 执行第三方安装器**。所以：
- 版本与加载器都从**用户显式给的**参数来，不猜、不用「最新」；
- 安装产物必须**校验**（目录结构 + 关键文件存在）再报成功；
- 任何失败都返回 `{ok:false, why, how_to}`，不抛异常。
"""
from __future__ import annotations

import json
import os
import shutil
import subprocess
import time
from pathlib import Path
from typing import Any


BASE = Path(os.environ.get("MCVIEW_BASE") or "/home/newqqagent/mcview")
STORE = BASE / "store"
INSTANCES = BASE / "instances"
RUNTIME = BASE / "runtime"
LOADER_CACHE = BASE / "loaders"

#: 各加载器在 store 里的**目录名规则**（与既有内容根一致）
def content_root_name(loader: str, version: str) -> str:
    return "vanilla" if loader in ("", "vanilla") else f"{loader}-{version}"


# ---------- 安装器资产（下载一次，按版本/加载器归档） ----------

def installer_path(loader: str, version: str) -> Path:
    LOADER_CACHE.mkdir(parents=True, exist_ok=True)
    return LOADER_CACHE / f"{loader}-installer-{version}.jar"


def _download(url: str, dest: Path, timeout: int = 600) -> dict[str, Any]:
    """下载安装器到缓存目录。返回结构化结果（不抛异常）。"""
    import urllib.request
    dest.parent.mkdir(parents=True, exist_ok=True)
    tmp = dest.with_suffix(dest.suffix + ".part")
    try:
        with urllib.request.urlopen(url, timeout=timeout) as r, \
                open(tmp, "wb") as f:
            shutil.copyfileobj(r, f)
        size = tmp.stat().st_size
        if size < 10_000:
            tmp.unlink(missing_ok=True)
            return {"ok": False, "why": f"下载的文件太小（{size} 字节），多半不是安装器",
                    "how_to": "检查 URL 是否正确"}
        tmp.replace(dest)
        return {"ok": True, "path": str(dest), "bytes": size}
    except Exception as e:
        tmp.unlink(missing_ok=True)
        return {"ok": False, "why": f"下载失败 {type(e).__name__}: {e}",
                "how_to": "检查网络/镜像可达性"}


def fabric_installer(version: str) -> dict[str, Any]:
    """取 Fabric 安装器（已有缓存就复用）。"""
    p = installer_path("fabric", version)
    if p.exists() and p.stat().st_size > 10_000:
        return {"ok": True, "path": str(p), "cached": True}
    url = "https://maven.fabricmc.net/net/fabricmc/fabric-installer/1.0.1/fabric-installer-1.0.1.jar"
    return _download(url, p)


def fabric_server_launcher(version: str) -> dict[str, Any]:
    """取 `fabric-server-launch`，可直接 `java -jar` 起 Fabric 服务端。"""
    p = LOADER_CACHE / f"fabric-server-launch-{version}.jar"
    if p.exists() and p.stat().st_size > 10_000:
        return {"ok": True, "path": str(p), "cached": True}
    # 需要先按版本定 loader 版本
    meta = fabric_profile_meta(version)
    if not meta.get("ok"):
        return meta
    loader_v = meta["loader"]["version"]
    url = (f"https://maven.fabricmc.net/net/fabricmc/fabric-server-launch"
           f"/{loader_v}/fabric-server-launch-{loader_v}.jar")
    return _download(url, p)


def fabric_profile_meta(version: str) -> dict[str, Any]:
    """查 Fabric 对该 MC 版本的 loader/intermediary 组合。"""
    import urllib.request
    import json as _json
    try:
        with urllib.request.urlopen(
                f"https://meta.fabricmc.net/v2/versions/loader/{version}", timeout=60) as r:
            data = _json.loads(r.read())
    except Exception as e:
        return {"ok": False, "why": f"查 Fabric 元数据失败：{type(e).__name__}: {e}"}
    if not data:
        return {"ok": False, "why": f"Fabric 没有 {version} 的稳定 loader"}
    # 优先 stable
    for entry in data:
        if entry.get("loader", {}).get("stable"):
            return {"ok": True, **entry}
    return {"ok": True, **data[0]}


# ---------- 装一个「带 mod 的服」 ----------

def create_modded_server(
    name: str,
    *,
    version: str,
    loader: str,
    mod: list[str] | None = None,
    server_jar: str | os.PathLike[str] | None = None,
    java_heap: str = "2G",
    timeout: int = 900,
) -> dict[str, Any]:
    """在**实例层**里装一个 modded 服务端（内容根只读共享）。

    返回 `{ok, instance, mods_installed, ...}`；失败一律 `{ok:false, why, how_to}`。
    """
    import re
    if not re.match(r"^[A-Za-z0-9._-]{1,64}$", name or ""):
        return {"ok": False, "why": f"实例名不合法: {name!r}",
                "how_to": "只允许字母数字 . _ -，长度 1..64"}

    # ★ 先查加载器支持情况，**再建目录**：
    #   否则失败的加载器会留下一个「看着存在但没配好」的实例目录
    #   （实测：NeoForge 失败后实例目录仍在 ⇒ 以后 list_instances 会列出来，
    #    使用者以为装好了）。失败应当**不留痕**。
    if loader not in ("", "vanilla", "fabric"):
        return {
            "ok": False,
            "why": f"加载器 {loader!r} 的服务端还没接（当前支持 vanilla / fabric）",
            "how_to": "NeoForge/Forge 服务端需各自的 installer 流程；"
                      "客户端渲染路线见 scaffold/LOADERS.md",
            "left_nothing": True,
        }

    inst = INSTANCES / name
    mods_dir = inst / "mods"
    rt = RUNTIME / name
    inst.mkdir(parents=True, exist_ok=True)
    mods_dir.mkdir(parents=True, exist_ok=True)
    rt.mkdir(parents=True, exist_ok=True)

    # 实例元数据（补上之前缺的 —— 隔离的**依据**）
    meta_path = inst / "instance.json"
    meta = {
        "name": name,
        "version": version,
        "loader": loader,
        "role": "server",
        "created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
    }
    meta_path.write_text(json.dumps(meta, ensure_ascii=False, indent=2), encoding="utf-8")

    # 装 mod（★ 放实例层，内容根保持干净）
    installed: list[str] = []
    missing: list[str] = []
    for m in (mod or []):
        src = Path(m)
        if not src.exists():
            missing.append(str(m))
            continue
        shutil.copy2(src, mods_dir / src.name)
        installed.append(src.name)
    if missing:
        return {"ok": False, "why": f"这些 mod 找不到：{missing}",
                "how_to": "给绝对路径；已装的部分保留在 " + str(mods_dir),
                "installed": installed}

    if loader in ("", "vanilla"):
        if server_jar is None:
            return {"ok": False, "why": "vanilla 服务端需要 server_jar",
                    "how_to": "传 --server-jar 指向已有的 server.jar"}
        shutil.copy2(server_jar, inst / "server.jar")
        launch = {"cmd": ["java", f"-Xmx{java_heap}", f"-Xms{java_heap}",
                          "-jar", "server.jar", "nogui"]}
    elif loader == "fabric":
        got = fabric_server_launcher(version)
        if not got.get("ok"):
            return {"ok": False, "why": got.get("why"), "how_to": got.get("how_to"),
                    "installed": installed}
        shutil.copy2(got["path"], inst / "fabric-server-launch.jar")
        launch = {"cmd": ["java", f"-Xmx{java_heap}", f"-Xms{java_heap}",
                          "-jar", "fabric-server-launch.jar", "nogui"]}
    # 写启动脚本（内容根零写入）
    script = inst / "start.sh"
    script.write_text(
        "#!/usr/bin/env bash\n"
        f"# mcfold {loader}/{version} 服务端（实例 {name}）\n"
        "set -euo pipefail\n"
        'cd "$(dirname "$0")"\n'
        f"exec {' '.join(launch['cmd'])}\n", encoding="utf-8")
    script.chmod(0o755)

    # 验收：关键文件必须在位
    need = ["start.sh", "eula.txt"] if loader != "fabric" else ["start.sh"]
    missing_files = [f for f in need if not (inst / f).exists()]
    if loader != "fabric":
        (inst / "eula.txt").write_text("eula=true\n", encoding="utf-8")

    return {
        "ok": True,
        "instance": str(inst),
        "runtime": str(rt),
        "loader": loader,
        "version": version,
        "mods_installed": installed,
        "mods_dir": str(mods_dir),
        "launch_cmd": launch["cmd"],
        "start_script": str(script),
        "note": ("内容根保持只读（mods 在实例层）⇒ 同一内容根可起多个不同 mod 组合"
                 if installed else "未装 mod"),
        "next": f"cd {inst} && ./start.sh   （首次会生成世界，约 1-2 分钟）",
    }