add delete save

Signed-off-by: anima <p.schreiner@4nima.de>

.gitignore

@@ -1,3 +1,8 @@
+*.conf
+*.ymal
+*.yml
+saves/
+
 # ---> Python
 # Byte-compiled / optimized / DLL files
 __pycache__/

SatiAPI.py

@@ -0,0 +1,561 @@
+#!/usr/bin/env python3
+"""
+https://github.com/wolveix/satisfactory-server/wiki/Official-API-Docs
+https://satisfactory.wiki.gg/wiki/Dedicated_servers/HTTPS_API
+"""
+
+import requests
+import logging
+import json
+import yaml
+from os.path import exists
+from os import makedirs
+
+
+import urllib3
+urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
+
+class SatiAPI:
+    """A API wrapper for Satisfactory dedicated server"""
+    __AUTHOR__      = 'anima'
+    __VERSION__     = '0.17.2'
+
+    def __init__(self, host: str = None, port: int = 7777, token: str = None, conffile: str = 'conf.yml', logfile: str = 'SatiAPI.log', loglevel: int = 20) -> None:
+        """create a wrapper for satisfactory dedicated server
+        
+        :param host: host ip or dns
+        :type host: string
+        :param port: port of host
+        :type port: int | str
+        :pram token: token for auth on server (can be get with `get_token`)
+        :type token: string
+        :param conffile: path to config file in yaml format
+        :type conffile: string
+        :param logfile: path to logfile
+        :type logfile: string
+        :param loglevel: level of logging (10 => debug, 20 => info (default), 30 => warning, 40 => error)
+
+        all server specific settings can be set via config file
+        """
+        self._log                       = logging.getLogger(__name__)
+        logHandler                      = logging.FileHandler(logfile)
+        logPrinter                      = logging.StreamHandler()
+        logFormat                       = logging.Formatter('%(asctime)s %(levelname)s \t %(name)s : %(message)s')
+        logHandler.setFormatter(logFormat)   # Format for File
+        logPrinter.setFormatter(logFormat)   # Format for Console
+        self._log.addHandler(logHandler)
+        self._log.addHandler(logPrinter)
+        self._log.setLevel(loglevel)
+
+        self.host = host
+        self.port = port
+        self.token = token
+        self.__map_config(conffile)
+
+    ## propertys
+    @property
+    def host(self):
+        return self.__host
+    
+    @host.setter
+    def host(self, value) -> bool:
+        """setter method for host. check if value valid. 
+
+        :param value: host ip or dns to dedicated server
+        :type value: string
+        :return: true if set successfull
+        :rtype: bool
+        """
+        if isinstance(value, str) or value is None:
+            self.__host = value
+            return True
+        else:
+            self._log.debug(f'wrong {value=} ({type(value)}) for host')
+            return False
+
+    @property
+    def port(self):
+        return self.__port
+    
+    @port.setter
+    def port(self, value) -> bool:
+        """setter method for port. check if value a valid network port
+
+        :param value: port of dedicated server instance
+        :type value: int | str
+        :return: true if set successfull
+        :rtype: bool
+        """
+        if not isinstance(value, int):
+            if isinstance(value, str):
+                try:
+                    value = int(value)
+                except ValueError:
+                    self._log.debug(f'wrong {value=} ({type(value)}) for port')
+                    return False
+            else:
+                self._log.debug(f'wrong {value=} ({type(value)}) for port')
+                return False
+        
+        if value > 0 and value < 65535:
+            self.__port = value
+            return True
+        else:
+            self._log.debug(f'out of port range {value=}')
+            return False
+
+    @property
+    def token(self):
+        return self.__token
+    
+    @token.setter
+    def token(self, value) -> bool:
+        """setter method for token. check if value a valid token
+        
+        :param value: token for auth on server
+        :type value: string
+        :return: true if set successfull
+        :rtype: bool
+        """
+        if isinstance(value, str) or value is None:
+            self.__token = value
+            return True
+        else:
+            self._log.debug(f'wrong {value=} ({type(value)}) for token')
+
+    ## methods
+    def __map_config(self, conffile) -> None:
+        """map config settings with multible usage in this class override only default values not changed.
+        
+        :param conffile: path to config file in yaml format
+        :type conffile: string
+        """
+        if exists(conffile):
+            with open(conffile, 'r') as f:
+                self._config = yaml.safe_load(f.read())
+
+            if 'host' in self._config and self.host is None:
+                self.host = self._config['host']
+            
+            if 'port' in self._config and self.port == 7777:
+                self.port = self._config['port']
+
+            if 'auth' in self._config:
+                if 'Administrator' in self._config['auth']:
+                    if 'token' in self._config['auth']['Administrator'] and self.token is None:
+                        self.token = self._config['auth']['Administrator']['token']
+
+    def _get_config_password(self, privilegeLevel: str = 'Administrator') -> str | None:
+        """internal helper method to get password from config file
+        
+        :param privilegeLevel: level of needed privilege
+        :type privilegeLevel: string
+        :return: password if exists
+        :rtype: string | None
+        """
+        if 'auth' in self._config:
+            if privilegeLevel in self._config['auth']:
+                if 'password' in self._config['auth'][privilegeLevel]:
+                    return self._config['auth'][privilegeLevel]['password']
+
+    def __query(self, query: str, data: dict = None, auth: bool = False) -> dict | bool | None:
+        """run a query to satisfacoty dedicated server
+        
+        :param query: name of querytype
+        :type query: string
+        :param data: (maybe optional) data to be send as dict
+        :type data: dict
+        :param auth: set true if authentication requiered
+        :type auth: bool
+        :return: response from server
+        :rtype: dict | None | bool
+        """
+        payload = dict()
+        payload['function'] = query
+        
+        if data is not None:
+            payload['data'] = data
+
+        headers = dict()
+        headers['Content-Type'] = 'application/json'
+        if auth:
+            if self.token is None:
+                self.get_token()
+                
+            if self.token is not None:
+                headers['Authorization'] = f'Bearer {self.token}'
+            else:
+                self._log.error('no data to authencitate')
+                return False
+
+        response = requests.post(f'https://{self.host}:{self.port}/api/v1', headers=headers, json=payload, verify=False)
+        if response.status_code == 200:
+            if response.headers['Content-Type'] == 'application/json':
+                json_response = json.loads(response.text)
+            else:
+                return response.content
+
+            if 'data' in json_response:
+                return json_response
+            else:
+                # if json_response['errorMessage'] == '':
+                        # self._log.debug(f'token expired')
+                        # self.get_token()
+                self._log.error(f'wrong response data {json_response=}')
+        elif response.status_code == 201:
+            # The function has been executed and returned no error. Returned by some functions to indicate that a new file has been created, such as UploadSaveGame
+            return True
+
+        elif response.status_code == 202:
+            # The function has been executed, but is still being processed. This is returned by some functions with side deffects, such as LoadGame
+            return True
+
+        elif response.status_code == 204:
+            # used from some querys as response with no response content
+            # The function has been executed successfully, but returned no data (and no error)
+            return True
+
+        else:
+            self._log.error(f'non successfull response {response.content} [{response.status_code}]')
+            self._log.debug(f'{response.request.url=}')
+            self._log.debug(f'{response.request.headers=}')
+            self._log.debug(f'{response.request.body=}')
+            return False
+
+    def _save_token(self, response: dict) -> bool:
+        """internal helper method to save token from response
+        
+        :param response: response after query
+        :type response: dict
+        :return: true if token successfull saved
+        :rtype: bool
+        """
+        if response:
+            if 'authenticationToken' in response['data']:
+                self.token = response['data']['authenticationToken']
+                return True
+        return False
+
+
+    def _passwordless_login(self, privilegeLevel: str = 'InitialAdmin') -> bool:
+        """Attempts to perform a passwordless login to the Dedicated Server as a player. Passwordless login is possible if the Dedicated Server is not claimed, or if Client Protection Password is not set for the Dedicated Server. This function requires no Authentication.
+        
+        :param privilegeLevel: Minimum privilege level to attempt to acquire by logging in. See Privilege Level enum for possible values
+        :type privilegeLevel: string
+        :return: true if successfull
+        :rtype: bool
+        """
+        data = dict()
+        data['MinimumPrivilegeLevel'] = privilegeLevel
+
+        response = self.__query('PasswordlessLogin', data)
+        return self._save_token(response)
+
+    def claim_server(self, servername: str, password: str = None) -> bool:
+        """Claims this Dedicated Server if it is not claimed. Requires InitialAdmin privilege level, which can only be acquired by attempting passwordless login while the server does not have an Admin Password set, e.g. it is not claimed yet. Function does not return any data in case of success, and the server is claimed. The client should drop InitialAdmin privileges after that and use returned AuthenticationToken instead, and update it's cached server game state by calling QueryServerState.
+        
+        :param servername: New name of the Dedicated Server
+        :type servername: string
+        :param password: Admin Password to set on the Dedicated Server, in plaintext. If empty the password will cread from conf file
+        :type password: string
+        :return: true if claim successfull
+        :rtype: bool
+
+        after claim it get a new token automaticly
+        """
+        if self._passwordless_login():
+            data = dict()
+            if password is None: 
+                password = self._get_config_password()
+                self._log.debug('password used from conf file')
+
+            if password is None: 
+                self._log.error('no password for server claim set')
+                return False
+            data['ServerName']      = servername
+            data['AdminPassword']   = password
+
+            response = self.__query('ClaimServer', data, True)
+            return self._save_token(response)
+        else: 
+            self._log.error('server can not be clamed because no authentication possable')
+            return False
+
+    def get_token(self, password: str = None, privilegeLevel: str = 'Administrator') -> bool:
+        """get token from satisfacory dedicated server, password is needed!
+        Attempts to log in to the Dedicated Server as a player using either Admin Password or Client Protection Password. This function requires no Authentication.
+        
+        :param password: password of admin level
+        :type password: string
+        :param privilegeLevel: level of authentication
+        :type privilegeLevel: string
+        :return: true if successfull get a token
+        :rtype: bool
+        """
+        if self.token is not None: 
+            self._log.warning('you have already a token')
+        if isinstance(password, str):
+            self._log.debug('password used from parameter')
+        else:
+            password = self._get_config_password()
+            if password is not None:
+                self._log.debug('password used from conf file')
+            else:
+                self._log.error(f'no password for auth given!')
+                return False
+
+        data = dict()
+        data['password'] = password
+        data['MinimumPrivilegeLevel'] = privilegeLevel
+
+        response = self.__query('PasswordLogin', data)
+        return self._save_token(response)
+
+    def shutdown(self) -> bool:
+        """Shuts down the Dedicated Server. If automatic restart script is setup, this allows restarting the server to apply new settings or update. Requires Admin privileges. Shutdowns initiated by remote hosts are logged with their IP and their token. Function does not return any data on success, and does not take any parameters.
+
+        :return: true is shutdown successfull init
+        :rtype: bool
+        """
+        return self.__query('Shutdown', auth=True)
+
+    def set_admin_password(self, password: str = None):
+        """Updates the currently set Admin Password. This will invalidate all previously issued Client and Admin authentication tokens. Requires Admin privileges. Function does not return any data on success.
+        
+        :param password: the new passwort to set will be read from config file it not set
+        :type password: string
+        :return: true if new password successfull set
+        :rtype: bool
+        """
+        if password is None: 
+            password = self._get_config_password()
+
+        if password is None:
+            self._log.error('can not set new admin password because no passwort is set')
+            return False
+        
+        data = dict()
+        data['Password'] = password
+        response = self.__query('SetAdminPassword', data, True)
+        return self._save_token(response)
+
+    def set_client_password(self, password: str = None):
+        """Updates the currently set Client Protection Password. This will invalidate all previously issued Client authentication tokens. Pass empty string to remove the password, and let anyone join the server as Client. Requres Admin privileges. Function does not return any data on success.
+
+        :param password: the new passwort to set will be read from config file it not set
+        :type password: string
+        :return: true if new password successfull set
+        :rtype: bool
+        """
+        if password is None: 
+            password = self._get_config_password('Client')
+
+        if password is None:
+            self._log.error('can not set new client password because no passwort is set')
+            return False
+        
+        data = dict()
+        data['Password'] = password
+        response = self.__query('SetClientPassword', data, True)
+        return True
+
+    def set_servername(self, servername: str) -> bool:
+        """Renames the Dedicated Server once it has been claimed. Requires Admin privileges. Function does not return any data on success.
+        
+        :param servername: New name of the Dedicated Server
+        :type servername: string
+        :return: true if set new name successfull
+        :rtype: bool
+        """
+        if isinstance(servername, str):
+            # TODO: add check if contain non allowed characters
+            data = dict() 
+            data['ServerName'] = servername
+
+            response = self.__query('RenameServer', data, True)
+            return response
+        else:
+            self._log.error(f'{servername=} is not a valid name')
+            return False
+
+    def get_status(self) -> dict:
+        """Retrieves the current state of the Dedicated Server. Does not require any input parameters.
+        
+        :return: full return from server
+        :rtpye: dict
+        """
+        return self.__query('QueryServerState', auth=True)
+
+    def get_health(self, ClientCustomData: str = '') -> dict:
+        """get health check
+        
+        :param ClientCustomData: Custom Data passed from the Game Client or Third Party service. Not used by vanilla Dedicated Servers
+        :type ClientCustomData: string
+        :return: full return from server
+        :rtpye: dict
+        """
+        data = dict()
+        data['ClientCustomData'] = ClientCustomData
+        
+        return self.__query('HealthCheck', data)
+
+    def get_sessions(self):
+        """Enumerates all save game files available on the Dedicated Server. Requires Admin privileges. Function does not require any additional parameters.
+
+        :return: List of sessions available on the Dedicated Server
+        :rtype: dict
+        """
+        return self.__query('EnumerateSessions', auth=True)
+
+    def create_new_game(self, sessionname: str = None, startLocation: str | int = None) -> bool:
+        """Creates a new session on the Dedicated Server, and immediately loads it. HTTPS API becomes temporarily unavailable when map loading is in progress | Function does not return any data on success.
+
+        :param sessionname: name of session / world
+        :type sessionname: string
+        :param startLocation: selected starting location int or full name (1 = Grass Fields, 2 = Rocky Desert, 3 = Northern Forest, 4 = Dune Destert)
+        :type startLocation: str | int
+        :return: true if new session successfull created
+        :rtype: bool
+        """
+        if not isinstance(sessionname, str):
+            self._log.error('can not create new game because session name is invalid')
+            return False
+
+        match startLocation:
+            case 1 | '1' | 'Grass Fields':
+                startLocation = 'Grass Fields'
+            case 2 | '2' | 'Rocky Desert':
+                startLocation = 'Rocky Desert'
+            case 3 | '3' | 'Northern Forest':
+                startLocation = 'Northern Forest'
+            case 4 | '4' | 'Dune Desert':
+                startLocation = 'Dune Desert' # not working ... 
+            case _:
+                startLocation = ''
+        
+        self._log.debug(f'matched {startLocation=}')
+
+        data = dict()
+        data['NewGameData'] = dict()
+        data['NewGameData']['SessionName'] = sessionname
+        data['NewGameData']['MapName'] = ''
+        data['NewGameData']['StartingLocation'] = startLocation
+
+        # TODO: add advanced start options
+        response = self.__query('CreateNewGame', data, True)
+        return response
+
+    def save_game(self, savename: str = None) -> bool:
+        """Saves the currently loaded session into the new save game file named as the argument. Requires Admin privileges. SaveName might be changed to satisfy file system restrictions on file names. Function does not return any data on success.
+
+        :param savename: Name of the save game file to create. Passed name might be sanitized (default: Sessionname_APISave)
+        :type savename: string
+        :return: return true if save successfull init
+        :rtype: bool
+        """
+        if savename is None: 
+            state = self.get_status()
+            savename = f'{state["data"]["serverGameState"]["activeSessionName"]}_APISave'
+        data = dict()
+        data['SaveName'] = savename
+
+        return self.__query('SaveGame', data, True)
+
+    def download_save(self, savename: str = None, savepath: str = 'saves/') -> bool:
+        """Downloads save game with the given name from the Dedicated Server. Requires Admin privileges. This function responds with the file attachment containing the save game file on success, and with normal error response in case of error.
+
+        :param savename: Name of the save game file to download from the Dedicated Server. Default: Sessionname_APISave (will be create before)
+        :type savename: string
+        :param savepath: path to save the savefile
+        :type savepath: sting
+        :return: true if save successfull downloaded
+        :rtype: bool
+        """
+        if savename is None: 
+            pass
+
+        data = dict()
+        data['SaveName'] = savename
+        response = self.__query('DownloadSaveGame', data, True)
+        if not exists(savepath):
+            makedirs(savepath)
+        with open(f'{savepath}/{savename}.sav', 'wb') as f:
+            f.write(response)
+        if exists(f'{savepath}/{savename}.sav'):
+            return True
+        else: 
+            return False
+
+    def delete_save(self, savename: str) -> bool:
+        """Deletes the existing save game file from the server. Requires Admin privileges. SaveName might be changed to satisfy file system restrictions on file names. Function does not return any data on success.
+        
+        :param savename: Name of the save game file to delete. Passed name might be sanitized
+        :type savename: sting
+        :return: true if save successfull deleted
+        :rtype: bool
+        """
+        if isinstance(savename, str):
+            data = dict()
+            data['SaveName'] = savename
+
+            response = self.__query('DeleteSaveFile', data, True)
+            return response
+
+    def load_game(self, savename: str, enableAdvancedSettings: bool = True) -> bool:
+        """Loads the save game file by name, optionally with Advanced Game Settings enabled. Requires Admin privileges. Dedicated Server HTTPS API will become temporarily unavailable when save game is being loaded. Function does not return any data on succcess.
+        
+        :param savename: Name of the save game file to load
+        :type savename: sting
+        :param enableAdvancedSettings: True if save game file should be loaded with Advanced Game Settings enabled
+        :type enableAdvancedSettings: bool
+        """
+        if isinstance(savename, str):
+            data = dict()
+            data['SaveName'] = savename
+            data['EnableAdvancedGameSettings'] = enableAdvancedSettings
+
+            response = self.__query('LoadGame', data, True)
+            return response
+
+        return False
+
+    def get_server_options(self) -> dict:
+        """Retrieves currently applied server options and server options that are still pending application (because of needing session or server restart) Does not require input parameters.
+        
+        :return: full return from server
+        :rtpye: dict
+        """
+        return self.__query('GetServerOptions')
+
+    def get_advanced_game_settings(self):
+        """Retrieves currently applied advanced game settings. Does not require input parameters.
+        
+        :return: full return from server
+        :rtpye: dict
+        """
+        return self.__query('GetAdvancedGameSettings')
+    
+    
+    
+    ##  todo
+    #RunCommand
+    
+    #SetAutoLoadSessionName
+    #ApplyServerOptions
+    #ApplyAdvancedGameSettings
+    
+    #UploadSaveGame
+    
+    #DeleteSaveFile
+    #DeleteSaveSession
+    
+
+if __name__ == "__main__":
+    sati = SatiAPI(loglevel=10)
+    # print('Verbundene Spieler: ', sati.get_status()['serverGameState']['numConnectedPlayers'])
+    print(sati.get_status())
+    print()
+    print(sati.get_sessions())
+    print()
+    # print(sati.download_save('MyWorld_APISave'))
+    print(sati.delete_save('MyWorld_APISave'))

conf.yml.sample

@@ -0,0 +1,6 @@
+host: ServerIP
+port: 7777
+auth: 
+  token: myToken
+  password: myAdminPassword         # not need if token existend
+  privilegeLevel: Administrator

requirements.txt

@@ -0,0 +1,2 @@
+requests
+PyYAML