pristontale-eu-linux/ptlinux/server_status.py

"""Official realm health used by the native launcher dashboard."""

from __future__ import annotations

from dataclasses import dataclass
import json
from typing import Callable, Mapping, Protocol
from urllib.request import Request, urlopen

from . import __version__


# This is the same publisher endpoint embedded in Windows launcher 1.0.0.64.
SERVER_STATUS_ENDPOINT = (
	"https://pristontale.eu/api/pristontale/developer"
	"?api_key=d9OlDSvworitU4ajDiH6dU6cxXApc4M0"
)
MAX_SERVER_STATUS_BYTES = 1024 * 1024


class ServerStatusError(RuntimeError):
	"""Raised when the official status response cannot be interpreted safely."""


class ServerStatusResponse(Protocol):
	def read(self, size: int = -1) -> bytes: ...

	def __enter__(self) -> ServerStatusResponse: ...

	def __exit__(
		self,
		exception_type: type[BaseException] | None,
		exception: BaseException | None,
		traceback: object | None,
	) -> bool | None: ...


ServerStatusOpener = Callable[..., ServerStatusResponse]


@dataclass(frozen=True)
class RealmServerStatus:
	name: str
	online: bool | None
	players_online: int | None
	login_online: bool | None = None
	game_online: bool | None = None


def fetch_server_status(
	*,
	timeout_seconds: float = 10.0,
	opener: ServerStatusOpener = urlopen,
) -> tuple[RealmServerStatus, ...]:
	request = Request(
		SERVER_STATUS_ENDPOINT,
		headers={"User-Agent": f"PT-Linux/{__version__} (+https://pristontale.eu/)"},
		method="GET",
	)
	try:
		response = opener(request, timeout=timeout_seconds)
		with response:
			payload = response.read(MAX_SERVER_STATUS_BYTES + 1)
	except OSError as exc:
		raise ServerStatusError("official server status is unavailable") from exc
	if len(payload) > MAX_SERVER_STATUS_BYTES:
		raise ServerStatusError("official server status exceeds the 1 MiB limit")
	return parse_server_status(payload)


def parse_server_status(payload: bytes) -> tuple[RealmServerStatus, ...]:
	try:
		value = json.loads(payload.decode("utf-8-sig"))
	except (UnicodeError, json.JSONDecodeError) as exc:
		raise ServerStatusError("official server status is not valid JSON") from exc
	# Historical versions returned an array containing the status object. The
	# current endpoint returns that object directly.
	if isinstance(value, list) and value:
		value = value[0]
	if not isinstance(value, dict):
		raise ServerStatusError("official server status is not a realm object")

	statuses: list[RealmServerStatus] = []
	for raw_name, raw_components in value.items():
		if not isinstance(raw_name, str) or raw_name.casefold() == "website":
			continue
		name = raw_name.strip()
		if not name or len(name) > 64 or not isinstance(raw_components, Mapping):
			continue
		status = _parse_realm(name, raw_components)
		if status is not None:
			statuses.append(status)
	if not statuses:
		raise ServerStatusError("official server status contains no realms")
	return tuple(statuses)


def _parse_realm(
	name: str,
	components: Mapping[object, object],
) -> RealmServerStatus | None:
	service_states: list[bool] = []
	login_online: bool | None = None
	game_online: bool | None = None
	players_online: int | None = None
	for raw_key, raw_value in components.items():
		if not isinstance(raw_key, str):
			continue
		key = raw_key.casefold()
		if key == "users.online":
			players_online = _non_negative_integer(raw_value)
			continue
		if not key.startswith("server."):
			continue
		state = _online_state(raw_value)
		if state is None:
			continue
		service_states.append(state)
		if "login" in key:
			login_online = state
		elif "game" in key:
			game_online = state
	if not service_states and players_online is None:
		return None
	online = all(service_states) if service_states else None
	display_name = name[0].upper() + name[1:]
	return RealmServerStatus(
		name=display_name,
		online=online,
		players_online=players_online,
		login_online=login_online,
		game_online=game_online,
	)


def _online_state(value: object) -> bool | None:
	if isinstance(value, bool):
		return value
	if isinstance(value, (int, float)):
		return value > 0
	if isinstance(value, str):
		text = value.strip().casefold()
		if text in {"online", "up", "true", "1"}:
			return True
		if text in {"offline", "down", "false", "0"}:
			return False
	return None


def _non_negative_integer(value: object) -> int | None:
	if isinstance(value, bool):
		return None
	if isinstance(value, int) and value >= 0:
		return value
	if isinstance(value, str) and value.isascii() and value.isdecimal():
		return int(value)
	return None