132 lines
4.2 KiB
Python
132 lines
4.2 KiB
Python
import asyncio
|
|
import datetime
|
|
from typing import Literal
|
|
import mcstatus
|
|
|
|
from nonebot import on_command
|
|
from nonebot.adapters import Event
|
|
from nonebot_plugin_alconna import Alconna, Args, UniMessage, on_alconna
|
|
from mcstatus.responses import JavaStatusResponse
|
|
from nonebot_plugin_apscheduler import scheduler
|
|
|
|
from konabot.common.permsys import DepPermManager, require_permission
|
|
from konabot.plugins.minecraft_servers.simpfun_server import SimpfunServer
|
|
|
|
|
|
cmd = on_command(
|
|
"宾几人",
|
|
aliases=set(("宾人数", "mcbingo")),
|
|
rule=require_permission("minecraft.bingo.check"),
|
|
)
|
|
|
|
|
|
def parse_status(motd: str) -> str:
|
|
if "[PRE-GAME]" in motd:
|
|
return "[✨ 空闲]"
|
|
if "[IN-GAME]" in motd:
|
|
return "[🕜 游戏中]"
|
|
if "[POST-GAME]" in motd:
|
|
return "[🕜 游戏中]"
|
|
return "[✨ 开放]"
|
|
|
|
|
|
def dump_server_status(name: str, status: JavaStatusResponse | BaseException) -> str:
|
|
if isinstance(status, JavaStatusResponse):
|
|
motd = status.motd.to_plain()
|
|
# Bingo Status: [PRE-GAME], [IN-GAME], [POST-GAME]
|
|
st = parse_status(motd)
|
|
players_sample = status.players.sample or []
|
|
players_sample_suffix = ""
|
|
if len(players_sample) > 0:
|
|
player_list = [s.name for s in players_sample]
|
|
players_sample_suffix = " (" + ", ".join(player_list) + ")"
|
|
return f"{name}: {st} {status.players.online} 人在线{players_sample_suffix}"
|
|
else:
|
|
return f"{name}: 好像没开"
|
|
|
|
|
|
@cmd.handle()
|
|
async def _(evt: Event, pm: DepPermManager):
|
|
servers = (
|
|
(mcstatus.JavaServer("play.simpfun.cn", 11495), "小帕 Bingo"),
|
|
(mcstatus.JavaServer("bingo.mujica.tech"), "坏枪 Bingo"),
|
|
(mcstatus.JavaServer("mc.mujica.tech", 11456), "齿轮盛宴"),
|
|
)
|
|
|
|
responses = await asyncio.gather(
|
|
*map(lambda s: s[0].async_status(), servers),
|
|
return_exceptions=True,
|
|
)
|
|
messages = "\n".join(
|
|
(
|
|
dump_server_status(n, r)
|
|
for n, r in zip(map(lambda s: s[1], servers), responses)
|
|
)
|
|
)
|
|
|
|
if await pm.check_has_permission(evt, "minecraft.bingo.manipulate"):
|
|
messages += "\n\n---\n\n你可以使用 bingoman start 开启小帕的 bingo 服，用 bingoman stop 关闭小帕的 bingo 服"
|
|
|
|
await UniMessage.text(messages).finish(evt, at_sender=False)
|
|
|
|
|
|
cmd_bingo_manipulate = on_alconna(
|
|
Alconna("bingoman", Args["action", str]),
|
|
aliases=("宾服务器", "bingo服"),
|
|
rule=require_permission("minecraft.bingo.manipulate"),
|
|
)
|
|
|
|
actions: dict[str, Literal["start", "stop", "restart", "kill"]] = {
|
|
"up": "start",
|
|
"down": "stop",
|
|
"start": "start",
|
|
"stop": "stop",
|
|
"开机": "start",
|
|
"关机": "stop",
|
|
"restart": "restart",
|
|
"kill": "kill",
|
|
"重启": "restart",
|
|
}
|
|
|
|
|
|
@cmd_bingo_manipulate.handle()
|
|
async def _(action: str, event: Event):
|
|
server = SimpfunServer.new() # 使用默认配置管理服务器
|
|
a = actions.get(action.lower().strip())
|
|
if a is None:
|
|
await UniMessage.text(f"操作 {action} 不存在").send(event, at_sender=True)
|
|
return
|
|
resp = await server.power(a)
|
|
if resp.code == 200:
|
|
await UniMessage.text("好了").send(event, at_sender=True)
|
|
else:
|
|
await UniMessage.text(f"不好：{resp}").send(event, at_sender=True)
|
|
|
|
|
|
@scheduler.scheduled_job("cron", hour="4,23")
|
|
async def _():
|
|
server = SimpfunServer.new()
|
|
today = datetime.datetime.now()
|
|
|
|
# 获取服务器当前状态，重试多次以保证不会误判服务器未开启
|
|
server_up = False
|
|
server_players = 0
|
|
for _ in range(3):
|
|
mcs = mcstatus.JavaServer("play.simpfun.cn", 11495)
|
|
try:
|
|
resp = await mcs.async_status()
|
|
server_up = True
|
|
server_players = resp.players.online
|
|
except Exception:
|
|
pass
|
|
|
|
if today.weekday() == 5 and today.hour < 12:
|
|
# 每周六开机一天，保证可以让服务器不被自动销毁
|
|
if not server_up:
|
|
await server.power("start")
|
|
else:
|
|
# 每用一个自然日都会计费，所以要赶在这一天结束之前关服
|
|
# 平时如果没人，也自动关上
|
|
if server_up and server_players == 0:
|
|
await server.power("stop")
|