#!/bin/env python3 from base64 import b64encode from requests.api import get, post # pyright:ignore[reportUnknownVariableType] from requests.models import Response from datetime import timedelta from pathlib import Path from time import sleep, time from os import system, WEXITSTATUS server_address = "http://:8212" admin_username = "admin" admin_password = "" class PalworldAPI: def __init__(self, server_url: str, password: str, username: str = "admin"): self.server_url: str = server_url token = b64encode(f"{username}:{password}".encode()).decode() self.headers: dict[str, str] = { "Accept": "application/json", "Content-Type": "application/json", "Authorization": f"Basic {token}", } def get(self, endpoint: str) -> Response: response = get(f"{self.server_url}{endpoint}", headers=self.headers) # pyright:ignore[reportUnknownVariableType] response.raise_for_status() # pyright:ignore[reportUnknownMemberType] return response # pyright:ignore[reportUnknownVariableType] def post(self, endpoint: str, payload: object = None) -> Response: response = post(f"{self.server_url}{endpoint}", json=payload, headers=self.headers) # pyright:ignore[reportUnknownVariableType] response.raise_for_status() # pyright:ignore[reportUnknownMemberType] return response # pyright:ignore[reportUnknownVariableType] def get_server_info(self) -> dict[str, object]: return self.get("/v1/api/info").json() # pyright:ignore[reportUnknownMemberType, reportAny] def get_player_list(self) -> dict[str, object]: return self.get("/v1/api/players").json() # pyright:ignore[reportUnknownMemberType, reportAny] def get_server_metrics(self) -> dict[str, object]: return self.get("/v1/api/metrics").json() # pyright:ignore[reportUnknownMemberType, reportAny] def get_server_settings(self) -> dict[str, object]: return self.get("/v1/api/settings").json() # pyright:ignore[reportUnknownMemberType, reportAny] def kick_player(self, userid: str, message: str) -> None: _ = self.post("/v1/api/kick", { "userid": userid, "message": message }) def ban_player(self, userid: str, message: str) -> None: _ = self.post("/v1/api/ban", {"userid": userid, "message": message}) def unban_player(self, userid: str) -> None: _ = self.post("/v1/api/unban", { "userid": userid }) def save_server_state(self) -> None: _ = self.post("/v1/api/save") def make_announcement(self, message: str) -> None: _ = self.post("/v1/api/announce", { "message": message }) def shutdown_server(self, waittime: int, message: str) -> None: _ = self.post("/v1/api/shutdown", { "waittime": waittime, "message": message }) def stop_server(self) -> None: _ = self.post("/v1/api/stop") def was_saved_since(save_path: Path, unix_timestamp: int | float) -> bool: try: level_sav = save_path / "Level.sav" level_meta_sav = save_path / "LevelMeta.sav" players_path = save_path / "Players" return level_sav.stat().st_ctime > unix_timestamp \ and level_meta_sav.stat().st_ctime > unix_timestamp \ and all(player_path.stat().st_ctime > unix_timestamp \ for player_path in players_path.iterdir() \ if "_" not in player_path.name) except: return False def was_backed_up_since(save_path: Path, unix_timestamp: int | float) -> bool: try: backups_path = save_path / "backup" / "world" last_backup_path = max(backups_path.iterdir(), key=lambda p: p.stat().st_ctime) return last_backup_path.stat().st_ctime > unix_timestamp except: return False def is_shutdown() -> bool: try: return WEXITSTATUS(system("pgrep PalServer-Linux")) == 1 except: return False def force_shutdown() -> bool: try: return WEXITSTATUS(system("pkill PalServer-Linux")) == 0 except: return is_shutdown() def stop_server(api: PalworldAPI) -> bool: try: api.stop_server() sleep(30) if is_shutdown(): return True else: return force_shutdown() except: return force_shutdown() def perform_shutdown(api: PalworldAPI, reason: str) -> bool: try: api.save_server_state() sleep(60) api.shutdown_server(30, f"WARNING: Shutting down in 30 seconds - Reason: {reason}") sleep(60) if is_shutdown(): return True else: return stop_server(api) except: return stop_server(api) def perform_shutdown_and_exit(api: PalworldAPI, reason: str): if perform_shutdown(api, reason): print("Successfully shut down") exit(1) else: print("Failed to shut down") exit(2) def main(): api = PalworldAPI(server_address, admin_password, admin_username) info = api.get_server_info() name = info["servername"] print(f"Name: {name}") description = info["description"] print(f"Description: {description}") version = info["version"] print(f"Version: {version}") guid: str = info["worldguid"] # pyright:ignore[reportAssignmentType] print(f"GUID: {guid}") metrics = api.get_server_metrics() uptime = timedelta(seconds=metrics["uptime"]) # pyright:ignore[reportArgumentType] print(f"Uptime: {uptime}") days = metrics["days"] print(f"Days: {days}") players = metrics["currentplayernum"] print(f"Players: {players}") fps = metrics["serverfps"] print(f"FPS: {fps}") frame_time = metrics["serverframetime"] print(f"Frame Time: {frame_time}") save_path = Path("Pal") / "Saved" / "SaveGames" / "0" / guid while True: sleep(60) try: print("Saving server...") api.save_server_state() print("Server save completed") except: print("ERROR: Failed to save server state, shutting down") perform_shutdown_and_exit(api, "Failed to save") sleep(120) print("Checking save modification dates...") if was_saved_since(save_path, time() - 600): print("Has been saved in the last 10 minutes") else: print("ERROR: Has not been saved in the last 10 minutes, shutting down") perform_shutdown_and_exit(api, "Has not saved in the last 10 minutes") print("Checking backup modification dates...") if was_backed_up_since(save_path, time() - 3600): print("Has been backed up in the last 60 minutes") else: print("ERROR: Has not been backed up in the last 60 minutes, shutting down") perform_shutdown_and_exit(api, "Has not backed up in the last 60 minutes") sleep(120) if __name__ == "__main__": main()