#!/usr/bin/env python3 """Upgrade an existing LuckyGame installation in a single-file Docker Compose project. Ubuntu prerequisites: python3, python3-ruamel.yaml, curl, tar, Docker Compose v2. Run from /opt/minecraft: python3 upgrade-luckygame.py [VERSION] """ import argparse import copy import fcntl import hashlib import io import json import os from pathlib import Path import re import shutil import subprocess import sys import tempfile import time from urllib.parse import urlsplit from zipfile import BadZipFile, ZipFile DEFAULT_SITE = "https://luckygame.saxontech01.com" LEGACY = {"ItemMultiplier", "LaserGuard", "PasswordDoor", "ExpandedCrafting", "WearableBackpacks", "WonderLever"} VERSION_RE = r"(?:0|[1-9][0-9]*)\.(?:0|[1-9][0-9]*)\.(?:0|[1-9][0-9]*)" JAR_RE = re.compile(r"LuckyGame(?:-" + VERSION_RE + r")?\.jar") class UpgradeError(Exception): pass def require(condition, message): if not condition: raise UpgradeError(message) def version_tuple(value): require(isinstance(value, str) and re.fullmatch(VERSION_RE, value), "仅支持正式版本号，例如 2.2.0。") return tuple(map(int, value.split("."))) def yaml_parser(): try: from ruamel.yaml import YAML except ImportError: raise UpgradeError("缺少 YAML 编辑依赖。先运行：apt-get install python3-ruamel.yaml") parser = YAML() parser.preserve_quotes = True parser.width = 4096 return parser def run(command, timeout=120, cwd=None): result = subprocess.run(command, capture_output=True, text=True, timeout=timeout, cwd=cwd) if result.returncode: # Compose can echo secrets from configuration. Keep its raw output out of errors. raise UpgradeError(f"命令失败（退出码 {result.returncode}）：{' '.join(map(str, command[:6]))}") return result.stdout def digest(path, algorithm="sha256"): result = hashlib.new(algorithm) with path.open("rb") as stream: for chunk in iter(lambda: stream.read(1024 * 1024), b""): result.update(chunk) return result.hexdigest() def download(url, path): require(urlsplit(url).scheme == "https", "下载地址必须使用 HTTPS。") run(["curl", "--fail", "--silent", "--show-error", "--location", "--proto", "=https", "--proto-redir", "=https", "--connect-timeout", "10", "--max-time", "60", "--retry", "3", "--retry-delay", "2", "--max-filesize", "52428800", "--output", str(path), url], timeout=270) def fetch_json(url, path): download(url, path) require(path.stat().st_size <= 1024 * 1024, "发布清单过大。") return json.loads(path.read_text()) def plugin_identity(path): try: with ZipFile(path) as archive: descriptor = "plugin.yml" if "plugin.yml" in archive.namelist() else "paper-plugin.yml" info = yaml_parser().load(archive.read(descriptor).decode("utf-8")) return str(info.get("name", "")), str(info.get("version", "")) except (OSError, KeyError, ValueError, BadZipFile): raise UpgradeError(f"无法读取插件身份：{path}") def environment(service): value = service.get("environment", {}) if isinstance(value, list): require(all(isinstance(item, str) and "=" in item for item in value), "environment 的列表形式须显式提供 KEY=VALUE。") return dict(item.split("=", 1) for item in value) require(isinstance(value, dict), "无法识别 Compose environment。") return copy.deepcopy(value) def mount_parts(mount): if isinstance(mount, dict): return str(mount.get("source", "")), str(mount.get("target", "")), mount.get("type", "bind") require(isinstance(mount, str) and "${" not in mount, "请将 volumes 写为明确路径后再升级。") parts = mount.split(":") require(len(parts) in (2, 3), "无法识别 Compose volume。") return parts[0], parts[1], "bind" if parts[0].startswith((".", "/")) else "volume" def safe_path(root, path): """Reject symlink components, including dangling links, before writing/backing up.""" path = Path(os.path.abspath(path)) require(path != root and root in path.parents, f"路径不在服务器目录内：{path}") for item in (path, *path.parents): if item == root: break require(not item.is_symlink(), f"暂不支持符号链接：{item}") return path def atomic_write(path, data, mode=0o644): previous = path.stat() if path.exists() else None fd, temporary = tempfile.mkstemp(prefix=".luckygame-", dir=path.parent) try: with os.fdopen(fd, "wb") as stream: stream.write(data) stream.flush() os.fsync(stream.fileno()) os.chmod(temporary, previous.st_mode & 0o777 if previous else mode) if previous and hasattr(os, "geteuid") and os.geteuid() == 0: os.chown(temporary, previous.st_uid, previous.st_gid) os.replace(temporary, path) finally: if os.path.exists(temporary): os.unlink(temporary) def readable_directory(path): missing = [] while not path.exists(): missing.append(path) path = path.parent for directory in reversed(missing): directory.mkdir(mode=0o755) os.chmod(directory, 0o755) class Upgrader: def __init__(self, args): self.args = args self.root = Path(args.directory).resolve() require(self.root.is_dir(), "服务器目录不存在。") require(not os.environ.get("COMPOSE_FILE"), "请取消 COMPOSE_FILE；目前支持单个 Compose 文件。") names = ["compose.yaml", "compose.yml", "docker-compose.yaml", "docker-compose.yml"] found = [self.root / name for name in names if (self.root / name).exists()] require(len(found) == 1, "服务器目录必须只有一个 compose.yaml/yml 或 docker-compose.yaml/yml。") require(not any(self.root.glob("*compose.override.y*ml")), "存在 Compose override 文件，暂不支持自动升级。") self.compose = safe_path(self.root, found[0]) self.original = self.compose.read_bytes() self.parser = yaml_parser() self.config = self.parser.load(self.original.decode()) require(not any(key in self.config for key in ("include", "extends")), "暂不支持多文件 Compose。") service = self.config.get("services", {}).get("minecraft") require(isinstance(service, dict), "Compose 中找不到 minecraft 服务。") require(not any(service.get(key) for key in ("extends", "env_file", "profiles")), "暂不支持 minecraft 服务的 extends/env_file/profiles。") self.service = copy.deepcopy(service) self.env = environment(self.service) mounts = self.service.get("volumes", []) self.jar_mounts = [] self.keep_mounts = [] data_mounts = 0 for mount in mounts: source, target, kind = mount_parts(mount) if target == "/data": require(kind == "bind" and Path(os.path.abspath(self.root / source)) == self.root / "data", "目前要求 ./data:/data 的本地持久化目录。") data_mounts += 1 elif target.startswith(("/data/", "/plugins")): require(kind == "bind" and target.startswith("/data/plugins/") and JAR_RE.fullmatch(target.removeprefix("/data/plugins/")), f"需先人工核对额外挂载：{target}；脚本不会修改此布局。") self.jar_mounts.append(safe_path(self.root, self.root / source)) continue self.keep_mounts.append(mount) require(data_mounts == 1 and len(self.jar_mounts) == 1, "目前支持 ./data:/data 加一个 LuckyGame JAR 挂载；六个旧插件请先按 README 迁移。") identity, self.current = plugin_identity(self.jar_mounts[0]) require(identity == "LuckyGame", "挂载文件不是 LuckyGame 插件。") version_tuple(self.current) self.data = safe_path(self.root, self.root / "data") require(self.data.is_dir(), "data 目录不存在，拒绝当作空服务器安装。") self.local = safe_path(self.root, self.root / "local-plugins") current_url = str(self.env.get("RESOURCE_PACK", "")) self.site = (args.base_url or current_url.split("/releases/", 1)[0] or DEFAULT_SITE).rstrip("/") parsed = urlsplit(self.site) require(parsed.scheme == "https" and parsed.netloc and not parsed.username and not parsed.query and not parsed.fragment, "下载站地址必须是无凭据的 HTTPS URL；可使用 --base-url 指定。") def docker(self, *args, compose=None, timeout=180): return run(["docker", "compose", "--project-directory", str(self.root), "-f", str(compose or self.compose), *args], timeout, cwd=self.root) def container(self): ids = self.docker("ps", "--all", "--quiet", "minecraft").split() require(len(ids) == 1, "minecraft 必须已有且仅有一个容器。") return json.loads(run(["docker", "inspect", ids[0]]))[0] def preflight(self): effective = json.loads(self.docker("config", "--format", "json")) env = effective["services"]["minecraft"].get("environment", {}) require(env.get("TYPE") == "PAPER" and str(env.get("VERSION")) == "26.2", "此脚本面向 Paper 26.2，服务器类型或版本不匹配。") require(str(env.get("ENABLE_RCON", "TRUE")).upper() == "TRUE", "启动验证需要 RCON，请先在 Compose 中启用 ENABLE_RCON。") require(self.container()["State"]["Running"], "minecraft 当前未运行；请先确认旧服务器可正常启动。") # Imports on restart must not resurrect old plugins downloaded via environment. managed = " ".join(str(env.get(key, "")) for key in ("PLUGINS", "MODRINTH_PROJECTS", "SPIGET_RESOURCES")) require(not any(name.lower() in managed.lower() for name in LEGACY | {"LuckyGame"}), "请先移除环境变量中的自制插件下载项，避免重新下载旧版。") for directory in (self.data / "plugins", self.data / "plugins/update"): if directory.exists(): for path in directory.glob("*.jar"): safe_path(self.root, path) if not path.stat().st_size: continue # A bind mount may leave an empty host placeholder. identity, _ = plugin_identity(path) require(identity not in LEGACY, f"仍存在旧独立插件：{path.name}。请先完成迁移。") require(identity != "LuckyGame" or JAR_RE.fullmatch(path.name), f"LuckyGame 使用了非标准文件名：{path.name}；请先人工整理。") geyser = safe_path(self.root, self.data / "plugins/Geyser-Spigot/config.yml") require(geyser.is_file(), "找不到 Geyser-Spigot/config.yml，请先确认基岩版已配置。") config = self.parser.load(geyser.read_text()) custom = config.get("gameplay", {}).get("enable-custom-content", config.get("enable-custom-content")) require(custom is True, "请先在 Geyser config.yml 中启用 enable-custom-content: true。") def prepare(self, stage): requested = self.args.version if requested is None: latest = fetch_json(self.site + "/latest.json", stage / "latest.json") requested = latest.get("version") target = version_tuple(requested) current = version_tuple(self.current) require(target >= current, "拒绝降级；恢复旧版本请使用完整停服备份。") require(self.args.version is not None or target[0] == current[0], f"发现主版本 {requested}；请先阅读迁移说明，再显式指定版本运行。") self.version = requested print(f"LuckyGame：{self.current} → {self.version}；下载站：{self.site}", flush=True) if self.args.check or target == current: print("当前已是该版本，无需升级。" if target == current else "存在可升级版本；本次仅检查。") return False manifest = fetch_json(f"{self.site}/releases/{requested}/manifest-{requested}.json", stage / "manifest.json") require(manifest.get("name") == "LuckyGame" and manifest.get("version") == requested and manifest.get("minecraft_java") == "26.2" and manifest.get("java") == 25, "发布清单的项目、版本或运行环境不匹配。") self.assets = {} names = [f"LuckyGame-{requested}.jar", f"LuckyGame-Java-{requested}.zip", f"LuckyGame-Bedrock-{requested}.mcpack", f"LuckyGame-Geyser-{requested}.json"] for name in names: matches = [item for item in manifest.get("assets", []) if item.get("name") == name] require(len(matches) == 1, f"发布清单缺少或重复文件：{name}") item = matches[0] require(item.get("path") == f"/releases/{requested}/{name}", "发布文件路径不匹配。") require(isinstance(item.get("size"), int) and 0 < item["size"] <= 50 * 1024 * 1024 and re.fullmatch(r"[a-f0-9]{64}", str(item.get("sha256", ""))) and re.fullmatch(r"[a-f0-9]{40}", str(item.get("sha1", ""))), "发布清单校验字段无效。") print(f"下载并校验 {name} …", flush=True) download(self.site + item["path"], stage / name) require((stage / name).stat().st_size == item["size"] and digest(stage / name) == item["sha256"] and digest(stage / name, "sha1") == item["sha1"], f"校验失败：{name}；服务器未停服。") self.assets[name] = item require(plugin_identity(stage / names[0]) == ("LuckyGame", requested), "新 JAR 的插件身份或版本错误。") self.env["RESOURCE_PACK"] = self.site + self.assets[names[1]]["path"] self.env["RESOURCE_PACK_SHA1"] = self.assets[names[1]]["sha1"] self.env.setdefault("RESOURCE_PACK_ENFORCE", "TRUE") self.env["RESOURCE_PACK_PROMPT"] = '{"text":"LuckyGame：激光器与旅行背包外观"}' self.service["environment"] = self.env self.service["volumes"] = self.keep_mounts + [f"./local-plugins/{names[0]}:/data/plugins/LuckyGame.jar:ro"] self.config["services"]["minecraft"] = self.service stream = io.StringIO() self.parser.dump(self.config, stream) self.candidate = stream.getvalue().encode() candidate_path = stage / "compose.yaml" candidate_path.write_bytes(self.candidate) self.docker("config", "--quiet", compose=candidate_path) self.to_remove = self.old_files() self.destinations = {names[0]: self.local / names[0], names[2]: self.data / "plugins/Geyser-Spigot/packs" / names[2], names[3]: self.data / "plugins/Geyser-Spigot/custom_mappings" / names[3]} for path in [*self.destinations.values(), self.root / "releases" / requested, self.root / "backups", self.root / ".env"]: safe_path(self.root, path) for name, path in self.destinations.items(): require(not path.exists() or (path.is_file() and digest(path) == self.assets[name]["sha256"]), f"目标文件已存在且内容不一致：{path}") for name in self.assets: cached = safe_path(self.root, self.root / "releases" / requested / name) require(not cached.exists() or (cached.is_file() and digest(cached) == self.assets[name]["sha256"]), f"本地 releases 中已有不同内容：{name}") self.backup_names = [self.compose.name, "data"] self.backup_names += [name for name in ("local-plugins", ".env") if (self.root / name).exists()] # Include an existing JAR mounted from another project-local directory. for jar in self.jar_mounts: relative = jar.relative_to(self.root) if relative.parts[0] not in self.backup_names: self.backup_names.append(str(relative)) total = 0 for name in self.backup_names: path = safe_path(self.root, self.root / name) for item in ([path] if path.is_file() else path.rglob("*")): safe_path(self.root, item) if item.is_file(): total += item.stat().st_size require(shutil.disk_usage(self.root).free > total * 1.10 + 512 * 1024 * 1024, "可用磁盘空间不足以安全创建完整备份；请先迁走历史备份。") print("预检完成：将停服，备份整个 data、Compose、.env 和本地插件，然后更新并检查启动。", flush=True) return True def old_files(self): paths = [] for directory in (self.data / "plugins", self.data / "plugins/update"): if directory.exists(): paths += [path for path in directory.glob("*.jar") if JAR_RE.fullmatch(path.name)] geyser = self.data / "plugins/Geyser-Spigot" for directory, suffix in (("packs", ".mcpack"), ("packs", ".zip"), ("custom_mappings", ".json")): folder = geyser / directory if folder.exists(): for path in folder.glob("*" + suffix): # Known project prefixes only. Never move other plugins' packs or maps. if re.match(r"^(LuckyGame|WearableBackpacks|LaserGuard|OurWorld)(?:[-_.]|$)", path.stem, re.I): paths.append(path) for path in paths: safe_path(self.root, path) require(path.is_file(), f"预期为文件：{path}") return paths def backup(self, folder): archive = folder / "before-upgrade.tar.gz" partial = folder / "before-upgrade.tar.gz.partial" run(["tar", "-czf", str(partial), "-C", str(self.root), "--", *self.backup_names], timeout=3600) run(["tar", "-tzf", str(partial)], timeout=3600) partial.rename(archive) (folder / "SHA256SUMS.txt").write_text(f"{digest(archive)} {archive.name}\n") def wait_ready(self): deadline = time.monotonic() + self.args.timeout while time.monotonic() < deadline: info = self.container() require(info["State"]["Running"] and not info["State"].get("Restarting") and info.get("RestartCount", 0) == 0, "新容器已退出或正在重启。") logs = self.docker("logs", "--no-color", "--tail", "2000", "minecraft") enabled = f"LuckyGame {self.version} 已启用，六项功能统一加载" in logs ready = re.search(r"Done \([0-9.]+s\)!", logs) disabled = re.search(r"\[LuckyGame\].*(?:Disabling LuckyGame|Error|Exception)", logs) require(not disabled, "LuckyGame 启用期间出现错误。") if enabled and ready: # Verify the plugin is still enabled, not merely mentioned in startup logs. output = self.docker("exec", "-T", "minecraft", "rcon-cli", "luckygame", timeout=30) require(f"LuckyGame {self.version}" in output, "RCON 未确认 LuckyGame 的运行版本。") return time.sleep(5) raise UpgradeError(f"等待 {self.args.timeout} 秒仍未确认服务器启动完成。") def deploy(self, stage): require(self.compose.read_bytes() == self.original, "Compose 在预检后被修改，请重新执行。") backups = self.root / "backups" backups.mkdir(exist_ok=True) folder = Path(tempfile.mkdtemp(prefix=f"luckygame-{self.current}-to-{self.version}-{time.strftime('%Y%m%d-%H%M%S')}-", dir=backups)) os.chmod(folder, 0o700) print(f"备份与日志目录：{folder}", flush=True) changed = False try: self.docker("stop", "-t", "120", "minecraft") # Do not modify files until Docker confirms that no writer is still running. require(not self.container()["State"]["Running"], "未确认服务停止，拒绝备份和替换文件。") print("正在创建停服备份 …", flush=True) self.backup(folder) changed = True for path in self.to_remove: destination = folder / "replaced-files" / path.relative_to(self.root) destination.parent.mkdir(parents=True, exist_ok=True) shutil.move(str(path), str(destination)) release = self.root / "releases" / self.version release.mkdir(parents=True, exist_ok=True) for name in self.assets: path = release / name safe_path(self.root, path) require(not path.exists() or digest(path) == self.assets[name]["sha256"], f"本地 releases 中已有不同内容：{name}") if not path.exists(): atomic_write(path, (stage / name).read_bytes()) for name, path in self.destinations.items(): readable_directory(path.parent) atomic_write(path, (stage / name).read_bytes()) atomic_write(self.compose, self.candidate) self.docker("config", "--quiet") print("正在启动并检查 Paper、LuckyGame 和 RCON …", flush=True) self.docker("up", "-d", "--no-deps", "--pull", "never", "--force-recreate", "minecraft") self.wait_ready() except BaseException: if not changed: print("备份未完成，未替换文件；尝试重新启动原容器。", flush=True) self.docker("start", "minecraft") else: try: self.docker("stop", "-t", "120", "minecraft") finally: try: (folder / "failed-startup.log").write_text(self.docker("logs", "--no-color", "--tail", "2000", "minecraft")) finally: print(f"升级未完成，已请求停服。备份：{folder / 'before-upgrade.tar.gz'}\n" "请按升级说明恢复完整备份；不要直接换回旧 JAR 后启动。", file=sys.stderr) raise (folder / "result.json").write_text(json.dumps({"from": self.current, "to": self.version, "status": "ready", "site": self.site}, indent=2) + "\n") print(f"升级成功：LuckyGame {self.version}，Paper 启动完成，RCON 已确认版本。\n备份保留在：{folder}") def main(argv=None): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("version", nargs="?", help="目标正式版本；省略时查询最新版本，同主版本内升级") parser.add_argument("--directory", default=".", help="服务器 Compose 目录，默认当前目录") parser.add_argument("--base-url", help="下载站；默认沿用 Compose 资源包地址的域名") parser.add_argument("--check", action="store_true", help="仅查询版本，不停服") parser.add_argument("--dry-run", action="store_true", help="下载、校验并预检，不修改服务器文件或停服") parser.add_argument("--timeout", type=int, default=300, help="启动检查时限（秒），默认 300") args = parser.parse_args(argv) old_umask = os.umask(0o077) try: require(args.timeout >= 30, "启动检查时限须至少为 30 秒。") for command in ("docker", "curl", "tar"): require(shutil.which(command), f"缺少依赖：{command}") root = Path(args.directory).resolve() lock_path = safe_path(root, root / ".luckygame-upgrade.lock") with lock_path.open("a") as lock: try: fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError: raise UpgradeError("另一个 LuckyGame 升级进程正在运行。") upgrade = Upgrader(args) upgrade.preflight() with tempfile.TemporaryDirectory(prefix="luckygame-upgrade-") as temporary: stage = Path(temporary) if upgrade.prepare(stage): if args.dry_run: print("预演通过；服务器继续运行，未替换文件。") else: upgrade.deploy(stage) return 0 except (UpgradeError, OSError, ValueError, subprocess.SubprocessError) as error: print(f"升级中止：{error}", file=sys.stderr) return 1 except KeyboardInterrupt: print("操作已中断。请查看上方备份路径和服务器状态。", file=sys.stderr) return 130 finally: os.umask(old_umask) if __name__ == "__main__": sys.exit(main())