squaddocker/scripts/mcsm.py

#!/usr/bin/env python3
"""
MCSManager API 封装 —— 用于编程化管理 Squad 服务器实例。

只用 Python 标准库,无第三方依赖(目标机不保证有 pip)。

凭据从环境变量读取,不接受命令行参数传入 —— 命令行参数会泄漏到进程列表:
    export MCSM_URL='http://<面板地址>:23333'
    export MCSM_APIKEY='<你的 apikey>'

用法:
    ./mcsm.py daemons                       列出 daemon 节点(拿 daemonId)
    ./mcsm.py instances <daemonId>          列出该节点上的实例
    ./mcsm.py create-squad <daemonId> [选项] 创建 Squad 实例
    ./mcsm.py start   <daemonId> <uuid>
    ./mcsm.py stop    <daemonId> <uuid>
    ./mcsm.py restart <daemonId> <uuid>
    ./mcsm.py kill    <daemonId> <uuid>
    ./mcsm.py status  <daemonId> <uuid>
    ./mcsm.py command <daemonId> <uuid> "<控制台命令>"
    ./mcsm.py logs    <daemonId> <uuid> [大小KB]
    ./mcsm.py delete  <daemonId> <uuid> [--with-files]

注意:MCSManager 文档明确警告「提交所有参数,即使参数未定义,否则有概率触发
bug」。因此 create-squad 会提交完整请求体,而不是只提交要改的字段。
"""

import json
import os
import ssl
import sys
import urllib.error
import urllib.parse
import urllib.request

# 实例状态码,来自 MCSManager 文档
STATUS = {
    -1: "busy(忙 / 安装或更新中)",
    0: "stopped(已停止)",
    1: "stopping(正在停止)",
    2: "starting(正在启动)",
    3: "running(运行中)",
}


def _config():
    url = os.environ.get("MCSM_URL", "").rstrip("/")
    key = os.environ.get("MCSM_APIKEY", "")
    if not url or not key:
        sys.exit("错误: 需要设置环境变量 MCSM_URL 和 MCSM_APIKEY")
    return url, key


def call(method, path, params=None, body=None):
    """调用 MCSManager API。apikey 走 query 参数,并带上文档要求的请求头。"""
    base, key = _config()
    q = dict(params or {})
    q["apikey"] = key
    target = f"{base}{path}?{urllib.parse.urlencode(q)}"

    data = json.dumps(body).encode() if body is not None else None
    req = urllib.request.Request(target, data=data, method=method)
    # 这两个头是文档明确要求的,缺了会被拒
    req.add_header("X-Requested-With", "XMLHttpRequest")
    req.add_header("Content-Type", "application/json; charset=utf-8")

    ctx = ssl._create_unverified_context() if base.startswith("https") else None
    try:
        with urllib.request.urlopen(req, timeout=30, context=ctx) as r:
            return json.loads(r.read().decode() or "{}")
    except urllib.error.HTTPError as e:
        detail = e.read().decode(errors="replace")[:500]
        # 不要把 target 打出来 —— 它含 apikey
        sys.exit(f"HTTP {e.code} {method} {path}\n{detail}")
    except urllib.error.URLError as e:
        sys.exit(f"连接失败 {method} {path}: {e.reason}")


def squad_instance_body(nickname, port, queryport, beaconport, rconport,
                        maxplayers, tickrate, image, data_dir, mem_mb,
                        server_name, rcon_password):
    """
    构造 Squad 实例的完整配置。

    要点:
    - processType=docker + networkMode=host:Squad 依赖 Steam 查询协议,
      端口映射模式会导致服务器在游戏内浏览器搜不到
    - ie/oe/fileCode 用 utf8:文档示例给的 gbk 是 Windows 中文环境用的,
      Linux 上会导致控制台日志乱码
    - stopCommand 用 ^C:Squad 没有 stop 这类控制台命令,只能靠信号
    - 不传 RANDOM:该变量在 cm2network/squad 镜像中无效(bash 内置特殊变量)
    """
    return {
        "nickname": nickname,
        "startCommand": "bash entry.sh",
        "stopCommand": "^C",
        "cwd": "/home/steam",
        "ie": "utf8",
        "oe": "utf8",
        "fileCode": "utf8",
        "type": "universal",
        "tag": [],
        "endTime": 0,
        "processType": "docker",
        "updateCommand": "",
        "actionCommandList": [],
        "crlf": 1,
        "enableRcon": True,
        "rconPassword": rcon_password,
        "rconPort": rconport,
        "rconIp": "127.0.0.1",
        "terminalOption": {"haveColor": False, "pty": True},
        "eventTask": {"autoStart": False, "autoRestart": True, "ignore": False},
        # pingConfig 默认端口是 Minecraft 的 25565,Squad 要指向 Steam 查询端口
        "pingConfig": {"ip": "127.0.0.1", "port": queryport, "type": 1},
        "docker": {
            "containerName": nickname,
            "image": image,
            "memory": mem_mb,
            "ports": [],              # host 网络下端口映射无意义
            "extraVolumes": [f"{data_dir}:/home/steam/squad-dedicated"],
            "maxSpace": None,
            "network": None,
            "io": None,
            "networkMode": "host",    # Squad 的硬要求
            "networkAliases": [],
            "cpusetCpus": "",
            "cpuUsage": 0,
            "workingDir": "/home/steam",
            "changeWorkdir": False,
            "env": [
                f"PORT={port}",
                f"QUERYPORT={queryport}",
                f"BEACONPORT={beaconport}",
                f"RCONPORT={rconport}",
                f"FIXEDMAXPLAYERS={maxplayers}",
                f"FIXEDMAXTICKRATE={tickrate}",
                f"SERVER_NAME={server_name}",
            ],
        },
    }


