155 lines
4 KiB
Python
155 lines
4 KiB
Python
import shlex
|
|
from collections import deque
|
|
from os import setsid
|
|
from pathlib import Path
|
|
from subprocess import PIPE, Popen
|
|
from time import sleep
|
|
from typing import Generator
|
|
|
|
from mcrcon import MCRcon
|
|
from mcstatus import JavaServer
|
|
from psutil import NoSuchProcess, Process
|
|
|
|
from .classes import ServerStatus
|
|
from .config import Config
|
|
|
|
|
|
class ProcessController:
|
|
def __init__(self):
|
|
self.pid_file: Path = Path(Config.PID_FILE)
|
|
self.start_command: list[str] = shlex.split(Config.START_COMMAND)
|
|
self.process: Process | None = None
|
|
|
|
if self.pid_file.is_file():
|
|
pid = int(self.pid_file.read_text())
|
|
if self._is_pid_alive(pid):
|
|
self.process = Process(pid)
|
|
else:
|
|
self.pid_file.unlink()
|
|
|
|
def start(self) -> None:
|
|
"Starts the process."
|
|
if self.is_started():
|
|
return
|
|
|
|
process = Popen(
|
|
self.start_command,
|
|
stdout=PIPE,
|
|
stderr=PIPE,
|
|
preexec_fn=setsid,
|
|
)
|
|
self.process = Process(process.pid)
|
|
self.pid_file.write_text(str(self.process.pid))
|
|
|
|
def is_started(self) -> bool:
|
|
"Check if the process is running."
|
|
if self.process:
|
|
return self.process.is_running()
|
|
return False
|
|
|
|
def is_crashed(self) -> bool:
|
|
"Check if the process has crashed."
|
|
return False # TODO
|
|
|
|
def kill(self) -> None:
|
|
"Kill the process."
|
|
if self.process:
|
|
self.process.terminate()
|
|
self.process = None
|
|
self.pid_file.unlink()
|
|
|
|
def _is_pid_alive(self, pid: int) -> bool:
|
|
"Check if a process with the given PID is alive."
|
|
try:
|
|
p = Process(pid)
|
|
return p.is_running()
|
|
except NoSuchProcess:
|
|
return False
|
|
|
|
|
|
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:
|
|
self.rcon.connect()
|
|
output = self.rcon.command(command)
|
|
self.rcon.disconnect()
|
|
return output
|
|
|
|
|
|
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()
|