"""MC 服务器状态 —— 业务逻辑(Service 层)。""" from __future__ import annotations from base64 import b64decode from io import BytesIO from re import findall from mcstatus import BedrockServer, JavaServer from nonebot.adapters.onebot.v11 import Message, MessageSegment as MS def _server_icon(status) -> bytes | None: """取服务器图标字节;没有图标或格式不认识就返回 None。 属性名跨版本换过:mcstatus ≥13 叫 `icon`,11.x/12.x 叫 `favicon` (13.0.0 的 changelog:`JavaStatusResponse.favicon` → `icon`,值没变)。 值可能是完整的 data URI,也可能是不带前缀的裸 base64,两种都认。 """ raw = getattr(status, "icon", None) or getattr(status, "favicon", None) if not raw: return None data = raw.split("base64,", 1)[-1] if "base64," in raw else raw try: return b64decode(data) except Exception: # noqa: BLE001 — 图标坏了不影响其它字段 return None async def check_mc_status( name: str, host: str, server_type: str ) -> str | Message: try: if server_type == "js": js = await JavaServer.async_lookup(host, timeout=2) status = js.status() version_list = findall(r"\d+\.\d+(?:\.[\dxX]+)?", status.version.name) if len(version_list) != 1: version = f"{version_list[0]}-{version_list[-1]}" else: version = version_list[0] online = f"{status.players.online}/{status.players.max}" if status.players.online and status.players.sample: anonymous_player = 0 _player_list = [] for p in status.players.sample: if p.id == "00000000-0000-0000-0000-000000000000": anonymous_player += 1 else: _player_list.append(p.name) if anonymous_player: _player_list.append(f"[{anonymous_player}个匿名玩家]") player_list = "\n".join(_player_list) if _player_list else "没返回玩家列表" else: player_list = "没人在线" latency = round(status.latency) icon_data = _server_icon(status) icon = MS.image(BytesIO(icon_data)) + "\n" if icon_data else "" msg = ( icon + f"名称:{name}\n版本:{version}\n在线:{online}\n延迟:{latency}ms\n在线列表:\n{player_list}" ) else: if host.find(":") != -1: host, port = host.split(":") else: host, port = host, 19132 bds = BedrockServer(host=host, port=int(port)) status = await bds.async_status() online = f"{status.players_online}/{status.players_max}" latency = round(status.latency) version = status.version.version msg = f"名称:{name} 【{version}】\n在线:{online} 延迟:{latency}ms" except Exception as e: msg = f"名称:{name} 查询失败!\n错误:{repr(e)}" return msg