squaddocker/scripts/squad-instance.py

#!/usr/bin/env python3
"""
通过 MCSManager API 批量创建/管理 Squad 实例 —— 为「方便多开」而写。

端口按实例号自动排布,不会撞车:
    实例 N:  游戏 7787+(N-1)*10   查询 27165+(N-1)*10
             Beacon 15000+(N-1)*10  RCON 21114+(N-1)*10
间隔取 10 是因为游戏端口和查询端口各占 N 和 N+1 两个,只差 1 会重叠。

前置条件(都是实测踩出来的坑,缺一个 API 就不通):
  1. 面板 SystemConfig/config.json 里 enableApiKey 必须为 true。
     默认是 false,API 文档没提。改完 systemctl restart mcsm-web。
  2. 目标机若无法访问 Docker Hub,镜像必须先搬到私有注册表,
     并且面板 daemon 以 root 运行 —— 需 sudo docker login。
  3. Squad 必须用 networkMode=host,bridge 模式下游戏内浏览器搜不到服务器。
  4. 端口必须写进 startCommand 作为命令行参数。面板模板默认只有
     ./SquadGameServer.sh 不带参数,多开时第二个实例会抢第一个的端口。

环境变量:
    MCSM_URL       面板地址,如 http://1.2.3.4:23333
    MCSM_APIKEY    面板用户信息页生成的 apikey
    MCSM_DAEMON    daemonId(用 `list-daemons` 查)
    SQUAD_IMAGE    镜像地址(默认走 ACR)
    SQUAD_DATA_ROOT 数据根目录(默认 /srv/squad),实例 N 用 <root>/instN

用法:
    ./squad-instance.py list-daemons
    ./squad-instance.py list
    ./squad-instance.py create <N> [人数上限]
    ./squad-instance.py start|stop|restart|kill <uuid>
    ./squad-instance.py status <uuid>
    ./squad-instance.py logs <uuid> [KB]
    ./squad-instance.py cmd <uuid> "<控制台命令>"
    ./squad-instance.py delete <uuid> [--with-files]

创建后必须做的一步:
    RCON 密码要同时写进服务端的 ServerConfig/Rcon.cfg,否则服务端不会
    启动 RCON(端口不监听)。本脚本会打印需要执行的命令。
"""

import http.client
import json
import os
import pathlib
import secrets
import stat
import string
import sys
import time
import urllib.error
import urllib.parse
import urllib.request

STATUS = {-1: "busy(忙)", 0: "stopped(已停止)", 1: "stopping(停止中)",
          2: "starting(启动中)", 3: "running(运行中)"}

BASE = {"port": 7787, "query": 27165, "beacon": 15000, "rcon": 21114}
STEP = 10


def ports_for(n):
    """实例号 → 四个端口。n 从 1 开始。"""
    if n < 1:
        sys.exit("实例号必须 >= 1")
    off = (n - 1) * STEP
    return {k: v + off for k, v in BASE.items()}


def env(name, default=None, required=False):
    v = os.environ.get(name, default)
    if required and not v:
        sys.exit(f"错误: 需要设置环境变量 {name}")
    return v


