HeXi/hexi/plugins/nonebot_plugin_mc_server_status/services/servers.py

"""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)