minecraftd/controllers.py

import shlex
from collections import deque
from pathlib import Path
from subprocess import Popen
from threading import Lock
from time import sleep
from typing import Literal

from mcrcon import MCRcon
from mcstatus import JavaServer

from classes import ProcessStatus, ServerStatus
from config import Config
from logs import logger


class ProcessController:
    def __init__(self):
        self.start_command: list[str] = shlex.split(
            Path(Config.START_COMMAND_FILE).read_text()
        )
        self.cwd = Config.SERVER_PATH
        self.process: Popen | None = None
        self.last_status: Literal[ProcessStatus.STOPPED, ProcessStatus.CRASHED] = (
            ProcessStatus.STOPPED
        )
        self._lock = Lock()

    def start(self) -> None:
        "Start the process."
        logger.debug("ProcessController.start()")
        with self._lock:
            if self.status() == ProcessStatus.RUNNING:
                if self.process:
                    logger.debug(
                        "ProcessController.start() - Process was already running with PID: %s",
                        self.process.pid,
                    )
                    return

            self.process = Popen(
                self.start_command,
                stdout=None,
                stderr=None,
                stdin=None,
                start_new_session=True,
                cwd=self.cwd,
            )
            logger.info(
                "ProcessController.start() - Started process with PID: %s",
                self.process.pid,
            )

    def status(self) -> ProcessStatus:
        "Check the process' status."
        logger.debug("ProcessController.status()")

        if not self.process:
            logger.debug("ProcessController.status() => %s", self.last_status)
            return self.last_status
        match self.process.poll():
            case None:
                logger.debug("ProcessController.status() => ProcessStatus.RUNNING")
                return ProcessStatus.RUNNING
            case 0:
                logger.debug("ProcessController.status() => ProcessStatus.STOPPED")
                self.last_status = ProcessStatus.STOPPED
                return ProcessStatus.STOPPED
            case _:
                logger.debug("ProcessController.status() => ProcessStatus.CRASHED")
                self.last_status = ProcessStatus.CRASHED
                return ProcessStatus.CRASHED

    def kill(self) -> None:
        "Kill the process."
        logger.debug("ProcessController.kill()")
        with self._lock:
            if self.process:
                pid = self.process.pid
                self.process.kill()
                code = self.process.wait()
                self.process = None
                logger.info(
                    "ProcessController.kill() - Process with PID %s killed with return code: %s",
                    pid,
                    code,
                )
                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:
        logger.debug("ServerController.status()")
        try:
            status = self.server.status()
        except Exception:
            logger.debug("ServerController.status() - Server is offline")
            return {"online": False}

        players = []
        if status.players.sample:
            for player in status.players.sample:
                players.append(player.name)

        logger.debug("ServerController.status() - Server is online")
        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:
        logger.debug('ServerController.command(command="%s")', command)
        try:
            self.rcon.connect()
            output = self.rcon.command(command)
            self.rcon.disconnect()
            logger.info('ServerController.command(command="%s") => %s', command, output)
            return output
        except Exception:
            logger.exception(
                'ServerController.command(command="%s") - Command execution failed',
                command,
            )
            return ""


class MaintainanceController:
    def __init__(self):
        self.mnt_file = Path(Config.MAINTAINANCE_FILE)

    def set(self, reason: str):
        logger.debug('MaintainanceController.set(reason="%s")', reason)
        self.mnt_file.write_text(reason)

    def is_set(self) -> bool:
        logger.debug("MaintainanceController.is_set() => %s", self.mnt_file.is_file())
        return self.mnt_file.is_file()

    def get(self) -> str:
        if self.is_set():
            logger.debug(
                "MaintainanceController.get() => %s", self.mnt_file.read_text()
            )
            return self.mnt_file.read_text()

        logger.debug("MaintainanceController.get() => ")
        return ""

    def unset(self):
        logger.debug("MaintainanceController.unset()")
        if self.is_set():
            self.mnt_file.unlink()
            logger.debug("MaintainanceController.unset() - Maintainance file was unset")


class LogsController:
    def __init__(self):
        self.log_file = Path(Config.LOG_FILE)

    def stream(self):
        logger.debug("LogsController.stream()")
        i = 0
        with self.log_file.open() as f:
            f.seek(0, 2)
            while True:
                line = f.readline()
                if line:
                    logger.debug("LogsController.stream() - Yielding line %s", i)
                    i += 1
                    yield line
                else:
                    sleep(0.1)

    def tail(self, back: int = 10):
        logger.debug("LogsController.tail(back=%s)", back)
        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()