palworld-watchdog/palworld_watchdog.py

#!/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()