commit 6e6232205e3dbc00bd8fd62a5a5683257dcb213c Author: Lily Rose Date: Thu Jul 10 18:07:55 2025 +1000 Initial commit diff --git a/palworld_watchdog.py b/palworld_watchdog.py new file mode 100755 index 0000000..aab025c --- /dev/null +++ b/palworld_watchdog.py @@ -0,0 +1,181 @@ +#!/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()