def call(method, path, params=None, body=None):
    url_base = env("MCSM_URL", required=True).rstrip("/")
    q = dict(params or {})
    q["apikey"] = env("MCSM_APIKEY", required=True)
    url = f"{url_base}{path}?{urllib.parse.urlencode(q)}"
    data = json.dumps(body).encode() if body is not None else None
    req = urllib.request.Request(url, data=data, method=method)
    # 这两个头是 MCSManager 强制要求的,缺了直接被拒
    req.add_header("X-Requested-With", "XMLHttpRequest")
    req.add_header("Content-Type", "application/json; charset=utf-8")
    # 面板偶发 RemoteDisconnected,重试两次
    last = None
    for attempt in range(3):
        try:
            with urllib.request.urlopen(req, timeout=60) as r:
                out = json.loads(r.read().decode() or "{}")
            break
        except urllib.error.HTTPError as e:
            # 不打印 url —— 它含 apikey
            sys.exit(f"HTTP {e.code} {method} {path}\n"
                     f"{e.read().decode(errors='replace')[:400]}")
        except (urllib.error.URLError, ConnectionError,
                http.client.RemoteDisconnected) as e:
            last = e
            if attempt < 2:
                time.sleep(2)
    else:
        sys.exit(f"连接失败 {method} {path}: {last}")
    if out.get("status") != 200:
        detail = out.get("data")
        if isinstance(detail, str) and "enableApiKey" in detail:
            sys.exit("API key 被面板禁用。请在目标机执行:\n"
                     "  sudo python3 - <<'EOF'\n"
                     "  import json,pathlib\n"
                     "  p=pathlib.Path('/opt/mcsmanager/web/data/SystemConfig/config.json')\n"
                     "  d=json.loads(p.read_text()); d['enableApiKey']=True\n"
                     "  p.write_text(json.dumps(d,indent=2,ensure_ascii=False))\n"
                     "  EOF\n"
                     "  sudo systemctl restart mcsm-web")
        sys.exit(f"API 返回 {out.get('status')}: {detail}")
    return out


def gen_password(n=24):
    alphabet = string.ascii_letters + string.digits
    return "".join(secrets.choice(alphabet) for _ in range(n))


def instance_body(n, maxplayers):
    p = ports_for(n)
    data_dir = f"{env('SQUAD_DATA_ROOT', '/srv/squad')}/inst{n}"
    pw = gen_password()
    return pw, data_dir, p, {
        "nickname": f"Squad-{n}",
        # 端口必须作为命令行参数 —— 这是多开不冲突的关键
        "startCommand": (f"./SquadGameServer.sh Port={p['port']} "
                         f"QueryPort={p['query']} RCONPORT={p['rcon']} "
                         f"beaconport={p['beacon']} "
                         f"FIXEDMAXPLAYERS={maxplayers} FIXEDMAXTICKRATE=50"),
        "stopCommand": "^C",          # Squad 没有 stop 命令,只能靠信号
        "cwd": data_dir,              # 面板会把它挂载到容器内 /data
        # Linux 上必须 utf8。面板模板默认 gbk(Windows 中文环境),会乱码
        "ie": "utf8", "oe": "utf8", "fileCode": "utf8",
        "type": "universal",
        "tag": [],
        "endTime": 0,
        "processType": "docker",
        "updateCommand": ('/home/steam/steamcmd/steamcmd.sh -net_ipv4 '
                          '+force_install_dir "/data" +login anonymous '
                          '"+app_update 403240 validate" +quit'),
        "actionCommandList": [],
        "crlf": 1,
        "enableRcon": True,
        "rconPassword": pw,
        "rconPort": p["rcon"],
        "rconIp": "127.0.0.1",
        "terminalOption": {"haveColor": False, "pty": True},
        "eventTask": {"autoStart": False, "autoRestart": True, "ignore": False},
        # 面板模板默认 25565 是 Minecraft 的,Squad 要指向 Steam 查询端口
        "pingConfig": {"ip": "127.0.0.1", "port": p["query"], "type": 1},
        "docker": {
            "containerName": f"squad-{n}",
            "image": env("SQUAD_IMAGE",
                         "squaddocker.azurecr.cn/steam-game-runtime:latest"),
            # 0 表示无上限。内存小的机器不设会被 OOM 拖垮整机
            "memory": int(env("SQUAD_MEM_MB", "6144")),
            "ports": [],              # host 网络下端口映射无意义
            "extraVolumes": [],
            "maxSpace": None,
            "network": None,
            "io": None,
            "networkMode": "host",    # Squad 的硬要求
            "networkAliases": [],
            "cpusetCpus": "",
            "cpuUsage": 0,
            "workingDir": "/data",
            "changeWorkdir": False,
            "env": [],
        },
    }


