64 lines
2.5 KiB
Python
64 lines
2.5 KiB
Python
 
|
"""MC 服务器状态 —— 业务逻辑(Service 层)。"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from base64 import b64decode
|
||
|
|
from io import BytesIO
|
||
|
|
from re import findall
|
||
|
|
from typing import Union
|
||
|
|
|
||
|
|
from mcstatus import BedrockServer, JavaServer
|
||
|
|
from nonebot.adapters.onebot.v11 import Message, MessageSegment as MS
|
||
|
|
|
||
|
|
|
||
|
|
async def check_mc_status(
|
||
|
|
name: str, host: str, server_type: str
|
||
|
|
) -> Union[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)
|
||
|
|
if status.favicon:
|
||
|
|
aa, bb = status.favicon.split("base64,")
|
||
|
|
icon = MS.image(BytesIO(b64decode(bb))) + "\n"
|
||
|
|
else:
|
||
|
|
icon = ""
|
||
|
|
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
|