def show(obj):
    print(json.dumps(obj, ensure_ascii=False, indent=2))


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

    if cmd == "daemons":
        r = call("GET", "/api/service/remote_services_list")
        for d in r.get("data", []) or []:
            print(f"  daemonId={d.get('uuid')}  {d.get('ip')}:{d.get('port')}  "
                  f"available={d.get('available')}  remarks={d.get('remarks','')}")
        if not r.get("data"):
            show(r)

    elif cmd == "instances":
        if not a:
            sys.exit("用法: instances <daemonId>")
        r = call("GET", "/api/service/remote_service_instances",
                 {"daemonId": a[0], "page": 1, "page_size": 50,
                  "instance_name": "", "status": ""})
        for i in (r.get("data", {}) or {}).get("data", []) or []:
            st = i.get("status")
            print(f"  uuid={i.get('instanceUuid')}  "
                  f"{(i.get('config') or {}).get('nickname')}  "
                  f"status={st} {STATUS.get(st, '?')}")

    elif cmd == "create-squad":
        if not a:
            sys.exit("用法: create-squad <daemonId> [nickname] [port] ...")
        daemon = a[0]
        body = squad_instance_body(
            nickname=a[1] if len(a) > 1 else "squad1",
            port=int(a[2]) if len(a) > 2 else 7787,
            queryport=int(a[3]) if len(a) > 3 else 27165,
            beaconport=int(a[4]) if len(a) > 4 else 15000,
            rconport=int(a[5]) if len(a) > 5 else 21114,
            maxplayers=int(a[6]) if len(a) > 6 else 50,
            tickrate=50,
            image=os.environ.get("SQUAD_IMAGE",
                                 "squaddocker.azurecr.cn/squad-server:v1"),
            data_dir=os.environ.get("SQUAD_DATA_DIR", "/srv/squad/inst1"),
            mem_mb=int(os.environ.get("SQUAD_MEM_MB", "6144")),
            server_name=os.environ.get("SQUAD_SERVER_NAME", "Squad Server"),
            # RCON 密码从环境变量读,不走命令行
            rcon_password=os.environ.get("SQUAD_RCON_PASSWORD", ""),
        )
        if not body["rconPassword"]:
            print("警告: 未设置 SQUAD_RCON_PASSWORD。RCON 密码为空时服务端不会"
                  "启动 RCON,21114 端口不会监听。", file=sys.stderr)
        r = call("POST", "/api/instance", {"daemonId": daemon}, body)
        show(r)

    elif cmd in ("start", "stop", "restart", "kill"):
        if len(a) < 2:
            sys.exit(f"用法: {cmd} <daemonId> <uuid>")
        r = call("GET", f"/api/protected_instance/{'open' if cmd == 'start' else cmd}",
                 {"uuid": a[1], "daemonId": a[0]})
        show(r)

    elif cmd == "status":
        if len(a) < 2:
            sys.exit("用法: status <daemonId> <uuid>")
        r = call("GET", "/api/instance", {"uuid": a[1], "daemonId": a[0]})
        st = (r.get("data") or {}).get("status")
        print(f"  status={st}  {STATUS.get(st, '未知')}")

    elif cmd == "command":
        if len(a) < 3:
            sys.exit('用法: command <daemonId> <uuid> "<命令>"')
        r = call("GET", "/api/protected_instance/command",
                 {"uuid": a[1], "daemonId": a[0], "command": a[2]})
        show(r)

    elif cmd == "logs":
        if len(a) < 2:
            sys.exit("用法: logs <daemonId> <uuid> [大小KB]")
        r = call("GET", "/api/protected_instance/outputlog",
                 {"uuid": a[1], "daemonId": a[0], "size": a[2] if len(a) > 2 else 512})
        print(r.get("data") or r)

    elif cmd == "delete":
        if len(a) < 2:
            sys.exit("用法: delete <daemonId> <uuid> [--with-files]")
        r = call("DELETE", "/api/instance", {"daemonId": a[0]},
                 {"uuids": [a[1]], "deleteFile": "--with-files" in a})
        show(r)

    else:
        sys.exit(__doc__)


if __name__ == "__main__":
    main()