Initial commit

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://<ip_address>:8212"
+admin_username = "admin"
+admin_password = "<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()