"""MC 服务器状态 —— 服务器列表数据层（校验 + 读写 + 增删改）。 内存态就是 `config.var.group_list`，形态是 `{int 群号: {服务器名: [地址, 类型]}}` （类型 `js`=Java / `bds`=基岩）。群命令（handlers/server.py）与 Web 管理台 （web_hub.py）都只经这里改数据，校验只有这一份，两边不会各自跑偏。 本模块顶层**不 import nonebot**，运行态 config 到函数里再延迟 import （见 `_runtime()`），这样数据层能脱离 NoneBot 运行时被单测加载。 """ from __future__ import annotations import json import logging from pathlib import Path from typing import Any try: # 单测里可能拿不到 nonebot.logger，退化成标准库 from nonebot.log import logger except Exception: # noqa: BLE001 — 拿不到日志不影响数据层可用 logger = logging.getLogger(__name__) #: 服务器类型 → 中文说明（排序即前端下拉顺序） SERVER_TYPES: dict[str, str] = {"js": "Java 版", "bds": "基岩版"} #: 地址里写了端口才连得快，否则 Java 版要多探几轮（上游加的注释，保留） ADDRESS_HINT = "地址带端口查得更快，如 mc.example.com:25565、mc.example.com:19132" MAX_NAME_LEN = 24 MAX_GROUP_LEN = 12 MAX_ADDRESS_LEN = 64 class ServerError(ValueError): """校验失败 / 目标不存在，`str(e)` 就是给用户看的话术。""" def _runtime() -> Any: """取运行态 `config` 模块（数据目录、配置、内存态都在那边）。 到调用时才 import：本模块要能在"插件包还没初始化"的场景下被加载， 只有真正读写数据时才需要 config（单测按包结构加载本模块，见 tests/test_mc_servers.py）。 """ from .. import config return config # ───────────────────────── 校验 / 归一化 ───────────────────────── def normalize_type(value: Any, default: str = "js") -> str: """类型归一：`js`/`java` → js，`bds`/`be`/`bedrock` → bds；空值取 default。""" if value is None or str(value).strip() == "": return default text = str(value).strip().lower() if text in ("js", "java"): return "js" if text in ("bds", "be", "bedrock"): return "bds" raise ServerError("类型请填 js 或 bds（js = Java 版，bds = 基岩版）") def normalize_address(value: Any) -> str: """地址归一：去空白、去协议前缀与尾部斜杠；`bds` 默认端口 19132 不写进数据里。""" text = str(value or "").strip() if not text: raise ServerError("服务器地址不能为空") for prefix in ("minecraft://", "mc://", "http://", "https://"): if text.lower().startswith(prefix): text = text[len(prefix) :] break text = text.strip().strip("/").strip() if not text: raise ServerError("服务器地址不能为空") if any(ch.isspace() for ch in text) or "/" in text: raise ServerError("服务器地址里不能有空格或斜杠") if len(text) > MAX_ADDRESS_LEN: raise ServerError(f"服务器地址太长了（最多 {MAX_ADDRESS_LEN} 个字符）") if text.startswith(":") or text.endswith(":"): raise ServerError("服务器地址的端口不能为空") return text def normalize_name(value: Any) -> str: """服务器名归一：去首尾空白、单双引号；名字里有空格是允许的。""" text = str(value or "").strip() if len(text) >= 2 and text[0] == text[-1] and text[0] in "\"'“”": text = text[1:-1].strip() if not text: raise ServerError("服务器名称不能为空") if len(text) > MAX_NAME_LEN: raise ServerError(f"服务器名称太长了（最多 {MAX_NAME_LEN} 个字符）") return text def parse_group_id(value: Any) -> int: """群号：必须是纯数字（DataStore/命令层都按 int 存）。""" text = str(value if value is not None else "").strip() if not text: raise ServerError("群号不能为空") if len(text) > MAX_GROUP_LEN: raise ServerError("群号看上去不对（最多 12 位数字）") if not text.isdigit(): raise ServerError("群号必须是纯数字") if not int(text): raise ServerError("群号看上去不对") return int(text) # ───────────────────────── 数据读写 ───────────────────────── def data_path(filename: str) -> Path: return _runtime().DATA_DIR / filename def load_servers(filename: str) -> dict[int, dict[str, list[str]]]: """读数据文件；文件不存在/内容坏了按"空数据 + 警告"处理，绝不让插件起不来。""" path = data_path(filename) if not path.exists(): return {} try: with open(path, encoding="utf-8") as f: raw = json.load(f) except Exception as e: # noqa: BLE001 logger.warning(f"MC 服务器数据读取失败（按空数据处理）：{path} {e!r}") return {} data: dict[int, dict[str, list[str]]] = {} if not isinstance(raw, dict): logger.warning(f"MC 服务器数据格式不是对象，已忽略：{path}") return data for group_key, servers in raw.items(): try: group_id = int(group_key) except (TypeError, ValueError): logger.warning(f"MC 服务器数据里有个非数字群号，已忽略：{group_key!r}") continue if not isinstance(servers, dict): continue clean: dict[str, list[str]] = {} for name, value in servers.items(): if isinstance(value, (list, tuple)) and len(value) >= 2: clean[str(name)] = [str(value[0]), str(value[1])] data[group_id] = clean return data def save_servers(filename: str, data: dict[int, dict[str, list[str]]]) -> None: """整表落盘（先写 .tmp 再替换，避免写一半断电留下半截 JSON）。""" path = data_path(filename) path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_suffix(path.suffix + ".tmp") with open(tmp, "w", encoding="utf-8") as f: json.dump(data, f, indent=4, ensure_ascii=False) tmp.replace(path) def _state() -> tuple[Any, dict[int, dict[str, list[str]]]]: config = _runtime() return config, config.var.group_list def snapshot() -> dict[int, dict[str, list[str]]]: """整表深拷贝（给 Web 只读接口用，避免调用方误改内存态）。""" _, groups = _state() return {gid: {name: list(value) for name, value in servers.items()} for gid, servers in groups.items()} def mutate(func, *args, **kwargs): """校验通过后才落盘；冲突类错误不要走到这里（改内存前就该抛）。""" config, groups = _state() result = func(groups, *args, **kwargs) save_servers(config.pc.mc_status_data_filename, groups) return result # ───────────────────────── 增删改 ───────────────────────── def _upsert( groups: dict[int, dict[str, list[str]]], group_id: int, old_name: str | None, name: str, address: str, server_type: str, ) -> bool: servers = groups.setdefault(group_id, {}) if old_name and old_name != name: # 改名：先查旧名在不在、新名会不会撞，确认没问题再动数据 if old_name not in servers: raise ServerError(f"找不到要改名的服务器「{old_name}」") if name in servers: raise ServerError(f"群里已经有一个叫「{name}」的服务器了") servers.pop(old_name) servers[name] = [address, server_type] return True created = name not in servers servers[name] = [address, server_type] return created def set_server( group_id: int, name: str, address: str, server_type: str, old_name: str | None = None, ) -> bool: """新增/更新一台服务器，返回是否新建。`old_name` 传了且与 name 不同即为改名。""" group_id = parse_group_id(group_id) name = normalize_name(name) address = normalize_address(address) server_type = normalize_type(server_type) old = normalize_name(old_name) if old_name else None return mutate(_upsert, group_id, old, name, address, server_type) def _delete( groups: dict[int, dict[str, list[str]]], group_id: int, name: str | None ) -> None: if group_id not in groups: raise ServerError(f"群 {group_id} 还没有配置服务器") if name is None: # 整个群一起删 groups.pop(group_id) return if name not in groups[group_id]: raise ServerError(f"群里没有叫「{name}」的服务器") groups[group_id].pop(name) if not groups[group_id]: # 删空了就把群一起收掉，别留空壳 groups.pop(group_id) def delete_server(group_id: int, name: str) -> None: """删掉某群里的一台服务器（删空了连带删群）。""" group_id = parse_group_id(group_id) mutate(_delete, group_id, normalize_name(name)) def delete_group(group_id: int) -> None: """删掉整个群的服务器配置。""" group_id = parse_group_id(group_id) mutate(_delete, group_id, None) def lookup(group_id: int, name: str) -> list[str] | None: """按群号 + 服务器名取 `[地址, 类型]`（给 Web 的「测试连接」用）。""" _, groups = _state() return groups.get(int(group_id), {}).get(name)