144 lines
4.7 KiB
Python
144 lines
4.7 KiB
Python
from fastapi import FastAPI, status
|
|
from pydantic import BaseModel, Field
|
|
from pathlib import Path
|
|
from datetime import datetime, timedelta
|
|
import docker
|
|
|
|
app = FastAPI()
|
|
client = docker.from_env()
|
|
|
|
|
|
class PyPistonServer(BaseModel):
|
|
server_id: str | None = None
|
|
server_name: str | None = None
|
|
container_name: str
|
|
|
|
image: str = Field(
|
|
title="The base docker image that was used to create this server"
|
|
)
|
|
version: str = Field(
|
|
title="The version of the docker image that was used to create this server"
|
|
)
|
|
description: str | None = Field(
|
|
default=None,
|
|
title="Description or notes for the server instance",
|
|
max_length=300
|
|
)
|
|
motd: str | None = Field(
|
|
default=None,
|
|
title="Message of the Day that will appear on the Minecraft server listing page in the minecraft client",
|
|
max_length=300
|
|
)
|
|
rcon_enabled: bool = Field(
|
|
default=True,
|
|
title="Whether or not this container can be managed via rcon"
|
|
)
|
|
ports: dict = Field(
|
|
default={25565: 25565},
|
|
title="List of ports that have been published to the host"
|
|
)
|
|
is_vanilla: bool = Field(
|
|
default=False,
|
|
title="Whether or not the server is running a modded vanilla (non-modded/custom) of Minecraft"
|
|
)
|
|
snooper_enabled: bool = True
|
|
status: str | None = Field(
|
|
default=None,
|
|
title="State of the docker container"
|
|
)
|
|
uptime: str | None = Field(
|
|
default=None,
|
|
title="Time the container has been up"
|
|
)
|
|
world_name: str = "world"
|
|
online_mode: bool = True
|
|
allow_flight: bool = False
|
|
|
|
|
|
def get_env_value(env_key: str, env_list: list):
|
|
val = [e for e in env_list if f"{env_key}=" in e]
|
|
return val[0].split('=')[-1] if val else None
|
|
|
|
|
|
def get_servers():
|
|
for container in client.containers.list(filters={"ancestor": "itzg/minecraft-server"}, all=True):
|
|
image = container.image.tags[0]
|
|
env_list = container.attrs['Config']['Env']
|
|
yield PyPistonServer(
|
|
server_id=container.id,
|
|
server_name=get_env_value("SERVER_NAME", env_list),
|
|
container_name=container.name,
|
|
image=image,
|
|
version=get_env_value("VERSION", env_list) or "Vanilla",
|
|
description="",
|
|
motd=get_env_value("MOTD", env_list),
|
|
rcon_enabled=get_env_value("ENABLE_RCON", env_list) != "FALSE",
|
|
ports={k: v[0]['HostPort'] for k, v in container.ports.items() if v},
|
|
is_vanilla=get_env_value("TYPE", env_list) is None,
|
|
snooper_enabled=get_env_value("SNOOPER_ENABLED", env_list) == "TRUE",
|
|
status=([s for s in container.attrs['State'] if container.attrs['State'][s] is True] or ["Starting"])[0],
|
|
uptime=str(datetime.now().astimezone() - datetime.fromisoformat(container.attrs['Created'])),
|
|
world_name=get_env_value("LEVEL", env_list) or "world",
|
|
online_mode=get_env_value("ONLINE_MODE", env_list) != "FALSE",
|
|
allow_flight=get_env_value("ALLOW_FLIGHT", env_list) == "TRUE"
|
|
)
|
|
|
|
|
|
@app.get("/", status_code=status.HTTP_200_OK)
|
|
def read_root():
|
|
return {"version": "0.1-dev"}
|
|
|
|
|
|
@app.get("/servers/")
|
|
def read_active_servers():
|
|
mc_servers = list(get_servers())
|
|
return {
|
|
"total": len(mc_servers),
|
|
"servers": mc_servers
|
|
}
|
|
|
|
|
|
@app.post("/servers/new")
|
|
def create_new_server(server: PyPistonServer):
|
|
env = {
|
|
"SERVER_NAME": server.server_name,
|
|
"MOTD": server.motd,
|
|
"ENABLE_RCON": server.rcon_enabled,
|
|
"SNOOPER_ENABLED": server.snooper_enabled,
|
|
"LEVEL": server.world_name,
|
|
"ONLINE_MODE": server.online_mode,
|
|
"ALLOW_FLIGHT": server.allow_flight
|
|
}
|
|
if server.version != "Vanilla":
|
|
env["VERSION"] = server.version
|
|
new_srv = client.containers.run(
|
|
server.image,
|
|
ports=server.ports,
|
|
restart_policy={"Name": "on-failure", "MaximumRetryCount": 5},
|
|
labels={"pyPistonManged": "True"},
|
|
environment=env,
|
|
detach=True
|
|
)
|
|
new_srv = [s for s in get_servers() if s.server_id == new_srv.id][0]
|
|
return new_srv
|
|
|
|
|
|
@app.get("/servers/{server_id}")
|
|
def get_server_by_id(server_id: str):
|
|
return [s for s in get_servers() if s.server_id == server_id][0]
|
|
|
|
|
|
@app.post("/servers/{server_id}/exec")
|
|
def run_cmd_on_server(server_id: str, command: str):
|
|
server = [s for s in get_servers() if s.server_id == server_id][0]
|
|
container = client.containers.get(server.server_id)
|
|
if server.rcon_enabled:
|
|
cmd = f"rcon-cli {command}"
|
|
else:
|
|
cmd = f"mc-send-to-console {command}"
|
|
result = container.exec_run(cmd)
|
|
return {
|
|
"exit_code": result.exit_code,
|
|
"output": result.output.decode("utf-8")
|
|
}
|