minecraftd/main.py

from typing import Annotated, Optional

import uvicorn
from classes import ProcessStatus
from controllers import Controllers
from fastapi import BackgroundTasks, FastAPI, Header
from fastapi.responses import StreamingResponse
from logs import logger
from models import Models
from responses import Responses
from util import check_password, stop_server

app = FastAPI()


@app.get("/start")
async def start(tasks: BackgroundTasks) -> Responses.StartResponse:
    """Starts the Server's process if it is not already running.
    Returns:
        status: "started" or "running".
        message: The Server's response.
    """

    logger.debug("/start")

    if Controllers.process.status() == ProcessStatus.RUNNING:
        logger.debug("/start - The Server was already running")
        return {
            "status": "running",
            "message": "The Server was already running.",
        }

    if Controllers.maintainance.is_set():
        Controllers.maintainance.unset()
        logger.debug("/start - Unset maintainance")

    tasks.add_task(Controllers.process.start)
    logger.info("/start - Starting server")
    return {
        "status": "starting",
        "message": "The Server is starting.",
    }


@app.get("/status")
async def status() -> Responses.StatusResponse:
    """Checks whether the Server is running and returns its information.
    Returns:
        status: One of "online", "offline", "crashed", "maintainance", or "starting".
        message: The Server's response.
        reason: if status is "maintainance", contains the reason for the Server's maintainance state.
        motd: if status is "online", contains the Server's MOTD as an HTML string.
        icon: if status is "online", contains the Server's icon as a base64 string.
        players: if status is "online", contains:
            online: the number of online players.
            max: the number of max allowed players.
            list: the list of online players' usernames as strings.
    """

    logger.debug("/status")

    process_status = Controllers.process.status()
    if process_status != ProcessStatus.RUNNING:
        # Crashed
        if process_status == ProcessStatus.CRASHED:
            logger.debug("/status - The Server has crashed")
            return {
                "status": "crashed",
                "message": "The Server has crashed.",
            }
        # Maintainance
        if Controllers.maintainance.is_set():
            reason = Controllers.maintainance.get()
            logger.debug(
                '/status - The Server is offline due to maintainance ("%s")', reason
            )
            return {
                "status": "maintainance",
                "message": "The Server is offline due to maintainance.",
                "reason": reason,
            }
        # Offline
        logger.debug("/status - The Server is offline")
        return {
            "status": "offline",
            "message": "The Server is offline.",
        }

    server_status = Controllers.server.status()

    # Starting
    if not server_status["online"]:
        logger.debug("/status - The Server is starting")
        return {
            "status": "starting",
            "message": "The Server is starting.",
        }

    # Online
    logger.debug("/status - The Server is online")
    return {
        "status": "online",
        "message": "The Server is online.",
        "motd": server_status.get("motd", ""),
        "icon": server_status.get("icon", None),
        "players": server_status.get(
            "players",
            {
                "online": 0,
                "max": 20,
                "list": [],
            },
        ),
    }


@app.get("/stop")
async def stop(
    data: Models.StopModel,
    authorization: Annotated[str, Header()],
    tasks: BackgroundTasks,
) -> Responses.StopResponse:
    """Stops the Server.
    It waits for `countdown` seconds, then runs `/stop` on the Server, and kills it after `timeout` seconds if it's still alive.
    Args:
        countdown: the time in seconds to give the players to leave the Server before initiating the shutdown.
                   if set to 0, the shutdown phase is started immediately. Defaults to 60.
        reason: a brief message explaining why the Server is shutting down. Only shown if countdown is non-zero. Defaults to "".
        timeout: the time in seconds to wait before killing the Server if its process hasn't stopped. Defaults to 10.
    Headers:
        Authorization: the Authorization token.
    Returns:
        status: "stopping"
        message: The Server's response.
    """

    logger.debug("/stop")

    check_password(authorization)
    tasks.add_task(stop_server, "STOPPING", data.countdown, data.reason, data.timeout)

    logger.info("/stop - The Server is stopping")
    return {
        "status": "stopping",
        "message": "The Server is stopping.",
    }