def main():
    if len(sys.argv) < 2:
        sys.exit(__doc__)
    cmd, a = sys.argv[1], sys.argv[2:]

    if cmd == "list-daemons":
        for d in call("GET", "/api/service/remote_services_list")["data"]:
            print(f"  daemonId={d['uuid']}  {d['ip']}:{d['port']}  "
                  f"available={d['available']}")
        return

    daemon = env("MCSM_DAEMON", required=True)

    if cmd == "list":
        r = call("GET", "/api/service/remote_service_instances",
                 {"daemonId": daemon, "page": 1, "page_size": 50,
                  "instance_name": "", "status": ""})
        for i in (r["data"] or {}).get("data") or []:
            c = i.get("config") or {}
            print(f"  {i['instanceUuid']}  {c.get('nickname'):<12} "
                  f"{STATUS.get(i.get('status'), '?')}")
            print(f"      {c.get('startCommand')}")
        return

    if cmd == "create":
        if not a:
            sys.exit("用法: create <实例号N> [人数上限]")
        n = int(a[0])
        maxplayers = int(a[1]) if len(a) > 1 else 50
        pw, data_dir, p, body = instance_body(n, maxplayers)
        r = call("POST", "/api/instance", {"daemonId": daemon}, body)
        uuid = r["data"]["instanceUuid"]
        print(f"✅ 实例已创建: {uuid}  (Squad-{n})")
        print(f"   端口  游戏={p['port']}/{p['port']+1}  查询={p['query']}/{p['query']+1}  "
              f"Beacon={p['beacon']}  RCON={p['rcon']}")
        print(f"   目录  {data_dir}  (容器内 /data)")
        print()
        print("⚠️  还需两步,否则起不来或 RCON 不可用:")
        print(f"   1) 建目录:  sudo mkdir -p {data_dir}")
        print(f"   2) 首次启动会下载约 14GB 游戏文件(实测十几分钟)。")
        print(f"      下载完成后把 RCON 密码写进服务端配置,否则服务端不启动 RCON:")
        print(f"      sudo sed -i -E 's/^Password=.*$/Password={pw}\\r/' \\")
        print(f"        {data_dir}/SquadGame/ServerConfig/Rcon.cfg")
        print(f"      然后 restart 实例。")
        return

    if not a:
        sys.exit(f"用法: {cmd} <uuid>")
    uuid = a[0]

    if cmd in ("start", "stop", "restart", "kill"):
        ep = "open" if cmd == "start" else cmd
        call("GET", f"/api/protected_instance/{ep}",
             {"uuid": uuid, "daemonId": daemon})
        print(f"  已发出 {cmd} 指令")
    elif cmd == "status":
        d = call("GET", "/api/instance", {"uuid": uuid, "daemonId": daemon})["data"]
        print(f"  {d['config'].get('nickname')}: "
              f"{STATUS.get(d.get('status'), d.get('status'))}")
    elif cmd == "logs":
        size = a[1] if len(a) > 1 else 64
        t = call("GET", "/api/protected_instance/outputlog",
                 {"uuid": uuid, "daemonId": daemon, "size": size})["data"] or ""
        for line in [x for x in str(t).replace("\r", "\n").split("\n") if x.strip()][-40:]:
            print(" ", line[:160])
    elif cmd == "cmd":
        if len(a) < 2:
            sys.exit('用法: cmd <uuid> "<命令>"')
        call("GET", "/api/protected_instance/command",
             {"uuid": uuid, "daemonId": daemon, "command": a[1]})
        print("  已发送")
    elif cmd == "delete":
        call("DELETE", "/api/instance", {"daemonId": daemon},
             {"uuids": [uuid], "deleteFile": "--with-files" in a})
        print("  已删除")
    else:
        sys.exit(__doc__)


if __name__ == "__main__":
    main()