Added debugger class for dashboard and other launcher util classes.

ptlinux/server_status.py

@@ -0,0 +1,157 @@
+"""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