@app.get("/restart")
async def restart(
    data: Models.RestartModel,
    authorization: Annotated[str, Header()],
    tasks: BackgroundTasks,
) -> Responses.RestartResponse:
    """Restarts the Server.
    It waits for `countdown` seconds, then runs `/stop` on the Server, and kills it after `timeout` seconds if it's still alive.
    Then, it starts the Server again.
    Args:
        countdown: the time in seconds to give the players to leave the Server before initiating the shutdown.
                   if set to 0, the shutdown phase is started immediately. Defaults to 60.
        reason: a brief message explaining why the Server is restarting. Only shown if countdown is non-zero. Defaults to "".
        timeout: the time in seconds to wait before killing the Server if its process hasn't stopped. Defaults to 10.
    Headers:
        Authorization: the Authorization token.
    Returns:
        status: "restarting"
        message: The Server's response.
    """

    check_password(authorization)

    logger.debug("/restart")

    tasks.add_task(
        stop_server,
        "RESTARTING",
        data.countdown,
        data.reason,
        data.timeout,
        Controllers.process.start,
    )

    logger.info("/restart - The Server is restarting")

    return {
        "status": "restarting",
        "message": "The Server is restarting.",
    }


@app.get("/maintainance")
async def maintainance(
    data: Models.MaintainanceModel,
    authorization: Annotated[str, Header()],
    tasks: BackgroundTasks,
) -> Responses.MaintainanceResponse:
    """Stops the Server and sets it to maintainance status.
    It waits for `countdown` seconds, then runs `/stop` on the Server, and kills it after `timeout` seconds if it's still alive.
    Args:
        countdown: the time in seconds to give the players to leave the Server before initiating the shutdown.
                   if set to 0, the shutdown phase is started immediately. Defaults to 60.
        reason: a brief message explaining why the Server is entering maintanance mode. Only shown if countdown is non-zero. Defaults to "".
        timeout: the time in seconds to wait before killing the Server if its process hasn't stopped. Defaults to 10.
    Headers:
        Authorization: the Authorization token.
    Returns:
        status: "stopping"
        message: The Server's response.
    """

    check_password(authorization)

    logger.debug("/maintainance")

    tasks.add_task(
        stop_server,
        "STOPPING FOR MAINTAINANCE",
        data.countdown,
        data.reason,
        data.timeout,
        Controllers.maintainance.set,
        data.reason,
    )

    logger.info("/maintainance - The Server is stopping for maintainance")

    return {
        "status": "stopping",
        "message": "The Server is stopping for maintainance.",
    }


@app.get("/command")
async def command(
    data: Models.CommandModel, authorization: Annotated[str, Header()]
) -> Responses.CommandResponse:
    """
    Executes a command on the Server and returns its output.
    Args:
        command: The command to execute.
    Headers:
        Authorization: the Authorization token.
    Returns:
        status: "executed"
        message: The Server's response.
        output: The command's output.
    """

    logger.debug('/command {command: "%s"}', data.command)
    check_password(authorization)

    output = Controllers.server.command(data.command)
    return {
        "status": "executed",
        "message": "The command was executed.",
        "output": output,
    }


@app.get("/logs")
@app.get("/logs/stream")
async def logs_stream(authorization: Annotated[str, Header()]) -> StreamingResponse:
    """Streams Server logs in real-time using SSE.
    Headers:
        Authorization: the Authorization token.
    Returns: text/event-stream
    """

    logger.debug("/logs/stream")
    check_password(authorization)
    return StreamingResponse(Controllers.logs.stream(), media_type="text/event-stream")


@app.get("/logs/tail")
async def logs_tail(
    authorization: Annotated[str, Header()],
    data: Optional[Models.LogsTailModel] = None,
) -> StreamingResponse:
    """Streams the last few lines of the Server's logs.
    Args:
        back: The number of lines to stream.
    Headers:
        Authorization: the Authorization token.
    Returns: text/event-stream
    """

    logger.debug("/logs/tail")
    check_password(authorization)
    return StreamingResponse(Controllers.logs.tail(data.back if data else 10))


if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=42101)