144 lines
3.8 KiB
Python
144 lines
3.8 KiB
Python
import shlex
|
|
from collections import deque
|
|
from pathlib import Path
|
|
from subprocess import PIPE, Popen
|
|
from time import sleep
|
|
from typing import Generator, Literal
|
|
|
|
from mcrcon import MCRcon
|
|
from mcstatus import JavaServer
|
|
|
|
from .classes import ProcessStatus, ServerStatus
|
|
from .config import Config
|
|
|
|
|
|
class ProcessController:
|
|
def __init__(self):
|
|
self.start_command: list[str] = shlex.split(Config.START_COMMAND)
|
|
self.process: Popen | None = None
|
|
self.last_status: Literal[ProcessStatus.STOPPED, ProcessStatus.CRASHED] = (
|
|
ProcessStatus.STOPPED
|
|
)
|
|
|
|
def start(self) -> None:
|
|
"Start the process."
|
|
if self.status() == ProcessStatus.RUNNING:
|
|
return
|
|
|
|
self.process = Popen(
|
|
self.start_command,
|
|
stdout=PIPE,
|
|
stderr=PIPE,
|
|
)
|
|
|
|
def status(self) -> ProcessStatus:
|
|
"Check the process' status."
|
|
if not self.process:
|
|
return self.last_status
|
|
match self.process.poll():
|
|
case None:
|
|
return ProcessStatus.RUNNING
|
|
case 0:
|
|
self.last_status = ProcessStatus.STOPPED
|
|
return ProcessStatus.STOPPED
|
|
case _:
|
|
self.last_status = ProcessStatus.CRASHED
|
|
return ProcessStatus.CRASHED
|
|
|
|
def kill(self) -> None:
|
|
"Kill the process."
|
|
if self.process:
|
|
self.process.terminate()
|
|
self.process = None
|
|
self.last_status = ProcessStatus.STOPPED
|
|
|
|
|
|
class ServerController:
|
|
def __init__(self):
|
|
self.server: JavaServer = JavaServer(
|
|
Config.SERVER_HOST,
|
|
Config.SERVER_PORT,
|
|
)
|
|
self.rcon = MCRcon(
|
|
Config.SERVER_HOST,
|
|
Config.SERVER_RCON_PASSWORD,
|
|
Config.SERVER_RCON_PORT,
|
|
)
|
|
|
|
def status(self) -> ServerStatus:
|
|
try:
|
|
status = self.server.status()
|
|
except Exception:
|
|
return {"online": False}
|
|
|
|
players = []
|
|
if status.players.sample:
|
|
for player in status.players.sample:
|
|
players.append(player.name)
|
|
|
|
return {
|
|
"online": True,
|
|
"icon": status.icon,
|
|
"motd": status.motd.to_html(),
|
|
"players": {
|
|
"online": status.players.online,
|
|
"max": status.players.max,
|
|
"list": players,
|
|
},
|
|
}
|
|
|
|
def command(self, command: str) -> str:
|
|
try:
|
|
self.rcon.connect()
|
|
output = self.rcon.command(command)
|
|
self.rcon.disconnect()
|
|
return output
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
class MaintainanceController:
|
|
def __init__(self):
|
|
self.mnt_file = Path(Config.MAINTAINANCE_FILE)
|
|
|
|
def set(self, reason: str):
|
|
self.mnt_file.write_text(reason)
|
|
|
|
def is_set(self) -> bool:
|
|
return self.mnt_file.is_file()
|
|
|
|
def get(self) -> str:
|
|
if self.is_set():
|
|
return self.mnt_file.read_text()
|
|
return ""
|
|
|
|
def unset(self):
|
|
if self.is_set():
|
|
self.mnt_file.unlink()
|
|
|
|
|
|
class LogsController:
|
|
def __init__(self):
|
|
self.log_file = Path(Config.LOG_FILE)
|
|
|
|
def stream(self) -> Generator[str]:
|
|
with self.log_file.open() as f:
|
|
f.seek(0, 2)
|
|
while True:
|
|
line = f.readline()
|
|
if line:
|
|
yield line
|
|
else:
|
|
sleep(0.1)
|
|
|
|
def tail(self, back: int = 10) -> Generator[str]:
|
|
with self.log_file.open() as f:
|
|
for line in deque(f, maxlen=back):
|
|
yield line
|
|
|
|
|
|
class Controllers:
|
|
process = ProcessController()
|
|
server = ServerController()
|
|
maintainance = MaintainanceController()
|
|
logs = LogsController()
|